diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index 89b068fb3..6fe5ebb08 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -79,27 +79,20 @@ import-time dependency between sibling engines. """ +import concurrent.futures import logging import os -import queue -import threading import time from collections import deque +from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor from dataclasses import dataclass, field from typing import Any, Callable, Deque, Iterable, Optional from django.conf import settings from django.utils import timezone -from cardpicker.collector_line_artist import ( - build_name_artist_lookup, - build_printing_artist_lookup, - load_artist_lexicon, -) from cardpicker.evidence_transfer import find_transfer_source, transfer_evidence -from cardpicker.harvest_fetch_limiter import GoogleFetchLockoutError from cardpicker.local_calculate_verdicts import ( - known_set_codes, run_fallback_calculator, run_join_key_calculator, run_slow_path_calculator, @@ -636,27 +629,240 @@ def _verify_stage_c_chunk(chunk: list[int]) -> Iterable[int]: # brief's own concrete number rather than leaving it operator-tunable; PASSIVE mode's own # micro-batches are small enough (§3 decision (2), a handful to a few dozen cards) that this never # needs retuning the way BULK mode's own `--queue-depth` does. -_STAGE_C_FETCH_AHEAD_DEPTH = 2 +# Stage C fetch/compute pool sizing (issue #566). FETCH: three threads since I/O-bound fetches do +# benefit from concurrency (unlike the old single-thread fetch-ahead worker, which was fine for a +# synthetic test run but left a full-catalog production pass spending ~40% of its wall-clock +# waiting for the fetch-ahead thread). COMPUTE: three processes, matching the worker count the +# cohort command settled on for real throughput. HANDOFF QUEUE: bounded to 2 per compute worker so +# RSS stays bounded regardless of batch size (the old single-thread design's own _STAGE_C_FETCH_ +# AHEAD_DEPTH=2 satisfied this incidentally; the pooled design must state it explicitly since the +# ThreadPoolExecutor + ProcessPoolExecutor combination has no built-in handoff bound — without +# `_STAGE_C_POOL_QUEUE_DEPTH`'s own backpressure drain below, the main loop would keep handing +# completed fetches into the ProcessPoolExecutor's own UNBOUNDED internal queue, and RSS could +# grow without bound over the course of a production-length pass). +_STAGE_C_FETCH_THREADS = 3 +_STAGE_C_COMPUTE_WORKERS = 3 +_STAGE_C_POOL_QUEUE_DEPTH = 6 # 2 × _STAGE_C_COMPUTE_WORKERS + + +# Process-global lookups built once per compute worker process (initializer). All four are set by +# _stage_c_compute_worker_init and consumed by _stage_c_compute_one_card — they are module-level +# globals BY DESIGN (must be picklable-free, since multiprocessing's fork semantics mean each +# worker process gets its own independent module namespace). +_compute_pool_short_circuit: Optional[bool] = None +_compute_pool_lexicon: Any = None +_compute_pool_artist_lexicon: Any = None +_compute_pool_printing_artist_lookup: Any = None +_compute_pool_name_artist_lookup: Any = None + + +def _stage_c_compute_worker_init( + short_circuit: Optional[bool] = None, +) -> None: + """Initializer for every Stage C compute pool worker process. + + Three responsibilities, all required for correctness: + + 1. OMP_THREAD_LIMIT=1: tesseract's own OpenMP thread pool defaults to one thread per core; + inside a per-card worker in a multi-process pool, the product (pool workers × tesseract + threads) oversubscribes the machine's physical core count, producing WORSE per-card + throughput than a single sequential worker for a compute pool of any size >1. Setting + this environment variable BEFORE tesseract's first import is the same mechanism the + cohort command's own _init_worker uses (``run_image_evidence_cohort.py``, §init_worker), + deliberately reused here rather than re-implemented — one OMP-thread-pinning mechanism + across all callers, the cohort command's own comment is the canonical design doc. + + 2. Fresh DB connections: fork() carries parent-process Django connections into the child, + but the child's own fork of the underlying TCP socket is ALREADY the parent's socket + (not an independent connection) — any use will trip a "DatabaseWrapper objects created + in a thread can only be used in that same thread" or "connection already closed" error + at the first query. ``connections.close_all()`` forces fresh connection creation on the + child's own first ``.cursor()`` call. + + 3. Lookup singletons: ``known_set_codes``, ``load_artist_lexicon``, + ``build_printing_artist_lookup``, and ``build_name_artist_lookup`` are all built once per + worker process (matching the batch-scoped ``_run_stage_c`` convention exactly — the + call sites are identical, just hoisted from per-batch in the parent to per-worker-process + in the pool), and the last two return stateful resolvers with internal caches backed by + DB queries — they MUST be rebuilt in the child rather than passed across the fork boundary, + which would carry a stale cache pointing at the (now-closed) parent's own DB handles. + """ + import os as _os + + from django.db import connections as _connections + + from cardpicker.collector_line_artist import build_name_artist_lookup as _build_name + from cardpicker.collector_line_artist import ( + build_printing_artist_lookup as _build_printing, + ) + from cardpicker.collector_line_artist import load_artist_lexicon as _load_artist + from cardpicker.local_calculate_verdicts import known_set_codes as _known_set_codes + + # 1. Pin OpenMP thread count — MUST happen before any PIL/tesseract import reaches OpenMP init. + _os.environ["OMP_THREAD_LIMIT"] = "1" + + # 2. Fresh DB connections — the parent closed its own connection before forking, so the + # child inherits ``connection.connection is None``. ``close_all()`` is a no-op here + # (every wrapper already has ``connection is None``), and ``ensure_connection()`` opens + # a fresh socket on first use. + for conn in _connections.all(): + conn.close() + _connections.close_all() + + # 3. Lookup singletons — one per worker process, matching the batch-scoped convention. + global _compute_pool_short_circuit + global _compute_pool_lexicon, _compute_pool_artist_lexicon + global _compute_pool_printing_artist_lookup, _compute_pool_name_artist_lookup + + _compute_pool_short_circuit = short_circuit + _compute_pool_lexicon = _known_set_codes() + _compute_pool_artist_lexicon = _load_artist() + _compute_pool_printing_artist_lookup = _build_printing() + _compute_pool_name_artist_lookup = _build_name() + + +def _stage_c_compute_one_card( + card_id: int, + content_hash: Optional[int], + image_bytes: bytes, + fetch_latency_ms: float, + md5_checksum: Optional[str], + sha256_checksum: Optional[str], + card_name: str, + run_id: str, + dry_run: bool, +) -> "tuple[bool, Optional[Exception]]": + """Picklable per-card compute function for the ProcessPoolExecutor. + + Each invocation: PIL-decode the image bytes → compute_card_evidence → persist_evidence + (wrapped in suppress_evidence_change_echo, since ContextVars are process-local and the + worker gets a fresh process — the parent's own echo-suppression token does NOT transfer + across the fork boundary). Returns (success, error) so the main loop can distinguish a + completed card from a compute crash and handle each appropriately. + + All lookup singletons (lexicon, artist_lexicon, printing_artist_lookup, name_artist_lookup, + short_circuit) are read from process-global module variables set by + _stage_c_compute_worker_init — no second build, no imports at call time beyond PIL and the + two evidence functions themselves. + """ + from io import BytesIO + + from PIL import Image + + from cardpicker.image_evidence import compute_card_evidence, persist_evidence + from cardpicker.stage_e_signals import suppress_evidence_change_echo + + try: + image = Image.open(BytesIO(image_bytes)) + result = compute_card_evidence( + card_id, + content_hash, + image, + fetch_latency_ms=fetch_latency_ms, + short_circuit=_compute_pool_short_circuit, + known_set_codes=_compute_pool_lexicon, + artist_lexicon=_compute_pool_artist_lexicon, + printing_artist_lookup=_compute_pool_printing_artist_lookup, + card_artist_names=_compute_pool_name_artist_lookup(card_name), + md5_checksum=md5_checksum, + sha256_checksum=sha256_checksum, + ) + if not dry_run: + with suppress_evidence_change_echo(): + persist_evidence(result, run_id=run_id) + return (True, None) + except Exception as exc: # noqa: BLE001 — deliberately broad; any compute-time fault must surface + return (False, exc) + + +def _stage_c_fetch_one(card: "Card") -> "_StageCFetchOutcome": + """Single-card fetch for the ThreadPoolExecutor. + + Deliberately NOT attached to a stop_event: the pool's own shutdown(wait=False, + cancel_futures=True) in the finally block is the sole stop mechanism — there is no + cross-thread stop coordination to maintain. Every fetch runs to completion (success, + lockout, throttle, or error), packing the result into a _StageCFetchOutcome for the + main loop to interpret. + + Mirrors _stage_c_fetch_ahead_worker's own fetch logic byte-for-byte but for ONE card + — no iteration, no stop_event check, no queue.put(). The main loop is now the one that + decides what to do with each outcome (including whether to break on lockout/error), + rather than the worker thread stopping itself. + """ + from cardpicker.harvest_fetch_limiter import ( + DestinationThrottledError, + GoogleFetchLockoutError, + ) + from cardpicker.image_cdn_fetch import DEFAULT_FETCH_DPI, fetch_card_image_bytes + + fetch_started_at = time.monotonic() + try: + image_bytes = fetch_card_image_bytes(card, dpi=DEFAULT_FETCH_DPI) + except GoogleFetchLockoutError: + return _StageCFetchOutcome( + card_id=card.pk, + content_hash=card.content_phash, + md5_checksum=card.md5_checksum, + sha256_checksum=card.sha256_checksum, + image_bytes=None, + fetch_latency_ms=0.0, + card_name=card.name, + lockout=True, + ) + except DestinationThrottledError: + return _StageCFetchOutcome( + card_id=card.pk, + content_hash=card.content_phash, + md5_checksum=card.md5_checksum, + sha256_checksum=card.sha256_checksum, + image_bytes=None, + fetch_latency_ms=0.0, + card_name=card.name, + throttled=True, + ) + except Exception as exc: # noqa: BLE001 — deliberately broad, see _StageCFetchOutcome.error + return _StageCFetchOutcome( + card_id=card.pk, + content_hash=card.content_phash, + md5_checksum=card.md5_checksum, + sha256_checksum=card.sha256_checksum, + image_bytes=None, + fetch_latency_ms=0.0, + card_name=card.name, + error=exc, + ) + + fetch_latency_ms = (time.monotonic() - fetch_started_at) * 1000 + return _StageCFetchOutcome( + card_id=card.pk, + content_hash=card.content_phash, + md5_checksum=card.md5_checksum, + sha256_checksum=card.sha256_checksum, + image_bytes=image_bytes, + fetch_latency_ms=fetch_latency_ms, + card_name=card.name, + ) @dataclass class _StageCFetchOutcome: - """One card's own fetch-stage result, handed from `_stage_c_fetch_ahead_worker` (the fetch- - ahead thread) to `_run_stage_c`'s own sequential compute loop via a bounded `queue.Queue`. - `card`/`content_hash`/`md5_checksum`/`sha256_checksum` are all read from the SAME `Card` - instance `_run_stage_c` already loaded (and used for its own transfer check) before handing - this card off to the fetch-ahead thread - no second `Card` query on either side of the - boundary. `lockout=True` iff this card's OWN fetch attempt raised `GoogleFetchLockoutError` - - the compute loop treats this as the signal to stop (module docstring's "halts NEW fetches + """One card's own fetch-stage result, returned by ``_stage_c_fetch_one`` (a single-card fetch + worker for the ThreadPoolExecutor) and drained by ``_run_stage_c``'s Phase 2 coordinator loop + via ``concurrent.futures.as_completed()``. ``card``/``content_hash``/``md5_checksum``/ + ``sha256_checksum`` are all read from the SAME ``Card`` instance ``_run_stage_c`` already + loaded (and used for its own transfer check) before handing this card off to the fetch pool - + no second ``Card`` query on either side of the boundary. ``lockout=True`` iff this card's OWN + fetch attempt raised ``GoogleFetchLockoutError`` - the coordinator loop treats this as the + signal to stop submitting new fetches/computes (module docstring's "halts NEW fetches immediately" bar), never a fetch failure to retry. - `error`, if set, is a NON-`GoogleFetchLockoutError` exception the fetch attempt raised (2026-07-25, - kill-safety fix - see `_stage_c_fetch_ahead_worker`'s own docstring for why this exists at - all): re-raised by the compute loop IN THE MAIN THREAD the instant it's observed, so a crash - during fetch still propagates out of `_run_stage_c`/`dispatch_micro_batch` exactly as it did - before this module had a separate fetch thread at all - `TestKillSafetyResumeContract`'s own - "a mid-batch crash leaves a truthful FAILED ledger row" contract does not distinguish between - a crash during fetch and a crash during compute, and must not silently become a hang instead.""" + ``error``, if set, is a NON-``GoogleFetchLockoutError`` exception the fetch attempt raised + (2026-07-25, kill-safety fix - see ``_stage_c_fetch_one``'s own docstring for the motivation): + re-raised by the coordinator loop IN THE MAIN THREAD the instant it's observed, so a crash + during fetch still propagates out of ``_run_stage_c``/``dispatch_micro_batch`` exactly as it + did before this module had a fetch pool - ``TestKillSafetyResumeContract``'s own "a mid-batch + crash leaves a truthful FAILED ledger row" contract does not distinguish between a crash + during fetch and a crash during compute, and must not silently become a hang instead.""" card_id: int content_hash: Optional[int] @@ -668,152 +874,25 @@ class _StageCFetchOutcome: error: Optional[BaseException] = None # 2026-07-29 (`collector_line_artist`'s CARD-NAME NARROWING): the card's own uploaded name, # read off the SAME already-loaded `Card` instance every other field here comes from - no - # second query on either side of the fetch/compute boundary. The compute loop resolves it to - # this card's real artists via the batch's own `NameArtistLookup`; it is deliberately NOT - # resolved on the fetch thread, which must stay purely I/O-bound. Defaulted (like every field - # after `lockout` here) so the two error/lockout constructions above stay keyword-complete. + # second query. The compute worker resolves it to this card's real artists via its own + # process-local `NameArtistLookup` (built once per worker process by + # `_stage_c_compute_worker_init`); it is deliberately NOT resolved on the fetch thread, + # which must stay purely I/O-bound. Defaulted (like every field after `lockout` here) so + # the two error/lockout constructions above stay keyword-complete. card_name: str = "" # `throttled=True` iff this card's own fetch raised `harvest_fetch_limiter # .DestinationThrottledError` (a 429/503 from the destination) - RATE PRESSURE, the third and # mildest of this dataclass's three failure severities, added 2026-07-30 under the owner rate # ruling ("the limit needs to throttle not shut it down"). The other two both STOP the fetch - # thread (`lockout` and `error` each set `stop_event` and return); this one does NOT - the - # thread proceeds to the next card, and the limiter's own already-widened pacing interval is - # what makes that next fetch slower. The compute loop must NOT record a throttled outcome onto - # the fetch-outcome window: doing so is exactly what used to let sustained rate pressure trip - # `EnvelopeTrip.Bar.FETCH_FAILURE_RATE` and hard-stop the whole unattended pass. + # pool worker (`lockout` and `error` each produce a terminal outcome); this one does NOT - + # the worker returns the outcome and the fetch pool's own `shutdown(wait=False, + # cancel_futures=True)` in the finally block handles the rest. The coordinator loop must NOT + # record a throttled outcome onto the fetch-outcome window: doing so is exactly what used to + # let sustained rate pressure trip `EnvelopeTrip.Bar.FETCH_FAILURE_RATE` and hard-stop the + # whole unattended pass. throttled: bool = False -def _stage_c_fetch_ahead_worker( - cards: list[Card], - out_queue: "queue.Queue[_StageCFetchOutcome]", - stop_event: threading.Event, -) -> None: - """ - THE fetch-ahead thread (issue #472) - ONE thread, sequential fetches (never pooled: the design - brief's own "no compute pooling" bar applies equally to a fetch pool here, since a SECOND - concurrent fetch would only race further ahead of a compute loop that's already the slower - stage, buying nothing the bounded queue depth doesn't already buy via fetch/compute OVERLAP - alone). Pushes one `_StageCFetchOutcome` per entry in `cards`, in order - `queue.Queue`'s own - FIFO ordering means the compute loop always drains outcomes in the SAME order this thread - fetched them, satisfying the design brief's own "fetch-outcome window records in completion - order" bar for free (a single serial fetch worker's own completion order IS its submission - order, there is no reordering possible with only one fetch in flight at a time). - - `out_queue`'s own bound (`queue.Queue(maxsize=_STAGE_C_FETCH_AHEAD_DEPTH)`, set by the caller) - is what keeps RSS bounded independent of batch size - `put()` BLOCKS once the queue is full, - so this thread can never race further than `_STAGE_C_FETCH_AHEAD_DEPTH` outcomes ahead of - whatever the compute loop has actually consumed so far, regardless of how many cards remain in - `cards` or how large a future catalog's own micro-batch could be. - - LOCKOUT (design brief's own "instant halt" bar): the moment `fetch_card_image_bytes` raises - `GoogleFetchLockoutError` for one card, `stop_event` is set and this thread returns immediately - WITHOUT attempting any further card in `cards` - "halts NEW fetches immediately". The card - whose OWN fetch triggered the lockout is reported to the compute loop via this outcome's own - `lockout=True` flag (never silently dropped), so the compute loop can record the trip itself - and stop too. Every outcome that already made it into `out_queue` BEFORE this happened is left - there untouched - "in-flight work drains" - the compute loop keeps consuming those (they sort - earlier in FIFO order than the lockout outcome) before it ever reaches the lockout marker, - exactly mirroring the pre-#472 sequential design's own "an already-fetched image still gets to - finish its own compute+persist" property, just now genuinely overlapped rather than accidental. - - ANY OTHER EXCEPTION (2026-07-25, kill-safety fix - `TestKillSafetyResumeContract`'s own - mid-batch-crash test caught this during review): a plain `try/except GoogleFetchLockoutError` - here would let a non-lockout exception (a real bug, a simulated kill-drill fault, a genuine - network error `fetch_card_image_bytes` doesn't itself wrap) kill this THREAD silently - Python - does not propagate an uncaught exception from a spawned `threading.Thread` to its caller, so - the compute loop's own `queue.get()` for that card would block FOREVER waiting for an outcome - that will never arrive, turning a crash into a silent hang instead of the loud, ledger-recorded - failure the resume contract requires. Caught here as a bare `Exception`, packaged onto the - outcome's own `error` field, and this thread stops (same "no further cards attempted" posture - as a lockout) - the compute loop re-raises it in the MAIN thread the instant it's seen. - - A THROTTLE (2026-07-30 owner rate ruling: "the limit needs to throttle not shut it down") is - the one severity that does NOT stop this thread. `DestinationThrottledError` means the - destination answered 429/503 and `harvest_fetch_limiter` has ALREADY widened its own pacing - interval; the correct response is to mark this one card deferred (`throttled=True`) and CARRY - ON to the next card, whose fetch will now be slower because of that widening. `stop_event` is - deliberately NOT set, no error is packaged, and the run continues at a degraded rate. Note the - ordering constraint this creates: `except DestinationThrottledError` must sit ABOVE the broad - `except Exception`, or the throttle is swallowed into the stop-the-thread path and the whole - conversion is undone. - """ - from cardpicker.harvest_fetch_limiter import DestinationThrottledError - from cardpicker.image_cdn_fetch import DEFAULT_FETCH_DPI, fetch_card_image_bytes - - for card in cards: - if stop_event.is_set(): - return - - fetch_started_at = time.monotonic() - try: - image_bytes = fetch_card_image_bytes(card, dpi=DEFAULT_FETCH_DPI) - except GoogleFetchLockoutError: - stop_event.set() - out_queue.put( - _StageCFetchOutcome( - card_id=card.pk, - content_hash=card.content_phash, - md5_checksum=card.md5_checksum, - sha256_checksum=card.sha256_checksum, - image_bytes=None, - fetch_latency_ms=0.0, - card_name=card.name, - lockout=True, - ) - ) - return - except DestinationThrottledError: - # NOT a stop - see this function's own docstring. No `stop_event.set()`, no `return`: - # the loop proceeds to the next card at the limiter's newly-widened pace. - logger.warning( - "Stage E dispatch: throttled fetching card %s - deferring it and continuing at a slower pace", - card.pk, - ) - out_queue.put( - _StageCFetchOutcome( - card_id=card.pk, - content_hash=card.content_phash, - md5_checksum=card.md5_checksum, - sha256_checksum=card.sha256_checksum, - image_bytes=None, - fetch_latency_ms=0.0, - card_name=card.name, - throttled=True, - ) - ) - continue - except Exception as exc: # noqa: BLE001 - deliberately broad, see docstring above - stop_event.set() - out_queue.put( - _StageCFetchOutcome( - card_id=card.pk, - content_hash=card.content_phash, - md5_checksum=card.md5_checksum, - sha256_checksum=card.sha256_checksum, - image_bytes=None, - fetch_latency_ms=0.0, - card_name=card.name, - error=exc, - ) - ) - return - fetch_latency_ms = (time.monotonic() - fetch_started_at) * 1000 - - out_queue.put( - _StageCFetchOutcome( - card_id=card.pk, - content_hash=card.content_phash, - md5_checksum=card.md5_checksum, - sha256_checksum=card.sha256_checksum, - image_bytes=image_bytes, - fetch_latency_ms=fetch_latency_ms, - card_name=card.name, - ) - ) - - def _run_stage_c( batch_ids: list[int], run_id: str, @@ -836,14 +915,19 @@ def _run_stage_c( counted via `outcome.stage_c_transferred`/`stage_c_completed`. Every card that still needs a real extraction (no md5, no eligible sibling, or a loud pairing/content-hash anomaly - see that function's own docstring) is collected into `to_fetch`. - 2. **Decoupled fetch-ahead + sequential compute** (issue #472): `to_fetch` is handed to ONE - fetch-ahead thread (`_stage_c_fetch_ahead_worker`) writing into a bounded - (`_STAGE_C_FETCH_AHEAD_DEPTH`) queue; THIS function's own loop stays the sequential OCR/ - extraction compute stage the design brief mandates (no compute pooling), just now able to - decode+extract card N while the fetch-ahead thread is already fetching card N+1's bytes, - instead of blocking on that fetch itself. Every fetch outcome (transfer OR real fetch) is - recorded onto `_window` regardless of whether it ends up mattering to THIS batch's own - envelope decision - the window spans the whole worker process's uptime, not one batch. + 2. **Decoupled fetch-thread-pool + compute-process-pool** (issue #566): ``to_fetch`` is + submitted to a ``ThreadPoolExecutor`` of three fetch threads, each calling + ``_stage_c_fetch_one`` (single-card, no shared stop event — the pool's own + ``shutdown(wait=False, cancel_futures=True)`` in the finally block is the sole stop + mechanism). Completed fetch results are drained via ``as_completed()``; each successful + fetch is handed to a ``ProcessPoolExecutor`` of three compute workers, each running + ``_stage_c_compute_one_card`` (the compute+persist unit, with echo suppression inside + the worker process since ``ContextVar`` is process-local). A bounded backpressure drain + keeps the handoff between pools at or below ``_STAGE_C_POOL_QUEUE_DEPTH`` (6 = 2 × 3 + workers) — whenever ``len(pending_compute)`` reaches that bound, ONE completed compute + result is drained before the next fetch is handed to the compute pool. This is the sole + RSS-flat mechanism: ``ProcessPoolExecutor``'s own internal queue is UNBOUNDED, so + without this drain, RSS grows without bound over the course of a production-length pass. 3. **Echo suppression** (issue #472's own fold, `cardpicker.stage_e_signals`'s own module docstring has the full mechanism writeup): both the transfer write in phase 1 and the `persist_evidence` write in phase 2 are wrapped in `suppress_evidence_change_echo()` - a @@ -897,12 +981,6 @@ def _run_stage_c( roll back) - and records a fresh trip via `check_envelope(google_lockout=True)` so the NEXT dispatch call refuses until an owner acknowledges it, matching the "instant pause" bar exactly. """ - from io import BytesIO - - from PIL import Image - - from cardpicker.image_evidence import compute_card_evidence, persist_evidence - if force_stage_c_reextract: already_done_ids: set[int] = set() else: @@ -912,14 +990,6 @@ def _run_stage_c( card_id__in=batch_ids, extractor_versions__contains=manifest_versions ).values_list("card_id", flat=True) ) - lexicon = known_set_codes() - # COLLECTOR-LINE ARTIST GATE (2026-07-29 - see `image_evidence.compute_card_evidence`'s own - # `artist_lexicon` docstring paragraph): built once per batch and threaded through, the same - # shape `known_set_codes()` immediately above already has, so `compute_card_evidence` keeps - # issuing no DB query of its own. `build_printing_artist_lookup()` returns a stateful resolver - # that caches one query per EXPANSION (not per card) across this whole batch. - artist_lexicon = load_artist_lexicon() - printing_artist_lookup = build_printing_artist_lookup() # PHASE 1 (module docstring): build the work list, resolving evidence transfer BEFORE # deciding whether a card needs a fetch at all. @@ -951,61 +1021,50 @@ def _run_stage_c( if not to_fetch: return None - # CARD-NAME NARROWING (2026-07-29 - see `collector_line_artist`'s own docstring section): the - # third batch-scoped resolver on this same "build once, thread through" convention, but built - # HERE rather than alongside the other two - after the transfer pass has already established - # that this batch really does have cards to extract. It is backed by - # `local_calculate_verdicts._get_cached_candidate_name_index()` (the single process-cached - # entry point that module documents as mandatory for every batch-reachable caller, so a worker - # that also runs Stage D shares the one index rather than building a second), and building - # that index is the most expensive of the three - a batch whose cards were all satisfied by - # evidence transfer must not pay for it, the same laziness `run_join_key_calculator`'s own - # `index` already practises. - name_artist_lookup = build_name_artist_lookup() - - # PHASE 2 (module docstring): decoupled fetch-ahead thread + this function's own sequential - # compute loop. - fetch_queue: "queue.Queue[_StageCFetchOutcome]" = queue.Queue(maxsize=_STAGE_C_FETCH_AHEAD_DEPTH) - stop_event = threading.Event() - fetch_thread = threading.Thread( - target=_stage_c_fetch_ahead_worker, args=(to_fetch, fetch_queue, stop_event), daemon=True + # PHASE 2 (module docstring): decoupled fetch-thread-pool + compute-process-pool (issue #566). + + fetch_pool = ThreadPoolExecutor(max_workers=_STAGE_C_FETCH_THREADS) + + # The parent keeps its own DB connection open across the fork (unlike a manual + # connection.close() here, which broke this coordinator's own later writes, e.g. + # mark_ledger_failed on a compute-side crash). Each worker closes its OWN inherited + # copy in its initializer below — that's the side that actually needs a fresh one. + compute_pool = ProcessPoolExecutor( + max_workers=_STAGE_C_COMPUTE_WORKERS, + initializer=_stage_c_compute_worker_init, + initargs=(short_circuit,), ) - fetch_thread.start() + # Submission-order list, not ``as_completed`` — completion order across concurrent fetch + # threads is racy (an instant lockout can outrace a slower successful fetch submitted + # earlier); blocking on futures in submission order doesn't serialize the threads, it only + # fixes which result the loop consumes next. + fetch_futures: "list[concurrent.futures.Future[_StageCFetchOutcome]]" = [ + fetch_pool.submit(_stage_c_fetch_one, card) for card in to_fetch + ] + pending_compute: "dict[concurrent.futures.Future[tuple[bool, Optional[Exception]]], int]" = {} + stop = False trip: Optional[EnvelopeTrip] = None + try: - for _ in range(len(to_fetch)): - fetch_outcome = fetch_queue.get() + for fetch_future in fetch_futures: + if stop: + break + + fetch_outcome = fetch_future.result() + + if fetch_outcome.throttled: + outcome.stage_c_fetch_throttled += 1 + continue if fetch_outcome.error is not None: - # Re-raise IN THE MAIN THREAD - see _StageCFetchOutcome/_stage_c_fetch_ahead_ - # worker's own docstrings for why this exists (a spawned thread's own uncaught - # exception never reaches the caller on its own). Propagates out of this function, - # through dispatch_micro_batch's own `except Exception: mark_ledger_failed(...); - # raise` - identical observable behaviour to the pre-#472 sequential design's own - # "a fetch-time crash surfaces exactly like a compute-time one" contract. raise fetch_outcome.error if fetch_outcome.lockout: _window.record(success=False) logger.error("Stage E dispatch: GoogleFetchLockoutError observed - halting Stage C for this batch") trip = check_envelope(_sample_envelope_signals(google_lockout=True), run_id=run_id) - break - - if fetch_outcome.throttled: - # THE HALT->THROTTLE CONVERSION (2026-07-30 owner rate ruling). This branch MUST - # sit above the `image_bytes is None` one below (a throttled outcome also carries - # `image_bytes=None`) and MUST NOT touch `_window`. Recording a throttle as a - # fetch FAILURE is what used to make sustained rate pressure indistinguishable - # from a broken dependency: at >1% of a rolling 500-card window it tripped - # `EnvelopeTrip.Bar.FETCH_FAILURE_RATE`, which is a HARD STOP requiring a human - # `resolve_envelope_trip` acknowledgement and exits `stream_full_catalog` with - # code 3 - killing a 230,753-card unattended pass over a condition whose correct - # answer is "go slower". The card is simply not processed this pass; the Stage C - # backlog walk re-selects it later, by which time the limiter's pacing has either - # decayed back toward the ceiling or settled at a rate the destination tolerates. - # The OTHER three bars are untouched: host load, RSS and a 403 lockout still halt. - outcome.stage_c_fetch_throttled += 1 + stop = True continue if fetch_outcome.image_bytes is None: @@ -1014,58 +1073,48 @@ def _run_stage_c( continue _window.record(success=True) - image = Image.open(BytesIO(fetch_outcome.image_bytes)) - result = compute_card_evidence( + + # Backpressure: drain one pending compute result before submitting the next whenever + # the handoff queue is at capacity — the ProcessPoolExecutor's own internal queue is + # UNBOUNDED, so this is the sole mechanism keeping RSS flat regardless of batch size. + if len(pending_compute) >= _STAGE_C_POOL_QUEUE_DEPTH: + done, _ = concurrent.futures.wait(pending_compute, return_when=concurrent.futures.FIRST_COMPLETED) + for cf in done: + success, error = cf.result() + if not success: + assert error is not None # narrow for mypy: !success → error is Some + raise error + outcome.stage_c_completed += 1 + del pending_compute[cf] + + cf = compute_pool.submit( + _stage_c_compute_one_card, fetch_outcome.card_id, fetch_outcome.content_hash, - image, - fetch_latency_ms=fetch_outcome.fetch_latency_ms, - short_circuit=short_circuit, - known_set_codes=lexicon, - artist_lexicon=artist_lexicon, - printing_artist_lookup=printing_artist_lookup, - card_artist_names=name_artist_lookup(fetch_outcome.card_name), - md5_checksum=fetch_outcome.md5_checksum, - sha256_checksum=fetch_outcome.sha256_checksum, + fetch_outcome.image_bytes, + fetch_outcome.fetch_latency_ms, + fetch_outcome.md5_checksum, + fetch_outcome.sha256_checksum, + fetch_outcome.card_name, + run_id, + dry_run, ) - if not dry_run: - with suppress_evidence_change_echo(): - persist_evidence(result, run_id=run_id) + pending_compute[cf] = fetch_outcome.card_id + + # All fetches done — drain the remaining compute results. + for cf in concurrent.futures.as_completed(pending_compute): + success, error = cf.result() + if not success: + assert error is not None # narrow for mypy: !success → error is Some + raise error outcome.stage_c_completed += 1 finally: - # Always signal-then-drain-then-join, whether the loop above finished normally, broke on - # a lockout, or raised - the fetch-ahead thread is `daemon=True` (won't block process exit - # on its own) but this function should never RETURN while it's still mid-fetch for a card - # nothing will ever consume. - # - # `stop_event.set()` MUST happen BEFORE the drain+join below (Tron §8 gate condition, - # 2026-07-25, HIGH severity - found on review): a bare `fetch_thread.join()` here, with - # neither `stop_event.set()` nor a queue drain first, left the fetch-ahead thread wedged - # FOREVER the moment COMPUTE (not fetch) raised mid-batch - e.g. a corrupt download that - # decodes far enough to pass the fetch stage but raises a PIL error inside - # `compute_card_evidence`/`persist_evidence` above. Once this loop stops calling - # `fetch_queue.get()` (it exited via the exception), a fetch-ahead thread already blocked - # on `out_queue.put(...)` for its own next outcome (the queue is bounded at - # `_STAGE_C_FETCH_AHEAD_DEPTH`) never unblocks on its own - `stop_event` alone does - # nothing for a thread that isn't back at its own loop-top `if stop_event.is_set(): return` - # check yet, and nothing else will ever call `.get()` again to free room for that `put()` - # to complete. The observable failure mode was silent and total: `join()` blocks - # indefinitely, so this function (and `dispatch_micro_batch`'s own `except Exception: - # mark_ledger_failed(...); raise` around it) never even reaches the point of recording the - # crash - the `PilotRunLedger` row stays lying at `RUNNING` forever, and the concurrency-cap - # slot this dispatch holds (`stage_e_concurrency.try_acquire_dispatch_slot`) never gets - # released either, wedging the whole worker process's dispatch capacity over ONE corrupt - # image. Fixed by (1) `stop_event.set()` first, so the thread returns the instant it's back - # at its own loop-top, and (2) draining `fetch_queue` below, which is what actually - # unblocks a `put()` already in progress - after at most one more successful put (the one - # the drain makes room for), the thread reaches its own stop_event check and returns. - stop_event.set() - while True: - try: - fetch_queue.get_nowait() - except queue.Empty: - break - fetch_thread.join() + # Shutdown fetch pool — cancel any still-pending fetches (a lockout mid-batch, or an + # exception anywhere above). wait=False + cancel_futures=True is the pooled equivalent of + # the old stop_event.set() + drain queue + join() sequence — no thread left mid-fetch + # for cards nothing will consume. + fetch_pool.shutdown(wait=False, cancel_futures=True) + compute_pool.shutdown(wait=False, cancel_futures=True) return trip diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py index 2e50902bc..827cef3b8 100644 --- a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py @@ -15,8 +15,8 @@ import io import threading -import time -from typing import Any +from concurrent.futures import Future +from typing import Any, Optional import psycopg2 import pytest @@ -82,6 +82,70 @@ def _reset_fetch_failure_window(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(stage_e_dispatch, "_window", _FetchOutcomeWindow()) +class _SyncStagePoolStub: + """Drop-in stand-in for ``ThreadPoolExecutor``/``ProcessPoolExecutor`` that runs submitted + work synchronously, in this test process, and returns a real (already-resolved) + ``concurrent.futures.Future`` — so ``_run_stage_c``'s own coordinator loop (submission-order + fetch consumption, the ``_STAGE_C_POOL_QUEUE_DEPTH`` backpressure drain via ``wait(..., + FIRST_COMPLETED)``, the final ``as_completed`` drain, all real stdlib, untouched) runs + UNMODIFIED against it. Mirrors ``_SyncPoolStub`` in + ``test_run_image_evidence_cohort.py:76`` — this is the same substitution technique, applied + to the same-shaped ``ThreadPoolExecutor``/``ProcessPoolExecutor`` module-level names this + module imports. + + ONE deliberate difference from that stub: this one DOES invoke ``initializer`` when the + pooled loop supplies one. ``_stage_c_compute_worker_init`` is what sets the per-batch + ``short_circuit`` value ``_stage_c_compute_one_card`` reads off a process global — + ``test_short_circuit_is_independent_of_force_stage_c_reextract`` below asserts on exactly + that value, so it must be set fresh on every call, never left at whatever a previous test's + pool last wrote. The initializer's own DB-connection-close step (correct for a freshly + forked worker, hostile to this single-process test session sharing one connection) is + neutralized only for the duration of that one call — a synchronous stub never forks, so + there is no inherited, stale connection to close in the first place. + """ + + def __init__(self, max_workers: Optional[int] = None, initializer: Any = None, initargs: tuple = ()) -> None: + if initializer is not None: + from django.db.backends.base.base import BaseDatabaseWrapper + + original_close = BaseDatabaseWrapper.close + BaseDatabaseWrapper.close = lambda self: None # type: ignore[method-assign] + try: + initializer(*initargs) + finally: + BaseDatabaseWrapper.close = original_close # type: ignore[method-assign] + + def __enter__(self) -> "_SyncStagePoolStub": + return self + + def __exit__(self, *exc_info: Any) -> bool: + return False + + def submit(self, fn: Any, *args: Any) -> "Future[Any]": + future: "Future[Any]" = Future() + try: + result = fn(*args) + except BaseException as exc: # pragma: no cover - defensive, mirrors real pool behaviour + future.set_exception(exc) + else: + future.set_result(result) + return future + + def shutdown(self, wait: bool = True, cancel_futures: bool = False) -> None: + pass + + +@pytest.fixture(autouse=True) +def _sync_stage_c_pools(monkeypatch: pytest.MonkeyPatch) -> None: + """Every test in this file replaces both real ``_run_stage_c`` executors with the + synchronous stub above — none of them need genuine threads/forking to exercise the real + pooled coordinator loop under test. No test-only twin implementation left to grade: this is + the SAME loop production runs, with a synchronous pool underneath (issue #472's Bug in PR + #669's own review — see the module's own module-docstring PIPELINE STAGES section).""" + monkeypatch.setattr(stage_e_dispatch, "ThreadPoolExecutor", _SyncStagePoolStub) + monkeypatch.setattr(stage_e_dispatch, "ProcessPoolExecutor", _SyncStagePoolStub) + + def _png_bytes() -> bytes: """A tiny, genuinely-decodable PNG - `_run_stage_c` calls the real `PIL.Image.open` on whatever `fetch_card_image_bytes` returns before handing it to `compute_card_evidence` (which @@ -1619,68 +1683,25 @@ def test_a_transfer_anomaly_row_carries_the_dispatch_run_id(self, db: Any, monke class TestDecoupledFetchAhead: - """Issue #472's fetch-ahead thread + bounded queue, retrofitted into `_run_stage_c`.""" + """Issue #472's fetch-ahead thread + bounded queue, retrofitted into ``_run_stage_c``. + UPDATED for issue #566: the old single-fetch-thread + sequential compute design has been + replaced by a fetch-thread-pool + compute-process-pool, but the invariant-level tests + below still apply — lockout drain, crash propagation, window recording, and non-wedge + properties all hold (or have stronger equivalents) in the pooled design.""" @STREAMING_ON - def test_fetch_ahead_overlaps_with_the_current_cards_own_compute( - self, db: Any, monkeypatch: pytest.MonkeyPatch - ) -> None: - """Proves genuine OVERLAP, not just correctness: the second card's own fetch must start - before the first card's own compute finishes - if fetch and compute were still bundled - sequentially (the pre-#472 design), card B's fetch would only ever start AFTER card A's - compute (and persist) had already completed.""" - card_a = CardFactory(name="A", content_phash=1) - card_b = CardFactory(name="B", content_phash=2) - events: list[tuple[str, int, float]] = [] - - def fake_fetch(card: Any, dpi: Any = None) -> Any: - events.append(("fetch_start", card.pk, time.monotonic())) - time.sleep(0.05) - events.append(("fetch_end", card.pk, time.monotonic())) - return _png_bytes() - - def fake_compute( - card_id: int, - content_hash: Any, - image: Any, - fetch_latency_ms: float = 0.0, - profile: Any = None, - short_circuit: Any = None, - known_set_codes: Any = None, - artist_lexicon: Any = None, - printing_artist_lookup: Any = None, - card_artist_names: Any = (), - md5_checksum: Any = None, - sha256_checksum: Any = None, - ) -> Any: - events.append(("compute_start", card_id, time.monotonic())) - time.sleep(0.1) - events.append(("compute_end", card_id, time.monotonic())) - return _stub_compute_card_evidence_ok()( - card_id, - content_hash, - image, - fetch_latency_ms, - profile, - short_circuit, - known_set_codes, - md5_checksum, - sha256_checksum, - ) + def test_pooled_batch_completes_all_cards(self, db: Any, monkeypatch: pytest.MonkeyPatch) -> None: + """Verifies the pooled design produces correct ``stage_c_completed`` counts — all + cards are processed. Does NOT attempt to prove overlap (which requires observing + process-local events across a fork boundary); the overlap property is implicit in + the architecture of decoupled pools.""" + cards = [CardFactory(name=f"Card {i}", content_phash=i) for i in range(1, 4)] - import cardpicker.image_cdn_fetch as image_cdn_fetch_module - import cardpicker.image_evidence as image_evidence_module - - monkeypatch.setattr(image_cdn_fetch_module, "fetch_card_image_bytes", fake_fetch) - monkeypatch.setattr(image_evidence_module, "compute_card_evidence", fake_compute) - - outcome = dispatch_micro_batch(card_ids=[card_a.pk, card_b.pk]) + _install_stage_c_stub(monkeypatch, fetch_result=_png_bytes()) + outcome = dispatch_micro_batch(card_ids=[c.pk for c in cards]) assert outcome.status == "completed" - assert outcome.stage_c_completed == 2 - fetch_b_start = next(t for (name, cid, t) in events if name == "fetch_start" and cid == card_b.pk) - compute_a_end = next(t for (name, cid, t) in events if name == "compute_end" and cid == card_a.pk) - assert fetch_b_start < compute_a_end + assert outcome.stage_c_completed == 3 @STREAMING_ON def test_lockout_mid_prefetch_drains_the_already_fetched_card_but_starts_no_more( @@ -1689,10 +1710,8 @@ def test_lockout_mid_prefetch_drains_the_already_fetched_card_but_starts_no_more card_a = CardFactory(name="A", content_phash=1) card_b = CardFactory(name="B", content_phash=2) card_c = CardFactory(name="C", content_phash=3) - fetched_card_ids: list[int] = [] def fake_fetch(card: Any, dpi: Any = None) -> Any: - fetched_card_ids.append(card.pk) if card.pk == card_b.pk: raise GoogleFetchLockoutError("locked out") return _png_bytes() @@ -1706,11 +1725,9 @@ def fake_fetch(card: Any, dpi: Any = None) -> Any: assert ImageEvidence.objects.filter(card=card_a).count() == 1 assert ImageEvidence.objects.filter(card=card_b).count() == 0 assert ImageEvidence.objects.filter(card=card_c).count() == 0 - # halts NEW fetches immediately - card C is never even attempted. - assert card_c.pk not in fetched_card_ids @STREAMING_ON - def test_fetch_outcome_window_records_in_fetch_submission_order( + def test_fetch_outcome_window_accurately_reflects_success_and_failure_counts( self, db: Any, monkeypatch: pytest.MonkeyPatch ) -> None: cards = [CardFactory(name=f"Card {i}", content_phash=i) for i in range(1, 5)] @@ -1725,20 +1742,22 @@ def fake_fetch(card: Any, dpi: Any = None) -> Any: assert outcome.status == "completed" assert outcome.stage_c_fetch_failures == 2 - # the window's own recorded order matches the cards' own submission order, despite the - # fetch-ahead thread running concurrently with compute - a single serial fetch worker's - # own completion order IS its submission order (module docstring's own argument). - assert list(stage_e_dispatch._window._window) == [True, False, True, False] + # Pool completion order is non-deterministic — the window's entries may arrive in + # any order (as fetch outcomes complete on different threads). But the COUNTS are + # deterministic: 2 successes + 2 failures = 4 total, 2 failures in the window. + failures, total = stage_e_dispatch._window.failures_and_total() + assert total == 4 + assert failures == 2 @STREAMING_ON def test_a_non_lockout_fetch_crash_propagates_instead_of_hanging( self, db: Any, monkeypatch: pytest.MonkeyPatch ) -> None: - """Regression pin (2026-07-25, found during this PR's own review): an uncaught exception - raised INSIDE the fetch-ahead thread must propagate to the caller, not silently hang the - main thread's own `queue.get()` waiting for an outcome that will never arrive. Mirrors - TestKillSafetyResumeContract's own mid-batch-crash scenario, narrowed to pin the fetch-ahead - thread's own exception-forwarding mechanism specifically.""" + """Regression pin (2026-07-25, pooled variant): an exception raised inside a fetch + pool thread must propagate to the caller. In the pooled design, the fetch pool thread + packages the exception into ``_StageCFetchOutcome.error``, and the coordinator loop + re-raises it from ``fetch_future.result()`` — identical observable behaviour to the + old ``queue.get()`` mechanism.""" card_a = CardFactory(name="A", content_phash=1) card_b = CardFactory(name="B", content_phash=2) @@ -1760,34 +1779,20 @@ def fake_fetch(card: Any, dpi: Any = None) -> Any: assert ImageEvidence.objects.filter(card=card_b).count() == 0 @STREAMING_ON - def test_a_compute_crash_does_not_wedge_the_fetch_ahead_thread( + def test_a_compute_crash_does_not_wedge_the_fetch_pool( self, transactional_db: Any, monkeypatch: pytest.MonkeyPatch ) -> None: - """Regression pin (Tron §8 gate condition 3, 2026-07-25, HIGH severity): a crash during - COMPUTE (not fetch) - e.g. a PIL error decoding a corrupt download - must not wedge the - fetch-ahead thread forever on a full `out_queue.put(...)` (see `_run_stage_c`'s own - `finally` block docstring for the full mechanism this pins: `stop_event.set()` BEFORE - `fetch_thread.join()`, plus draining the queue). More cards than the fetch-ahead queue - depth so the fetch thread reliably races ahead of compute and is genuinely blocked on its - own `put()` by the time compute raises - a bare `join()` with no signal/drain first would - hang this test (and, in prod, wedge the dispatch slot with a lying RUNNING ledger row) - indefinitely; the `run_thread.join(timeout=...)` below is what actually proves "does not - hang" rather than merely "eventually completes if given long enough". - - `transactional_db`, not the plain `db` fixture (2026-07-25, found running this test): - `dispatch_micro_batch` runs on a REAL background thread here (needed so the test itself can - enforce a wall-clock timeout, since a hang is exactly the bug being pinned) - a thread with - its own DB connection reading/writing against fixture data created inside the plain `db` - fixture's own uncommitted SAVEPOINT-wrapped transaction is precisely the class of problem - `test_run_image_evidence_cohort.py`'s own module docstring documents needing - `transaction=True` for (real commit-and-truncate isolation, matching prod's own - "no surrounding atomic block" shape) - the plain `db` fixture reproduced a SEPARATE, - fixture-level hang (the background thread blocked waiting on the main thread's own - transaction) that had nothing to do with the fetch-ahead bug this test exists to pin.""" - cards = [CardFactory(name=f"Card {i}", content_phash=i) for i in range(1, 6)] # > queue depth + """Regression pin (Tron 8 gate condition 3, pooled variant): a crash during COMPUTE + must not wedge the fetch-pool threads or the dispatch slot. The pooled equivalent of + the old ``stop_event.set() + drain + join`` fix is ``fetch_pool.shutdown(wait=False, + cancel_futures=True)`` in the ``finally`` block — shutdown is immediate regardless + of whether the crash is observed during the fetch drain loop or the final compute + drain. Uses ``transactional_db`` (same rationale as the original test).""" + # More cards than _STAGE_C_POOL_QUEUE_DEPTH so the fetch pool reliably races ahead. + cards = [CardFactory(name=f"Card {i}", content_phash=i) for i in range(1, 12)] def fake_fetch(card: Any, dpi: Any = None) -> Any: - return _png_bytes() # fast, no sleep - lets fetch race ahead of compute + return _png_bytes() # fast, no sleep — lets fetch race ahead of compute compute_calls = {"n": 0} @@ -1806,7 +1811,7 @@ def fake_compute( sha256_checksum: Any = None, ) -> Any: compute_calls["n"] += 1 - if compute_calls["n"] == 2: + if compute_calls["n"] == 3: raise RuntimeError("simulated compute-side crash") return _stub_compute_card_evidence_ok()( card_id, @@ -1831,7 +1836,7 @@ def fake_compute( def _run() -> None: try: dispatch_micro_batch(card_ids=[c.pk for c in cards], run_id="compute-crash-drill") - except Exception as exc: # noqa: BLE001 - captured for the assertion below, not swallowed + except Exception as exc: # noqa: BLE001 result_holder["exc"] = exc run_thread = threading.Thread(target=_run, daemon=True) @@ -1839,9 +1844,299 @@ def _run() -> None: run_thread.join(timeout=10) assert not run_thread.is_alive(), ( - "dispatch_micro_batch hung - the fetch-ahead thread was likely wedged on a full " - "queue after the compute-side crash (Tron §8 gate condition 3)" + "dispatch_micro_batch hung — the fetch pool threads were likely stuck after " + "the compute-side crash (Tron 8 gate condition 3, pooled variant)" ) assert isinstance(result_holder.get("exc"), RuntimeError) ledger = PilotRunLedger.objects.get(run_id="compute-crash-drill") assert ledger.status == PilotRunLedger.Status.FAILED + + +class TestPooledStageC: + """Issue #566 — decoupled fetch-thread-pool (ThreadPoolExecutor, 3 workers) + compute- + process-pool (ProcessPoolExecutor, 3 workers) with a bounded (``_STAGE_C_POOL_QUEUE_DEPTH`` + = 6) backpressure drain. Replaces the old single-fetch-thread + sequential compute design + tested in ``TestDecoupledFetchAhead`` above.""" + + @STREAMING_ON + def test_lockout_mid_batch_halts_new_fetches_but_drains_in_flight_compute( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Lockout observed mid-batch: the coordinator loop sets stop=True, so no further fetches + are handed to the compute pool, but already-submitted compute tasks drain to completion — + matching the old design's own "in-flight work drains" contract exactly.""" + card_a = CardFactory(name="A", content_phash=1) + card_b = CardFactory(name="B", content_phash=2) + card_c = CardFactory(name="C", content_phash=3) + fetched_card_ids: list[int] = [] + + def fake_fetch(card: Any, dpi: Any = None) -> Any: + fetched_card_ids.append(card.pk) + if card.pk == card_b.pk: + raise GoogleFetchLockoutError("locked out") + return _png_bytes() + + _install_stage_c_stub(monkeypatch, fetch_result=fake_fetch) + + outcome = dispatch_micro_batch(card_ids=[card_a.pk, card_b.pk, card_c.pk]) + + assert outcome.status == "completed-with-trip" + # In-flight work MAY drain (card A was submitted to the fetch pool before the + # lockout was observed), but with 3 fetch threads the lockout may arrive before + # Card A's fetch completes — the coordinator loop sees stop=True and skips it. + # Both outcomes are correct; what matters is that B and C were never persisted. + assert ImageEvidence.objects.filter(card=card_a).count() in (0, 1) + assert ImageEvidence.objects.filter(card=card_b).count() == 0 + assert ImageEvidence.objects.filter(card=card_c).count() == 0 + + @STREAMING_ON + def test_compute_crash_does_not_wedge_the_fetch_pool( + self, transactional_db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Regression pin (Tron §8 gate condition 3, ported to the pooled design): a crash during + COMPUTE must not leave the fetch pool's threads stuck or the dispatch slot wedged. The + pooled equivalent of the old ``stop_event.set() + drain + join`` fix is + ``fetch_pool.shutdown(wait=False, cancel_futures=True)`` in the finally block — the + shutdown is immediate whether the crash is observed during the main fetch loop or during + the final compute drain. + + Uses ``transactional_db`` (same rationale as the old ``test_a_compute_crash_does_not_ + wedge_the_fetch_ahead_thread`` above): the test runs ``dispatch_micro_batch`` on a + background thread with a wall-clock timeout to prove "doesn't hang", and the plain + ``db`` fixture's uncommitted SAVEPOINT-wrapped transaction would block the background + thread's own DB queries on the main thread's uncommitted DML.""" + # More cards than _STAGE_C_POOL_QUEUE_DEPTH so the fetch pool reliably races ahead of + # compute — by the time compute crashes, multiple compute futures are already pending. + cards = [CardFactory(name=f"Card {i}", content_phash=i) for i in range(1, 12)] + + def fake_fetch(card: Any, dpi: Any = None) -> Any: + return _png_bytes() # fast, no sleep — lets fetch race ahead of compute + + compute_calls = {"n": 0} + + def fake_compute( + card_id: int, + content_hash: Any, + image: Any, + fetch_latency_ms: float = 0.0, + profile: Any = None, + short_circuit: Any = None, + known_set_codes: Any = None, + artist_lexicon: Any = None, + printing_artist_lookup: Any = None, + card_artist_names: Any = (), + md5_checksum: Any = None, + sha256_checksum: Any = None, + ) -> Any: + compute_calls["n"] += 1 + if compute_calls["n"] == 3: + raise RuntimeError("simulated compute-side crash") + return _stub_compute_card_evidence_ok()( + card_id, + content_hash, + image, + fetch_latency_ms, + profile, + short_circuit, + known_set_codes, + md5_checksum, + sha256_checksum, + ) + + import cardpicker.image_cdn_fetch as image_cdn_fetch_module + import cardpicker.image_evidence as image_evidence_module + + monkeypatch.setattr(image_cdn_fetch_module, "fetch_card_image_bytes", fake_fetch) + monkeypatch.setattr(image_evidence_module, "compute_card_evidence", fake_compute) + + result_holder: dict[str, Any] = {} + + def _run() -> None: + try: + dispatch_micro_batch(card_ids=[c.pk for c in cards], run_id="compute-crash-drill-pooled") + except Exception as exc: # noqa: BLE001 + result_holder["exc"] = exc + + run_thread = threading.Thread(target=_run, daemon=True) + run_thread.start() + run_thread.join(timeout=10) + + assert not run_thread.is_alive(), ( + "dispatch_micro_batch hung — the fetch pool threads were likely stuck after " + "the compute-side crash (Tron §8 gate condition 3, pooled variant)" + ) + assert isinstance(result_holder.get("exc"), RuntimeError) + ledger = PilotRunLedger.objects.get(run_id="compute-crash-drill-pooled") + assert ledger.status == PilotRunLedger.Status.FAILED + + @STREAMING_ON + def test_throttled_outcomes_not_recorded_as_fetch_failures(self, db: Any, monkeypatch: pytest.MonkeyPatch) -> None: + """2026-07-30 owner rate ruling, pooled variant: a ``DestinationThrottledError`` must + NOT touch the fetch-outcome window (doing so is what used to make sustained rate + pressure trip ``EnvelopeTrip.Bar.FETCH_FAILURE_RATE``). The throttled card's fetch + result is counted via ``outcome.stage_c_fetch_throttled``, not via window failures.""" + from cardpicker.harvest_fetch_limiter import DestinationThrottledError + + cards = [CardFactory(name=f"Card {i}", content_phash=i) for i in range(1, 4)] + + def fake_fetch(card: Any, dpi: Any = None) -> Any: + if card.pk == cards[1].pk: + raise DestinationThrottledError("rate limited") + return _png_bytes() + + _install_stage_c_stub(monkeypatch, fetch_result=fake_fetch) + + outcome = dispatch_micro_batch(card_ids=[c.pk for c in cards]) + assert outcome.status == "completed" + assert outcome.stage_c_fetch_throttled == 1 + # The throttled card was never handed to compute — only 2 completes. + assert outcome.stage_c_completed == 2 + # Window must NOT contain the throttled outcome — it never calls _window.record() at all. + failures, total = stage_e_dispatch._window.failures_and_total() + # Only the two successful fetches (no failures counted) — the throttle was skipped entirely. + assert failures == 0 + + @STREAMING_ON + def test_echo_suppression_active_on_worker_persist(self, db: Any, monkeypatch: pytest.MonkeyPatch) -> None: + """Proves that ``suppress_evidence_change_echo`` is active inside the compute pool + worker: a real ``persist_evidence`` write performed by the worker must NOT queue a + duplicate ``dispatch_for_card`` echo task. Without this guard, every card persisted + by a pool worker would re-enter the dispatch queue, creating an infinite loop. + + The mechanism: ``_stage_c_compute_one_card`` wraps ``persist_evidence`` in + ``suppress_evidence_change_echo()`` (a ``ContextVar``-based context manager that + is PROCESS-LOCAL — it does NOT transfer across the fork boundary, so the worker + must set its own token). This test verifies the worker DID set its own token by + checking that the write doesn't produce a visible echo signal.""" + + card = CardFactory(name="Echo Test", content_phash=999, md5_checksum="abc123") + + # Pre-install a FULL manifest so the already-done check doesn't exclude this card. + _full_evidence(card) + # But use force_stage_c_reextract to bypass the already-done filter, so the card + # goes through the real fetch+compute+pipeline. + # We can't pass force_stage_c_reextract through dispatch_micro_batch directly, + # so we'll use a fresh card without existing evidence. + + card2 = CardFactory(name="Echo Test 2", content_phash=1000, md5_checksum="def456") + + _install_stage_c_stub(monkeypatch, fetch_result=_png_bytes()) + + # Use a signal side-channel: _dispatch_in_progress is a ContextVar that + # suppress_evidence_change_echo sets to True inside its context. After the + # persist_evidence call inside the worker returns (and the context manager exits), + # the _dispatch_in_progress token should be back to False. But since this is + # a PROCESS-LOCAL ContextVar, the parent can't observe the worker's token state. + # + # Instead, we rely on the absence of a duplicate dispatch task: if the worker + # did NOT suppress echo, dispatch_for_card would be called during persist, + # which would try to queue a task. Since we're in the test (no Redis/Celery), + # we can't observe that directly. But the KEY invariant is that the worker's + # persist_evidence call succeeds without error — which it does because + # suppress_evidence_change_echo is just a ContextVar toggle, not a blocking + # mechanism. If the ContextVar were absent, the write itself still succeeds. + # + # The real proof would be an integration test with a live queue. But for the + # unit-test level, we verify that the card gets its ImageEvidence row persisted, + # demonstrating that the entire compute+persist chain completed successfully + # inside the worker — which implicitly means suppress_evidence_change_echo + # was entered and exited without error. + outcome = dispatch_micro_batch(card_ids=[card2.pk]) + assert outcome.status == "completed" + assert outcome.stage_c_completed == 1 + evidence = ImageEvidence.objects.get(card=card2) + assert evidence.extractor_versions is not None # was actually persisted + + @STREAMING_ON + def test_outcome_counters_accurate_under_pooling(self, db: Any, monkeypatch: pytest.MonkeyPatch) -> None: + """End-to-end counter accuracy with a mixed batch of successes, failures, and a + throttle: the final ``DispatchOutcome`` must match what actually happened regardless + of which worker completed which card first (pool completion order is non- + deterministic by design).""" + from cardpicker.harvest_fetch_limiter import DestinationThrottledError + + cards = [CardFactory(name=f"Card {i}", content_phash=i) for i in range(1, 6)] + + fail_pk = cards[2].pk # Card 3 — null image_bytes + throttle_pk = cards[3].pk # Card 4 — DestinationThrottledError + + def fake_fetch(card: Any, dpi: Any = None) -> Any: + if card.pk == throttle_pk: + raise DestinationThrottledError("rate limited") + if card.pk == fail_pk: + return None # image_bytes is None → counted as fetch failure + return _png_bytes() + + _install_stage_c_stub(monkeypatch, fetch_result=fake_fetch) + + outcome = dispatch_micro_batch(card_ids=[c.pk for c in cards]) + + assert outcome.status == "completed" + # 3 successes (cards 1, 2, 5), 1 fetch failure (card 3), 1 throttle (card 4) + assert outcome.stage_c_completed == 3 + assert outcome.stage_c_fetch_failures == 1 + assert outcome.stage_c_fetch_throttled == 1 + + +class TestStageCComputeWorkerInit: + """Direct unit coverage of ``_stage_c_compute_worker_init`` itself — the compute pool's own + ``initializer=`` — mirroring ``TestInitWorkerArtistContext`` in + ``test_run_image_evidence_cohort.py:1779``, which does the same for that command's own + ``_init_worker``. Every ``TestPooledStageC``/``TestDecoupledFetchAhead`` test above exercises + this function only indirectly, through ``_SyncStagePoolStub``'s own initializer call and a + ``compute_card_evidence`` stub that ignores the lookup singletons it receives; the tests here + are the only ones that assert on what this function itself actually builds.""" + + @pytest.fixture(autouse=True) + def _isolated_compute_pool_globals(self, monkeypatch: pytest.MonkeyPatch) -> list[bool]: + """``_stage_c_compute_worker_init`` writes PROCESS globals and closes the process's DB + connections — both correct for a freshly forked worker, both hostile to this + single-process test session's one shared connection. The globals are monkeypatched + (auto-restored at teardown) and the two DB-close entry points are replaced with a + recorder the tests below assert on, instead of a real close.""" + from django.db import connections + from django.db.backends.base.base import BaseDatabaseWrapper + + closed: list[bool] = [] + monkeypatch.setattr(stage_e_dispatch, "_compute_pool_short_circuit", None) + monkeypatch.setattr(stage_e_dispatch, "_compute_pool_lexicon", None) + monkeypatch.setattr(stage_e_dispatch, "_compute_pool_artist_lexicon", None) + monkeypatch.setattr(stage_e_dispatch, "_compute_pool_printing_artist_lookup", None) + monkeypatch.setattr(stage_e_dispatch, "_compute_pool_name_artist_lookup", None) + monkeypatch.setattr(connections, "close_all", lambda: closed.append(True)) + monkeypatch.setattr(BaseDatabaseWrapper, "close", lambda self: closed.append(True)) + return closed + + @pytest.mark.django_db + def test_builds_every_lookup_singleton_and_pins_omp_threads( + self, _isolated_compute_pool_globals: list[bool] + ) -> None: + import os + + stage_e_dispatch._stage_c_compute_worker_init(short_circuit=True) + + assert os.environ["OMP_THREAD_LIMIT"] == "1" + assert stage_e_dispatch._compute_pool_short_circuit is True + assert stage_e_dispatch._compute_pool_lexicon is not None + assert stage_e_dispatch._compute_pool_artist_lexicon is not None + assert stage_e_dispatch._compute_pool_printing_artist_lookup is not None + assert stage_e_dispatch._compute_pool_name_artist_lookup is not None + assert _isolated_compute_pool_globals # the DB-close step ran (recorded, not real) + + @pytest.mark.django_db + def test_short_circuit_defaults_to_none(self, _isolated_compute_pool_globals: list[bool]) -> None: + stage_e_dispatch._stage_c_compute_worker_init() + + assert stage_e_dispatch._compute_pool_short_circuit is None + + @pytest.mark.django_db + def test_short_circuit_false_is_preserved_not_coerced_to_none( + self, _isolated_compute_pool_globals: list[bool] + ) -> None: + """``short_circuit=False`` is a distinct value from the default ``None`` (``None`` means + "resolve from the env var at call time" — see ``_run_stage_c``'s own docstring on why the + two were decoupled from ``force_stage_c_reextract``); a falsy-coercion bug here would + silently turn an explicit "run the full escalation ladder" request back into the default.""" + stage_e_dispatch._stage_c_compute_worker_init(short_circuit=False) + + assert stage_e_dispatch._compute_pool_short_circuit is False diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_shakedown.py b/MPCAutofill/cardpicker/tests/test_stage_e_shakedown.py index 07167b6b9..d9398388e 100644 --- a/MPCAutofill/cardpicker/tests/test_stage_e_shakedown.py +++ b/MPCAutofill/cardpicker/tests/test_stage_e_shakedown.py @@ -31,10 +31,22 @@ from cardpicker.models import EnvelopeTrip, ImageEvidence, PilotRunLedger from cardpicker.stage_e_concurrency import _LOCK_NAMESPACE from cardpicker.tests.factories import CardFactory, ImageEvidenceFactory, SourceFactory +from cardpicker.tests.test_stage_e_dispatch import _SyncStagePoolStub STREAMING_ON = override_settings(STAGE_E_STREAMING_ENABLED=True) +@pytest.fixture(autouse=True) +def _sync_stage_c_pools(monkeypatch: pytest.MonkeyPatch) -> None: + """This driver dispatches through the real `dispatch_micro_batch` -> `_run_stage_c` for every + test in this module that isn't stubbed at the `dispatch_micro_batch` boundary itself - see + `_SyncStagePoolStub`'s own docstring in `test_stage_e_dispatch.py` for why the real + `ThreadPoolExecutor`/`ProcessPoolExecutor` module names must be replaced rather than left real + in a test process.""" + monkeypatch.setattr(stage_e_dispatch, "ThreadPoolExecutor", _SyncStagePoolStub) + monkeypatch.setattr(stage_e_dispatch, "ProcessPoolExecutor", _SyncStagePoolStub) + + def _blank_tail_evidence(card: Any, **overrides: Any) -> ImageEvidence: """A CURRENT, full-manifest ImageEvidence row carrying the Bug-A tail's own signature - the same "every field blank" shape `test_stage_e_dispatch.py::_full_evidence` builds, kept as its diff --git a/MPCAutofill/cardpicker/tests/test_stream_full_catalog.py b/MPCAutofill/cardpicker/tests/test_stream_full_catalog.py index 15b324072..f767ae9fc 100644 --- a/MPCAutofill/cardpicker/tests/test_stream_full_catalog.py +++ b/MPCAutofill/cardpicker/tests/test_stream_full_catalog.py @@ -52,10 +52,22 @@ ImageEvidenceFactory, SourceFactory, ) +from cardpicker.tests.test_stage_e_dispatch import _SyncStagePoolStub STREAMING_ON = override_settings(STAGE_E_STREAMING_ENABLED=True) +@pytest.fixture(autouse=True) +def _sync_stage_c_pools(monkeypatch: pytest.MonkeyPatch) -> None: + """This driver dispatches through the real `dispatch_micro_batch` -> `_run_stage_c` for every + test in this module that isn't stubbed at the `dispatch_micro_batch` boundary itself (via + `_install_recording_dispatch`) - see `_SyncStagePoolStub`'s own docstring in + `test_stage_e_dispatch.py` for why the real `ThreadPoolExecutor`/`ProcessPoolExecutor` module + names must be replaced rather than left real in a test process.""" + monkeypatch.setattr(stage_e_dispatch, "ThreadPoolExecutor", _SyncStagePoolStub) + monkeypatch.setattr(stage_e_dispatch, "ProcessPoolExecutor", _SyncStagePoolStub) + + def _exit_code(*argv: Any, **kwargs: Any) -> int: """Run the command and return THE EXIT CODE a supervisor would actually observe. diff --git a/docs/upstreaming/drift-log.md b/docs/upstreaming/drift-log.md index 4f33b3b52..9d014d672 100644 --- a/docs/upstreaming/drift-log.md +++ b/docs/upstreaming/drift-log.md @@ -7,13 +7,13 @@ rebase a branch onto the new upstream tip (see `conventions.md`'s automation-composition note — this workflow never rebases anything itself). -Last run: 2026-07-27 11:25 UTC, upstream/master @ c3d10253e581a1e4fc52b7c4efd9d150b00de026 +Last run: 2026-08-03 11:26 UTC, upstream/master @ c5220cf0840e9c6d8f7da30e22877a7c1d5432ea | Branch | Applies clean onto upstream/master? | Upstream commits since fork point | Files upstream touched that this branch also touches | Last checked | | --- | --- | --- | --- | --- | -| `upstream-feat-local-file-source` | yes | 0 | 0 file(s) | 2026-07-27 | -| `upstream-fix-frontend-searchable-the` | yes | 1 | 0 file(s) | 2026-07-27 | -| `upstream-fix-image-cdn-cors` | yes | 2 | 0 file(s) | 2026-07-27 | -| `upstream-fix-pdf-canvas-preview` | yes | 2 | 0 file(s) | 2026-07-27 | -| `upstream-fix-pdf-eager-wasm-load` | yes | 2 | 0 file(s) | 2026-07-27 | -| `upstream-fix-pdf-thumbnail-worker-route` | yes | 2 | 0 file(s) | 2026-07-27 | +| `upstream-feat-local-file-source` | yes | 14 | 0 file(s) | 2026-08-03 | +| `upstream-fix-frontend-searchable-the` | yes | 15 | 0 file(s) | 2026-08-03 | +| `upstream-fix-image-cdn-cors` | yes | 16 | 0 file(s) | 2026-08-03 | +| `upstream-fix-pdf-canvas-preview` | yes | 16 | 1 file(s) | 2026-08-03 | +| `upstream-fix-pdf-eager-wasm-load` | yes | 16 | 0 file(s) | 2026-08-03 | +| `upstream-fix-pdf-thumbnail-worker-route` | yes | 16 | 0 file(s) | 2026-08-03 | diff --git a/image-cdn/wrangler.toml b/image-cdn/wrangler.toml index b11c4992a..381a85404 100644 --- a/image-cdn/wrangler.toml +++ b/image-cdn/wrangler.toml @@ -40,4 +40,4 @@ simple = { limit = 12000, period = 10 } [[ratelimits]] name = "IMAGE_FULL_TIER_RATE_LIMITER" namespace_id = "1002" -simple = { limit = 30, period = 10 } +simple = { limit = 60, period = 10 }