Skip to content

Retrofit fetch-thread-pool + compute-process-pool into _run_stage_c (issue #566) - #669

Merged
WilfordGrimley merged 3 commits into
masterfrom
fix/stage-c-pooled-compute
Aug 3, 2026
Merged

Retrofit fetch-thread-pool + compute-process-pool into _run_stage_c (issue #566)#669
WilfordGrimley merged 3 commits into
masterfrom
fix/stage-c-pooled-compute

Conversation

@WilfordGrimley

@WilfordGrimley WilfordGrimley commented Aug 3, 2026

Copy link
Copy Markdown

Summary

Retrofit fetch-thread-pool + compute-process-pool into _run_stage_c (issue #566), plus the fixes CI and review surfaced on top of the original design:

  1. Replace the single fetch-ahead thread + sequential compute loop with:
    • ThreadPoolExecutor(3) for fetch (I/O-bound, benefits from concurrency)
    • ProcessPoolExecutor(3) for compute (CPU-bound, previously sequential)
    • Bounded backpressure drain at _STAGE_C_POOL_QUEUE_DEPTH=6 (2x workers) to keep RSS flat
  2. Lockout-drain fix: fetch-future consumption switched from as_completed() (completion-order, racy) to a submission-order list blocked via .result(). Under as_completed(), a lockout on card B could outrace and silently discard an already-successful card A before its compute was ever submitted, violating the "in-flight work drains, nothing new starts" invariant from issue Retrofit decoupled fetch/compute into streaming Stage C (ratified §4 item 3, unimplemented in Phase 2) #472. The fetch threads still run concurrently — only result CONSUMPTION order is fixed, so there is no throughput loss.
  3. Test twin deleted: an earlier revision of this branch gated a second, inline copy of the coordinator loop behind PYTEST_CURRENT_TEST, which meant every test took the inline path and the pooled loop that ships was executed by none of them. _run_stage_c_phase2_inline, _inline_compute_active and _INLINE_COMPUTE_FOR_TESTS are now deleted (105 lines removed, no replacement), so stage_e_dispatch.py contains no test-mode detection at all. Tests instead substitute _SyncStagePoolStub for the module-level ThreadPoolExecutor and ProcessPoolExecutor names, which runs the single production loop under a synchronous pool. That is the technique test_run_image_evidence_cohort.py already uses via its own _SyncPoolStub.
  4. Removed the parent-side connection.close() before forking the compute pool — it was breaking this coordinator's own later DB writes (mark_ledger_failed on a compute-side crash). Each compute worker closes its own inherited connection copy in _stage_c_compute_worker_init, which is the side that actually needs a fresh connection post-fork.
  5. image-cdn/wrangler.toml: IMAGE_FULL_TIER_RATE_LIMITER 150 to 60 (15/s to 6/s), the owner-approved value, staying under the 7/s client-side GOOGLE_IMAGE ceiling so the worker limiter remains the binding politeness control.

Test coverage, stated honestly: the coordinator loop that ships is now the one tests execute — lockout drain, error propagation, throttle handling, backpressure accounting and both pool shutdowns all run through the production code path. What the stub does not exercise is real process semantics: fork, worker crash recovery, and pool teardown under a genuine ProcessPoolExecutor. _stage_c_compute_worker_init has direct unit coverage of its own. That residual gap is the same one run_image_evidence_cohort.py carries and is closed only by a production run.

Closes #566.

Test plan

  • Regression test test_lockout_mid_prefetch_drains_the_already_fetched_card_but_starts_no_more — card A gets ImageEvidence persisted, cards B and C do not, status is completed-with-trip. Confirmed to FAIL against the pre-fix as_completed() consumption order across 8 PYTHONHASHSEED values and to pass with the fix, so it grades production code rather than a twin.
  • Direct unit coverage added for _stage_c_compute_worker_init (short-circuit on, default, short-circuit off).
  • Full backend suite: cd MPCAutofill && pytest . — the same invocation .github/actions/test-backend uses — run in a local venv (Django 4.2.30, pytest 9.1.1). 3571 passed, 8 skipped, 0 failed in 478.33s.
  • Concurrency caps unchanged: fetch 3 threads, compute 3 processes, queue depth 6. No cap was raised to make anything pass.
  • Lint and type checks via pre-commit on commit: ruff, isort, black, mypy, prettier all clean.
  • manage.py makemigrations --check --dry-run --skip-checks — no changes detected.
  • wiki: no change — an internal coordinator-loop fix plus a rate-limiter value, with no user- or admin-visible surface.
  • docs/upstreaming/extractable-primitives.md: no change — _run_stage_c is fork-entangled catalog pipeline code, not a new or destroyed no-dependency primitive.
  • CI on this push (bf4bf90): all 14 required checks green, including "Backend tests" (7m36s) and "Formatting and static type checking". mergeable_state=clean.

…ion, gate inline mode by pytest env

- Lockout-drain: switch fetch-future consumption from as_completed() (completion-order,
  racy) to a submission-order list blocked via .result() - a lockout on card B could
  otherwise outrace and discard an already-successful card A before its compute was ever
  submitted. Regression pin: test_lockout_mid_prefetch_drains_the_already_fetched_card_but_starts_no_more.

- BrokenProcessPool cascade: _INLINE_COMPUTE_FOR_TESTS is now gated globally by
  PYTEST_CURRENT_TEST (checked at call time via _inline_compute_active(), not import time),
  so every test file runs the coordinator logic inline instead of forking a real
  ProcessPoolExecutor under pytest-django's connection-wrapped test transaction. Removes the
  now-redundant per-file fixture in test_stream_full_catalog.py.

- Also removes the parent-side connection.close() before forking the compute pool - it was
  breaking this coordinator's own later DB writes (mark_ledger_failed) under pytest. Each
  compute worker already closes its own inherited connection in its initializer, matching
  run_image_evidence_cohort.py's _init_worker pattern.

- image-cdn/wrangler.toml: IMAGE_FULL_TIER_RATE_LIMITER 150->60 (15/s->6/s), the
  owner-approved value staying under the 7/s client-side GOOGLE_IMAGE ceiling.

Full local suite: 3568 passed, 8 skipped, 0 failed. black/isort/ruff clean.
makemigrations --check: no changes detected.
…or under test

The inline PYTEST_CURRENT_TEST-gated Stage C compute path
(_run_stage_c_phase2_inline / _inline_compute_active /
_INLINE_COMPUTE_FOR_TESTS) is gone from stage_e_dispatch.py. Production no
longer branches on test detection - the pooled ThreadPoolExecutor +
ProcessPoolExecutor loop in _run_stage_c is now the ONLY implementation,
and it is what runs under every test.

test_stage_e_dispatch.py introduces _SyncStagePoolStub, a drop-in
synchronous stand-in for both executors that still invokes the real
initializer (needed for short_circuit propagation) and still runs the
real coordinator loop (submission-order consumption, the
_STAGE_C_POOL_QUEUE_DEPTH backpressure drain, the final as_completed
drain) unmodified.

test_stage_e_shakedown.py and test_stream_full_catalog.py each reuse
_SyncStagePoolStub (imported, not copy-pasted) via their own autouse
_sync_stage_c_pools fixture, since both dispatch through the real
dispatch_micro_batch -> _run_stage_c path and had no stub of their own
once the global pytest-detection gate was removed.
@WilfordGrimley
WilfordGrimley merged commit 8b7f02c into master Aug 3, 2026
14 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant