fix(forms): stop the worklist producer OOM — bounded scan, then bounded batches

[sroussey]

Problem

The forms sweep OOM'd on a 6-way --shard launch. Two separate causes, fixed in two commits.

1. The producer materialized every filing of every form. ComputeFormsWorklistTask called filingRepo.query({ form }) for each of the 55 form symbols, applying the shard filter only afterwards. Form 4 alone is ~4.6M rows and the full worklist ~6.4M, at a measured ~460 bytes per 15-column row — ~3 GB per process. Because sharding ran after the load, every shard process paid that in full: ~17.8 GB across six processes, on a 26 GB box also running Postgres.

2. The worklist didn't need to exist at all. FORMS_SWEEP_CONCURRENCY_LIMIT is 10, so the sweep never holds more than ~10 filings — yet the producer computed all 1,017,448 entries for one shard (158 MB of arrays) before the first fetch, during a 7.7-minute scan in which no work happened. Both numbers grow without bound as filings accumulate.

Changes

e300acf — bound the scan. Page through each form and apply the shard filter first, before any other test, so only this shard's slice survives. Retain only the four fields the fan-out reads, never an intermediate Filing[]. successfulRunKeys is split out of listFilingsWithoutSuccessfulRun so the already-processed set is built once per form rather than re-queried per page (listFilingsWithoutSuccessfulRun is reimplemented on top of it and is unchanged for BackfillExtractorTask).

25c0d28 — bound the worklist. The producer now yields at most 5,000 filings per run and resumes where it left off; formsSweepLoop wraps it in a while whose body is producer → forEach over that batch. Peak retention is one page plus one batch, independent of corpus size. This drops queryPage from the path — last-key + limit makes plain query sufficient.

Design notes

Resume state is a plain last-key, not a cursor. SearchCriteria allows one condition per column and has no OR, so the exact keyset predicate (cik, accession) > (lastCik, lastAccession) is not expressible. Resuming at cik >= lastCik and dropping the already-emitted head in memory is equivalent; the re-read is bounded by the largest single (form, cik) group — 4,628 rows at the extreme, ~19 typical. FILING_PAGE_SIZE is held above that group size so the scan can always advance, and readPage throws rather than spinning if that assumption ever stops holding.

The state lives on the producer instance, not on output ports. WhileTask chains the body's merged output forward, and the body ends in forEach — so the producer's output never reaches the next iteration. A resume key routed through ports arrives stale and the loop never advances; I confirmed this with a probe graph (the loop ran exactly once) before settling on instance state with the condition closing over exhausted.

Verification

Measured against the live 6.4M-filing database, shard 1/6:

	before	after
time to first batch	463 s	472 ms
full scan	463 s	22.4 s
worklist arrays	158 MB	~0.8 MB
peak RSS	~3 GB (est. ~17.8 GB across 6)	554 MB

204 batches, largest exactly 5,000, 1,017,448 distinct filings, 0 duplicates — identical total to the pre-batching scan, so coverage is exact-once. The 24x faster scan is a side effect of the indexed cik >= range scan replacing keyset pagination.

Tests: 1986 pass / 0 fail, typecheck clean. New coverage asserts every filing read is bounded (verified to fail when a bare query({ form }) is reintroduced) and that batching resumes without loss or duplication.

Notes for review

• A bug shipped mid-build and only the full-scale run caught it: readPage returned filtered rows, so rows.length < FILING_PAGE_SIZE fired as soon as the resume head trimmed anything and marked forms drained early — the dry run reported 49,407 instead of 1,017,448. Fixed by reporting the raw page size separately. Unit tests passed throughout, which is why the 1,017,448 match is the number to trust.
• Each shard still reads all 6.4M rows to keep its 1/6. That's now 22 s spread across the run rather than a startup stall. Pushing the shard into SQL is possible (a stored shard column, or cik range windows) but is an optimization, not a fix.
• Bounded paging assumes the backend overrides queryPage/query with a real SQL LIMIT, which Postgres and SQLite do. BaseTabularStorage's generic pager sets fetchLimit = useFallback ? undefined : limit, so a backend falling back to it could still issue an unbounded scan.

🤖 Generated with Claude Code

https://claude.ai/code/session_01RuwHHhCEutPqwTF4861Ae3