[bulk] Low priority: let diff's lookahead grow until memory actually runs out, and diagnose it when it does

**Low priority / not scheduled.** Recording the design so it is not re-derived, and so the cases it covers are owned somewhere rather than forgotten.

#356 was the pressing symptom and is closed by raising `MAX_LOOKAHEAD_ITEMS` to 30,000. That bump is a mitigation, not a fix: everything it leaves behind is this issue, enumerated under "Residual scope" below. Closing #356 on the bump is deliberate — the residual cases are unlikely enough not to justify an issue sitting open for months describing them.

## Where this comes from

`diff`'s original stance was "try our best; run out of memory if it comes to that." #349/#351 replaced that with a hard bound, which traded OOM for a polite failure on inputs that previously succeeded (#356). Now that #347/#348 diagnose memory failures and name the fix, the original stance is affordable again — the reason to bound was that OOM produced a mystery, and it no longer does.

Slowness in pathological cases is explicitly acceptable; the cases that matter are seconds, not hours.

## Residual scope (inherited from #356)

Yes, 30,000 still is not enough for everything. These are the known gaps, each with what
to do about it, none of which we are doing now.

**1. Item collections wider than the cap.** Realigning across an unmatched item collection
requires peeking past all of it, and every item in it carries the same pk, so the whole
collection is buffered to learn one fact. At 30,000 a wider collection still fails with
"too different to diff accurately" on tables that may be otherwise identical.

*What to do:* step 3 below. Budget cumulative attribute count rather than item count —
measured cost is ~320-470 Python bytes per attribute regardless of value length, so an
attribute budget adapts to item shape where an item count cannot. It needs only a running
total maintained in `_load_page()` and `advance()`.

**2. Items with very large values.** No item count bounds these. Measured expansion is
**1.0x** for a single 400 KB value versus **6.8x** for 37 small attributes, because the cost
is per-attribute object overhead, not payload — so a table of 400 KB items is over any
budget worth having at almost any item count, and lowering the cap does not help. Already
noted in the constant's own comment.

*What to do:* the same attribute-count budget does not cover this either, since the cost is
in the values. Only real byte accounting does. Reaching it takes a segment holding
gigabytes, which is why it stays unfixed.

**3. The bound is still a guess about memory, not a measurement of it.** 30,000 is derived
from a per-worker budget of ~6 GB across 8 buffers (2 streams x 4 concurrent tasks) at the
worst normal item shape. Change the worker type, the task concurrency, or the item shape and
the real headroom moves; the constant does not.

*What to do:* steps 1-3 together, which replace "guess a safe cap" with "grow until memory
actually runs out, and diagnose it when it does." Step 2 is the part that makes being wrong
survivable, and is the prerequisite for the rest.

## Step 1 — remove the O(n^2) in the realignment loop

`diff_segment`'s realignment loop recomputes a full set intersection every iteration:

```python
while not (seen_pk_a & seen_pk_b) and (stream_a.peek(n) or stream_b.peek(n)):
```

Exactly one pk is added per iteration, so the intersection can only become non-empty via that pk. Replace with an O(1) membership test at insertion: keep `aligning_pk = None`, and when adding to `seen_pk_a` test `if pk in seen_pk_b` (symmetrically for b); loop while `aligning_pk is None`.

Two things fall out rather than being added:

- `aligning_pks = seen_pk_a & seen_pk_b` and `next(iter(aligning_pks), None)` disappear, since the found pk is held directly. That retires a latent fragility: the arbitrary `iter()` pick is correct today *only* because the loop adds one pk at a time and exits immediately, so the intersection always holds exactly one element. Nothing states that invariant, and a future edit could break it silently.
- Pre-loop seeding is unchanged. `pk_a != pk_b` is guaranteed (this is the `else` of the equality branch), so no initial check is needed — worth a comment, as it is the sort of thing that gets added back redundantly.

Behaviour is identical: same aligning pk, same emitted diffs, same exit conditions. The existing `test_diff.py` should pass untouched, which is the safety argument. Justification is complexity, not behaviour, so it belongs in a measured commit-message table (10k/50k/200k/500k, before and after) rather than a flaky timing assertion.

## Step 2 — diagnose Python-side memory death

Every memory signal in `client/src/utils/__init__.py` is JVM-shaped: watchdog pattern `"OutOfMemoryError:"`, and `MEMORY_FAILURE_MARKERS` of `OUT_OF_MEMORY_ERROR` / `OutOfMemoryError` / `out of memory`. The comment records that these were measured against an executor exhausting its **10 GB heap** -- `spark.executor.memory`, i.e. the JVM.

`diff`'s lookahead buffer is Python dicts in the PySpark worker, outside that heap. A Python-side blowup surfaces as `PythonException: MemoryError`, `Container killed by YARN for exceeding memory limits`, or a bare `ExecutorLostFailure` -- none of which contain any existing marker (`MemoryError` is not a substring of `OutOfMemoryError`). So a Python OOM today produces exactly the bare three-line "Job was stopped." that #348 was written to eliminate.

The watchdog already reads executor streams as well as the driver, so this is missing markers, not missing plumbing.

**Hard prerequisite:** the file's own rule is to write patterns exactly as printed, "case and trailing punctuation included," and the existing markers were measured. So this needs a real Python-side OOM on Glue with its logs captured -- easiest triggers are the wide-item generator from #344 or a disjoint `diff` with the cap lifted. Capture driver stream, executor streams, and the run's final `ErrorMessage`; check in particular whether Glue still categorises it as `OUT_OF_MEMORY_ERROR` (if so, only the watchdog pattern is missing).

**Deliberately out of scope:** bare `ExecutorLostFailure` or "Python worker exited unexpectedly" as signals. Executors are also lost to spot reclamation and network faults, and a matched signal *stops the job* and prints a cause -- a loose pattern would mislabel unrelated failures and kill runs that would have recovered. Only unambiguous memory wording.

`MEMORY_ADVICE` needs no change, and not by coincidence: G.1X is 16 GB with `spark.executor.memory=10g`, leaving ~6 GB for Python; R.1X is 32 GB with 20g, leaving ~12 GB. R.1X doubles the *Python-side* headroom too, so the R-first advice is correct for this failure rather than merely reused. Worth saying so in the comment, so nobody later assumes it is JVM-only.

Tests: one per new marker (fake log event -> summary + `MEMORY_ADVICE`), one for the final-`ErrorMessage` path, and a negative test that an unrelated executor loss does not trip a memory stop.

## Step 3 — relax the bound

With step 2 in place, the bound's job changes from *preventing* OOM to *giving a better diagnosis than OOM when we can tell why*. A genuinely disjoint pair is better served by "these tables are too different to diff accurately" than by a memory error plus generic advice — so keep a bound, but size it to admit any legitimate item collection rather than to defend the heap, and let OOM catch what slips past.

If it becomes byte-based rather than item-based, note that wire bytes are a poor proxy: measured expansion ranges from **1.0x** (one 400 KB value) to **6.8x** (37 small attributes), because the cost is per-attribute object overhead, not payload. Attribute count is a much better predictor than wire size — roughly 320-470 Python bytes per attribute regardless of value length (CPython 3.13: 184 B for each single-entry `{'S': ...}` dict, 48 B for the name string, 41 B of header on every value string). A cumulative-attribute-count budget is therefore the cheapest bound that adapts to item shape, and needs only a running total maintained in `_load_page` and `advance()`.

## Sequencing

Steps 1 and 2 are independently safe and valuable. Step 3 should be gated on step 2 -- it is the only one that reintroduces risk, and step 2 is what makes it affordable.

Server-side steps (1, 3) need `./bulk bootstrap` before live verification, then `make test-e2e-commands`.

## Open input needed for step 3

The widest item collection to treat as legitimate. "Tens of thousands" sizes a budget around 600 MB/stream; "millions" makes step 3 a full removal leaning entirely on step 2.

Start at diff_segment and the existing test_diff.py, then inspect client/src/utils/__init__.py for watchdog markers and memory-failure handling. Reproduce and capture a Python-side memory failure before adding narrowly matched diagnostics and their positive and negative tests. Verify the bound changes and server-side work with ./bulk bootstrap and make test-e2e-commands.

Written by the indexing model from the issue text.