From b633083c480d53d7721147add4e20b65e33c767f Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Fri, 24 Jul 2026 22:07:59 +0000 Subject: [PATCH 1/3] Fix Stage E concurrent-dispatch vote-collision IntegrityError Two concurrent dispatch_micro_batch invocations (django-q2's 8 workers, or the backstop sweep racing an event trigger) could both pass Stage D's per-identity eligibility check before either committed, then race to bulk_create the same (card, anonymous_id) CardPrintingTag - the loser hit IntegrityError and aborted its whole micro-batch (trip envtrip-20260724T214616-be6e5db9, failed run_ids stage-e-stream-20260724T2144*). Adds a pre-write skip-if-exists guard (_split_new_printing_tag_votes, mirroring PR #411's precedent) to run_join_key_calculator and run_fallback_calculator - skip-and-count, not retract-and-recast, since a concurrent race yields the same verdict from the same evidence, not a genuine conclusion change. run_slow_path_calculator needs no equivalent guard (CardScanLog carries no DB uniqueness constraint). Corrects stage-e-operations.md's overstated "eligibility exclude alone is idempotent" claim for the concurrent (not just sequential) case. Co-Authored-By: Claude Fable 5 --- .../cardpicker/local_calculate_verdicts.py | 113 +++++++++++++- .../commands/local_calculate_verdicts.py | 7 + MPCAutofill/cardpicker/stage_e_dispatch.py | 25 +++ .../tests/test_local_calculate_verdicts.py | 147 ++++++++++++++++++ .../cardpicker/tests/test_stage_e_dispatch.py | 49 ++++++ docs/features/stage-e-operations.md | 65 ++++++-- 6 files changed, 384 insertions(+), 22 deletions(-) diff --git a/MPCAutofill/cardpicker/local_calculate_verdicts.py b/MPCAutofill/cardpicker/local_calculate_verdicts.py index 841dd9d07..5c9d9ea98 100644 --- a/MPCAutofill/cardpicker/local_calculate_verdicts.py +++ b/MPCAutofill/cardpicker/local_calculate_verdicts.py @@ -845,6 +845,12 @@ class JoinKeyCalculatorResult: no_match_votes_would_cast: int = 0 votes_written: int = 0 no_match_votes_written: int = 0 + # Votes this run computed but did NOT write because a (card, anonymous_id) vote already + # existed under THIS SAME identity at write time (2026-07-24, trip + # envtrip-20260724T214616-be6e5db9 - see _split_new_printing_tag_votes' own docstring for the + # full incident/rationale). Counted, not silently dropped, matching PR #411's own + # already_voted convention in local_lands_identify.py. + already_voted: int = 0 skip_counts: dict[str, int] = field(default_factory=dict) # capped audit sample, mirroring PilotResult/FrameMismatchRecoveryResult's own "up to N, for # the report" convention elsewhere in this codebase. @@ -957,6 +963,76 @@ def _eligible_cards_queryset( return queryset +def _split_new_printing_tag_votes( + votes_batch: list[CardPrintingTag], +) -> tuple[list[CardPrintingTag], int]: + """ + CONCURRENT-DISPATCH VOTE COLLISION GUARD (2026-07-24, trip envtrip-20260724T214616-be6e5db9): + partitions `votes_batch` into (votes safe to `bulk_create`, count already voted) - a pre-write + skip-if-exists check against every (card, anonymous_id) pair a vote in this batch would write, + mirroring `local_lands_identify._split_new_votes`' own precedent (issue #408, PR #411) exactly + - `docs/proposals/stage-e-streaming.md` SS2 item 2 ("Vote-collision skip-if-exists guard") cites + that PR as the proven primitive Stage E's own per-card write path needs, but Phase 2's + `card_ids` scoping (this module, 2026-07-24) never actually wired it into this module's own + three calculators when it landed - `_eligible_cards_queryset`'s `.exclude(printing_tags__ + anonymous_id=anonymous_id)` was trusted alone as the idempotence mechanism, which holds for a + single sequential invocation but not for two CONCURRENT ones. + + ROOT CAUSE of the shakedown failure this guard fixes: `stage_e_dispatch.dispatch_micro_batch` + is invoked from django-q2 async tasks (`Q_CLUSTER["workers"] = 8`, MPCAutofill/settings.py) and + from the cron backstop sweep - two dispatches scoped to an overlapping card_ids set (e.g. an + evidence-change signal re-firing while the backstop sweep's own backlog fill has already picked + up the same card) can both read `_eligible_cards_queryset` BEFORE either has committed its own + vote, both compute a verdict for the same card from the same current `ImageEvidence` row, and + both then attempt to `bulk_create` a `CardPrintingTag` for it - the second `bulk_create` in the + race hits `cardprintingtag_unique_no_match_vote` (is_no_match=True) or + `cardprintingtag_unique_printing_vote` (is_no_match=False), aborting that WHOLE micro-batch's + write with an `IntegrityError`, per the four failed `PilotRunLedger` rows + (run_ids stage-e-stream-20260724T2144*) the shakedown sweep observed. BULK mode never hit this: + every existing `local_calculate_verdicts` invocation runs sequentially, one process, one + identity's eligibility exclude computed and consumed within a single query's lifetime - this + race is specific to Stage E's own concurrent PASSIVE-mode dispatch, the first caller ever to + make more than one of these calculators' write paths overlap in time. + + SKIP-AND-COUNT, NOT RETRACT-AND-RECAST: deliberately the OPPOSITE choice from + `reparse_collector_evidence.reparse_and_retract`'s own retract-then-recast pattern, because the + two scenarios are not the same shape. `reparse_collector_evidence` retracts a vote whose + CONCLUSION has genuinely changed - a parser fix, a lexicon-gate correction, freshly re-extracted + evidence - so the old vote is objectively stale and superseding it is correct. A concurrent + race here produces no such change: both racing invocations read the SAME current `ImageEvidence` + row under the SAME code version and necessarily compute the SAME verdict (deterministic pure + functions - `calculate_join_key_verdict`/`calculate_fallback_verdict` take no non-reproducible + input). Retracting the winner to let the loser recast would be pure churn - it would even risk + flip-flopping `resolve_and_persist_printing`'s own per-touch consensus recompute if a third + dispatch raced in between the retract and the recast - with no correctness benefit at all, since + there is nothing to correct. The loser's own vote is simply redundant, exactly the shape PR + #411's `_split_new_votes` already established a skip-and-count answer for. + + One batched existence query (not one query per card), scoped to just the card_ids/anonymous_ids + actually present in this batch - same "wasteful full-table scan" avoidance + `_split_new_votes`'s own docstring already cites. Checks (card_id, anonymous_id) only (not the + full (card, printing, anonymous_id) triple `_split_new_votes` checks) because this module's own + established invariant is stricter: `_eligible_cards_queryset`'s exclude already enforces AT MOST + ONE vote per (card, anonymous_id) regardless of match/no-match outcome (see that function's own + docstring's "Idempotence... comes entirely from the stable, per-calculator anonymous_id + exclusion" - matching CardPrintingTag's own two partial-unique constraints, which together + enforce exactly the same "one vote per identity" invariant, just via two different partial + indexes depending on is_no_match). + """ + if not votes_batch: + return [], 0 + + card_ids = {vote.card_id for vote in votes_batch} + anonymous_ids = {vote.anonymous_id for vote in votes_batch} + already_voted_pairs = set( + CardPrintingTag.objects.filter(card_id__in=card_ids, anonymous_id__in=anonymous_ids).values_list( + "card_id", "anonymous_id" + ) + ) + new_votes = [vote for vote in votes_batch if (vote.card_id, vote.anonymous_id) not in already_voted_pairs] + return new_votes, len(votes_batch) - len(new_votes) + + def run_join_key_calculator( run_id: Optional[str] = None, dry_run: bool = True, @@ -1049,13 +1125,18 @@ def run_join_key_calculator( touched_card_ids.append(card.pk) if not dry_run: - CardPrintingTag.objects.bulk_create(votes_batch) + # Pre-write skip-if-exists guard (2026-07-24) - see _split_new_printing_tag_votes' own + # docstring for the concurrent-dispatch collision this closes and why skip-and-count (not + # retract-and-recast) is the honest semantic here. + new_votes, result.already_voted = _split_new_printing_tag_votes(votes_batch) + if new_votes: + CardPrintingTag.objects.bulk_create(new_votes) CardScanLog.objects.bulk_create(scan_log_batch) for touched_card in Card.objects.filter(pk__in=touched_card_ids): resolve_and_persist_printing(touched_card) - result.votes_written = sum(1 for v in votes_batch if not v.is_no_match) - result.no_match_votes_written = sum(1 for v in votes_batch if v.is_no_match) + result.votes_written = sum(1 for v in new_votes if not v.is_no_match) + result.no_match_votes_written = sum(1 for v in new_votes if v.is_no_match) return result @@ -1231,6 +1312,11 @@ class FallbackCalculatorResult: cards_considered: int = 0 votes_would_cast: int = 0 votes_written: int = 0 + # Same "computed but not written, counted not dropped" convention as + # JoinKeyCalculatorResult.already_voted - see that field's own docstring and + # _split_new_printing_tag_votes for the shared collision this calculator is equally exposed + # to (it writes CardPrintingTag rows too, under its own STAGE_D_FALLBACK_ANONYMOUS_ID). + already_voted: int = 0 skip_counts: dict[str, int] = field(default_factory=dict) # capped audit sample, mirroring JoinKeyCalculatorResult.audit's own convention. audit: list[dict[str, object]] = field(default_factory=list) @@ -1354,12 +1440,18 @@ def run_fallback_calculator( touched_card_ids.append(card.pk) if not dry_run: - CardPrintingTag.objects.bulk_create(votes_batch) + # Same pre-write skip-if-exists guard as run_join_key_calculator - see + # _split_new_printing_tag_votes' own docstring (this calculator writes CardPrintingTag + # under its own STAGE_D_FALLBACK_ANONYMOUS_ID and is equally exposed to the concurrent- + # dispatch collision that fixes). + new_votes, result.already_voted = _split_new_printing_tag_votes(votes_batch) + if new_votes: + CardPrintingTag.objects.bulk_create(new_votes) CardScanLog.objects.bulk_create(scan_log_batch) for touched_card in Card.objects.filter(pk__in=touched_card_ids): resolve_and_persist_printing(touched_card) - result.votes_written = len(votes_batch) + result.votes_written = len(new_votes) return result @@ -1553,6 +1645,17 @@ def run_slow_path_calculator( matching `run_join_key_calculator`'s own convention) computes and counts everything without writing. `card_ids` (2026-07-24, Stage E Phase 2) is forwarded straight through to `_slow_path_eligible_cards_queryset` - see `_eligible_cards_queryset`'s own docstring. + + CONCURRENT-DISPATCH COLLISION - CHECKED, NOT VULNERABLE (2026-07-24, same investigation as + `_split_new_printing_tag_votes`): this calculator writes only `CardScanLog` rows, never a + `CardPrintingTag` - and `CardScanLog` carries no DB-level uniqueness constraint at all (see + that model's own docstring: "not deduplicated away... the scan_log table itself is an + append-only audit trail"; `review_clusters._review_queue_card_ids` already `.distinct()`s its + own read for exactly this reason). A concurrent race here can at most write two routing-marker + rows for the same card under two different run_ids - a harmless, already-tolerated audit-trail + duplicate, not an `IntegrityError` - so no skip-if-exists guard was added here; adding one + would fight this model's own by-design "append, never dedupe" contract for no correctness + gain. """ run_id = run_id or generate_run_id() result = SlowPathCalculatorResult(dry_run=dry_run, run_id=run_id) diff --git a/MPCAutofill/cardpicker/management/commands/local_calculate_verdicts.py b/MPCAutofill/cardpicker/management/commands/local_calculate_verdicts.py index a7658b9a4..b7c0b2de4 100644 --- a/MPCAutofill/cardpicker/management/commands/local_calculate_verdicts.py +++ b/MPCAutofill/cardpicker/management/commands/local_calculate_verdicts.py @@ -123,6 +123,12 @@ def handle(self, *args: Any, **kwargs: Any) -> None: f"[join-key] considered={result.cards_considered} " f"votes={'written=' + str(result.votes_written) if not dry_run else 'would_cast=' + str(result.votes_would_cast)} " f"no_match_votes={'written=' + str(result.no_match_votes_written) if not dry_run else 'would_cast=' + str(result.no_match_votes_would_cast)} " + # already_voted (2026-07-24, the Stage E concurrent-dispatch collision guard - + # see local_calculate_verdicts._split_new_printing_tag_votes' own docstring): + # always 0 for this sequential BULK-mode command in practice, printed anyway for + # the same observability parity local_lands_identify's own command established + # (PR #411) - a nonzero value here would itself be a signal worth investigating. + f"already_voted={result.already_voted} " f"skip_counts={dict(result.skip_counts)}" ) for entry in result.audit[:10]: @@ -161,6 +167,7 @@ def handle(self, *args: Any, **kwargs: Any) -> None: print( f"[fallback] considered={fallback_result.cards_considered} " f"votes={'written=' + str(fallback_result.votes_written) if not dry_run else 'would_cast=' + str(fallback_result.votes_would_cast)} " + f"already_voted={fallback_result.already_voted} " f"skip_counts={dict(fallback_result.skip_counts)}" ) for entry in fallback_result.audit[:10]: diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index 804e5155c..d95539190 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -175,6 +175,14 @@ class DispatchOutcome: stage_d_join_key_votes: int = 0 stage_d_fallback_votes: int = 0 stage_d_slow_path_routed: int = 0 + # A concurrent overlapping dispatch (this worker racing another django-q worker, or the + # backstop sweep racing an event trigger - local_calculate_verdicts._split_new_printing_tag_ + # votes' own docstring has the full incident) skipped a vote this batch computed because + # another dispatch had already cast it for the same (card, anonymous_id) first. Counted, not + # silently dropped - a healthy streaming deployment should see this occasionally, not never + # (zero forever would suggest the guard itself is dead code, not that races don't happen). + stage_d_join_key_already_voted: int = 0 + stage_d_fallback_already_voted: int = 0 trip_id: Optional[str] = None @@ -326,12 +334,27 @@ def _run_stage_d(batch_ids: list[int], run_id: str, outcome: DispatchOutcome) -> evidence and never needed Stage C at all this dispatch) - each calculator's own eligibility query simply finds nothing to do for a card with no current evidence (a "no-evidence" named skip, not an error), so this is always safe to call. + + CONCURRENT-DISPATCH VOTE COLLISION (2026-07-24, trip envtrip-20260724T214616-be6e5db9): this + is the FIRST caller ever to invoke `run_join_key_calculator`/`run_fallback_calculator` + concurrently (django-q2 runs `Q_CLUSTER["workers"] = 8`, and the cron backstop sweep can + overlap an event-driven dispatch too) - two dispatches scoped to the same card can both pass + that calculator's own eligibility check before either commits, race to `bulk_create` the same + (card, anonymous_id) vote, and the loser used to hit an `IntegrityError` that aborted its + WHOLE micro-batch. Both calculators now carry their own pre-write skip-if-exists guard + (`local_calculate_verdicts._split_new_printing_tag_votes`) - a losing race is now a counted + no-op (`already_voted`, surfaced on `DispatchOutcome`/this batch's own `PilotRunLedger` row), + never a crash. `run_slow_path_calculator` was checked too and needs no equivalent guard - it + writes only `CardScanLog` rows, which carry no DB uniqueness constraint at all (see that + calculator's own docstring). """ join_key_result = run_join_key_calculator(run_id=run_id, dry_run=False, card_ids=batch_ids) outcome.stage_d_join_key_votes = join_key_result.votes_written + join_key_result.no_match_votes_written + outcome.stage_d_join_key_already_voted = join_key_result.already_voted fallback_result = run_fallback_calculator(run_id=run_id, dry_run=False, card_ids=batch_ids) outcome.stage_d_fallback_votes = fallback_result.votes_written + outcome.stage_d_fallback_already_voted = fallback_result.already_voted slow_path_result = run_slow_path_calculator(run_id=run_id, dry_run=False, card_ids=batch_ids) outcome.stage_d_slow_path_routed = slow_path_result.routed_written @@ -431,7 +454,9 @@ def dispatch_micro_batch( "stage_c_completed": outcome.stage_c_completed, "stage_c_fetch_failures": outcome.stage_c_fetch_failures, "stage_d_join_key_votes": outcome.stage_d_join_key_votes, + "stage_d_join_key_already_voted": outcome.stage_d_join_key_already_voted, "stage_d_fallback_votes": outcome.stage_d_fallback_votes, + "stage_d_fallback_already_voted": outcome.stage_d_fallback_already_voted, "stage_d_slow_path_routed": outcome.stage_d_slow_path_routed, "peak_rss_mb": peak_rss_mb, "lockout_trip_id": lockout_trip.trip_id if lockout_trip is not None else None, diff --git a/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py b/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py index 8571d3939..77d5bcd5e 100644 --- a/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py +++ b/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py @@ -47,6 +47,7 @@ _eligible_cards_queryset, _filter_by_symbol_phash, _resolve_candidates_for_card, + _split_new_printing_tag_votes, _symbol_phash_tiebreak, calculate_fallback_verdict, calculate_join_key_verdict, @@ -486,6 +487,95 @@ def test_picks_the_unique_close_match(self): assert winner is not None and winner.pk == 1 +class TestSplitNewPrintingTagVotes: + """Direct unit coverage for the 2026-07-24 concurrent-dispatch collision guard + (trip envtrip-20260724T214616-be6e5db9), independent of either calculator's full + orchestration - mirrors test_local_lands_identify.py's own TestSplitNewVotes structure for + the sibling PR #411 guard.""" + + def test_empty_batch_returns_empty(self, db): + assert _split_new_printing_tag_votes([]) == ([], 0) + + def test_no_pre_existing_vote_keeps_everything(self, db): + card = CardFactory(name="Some Card") + vote = CardPrintingTag(card_id=card.pk, printing_id=None, is_no_match=True, anonymous_id=JOIN_KEY_ANONYMOUS_ID) + + new_votes, already_voted = _split_new_printing_tag_votes([vote]) + + assert new_votes == [vote] + assert already_voted == 0 + + def test_an_existing_no_match_vote_for_the_same_identity_is_skipped(self, db): + card = CardFactory(name="Some Card") + CardPrintingTag.objects.create( + card=card, printing=None, is_no_match=True, anonymous_id=JOIN_KEY_ANONYMOUS_ID, source=VoteSource.OCR + ) + vote = CardPrintingTag(card_id=card.pk, printing_id=None, is_no_match=True, anonymous_id=JOIN_KEY_ANONYMOUS_ID) + + new_votes, already_voted = _split_new_printing_tag_votes([vote]) + + assert new_votes == [] + assert already_voted == 1 + + def test_an_existing_match_vote_for_the_same_identity_is_skipped_even_with_a_different_printing(self, db): + """The invariant this guard enforces is 'at most one vote per (card, anonymous_id)', + matching _eligible_cards_queryset's own exclude - not 'at most one vote per (card, + printing, anonymous_id)', which is all the DB's own cardprintingtag_unique_printing_vote + constraint alone would check.""" + card = CardFactory(name="Some Card") + printing_a = CanonicalCardFactory(name="Some Card") + printing_b = CanonicalCardFactory(name="Some Card") + CardPrintingTag.objects.create( + card=card, printing=printing_a, is_no_match=False, anonymous_id=JOIN_KEY_ANONYMOUS_ID, source=VoteSource.OCR + ) + vote = CardPrintingTag( + card_id=card.pk, printing_id=printing_b.pk, is_no_match=False, anonymous_id=JOIN_KEY_ANONYMOUS_ID + ) + + new_votes, already_voted = _split_new_printing_tag_votes([vote]) + + assert new_votes == [] + assert already_voted == 1 + + def test_an_existing_vote_under_a_different_identity_is_not_a_collision(self, db): + card = CardFactory(name="Some Card") + CardPrintingTag.objects.create( + card=card, + printing=None, + is_no_match=True, + anonymous_id=STAGE_D_FALLBACK_ANONYMOUS_ID, + source=VoteSource.OCR, + ) + vote = CardPrintingTag(card_id=card.pk, printing_id=None, is_no_match=True, anonymous_id=JOIN_KEY_ANONYMOUS_ID) + + new_votes, already_voted = _split_new_printing_tag_votes([vote]) + + assert new_votes == [vote] + assert already_voted == 0 + + def test_mixed_batch_skips_only_the_colliding_vote(self, db): + collided_card = CardFactory(name="Some Card") + clean_card = CardFactory(name="Some Card") + CardPrintingTag.objects.create( + card=collided_card, + printing=None, + is_no_match=True, + anonymous_id=JOIN_KEY_ANONYMOUS_ID, + source=VoteSource.OCR, + ) + colliding_vote = CardPrintingTag( + card_id=collided_card.pk, printing_id=None, is_no_match=True, anonymous_id=JOIN_KEY_ANONYMOUS_ID + ) + clean_vote = CardPrintingTag( + card_id=clean_card.pk, printing_id=None, is_no_match=True, anonymous_id=JOIN_KEY_ANONYMOUS_ID + ) + + new_votes, already_voted = _split_new_printing_tag_votes([colliding_vote, clean_vote]) + + assert new_votes == [clean_vote] + assert already_voted == 1 + + class TestRunJoinKeyCalculator: def test_dry_run_counts_without_writing(self, db): card = CardFactory(name="Some Card", content_phash=42) @@ -607,6 +697,36 @@ def test_back_face_card_resolves_via_the_combined_scryfall_name(self, db, tmp_pa vote = CardPrintingTag.objects.get(card=card) assert vote.printing_id == printing.pk + def test_concurrent_dispatch_collision_is_skipped_not_crashed(self, db, monkeypatch): + """Regression for the Stage E Phase 2 shakedown's first live trip + (envtrip-20260724T214616-be6e5db9, failed run_ids stage-e-stream-20260724T2144*) - see + _split_new_printing_tag_votes' own docstring for the full root-cause writeup. Reproduces + the exact TOCTOU a concurrent streamed re-entry hits: this card was genuinely eligible + when ITS OWN eligibility read ran (no vote existed yet), but another, concurrent + dispatch's own write landed before this invocation reached its own bulk_create - + simulated here by monkeypatching the eligibility read to stay stale (representing the + already-consumed queryset a real concurrent caller would have) while a colliding vote is + seeded directly, reproducing the literal production key + (JOIN_KEY_ANONYMOUS_ID, is_no_match=True).""" + import cardpicker.local_calculate_verdicts as module + + card = CardFactory(name="Some Card", content_phash=42) + CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + _evidence(card, collector_line_collector_number="999") # no candidate match -> is_no_match=True + monkeypatch.setattr(module, "_eligible_cards_queryset", lambda *args, **kwargs: Card.objects.filter(pk=card.pk)) + + # the WINNER of the race: a vote already landed for this exact (card, anonymous_id) pair. + CardPrintingTag.objects.create( + card=card, printing=None, is_no_match=True, anonymous_id=JOIN_KEY_ANONYMOUS_ID, source=VoteSource.OCR + ) + + result = run_join_key_calculator(dry_run=False) # must not raise IntegrityError + + assert result.already_voted == 1 + assert result.votes_written == 0 + assert result.no_match_votes_written == 0 + assert CardPrintingTag.objects.filter(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID).count() == 1 + class TestEligibleCardsQueryset: """`_eligible_cards_queryset`'s two knowledge-inventory excludes (module docstring's @@ -1548,6 +1668,33 @@ def test_evidence_from_a_stale_content_hash_is_not_used(self, db): assert result.cards_considered == 0 assert result.skip_counts.get(FALLBACK_NO_EVIDENCE_SKIP_REASON) == 1 + def test_concurrent_dispatch_collision_is_skipped_not_crashed(self, db, monkeypatch): + """Same shakedown regression as TestRunJoinKeyCalculator's own version of this test - this + calculator writes CardPrintingTag under its own STAGE_D_FALLBACK_ANONYMOUS_ID identity too + and is equally exposed to the concurrent-dispatch collision + _split_new_printing_tag_votes' own docstring writes up.""" + import cardpicker.local_calculate_verdicts as module + + printing = CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + CanonicalPrintingMetadataFactory(canonical_card=printing, border_color="black") + card, _ = self._no_hit_card(layout_class="black") + monkeypatch.setattr(module, "_eligible_cards_queryset", lambda *args, **kwargs: Card.objects.filter(pk=card.pk)) + + # the WINNER of the race: a vote already landed for this exact (card, anonymous_id) pair. + CardPrintingTag.objects.create( + card=card, + printing=printing, + is_no_match=False, + anonymous_id=STAGE_D_FALLBACK_ANONYMOUS_ID, + source=VoteSource.OCR, + ) + + result = run_fallback_calculator(dry_run=False) # must not raise IntegrityError + + assert result.already_voted == 1 + assert result.votes_written == 0 + assert CardPrintingTag.objects.filter(card=card, anonymous_id=STAGE_D_FALLBACK_ANONYMOUS_ID).count() == 1 + class TestFallbackSlowPathInteraction: def test_a_card_the_fallback_calculator_resolved_is_not_routed_to_slow_path(self, db): diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py index 68df9f610..f2a4de165 100644 --- a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py @@ -35,6 +35,7 @@ ImageEvidence, PilotRunLedger, PrintingTagStatus, + VoteSource, ) from cardpicker.operating_envelope import ( FETCH_FAILURE_WINDOW, @@ -429,6 +430,54 @@ def _fetch_crashes_on_second_card(card, dpi=None): assert resumed_ledger.counters["stage_c_completed"] == 2 +class TestConcurrentDispatchVoteCollision: + """Regression for the Stage E Phase 2 shakedown's first live trip + (envtrip-20260724T214616-be6e5db9, failed run_ids stage-e-stream-20260724T2144*): two + CONCURRENT `dispatch_micro_batch` calls scoped to an overlapping card set (django-q2 runs 8 + workers; the backstop sweep can also overlap an event trigger - see + `local_calculate_verdicts._split_new_printing_tag_votes`' own docstring for the full + root-cause writeup) used to abort a WHOLE micro-batch with an `IntegrityError` the instant the + losing dispatch's own Stage D `bulk_create` raced a winner's. Reproduced here at the full + conveyor level (not just the calculator level `test_local_calculate_verdicts.py` covers) by + seeding the winner's vote directly and confirming the loser's own `dispatch_micro_batch` call + completes rather than raising.""" + + @STREAMING_ON + def test_a_losing_race_completes_instead_of_raising_integrity_error( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + import cardpicker.local_calculate_verdicts as local_calculate_verdicts_module + + card = CardFactory(name="Some Card", content_phash=42) + CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + _full_evidence(card, collector_line_collector_number="999") # doesn't match "158" -> is_no_match=True + monkeypatch.setattr( + local_calculate_verdicts_module, + "_eligible_cards_queryset", + lambda *args, **kwargs: local_calculate_verdicts_module.Card.objects.filter(pk=card.pk), + ) + + # the WINNER of the race: another (concurrent, not modeled here) dispatch's own vote + # already landed for this exact (card, anonymous_id) pair. + CardPrintingTag.objects.create( + card=card, + printing=None, + is_no_match=True, + anonymous_id=JOIN_KEY_ANONYMOUS_ID, + source=VoteSource.OCR, + ) + + outcome = dispatch_micro_batch(card_ids=[card.pk]) # must not raise IntegrityError + + assert outcome.status == "completed" + assert outcome.stage_d_join_key_already_voted == 1 + assert CardPrintingTag.objects.filter(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID).count() == 1 + + ledger = PilotRunLedger.objects.get(run_id=outcome.run_id) + assert ledger.status == PilotRunLedger.Status.COMPLETED + assert ledger.counters["stage_d_join_key_already_voted"] == 1 + + class TestBackstopSweep: @STREAMING_ON def test_sweep_processes_the_stage_d_backlog_and_is_idempotent_on_rerun( diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index 0960a6f54..82c692aa6 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -238,10 +238,14 @@ per-batch dry-run leg — see `stage-e-streaming.md` §3 decision (5)), and `counters` carrying `trigger_reason` (`"card-create"`/`"evidence-change"`/ `"backstop-sweep"`/`"backstop-sweep-stage-d"`), `batch_size`, `stage_c_completed`, `stage_c_fetch_failures`, `stage_d_join_key_votes`, -`stage_d_fallback_votes`, `stage_d_slow_path_routed`, `elapsed_s`, -`peak_rss_mb` (via the same `process_metrics.get_process_rss_mb` Phase 1 -wired in), and `lockout_trip_id` (non-null only when a Google lockout tripped -mid-batch). A halted call (`disabled`/`halted-open-trip`/`halted-new-trip`) +`stage_d_join_key_already_voted`, `stage_d_fallback_votes`, +`stage_d_fallback_already_voted` (2026-07-24 — a losing race against a +concurrent overlapping dispatch, see "Resume contract, extended to a +streamed micro-batch" below; non-zero occasionally is healthy, not a bug), +`stage_d_slow_path_routed`, `elapsed_s`, `peak_rss_mb` (via the same +`process_metrics.get_process_rss_mb` Phase 1 wired in), and `lockout_trip_id` +(non-null only when a Google lockout tripped mid-batch). A halted call +(`disabled`/`halted-open-trip`/`halted-new-trip`) writes NO ledger row at all — a halted dispatch never partially starts, so there's nothing to record beyond the `EnvelopeTrip` row `check_envelope` itself already persists. A crashed batch (any other exception) is marked @@ -253,19 +257,46 @@ no new failure-handling mechanism. Each card's own `persist_evidence` call is its own transaction — a crash mid-batch leaves every already-persisted card durably written and nothing -partially written for the card the crash interrupted. A re-invocation over -the same (or an overlapping) card-id set is idempotent: Stage C's resume -filter skips cards already fully processed, and Stage D's own -anonymous_id-exclusion eligibility queries skip cards already voted on — the -same "truthful ledger, idempotent re-entry, zero manual cleanup" property the -batch kill-test (`scripts/ops/crash_drill.sh`) already proves for BULK mode, -now covered for the streamed path by `cardpicker/tests/test_stage_e_dispatch.py`'s -`TestKillSafetyResumeContract` (a mid-batch exception, a truthful `FAILED` -ledger row, and an idempotent re-invocation, at unit-test granularity). The -LIVE, host-level dispatcher-kill drill `stage-e-streaming.md` §7(b) specs -(killing the dispatcher PROCESS itself, not a simulated exception) is still -open — see that section for why it's sequenced into the phase-3 shakedown, -not this change. +partially written for the card the crash interrupted. A **sequential** +re-invocation over the same (or an overlapping) card-id set is idempotent: +Stage C's resume filter skips cards already fully processed, and Stage D's +own anonymous_id-exclusion eligibility queries skip cards already voted on — +the same "truthful ledger, idempotent re-entry, zero manual cleanup" property +the batch kill-test (`scripts/ops/crash_drill.sh`) already proves for BULK +mode, now covered for the streamed path by +`cardpicker/tests/test_stage_e_dispatch.py`'s `TestKillSafetyResumeContract` +(a mid-batch exception, a truthful `FAILED` ledger row, and an idempotent +re-invocation, at unit-test granularity). + +**Correction (2026-07-24, trip `envtrip-20260724T214616-be6e5db9`): a +CONCURRENT overlapping re-invocation is not automatically a cheap no-op the +way a sequential one is — an earlier version of this page overstated the +eligibility exclude alone as sufficient.** django-q2 runs 8 workers +(`Q_CLUSTER["workers"]`), and the cron backstop sweep can overlap an +event-driven dispatch too — two dispatches scoped to the same card can both +pass `run_join_key_calculator`'s/`run_fallback_calculator`'s own eligibility +check before either commits, both compute the same verdict, and race to +write the same `(card, anonymous_id)` vote. The shakedown's first live run +hit exactly this (four failed `PilotRunLedger` rows, run_ids +`stage-e-stream-20260724T2144*`): the loser's `bulk_create` raised +`IntegrityError` and aborted its whole micro-batch. Fixed with a pre-write +skip-if-exists guard on both calculators +(`local_calculate_verdicts._split_new_printing_tag_votes` — see that +function's own docstring for why skip-and-count, not retract-and-recast, is +the honest semantic for a same-evidence concurrent duplicate, as opposed to +`reparse_collector_evidence`'s genuine-conclusion-change retractions). A +losing race is now a counted no-op (`already_voted` on the calculator +result, `stage_d_join_key_already_voted`/`stage_d_fallback_already_voted` on +the micro-batch's own `PilotRunLedger` counters) — real compute still ran +for the loser (evidence lookup, candidate resolution, verdict calculation), +it just doesn't write or crash. `run_slow_path_calculator` needs no +equivalent guard — it writes only `CardScanLog` rows, which carry no DB +uniqueness constraint (append-only by design; see that model's own +docstring), so a race there produces at most a harmless duplicate +routing-marker row, not an error. The LIVE, host-level dispatcher-kill drill +`stage-e-streaming.md` §7(b) specs (killing the dispatcher PROCESS itself, +not a simulated exception) is still open — see that section for why it's +sequenced into the phase-3 shakedown, not this change. ## Phase 3 (not yet built) From 7d948a571f670e77ac89f8194941277fae7b332b Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Fri, 24 Jul 2026 22:27:02 +0000 Subject: [PATCH 2/3] Correct incident facts, add ignore_conflicts belt-and-suspenders (Tron gate) Tron gate on PR #448: seven failed run_ids + one winner (not four) = Q_CLUSTER workers=8; separates the vote-collision failure (this guard) from the SEPARATE envtrip-20260724T214616 host-load trip (11.85 vs 7.0, 8 concurrent OCR dispatches on 7 cores, not fixed by this change); qualifies the "same verdict" premise as contingent on unchanged code/ evidence/lexicon, naming reparse_collector_evidence as the remedy otherwise; adds bulk_create(..., ignore_conflicts=True) as the actual crash-proofing against the guard's own residual check-then-insert race window (precedent: local_layout_class_cast.py:300, local_detect_ai_art.py:459, local_identify_printing_tags.py:1246), with a regression test defeating the pre-write check to prove it; adds an ops-doc runbook line against running BULK-mode writes while PASSIVE streaming is enabled. Co-Authored-By: Claude Fable 5 --- .../cardpicker/local_calculate_verdicts.py | 76 ++++++++++++++----- MPCAutofill/cardpicker/stage_e_dispatch.py | 37 ++++++--- .../tests/test_local_calculate_verdicts.py | 43 ++++++++++- .../cardpicker/tests/test_stage_e_dispatch.py | 21 ++--- docs/features/stage-e-operations.md | 71 ++++++++++++----- 5 files changed, 185 insertions(+), 63 deletions(-) diff --git a/MPCAutofill/cardpicker/local_calculate_verdicts.py b/MPCAutofill/cardpicker/local_calculate_verdicts.py index 5c9d9ea98..5dfb6beac 100644 --- a/MPCAutofill/cardpicker/local_calculate_verdicts.py +++ b/MPCAutofill/cardpicker/local_calculate_verdicts.py @@ -978,7 +978,7 @@ def _split_new_printing_tag_votes( anonymous_id=anonymous_id)` was trusted alone as the idempotence mechanism, which holds for a single sequential invocation but not for two CONCURRENT ones. - ROOT CAUSE of the shakedown failure this guard fixes: `stage_e_dispatch.dispatch_micro_batch` + ROOT CAUSE of the vote-collision failure this guard fixes: `stage_e_dispatch.dispatch_micro_batch` is invoked from django-q2 async tasks (`Q_CLUSTER["workers"] = 8`, MPCAutofill/settings.py) and from the cron backstop sweep - two dispatches scoped to an overlapping card_ids set (e.g. an evidence-change signal re-firing while the backstop sweep's own backlog fill has already picked @@ -987,26 +987,58 @@ def _split_new_printing_tag_votes( both then attempt to `bulk_create` a `CardPrintingTag` for it - the second `bulk_create` in the race hits `cardprintingtag_unique_no_match_vote` (is_no_match=True) or `cardprintingtag_unique_printing_vote` (is_no_match=False), aborting that WHOLE micro-batch's - write with an `IntegrityError`, per the four failed `PilotRunLedger` rows - (run_ids stage-e-stream-20260724T2144*) the shakedown sweep observed. BULK mode never hit this: - every existing `local_calculate_verdicts` invocation runs sequentially, one process, one - identity's eligibility exclude computed and consumed within a single query's lifetime - this - race is specific to Stage E's own concurrent PASSIVE-mode dispatch, the first caller ever to - make more than one of these calculators' write paths overlap in time. + write with an `IntegrityError`. The shakedown sweep observed exactly SEVEN failed `PilotRunLedger` + rows (run_ids stage-e-stream-20260724T2144*) alongside the one dispatch that won the race and + committed cleanly - seven losers plus one winner is eight total concurrent dispatches, exactly + `Q_CLUSTER["workers"] = 8`. BULK mode never hit this: every existing `local_calculate_verdicts` + invocation runs sequentially, one process, one identity's eligibility exclude computed and + consumed within a single query's lifetime - this race is specific to Stage E's own concurrent + PASSIVE-mode dispatch, the first caller ever to make more than one of these calculators' write + paths overlap in time. + + A SEPARATE FAILURE THIS GUARD DOES NOT ADDRESS: the same shakedown run also opened envelope trip + `envtrip-20260724T214616-be6e5db9` (`bar=host_load`, observed load 11.85 against the 7.0 ceiling) + 0.43s AFTER the winning vote landed - caused by all eight of those concurrent dispatches running + OCR/phash extraction at once against this host's 7 usable cores + (`docs/features/catalog-completion-plan.md` L1794/2248/2366's own hardware citation and 0.31x + concurrency finding). That is a HOST-RESOURCE-CONTENTION failure, not a vote-write correctness + one - fixing the vote collision (this function) does nothing to stop eight dispatches from + saturating a 7-core host again the moment streaming resumes. The trip itself still needs the + existing `resolve_envelope_trip` runbook (docs/features/stage-e-operations.md), and re-tripping + on resume is addressed by a companion change that caps concurrent `dispatch_micro_batch` + executions (`settings.STAGE_E_MAX_CONCURRENT_DISPATCHES`), not by anything in this function. + + NOT AN ABSOLUTE GUARANTEE AGAINST THE INTEGRITYERROR - A RESIDUAL WINDOW REMAINS: this is a + check-then-insert guard, not an atomic one - the existence query above and the `bulk_create` + call below are two separate statements, so two dispatches can both pass the check in the same + narrow window and still both attempt to insert. This function alone narrows that window (from + "the whole eligibility-query-to-bulk_create span" down to "the query-to-insert span"); it does + not close it. The actual crash-proofing comes from `bulk_create(..., ignore_conflicts=True)` at + both call sites below - the same belt-and-suspenders precedent this codebase already established + for the identical (eligibility-query-plus-partial-unique-constraint) shape on `CardTagVote` + (`local_layout_class_cast.py:300`, `local_detect_ai_art.py:459`, + `local_identify_printing_tags.py:1246`). The pre-write count this function returns stays useful + as an OBSERVABILITY signal (`already_voted` - a healthy streaming deployment should see it + occasionally) even though it is no longer the sole safety mechanism. SKIP-AND-COUNT, NOT RETRACT-AND-RECAST: deliberately the OPPOSITE choice from `reparse_collector_evidence.reparse_and_retract`'s own retract-then-recast pattern, because the two scenarios are not the same shape. `reparse_collector_evidence` retracts a vote whose CONCLUSION has genuinely changed - a parser fix, a lexicon-gate correction, freshly re-extracted - evidence - so the old vote is objectively stale and superseding it is correct. A concurrent - race here produces no such change: both racing invocations read the SAME current `ImageEvidence` - row under the SAME code version and necessarily compute the SAME verdict (deterministic pure - functions - `calculate_join_key_verdict`/`calculate_fallback_verdict` take no non-reproducible - input). Retracting the winner to let the loser recast would be pure churn - it would even risk - flip-flopping `resolve_and_persist_printing`'s own per-touch consensus recompute if a third - dispatch raced in between the retract and the recast - with no correctness benefit at all, since - there is nothing to correct. The loser's own vote is simply redundant, exactly the shape PR - #411's `_split_new_votes` already established a skip-and-count answer for. + evidence - so the old vote is objectively stale and superseding it is correct. A concurrent race + here produces no such change GIVEN its own operating assumptions hold: both racing invocations + read the SAME current `ImageEvidence` row under the SAME code version, the SAME + `CandidateNameIndex`, and the SAME `known_set_codes()` lexicon, and therefore compute the SAME + verdict (deterministic pure functions - `calculate_join_key_verdict`/`calculate_fallback_verdict` + take no non-reproducible input). If any of those three inputs genuinely differs between the two + racing reads (a mid-race code deploy, or a re-extraction landing between them), the loser's + dropped vote is no longer provably redundant - `reparse_collector_evidence` remains the correct + remedy for that case, exactly as it already is for any other stale-conclusion scenario, and this + guard does not attempt to detect or handle it. Retracting the winner to let the loser recast on + every race - the alternative this function rejects - would be pure churn in the common case where + the assumptions above DO hold, and would even risk flip-flopping `resolve_and_persist_printing`'s + own per-touch consensus recompute if a third dispatch raced in between the retract and the + recast, for no correctness benefit in that common case. One batched existence query (not one query per card), scoped to just the card_ids/anonymous_ids actually present in this batch - same "wasteful full-table scan" avoidance @@ -1127,10 +1159,13 @@ def run_join_key_calculator( if not dry_run: # Pre-write skip-if-exists guard (2026-07-24) - see _split_new_printing_tag_votes' own # docstring for the concurrent-dispatch collision this closes and why skip-and-count (not - # retract-and-recast) is the honest semantic here. + # retract-and-recast) is the honest semantic here. ignore_conflicts=True is the actual + # crash-proofing (the check above still leaves a narrow query-to-insert race window) - + # same belt-and-suspenders precedent as local_layout_class_cast.py:300/ + # local_detect_ai_art.py:459/local_identify_printing_tags.py:1246 for CardTagVote. new_votes, result.already_voted = _split_new_printing_tag_votes(votes_batch) if new_votes: - CardPrintingTag.objects.bulk_create(new_votes) + CardPrintingTag.objects.bulk_create(new_votes, ignore_conflicts=True) CardScanLog.objects.bulk_create(scan_log_batch) for touched_card in Card.objects.filter(pk__in=touched_card_ids): resolve_and_persist_printing(touched_card) @@ -1443,10 +1478,11 @@ def run_fallback_calculator( # Same pre-write skip-if-exists guard as run_join_key_calculator - see # _split_new_printing_tag_votes' own docstring (this calculator writes CardPrintingTag # under its own STAGE_D_FALLBACK_ANONYMOUS_ID and is equally exposed to the concurrent- - # dispatch collision that fixes). + # dispatch collision that fixes). ignore_conflicts=True is the actual crash-proofing - + # same rationale as run_join_key_calculator's own identical call site. new_votes, result.already_voted = _split_new_printing_tag_votes(votes_batch) if new_votes: - CardPrintingTag.objects.bulk_create(new_votes) + CardPrintingTag.objects.bulk_create(new_votes, ignore_conflicts=True) CardScanLog.objects.bulk_create(scan_log_batch) for touched_card in Card.objects.filter(pk__in=touched_card_ids): resolve_and_persist_printing(touched_card) diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index d95539190..969c413c7 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -335,18 +335,31 @@ def _run_stage_d(batch_ids: list[int], run_id: str, outcome: DispatchOutcome) -> query simply finds nothing to do for a card with no current evidence (a "no-evidence" named skip, not an error), so this is always safe to call. - CONCURRENT-DISPATCH VOTE COLLISION (2026-07-24, trip envtrip-20260724T214616-be6e5db9): this - is the FIRST caller ever to invoke `run_join_key_calculator`/`run_fallback_calculator` - concurrently (django-q2 runs `Q_CLUSTER["workers"] = 8`, and the cron backstop sweep can - overlap an event-driven dispatch too) - two dispatches scoped to the same card can both pass - that calculator's own eligibility check before either commits, race to `bulk_create` the same - (card, anonymous_id) vote, and the loser used to hit an `IntegrityError` that aborted its - WHOLE micro-batch. Both calculators now carry their own pre-write skip-if-exists guard - (`local_calculate_verdicts._split_new_printing_tag_votes`) - a losing race is now a counted - no-op (`already_voted`, surfaced on `DispatchOutcome`/this batch's own `PilotRunLedger` row), - never a crash. `run_slow_path_calculator` was checked too and needs no equivalent guard - it - writes only `CardScanLog` rows, which carry no DB uniqueness constraint at all (see that - calculator's own docstring). + CONCURRENT-DISPATCH VOTE COLLISION (2026-07-24, shakedown run tripping envelope trip + envtrip-20260724T214616-be6e5db9): this is the FIRST caller ever to invoke + `run_join_key_calculator`/`run_fallback_calculator` concurrently (django-q2 runs + `Q_CLUSTER["workers"] = 8`, and the cron backstop sweep can overlap an event-driven dispatch + too) - two dispatches scoped to the same card can both pass that calculator's own eligibility + check before either commits, race to `bulk_create` the same (card, anonymous_id) vote, and the + loser used to hit an `IntegrityError` that aborted its WHOLE micro-batch (seven of the eight + concurrent dispatches that run hit this; see `local_calculate_verdicts._split_new_printing_tag_ + votes`' own docstring for the exact incident numbers). Both calculators now carry their own + pre-write skip-if-exists guard PLUS `bulk_create(..., ignore_conflicts=True)` as the actual + crash-proofing (the pre-write check alone still leaves a narrow race window - see that + function's own docstring) - a losing race is now a counted no-op (`already_voted`, surfaced on + `DispatchOutcome`/this batch's own `PilotRunLedger` row), not a guaranteed-impossible one, but + no longer a crash either way. `run_slow_path_calculator` was checked too and needs no + equivalent guard - it writes only `CardScanLog` rows, which carry no DB uniqueness constraint + at all (see that calculator's own docstring). + + THIS DOES NOT ADDRESS THE HOST-LOAD ENVELOPE TRIP the same shakedown run also hit + (`envtrip-20260724T214616-be6e5db9`, `bar=host_load`, 11.85 observed against the 7.0 ceiling, + tripped 0.43s after the winning vote landed) - that trip is HOST RESOURCE CONTENTION from eight + concurrent OCR/phash dispatches on this box's 7 usable cores, a completely separate failure mode + from the vote-write race this function's own guard closes. Fixing the vote collision does not + stop eight concurrent dispatches from re-tripping the load bar the moment streaming resumes - + see `settings.STAGE_E_MAX_CONCURRENT_DISPATCHES` (companion change) for the actual fix to that, + and `docs/features/stage-e-operations.md`'s runbook for acknowledging the open trip itself. """ join_key_result = run_join_key_calculator(run_id=run_id, dry_run=False, card_ids=batch_ids) outcome.stage_d_join_key_votes = join_key_result.votes_written + join_key_result.no_match_votes_written diff --git a/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py b/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py index 77d5bcd5e..8286b2b28 100644 --- a/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py +++ b/MPCAutofill/cardpicker/tests/test_local_calculate_verdicts.py @@ -488,10 +488,12 @@ def test_picks_the_unique_close_match(self): class TestSplitNewPrintingTagVotes: - """Direct unit coverage for the 2026-07-24 concurrent-dispatch collision guard - (trip envtrip-20260724T214616-be6e5db9), independent of either calculator's full - orchestration - mirrors test_local_lands_identify.py's own TestSplitNewVotes structure for - the sibling PR #411 guard.""" + """Direct unit coverage for the 2026-07-24 concurrent-dispatch vote-collision guard (the + shakedown's failed run_ids stage-e-stream-20260724T2144*, a SEPARATE failure from that same + run's envtrip-20260724T214616-be6e5db9 host-load trip - see this guard's own docstring), + independent of either calculator's full orchestration - mirrors + test_local_lands_identify.py's own TestSplitNewVotes structure for the sibling PR #411 + guard.""" def test_empty_batch_returns_empty(self, db): assert _split_new_printing_tag_votes([]) == ([], 0) @@ -727,6 +729,39 @@ def test_concurrent_dispatch_collision_is_skipped_not_crashed(self, db, monkeypa assert result.no_match_votes_written == 0 assert CardPrintingTag.objects.filter(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID).count() == 1 + def test_ignore_conflicts_survives_a_race_the_pre_write_check_itself_missed(self, db, monkeypatch): + """The pre-write guard above (`_split_new_printing_tag_votes`) is check-then-insert, not + atomic - a residual query-to-insert race window remains where a colliding vote could land + between the guard's own existence check and this calculator's `bulk_create` call. Proves + the SECOND line of defense (`bulk_create(..., ignore_conflicts=True)`) alone survives that + residual window, by monkeypatching the guard itself to (falsely) report the vote as new - + exactly what it would see if the collision landed a moment after its own read - while the + colliding row is already in the DB by the time `bulk_create` runs.""" + import cardpicker.local_calculate_verdicts as module + + card = CardFactory(name="Some Card", content_phash=42) + CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + _evidence(card, collector_line_collector_number="999") # no candidate match -> is_no_match=True + + real_split = module._split_new_printing_tag_votes + + def _stale_split(votes_batch): + # Reports every vote as new (already_voted=0), then seeds the collision AFTER the + # guard's own "check" has already run - the exact residual window this test targets. + new_votes, _already_voted = real_split(votes_batch) + CardPrintingTag.objects.create( + card=card, printing=None, is_no_match=True, anonymous_id=JOIN_KEY_ANONYMOUS_ID, source=VoteSource.OCR + ) + return new_votes, 0 + + monkeypatch.setattr(module, "_split_new_printing_tag_votes", _stale_split) + + run_join_key_calculator(dry_run=False) # must not raise IntegrityError despite the stale guard + + # exactly one vote survives - ignore_conflicts=True silently dropped the duplicate insert + # attempt, it did not raise and did not create a second row. + assert CardPrintingTag.objects.filter(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID).count() == 1 + class TestEligibleCardsQueryset: """`_eligible_cards_queryset`'s two knowledge-inventory excludes (module docstring's diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py index f2a4de165..3f7f8b5f7 100644 --- a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py @@ -431,16 +431,19 @@ def _fetch_crashes_on_second_card(card, dpi=None): class TestConcurrentDispatchVoteCollision: - """Regression for the Stage E Phase 2 shakedown's first live trip - (envtrip-20260724T214616-be6e5db9, failed run_ids stage-e-stream-20260724T2144*): two - CONCURRENT `dispatch_micro_batch` calls scoped to an overlapping card set (django-q2 runs 8 - workers; the backstop sweep can also overlap an event trigger - see + """Regression for the VOTE-COLLISION half of the Stage E Phase 2 shakedown's first live + incident (failed run_ids stage-e-stream-20260724T2144*, seven losers + the one winner = eight + total concurrent dispatches, exactly Q_CLUSTER["workers"]=8) - see `local_calculate_verdicts._split_new_printing_tag_votes`' own docstring for the full - root-cause writeup) used to abort a WHOLE micro-batch with an `IntegrityError` the instant the - losing dispatch's own Stage D `bulk_create` raced a winner's. Reproduced here at the full - conveyor level (not just the calculator level `test_local_calculate_verdicts.py` covers) by - seeding the winner's vote directly and confirming the loser's own `dispatch_micro_batch` call - completes rather than raising.""" + root-cause writeup, INCLUDING why this is a separate failure from the same run's + `envtrip-20260724T214616-be6e5db9` host-load envelope trip (a resource-contention problem this + change does not address - see that docstring's own "SEPARATE FAILURE" section). Two CONCURRENT + `dispatch_micro_batch` calls scoped to an overlapping card set (django-q2 runs 8 workers; the + backstop sweep can also overlap an event trigger) used to abort a WHOLE micro-batch with an + `IntegrityError` the instant the losing dispatch's own Stage D `bulk_create` raced a winner's. + Reproduced here at the full conveyor level (not just the calculator level + `test_local_calculate_verdicts.py` covers) by seeding the winner's vote directly and + confirming the loser's own `dispatch_micro_batch` call completes rather than raising.""" @STREAMING_ON def test_a_losing_race_completes_instead_of_raising_integrity_error( diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index 82c692aa6..406f238da 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -277,26 +277,61 @@ event-driven dispatch too — two dispatches scoped to the same card can both pass `run_join_key_calculator`'s/`run_fallback_calculator`'s own eligibility check before either commits, both compute the same verdict, and race to write the same `(card, anonymous_id)` vote. The shakedown's first live run -hit exactly this (four failed `PilotRunLedger` rows, run_ids -`stage-e-stream-20260724T2144*`): the loser's `bulk_create` raised -`IntegrityError` and aborted its whole micro-batch. Fixed with a pre-write -skip-if-exists guard on both calculators +hit exactly this: **seven** failed `PilotRunLedger` rows (run_ids +`stage-e-stream-20260724T2144*`) plus the one dispatch that won the race and +committed cleanly — seven losers plus one winner is eight total concurrent +dispatches, exactly `Q_CLUSTER["workers"] = 8`. Each loser's `bulk_create` +raised `IntegrityError` and aborted its whole micro-batch. Fixed with a +pre-write skip-if-exists guard on both calculators (`local_calculate_verdicts._split_new_printing_tag_votes` — see that function's own docstring for why skip-and-count, not retract-and-recast, is -the honest semantic for a same-evidence concurrent duplicate, as opposed to -`reparse_collector_evidence`'s genuine-conclusion-change retractions). A -losing race is now a counted no-op (`already_voted` on the calculator -result, `stage_d_join_key_already_voted`/`stage_d_fallback_already_voted` on -the micro-batch's own `PilotRunLedger` counters) — real compute still ran -for the loser (evidence lookup, candidate resolution, verdict calculation), -it just doesn't write or crash. `run_slow_path_calculator` needs no -equivalent guard — it writes only `CardScanLog` rows, which carry no DB -uniqueness constraint (append-only by design; see that model's own -docstring), so a race there produces at most a harmless duplicate -routing-marker row, not an error. The LIVE, host-level dispatcher-kill drill -`stage-e-streaming.md` §7(b) specs (killing the dispatcher PROCESS itself, -not a simulated exception) is still open — see that section for why it's -sequenced into the phase-3 shakedown, not this change. +the honest semantic here, GIVEN both racing reads see the same evidence +under the same code version and lexicon — `reparse_collector_evidence` +remains the correct remedy if that assumption doesn't hold, e.g. a +mid-race code deploy or re-extraction) plus `bulk_create(..., ignore_conflicts=True)` — the pre-write check alone still leaves a narrow +query-to-insert race window, so `ignore_conflicts=True` (the same +belt-and-suspenders precedent already used for `CardTagVote` elsewhere in +this codebase) is the actual crash-proofing, not the check. A losing race is +now a counted no-op (`already_voted` on the calculator result, +`stage_d_join_key_already_voted`/`stage_d_fallback_already_voted` on the +micro-batch's own `PilotRunLedger` counters) — real compute still ran for +the loser (evidence lookup, candidate resolution, verdict calculation), it +just doesn't write or crash. `run_slow_path_calculator` needs no equivalent +guard — it writes only `CardScanLog` rows, which carry no DB uniqueness +constraint (append-only by design; see that model's own docstring), so a +race there produces at most a harmless duplicate routing-marker row, not an +error. The LIVE, host-level dispatcher-kill drill `stage-e-streaming.md` +§7(b) specs (killing the dispatcher PROCESS itself, not a simulated +exception) is still open — see that section for why it's sequenced into the +phase-3 shakedown, not this change. + +**A second, SEPARATE failure the same shakedown run hit — not fixed by the +above.** The same eight concurrent dispatches also tripped the envelope +itself: `envtrip-20260724T214616-be6e5db9` is `bar=host_load`, observed load +`11.85` against the `7.0` ceiling, tripped 0.43s AFTER the winning vote +landed — caused by all eight dispatches running OCR/phash extraction at once +against this host's 7 usable cores +(`docs/features/catalog-completion-plan.md` L1794/2248/2366's own hardware +citation and 0.31x concurrency finding), not by anything about how the vote +write path behaves. Fixing the vote collision does nothing to stop eight +concurrent dispatches from saturating the host and re-tripping the load bar +the instant streaming resumes — that trip still needs the ordinary +"Runbook: investigating and clearing a trip" steps above, and re-tripping on +resume is what the companion `settings.STAGE_E_MAX_CONCURRENT_DISPATCHES` +concurrency cap (a separate change) exists to prevent, not this page's own +vote-collision fix. + +**Runbook addition: do not run a BULK-mode write command while PASSIVE +streaming is enabled.** `run_image_evidence_cohort`/`local_calculate_verdicts`/`reparse_collector_evidence`/etc. are entirely outside +`operating_envelope`'s own bars (see "The envelope model" above — BULK mode +was never in scope for that primitive) and outside any per-worker +concurrency cap this page or its companion change adds — they run at their +own configured concurrency regardless of what PASSIVE streaming is doing at +the same time. Running one alongside an active streaming deployment adds +uncontrolled load the envelope has no visibility into and no ability to +throttle, which is exactly the class of resource contention that produced +the host-load trip above. Land a BULK write cleanly, or explicitly pause +PASSIVE streaming for its duration, rather than running both at once. ## Phase 3 (not yet built) From 313a87ad815fd6f8354d13c5c440980ec6e5a727 Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Fri, 24 Jul 2026 22:51:49 +0000 Subject: [PATCH 3/3] Add Stage E streaming dispatch concurrency cap Companion to PR #448's vote-collision fix (Tron gate round 1, COMPANION item), kept as a separate PR per that gate's own instruction. The shakedown's first live run had eight concurrent dispatch_micro_batch calls - all running CPU-bound OCR/phash extraction at once - trip the envelope's host-load bar (11.85 vs 7.0 ceiling) on a host with only 7 usable compute cores. The envelope only trips REACTIVELY, after load has already spiked; this cap is PROACTIVE, refusing to even start a dispatch once settings.STAGE_E_MAX_CONCURRENT_DISPATCHES (default 2) concurrent dispatches are already running. Mechanism: Postgres session-scoped advisory locks (cardpicker.stage_e_concurrency), not a cache-based counter (this app's cache is per-process LocMemCache, useless across django-q2's 8 separate worker processes) or a dedicated low-worker queue (disproportionate infra for a conservative cap). Chosen over a DB-row counter specifically for crash safety: a killed process's session-scoped lock auto-releases, no reconciliation code needed - no new migration either. New DispatchOutcome status "throttled-concurrency-cap" (no ledger row, matching the other halted statuses). Tests include genuine cross-session races (a raw psycopg2 connection standing in for a second django-q worker process) and a real-thread contention test with a shared max-concurrently-held counter, not just sequential calls. Co-Authored-By: Claude Fable 5 --- MPCAutofill/MPCAutofill/settings.py | 13 + MPCAutofill/cardpicker/stage_e_concurrency.py | 150 ++++++++ MPCAutofill/cardpicker/stage_e_dispatch.py | 155 ++++---- .../tests/test_stage_e_concurrency.py | 330 ++++++++++++++++++ docs/features/stage-e-operations.md | 63 +++- docs/upstreaming/extractable-primitives.md | 1 + 6 files changed, 643 insertions(+), 69 deletions(-) create mode 100644 MPCAutofill/cardpicker/stage_e_concurrency.py create mode 100644 MPCAutofill/cardpicker/tests/test_stage_e_concurrency.py diff --git a/MPCAutofill/MPCAutofill/settings.py b/MPCAutofill/MPCAutofill/settings.py index 6596d2f0e..119ea8baa 100755 --- a/MPCAutofill/MPCAutofill/settings.py +++ b/MPCAutofill/MPCAutofill/settings.py @@ -450,3 +450,16 @@ # measurement, not a considered answer. Tunable without a code change (env var) precisely so the # shakedown can adjust it without a redeploy. STAGE_E_MICRO_BATCH_SIZE = env.int("STAGE_E_MICRO_BATCH_SIZE", default=25) + +# Streaming concurrency cap (companion to the 2026-07-24 shakedown's vote-collision fix, PR #448 - +# see cardpicker/stage_e_concurrency.py's own module docstring for the full incident/mechanism +# writeup). Caps the number of CONCURRENTLY-EXECUTING dispatch_micro_batch calls across every +# django-q2 worker PROCESS (Q_CLUSTER["workers"] = 8 above) to this value - the shakedown's first +# live run had eight concurrent dispatches, each running CPU-bound OCR/phash extraction, trip the +# host-load envelope bar on a host with only 7 usable compute cores (docs/features/catalog- +# completion-plan.md L1794/2248/2366's hardware profile). Default 2 is a conservative starting +# point (well under the 7-core ceiling even accounting for other concurrent host activity - Stage +# C's own bulk driver, image-cdn fetch threads, etc.), not a measured/considered answer - tunable +# without a code change (env var) pending real shakedown data, matching STAGE_E_MICRO_BATCH_SIZE's +# own "placeholder, not invented precision" convention immediately above. +STAGE_E_MAX_CONCURRENT_DISPATCHES = env.int("STAGE_E_MAX_CONCURRENT_DISPATCHES", default=2) diff --git a/MPCAutofill/cardpicker/stage_e_concurrency.py b/MPCAutofill/cardpicker/stage_e_concurrency.py new file mode 100644 index 000000000..8dfedcf85 --- /dev/null +++ b/MPCAutofill/cardpicker/stage_e_concurrency.py @@ -0,0 +1,150 @@ +""" +Stage E Phase 2 companion - streaming dispatch concurrency cap (docs/features/stage-e-operations.md; +Tron gate round 1 on PR #448, "COMPANION" item). Caps the number of CONCURRENTLY-EXECUTING +`stage_e_dispatch.dispatch_micro_batch` calls, across every django-q2 worker PROCESS +(`Q_CLUSTER["workers"] = 8`, `MPCAutofill/settings.py`), to `settings.STAGE_E_MAX_CONCURRENT_ +DISPATCHES` (default 2). + +INCIDENT THIS ADDRESSES (2026-07-24 shakedown, distinct from PR #448's own vote-collision fix - see +that PR's own `local_calculate_verdicts._split_new_printing_tag_votes` docstring for the full +incident writeup): the shakedown's first live run had eight concurrent dispatches - all seven +django-q2 workers plus the backstop sweep, or some overlapping mix thereof - each running CPU-bound +OCR/phash extraction at once, which tripped the envelope's own host-load bar +(`envtrip-20260724T214616-be6e5db9`, observed load 11.85 against the 7.0 ceiling) 0.43s AFTER a +vote had already landed. This host has only 7 USABLE compute cores (1 of 8 OCPU is pinned to +network traffic - `docs/features/catalog-completion-plan.md` L1794/2248/2366's own hardware +citation), and that same doc's own local micro-benchmark found CPU-bound OCR work under +oversubscribed threading/concurrency measures at 0.31x-slower-than-sequential (L2226) - eight +dispatches racing for seven cores is exactly the oversubscription shape that number warns against. + +WHY A CAP, NOT JUST THE ENVELOPE: `operating_envelope.check_envelope` is REACTIVE - it only trips +AFTER a fresh signal sample crosses a bar, which is necessarily after the load has already spiked +(the incident's own trip landed 0.43s after the damage was done). This module is PROACTIVE: it +refuses to even START a dispatch once its own slots are all held, so the host is never driven past +a bounded concurrency level in the first place. The two mechanisms are complementary, not +redundant - this module bounds HOW MANY dispatches can run at once; the envelope still catches +whatever load a bounded number of dispatches produces anyway (a single card's OCR pass taking +unusually long, a host already loaded by something outside Stage E entirely, etc.). + +MECHANISM - Postgres session-scoped advisory locks (`pg_try_advisory_lock`/`pg_advisory_unlock`), +not a cache-based counter and not a dedicated low-worker django-q queue: + +- A cache-based slot counter was rejected: this app's own cache backend is Django's default, + per-PROCESS `LocMemCache` (`MPCAutofill/settings.py` has no `CACHES` override at all - the same + fact `cardpicker.review_clusters`' own module docstring already establishes for a different + feature, "the app runs a single gunicorn worker with Django's default (per-process) LocMemCache + backend"). django-q2's 8 workers are separate OS PROCESSES (`multiprocessing`, not threads) - a + cache-based counter would silently fail to coordinate across them, each process seeing its own + empty cache. Adding a shared cache backend (Redis/Memcached) purely to make this primitive work + would be new infrastructure this change has no mandate to introduce. +- A dedicated low-worker django-q queue (a second `Cluster` process pinned to a small worker count, + exclusively for Stage E tasks) was rejected as disproportionate: it needs its own supervisor + process/docker-compose service and deployment wiring, a much larger blast radius than a primitive + enforced INSIDE `dispatch_micro_batch` itself, for a change whose whole point is a conservative, + easily-tunable cap. +- A DB-row-based atomic counter (`UPDATE ... WHERE claimed_count < max_slots`) was considered and + rejected in favor of advisory locks specifically for CRASH SAFETY: a `kill -9`'d worker process + would leave a row-based counter's slot permanently "claimed" (nothing ties a table row to a + process's lifetime), requiring a whole separate staleness-reconciliation mechanism to match this + pipeline's own established "truthful ledger, idempotent re-entry, zero manual cleanup" ethos + (`scripts/ops/crash_drill.sh`, `test_stage_e_dispatch.py`'s `TestKillSafetyResumeContract`). A + Postgres SESSION-scoped advisory lock is tied to the underlying DB connection/session - the + moment that connection dies (process killed, network drop), Postgres auto-releases every advisory + lock that session held, with no reconciliation code needed. Postgres is also already the one + piece of shared, process-visible state this pipeline guarantees present (the same reason + `EnvelopeTrip` itself is a DB row, not a cache entry) - zero new infrastructure, zero new + migration (this module defines no model). + +CONNECTION-LIFECYCLE CONTRACT (why "same connection acquire-to-release" holds here): Postgres +advisory locks are per-SESSION, so acquiring on one DB connection and releasing on a DIFFERENT one +is a no-op that leaks the lock. `try_acquire_dispatch_slot` is a context manager that acquires and +releases within the SAME `with` block, using Django's own per-thread `django.db.connection` proxy +throughout - confirmed safe against django-q2's own connection-recycling behavior +(`django_q.worker.py`'s `close_old_django_connections()` call, which happens ONLY immediately +BEFORE the next task starts, never mid-task or between statements within one task) by direct +inspection of the installed `django-q2` package: a single `dispatch_micro_batch` call - and +therefore a single acquire/use/release cycle of this module - always executes as one uninterrupted +segment on one connection, whether triggered via a django-q `async_task` or a direct +`stream_backstop_sweep` command-line invocation. +""" + +import logging +from contextlib import contextmanager +from typing import Iterator, Optional + +from django.conf import settings +from django.db import connection + +logger = logging.getLogger(__name__) + +# Arbitrary but fixed and greppable namespace for this module's own advisory locks - the two-int +# pg_try_advisory_lock(key1, key2) form is used (not the single-bigint form) so this namespace and +# the slot index are both legible directly in a live `pg_locks` query during on-call debugging, +# rather than one opaque packed bigint. Confirmed (via grep across cardpicker/) that no other code +# in this repo uses Postgres advisory locks today, so there is no existing namespace to collide +# with. +_LOCK_NAMESPACE = 0x53746745 # "StgE" read as hex digits, chosen for exactly that mnemonic value + + +def _slot_count() -> int: + return getattr(settings, "STAGE_E_MAX_CONCURRENT_DISPATCHES", 2) + + +def _try_acquire_slot() -> Optional[int]: + """ + Tries every slot index in `[0, _slot_count())` in ascending order, returning the first one + whose `pg_try_advisory_lock` call succeeds, or `None` if every slot is already held elsewhere. + Never blocks - `pg_try_advisory_lock` is non-blocking by design (unlike the plain + `pg_advisory_lock`, which would queue), matching this primitive's own "refuse immediately, + never queue" posture - the same posture `operating_envelope.check_envelope` already + established for the envelope itself (a busy host should shed load, not build a backlog of + blocked dispatches waiting for a slot). + """ + with connection.cursor() as cursor: + for slot in range(_slot_count()): + cursor.execute("SELECT pg_try_advisory_lock(%s, %s)", [_LOCK_NAMESPACE, slot]) + (acquired,) = cursor.fetchone() + if acquired: + return slot + return None + + +def _release_slot(slot: int) -> None: + """Releases a slot this process previously acquired via `_try_acquire_slot` - MUST run on the + same `django.db.connection` the acquire happened on (see module docstring's own + "CONNECTION-LIFECYCLE CONTRACT" section). Logs rather than raises if Postgres reports the lock + wasn't held (`pg_advisory_unlock` returns `false`, never an error, for that case) - this would + only happen if the underlying connection was somehow recycled mid-dispatch, a condition this + module's own docstring argues shouldn't occur but that a release path should still fail soft + against rather than crash an otherwise-successful dispatch over. + """ + with connection.cursor() as cursor: + cursor.execute("SELECT pg_advisory_unlock(%s, %s)", [_LOCK_NAMESPACE, slot]) + (released,) = cursor.fetchone() + if not released: + logger.warning( + "stage_e_concurrency: pg_advisory_unlock reported slot %s was not held by this " + "connection - possible connection recycling mid-dispatch", + slot, + ) + + +@contextmanager +def try_acquire_dispatch_slot() -> Iterator[Optional[int]]: + """ + The primitive `stage_e_dispatch.dispatch_micro_batch` calls. Yields the acquired slot index (an + `int` in `[0, settings.STAGE_E_MAX_CONCURRENT_DISPATCHES)`), or `None` if every slot was already + held - the caller is expected to treat `None` as "throttled, do no work this call", the same + posture `current_trip() is not None` already gets in `dispatch_micro_batch`'s own no-self-resume + gate. ALWAYS releases whatever it acquired on exit, including when the `with` block raises - + a dispatch that crashes mid-batch must not permanently strand a slot (this module's own crash + line of defense is the connection-level auto-release described in the module docstring; this + `finally` is the FAST path for the ordinary "dispatch finished, successfully or not, without the + whole process dying" case). + """ + slot = _try_acquire_slot() + try: + yield slot + finally: + if slot is not None: + _release_slot(slot) diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index 969c413c7..a0dbba4ce 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -91,6 +91,7 @@ ) from cardpicker.pilot_run_lifecycle import mark_ledger_failed, merge_counters from cardpicker.process_metrics import get_process_rss_mb +from cardpicker.stage_e_concurrency import try_acquire_dispatch_slot from cardpicker.utils import get_baked_git_sha logger = logging.getLogger(__name__) @@ -159,6 +160,11 @@ class DispatchOutcome: - "halted-open-trip" - `current_trip()` was already non-None; no self-resume (module docstring). - "halted-new-trip" - this call's own fresh envelope sample breached a bar. - "empty" - streaming is enabled and the envelope is clear, but nothing was eligible. + - "throttled-concurrency-cap" - a real batch was selected, but every + `settings.STAGE_E_MAX_CONCURRENT_DISPATCHES` slot (`cardpicker.stage_e_concurrency`) was + already held by another concurrent dispatch - PROACTIVE throttling, distinct from the + envelope's own REACTIVE halted-new-trip (see that module's own docstring for why both + exist). No ledger row is written, matching the other halted statuses. - "completed" - did real work; does not itself guarantee zero failures inside the batch (a card can still fail its own fetch/extraction), only that the DISPATCH LOOP didn't halt. - "completed-with-trip" - did real work, but a `GoogleFetchLockoutError` observed mid-batch @@ -386,10 +392,12 @@ def dispatch_micro_batch( `stream_backstop_sweep` (`card_ids=None`, letting `_select_micro_batch` fill the whole batch from the backlog). - Ordering: default-off gate -> no-self-resume gate -> fresh envelope sample -> Stage C - (sequential, per-card) -> Stage D (AS-IS entry points, scoped) -> ledger write. Every gate below - returns WITHOUT touching the DB (aside from the envelope check's own trip-persist side effect) - the instant it applies - a halted dispatch never partially starts Stage C. + Ordering: default-off gate -> no-self-resume gate -> fresh envelope sample -> batch selection -> + concurrency-cap slot acquire (`cardpicker.stage_e_concurrency`) -> Stage C (sequential, per-card) + -> Stage D (AS-IS entry points, scoped) -> ledger write -> slot release. Every gate below returns + WITHOUT touching the DB (aside from the envelope check's own trip-persist side effect, and the + concurrency-cap check's own advisory-lock round trip, which writes nothing to any table) the + instant it applies - a halted or throttled dispatch never partially starts Stage C. """ if not getattr(settings, "STAGE_E_STREAMING_ENABLED", False): return DispatchOutcome(status="disabled", run_id=run_id) @@ -427,69 +435,86 @@ def dispatch_micro_batch( if not batch_ids: return DispatchOutcome(status="empty", run_id=run_id) - dispatch_run_id = run_id or f"stage-e-stream-{timezone.now().strftime('%Y%m%dT%H%M%S%f')}Z" - - # Micro-batch ledger row convention (task brief scope item 6, docs/features/stage-e-operations.md's - # "Phase 2" section): one PilotRunLedger row per micro-batch dispatch, `command= - # "stage_e_streaming_dispatch"`, `dry_run=False` always (PASSIVE mode has no dry-run leg - the - # per-envelope-change dry run §3 decision (5) describes is a one-off owner review of the - # envelope bounds themselves, not a per-batch gate the way BULK mode's forced-dry-run guard is). - ledger = PilotRunLedger.objects.create( - run_id=dispatch_run_id, - command="stage_e_streaming_dispatch", - dry_run=False, - status=PilotRunLedger.Status.RUNNING, - git_sha=get_baked_git_sha(), - counters={"trigger_reason": trigger_reason, "batch_size": len(batch_ids)}, - ) + # CONCURRENCY CAP (companion to PR #448's vote-collision fix - cardpicker.stage_e_concurrency's + # own module docstring has the full incident/mechanism writeup): acquired around exactly the + # CPU-heavy segment below (ledger create through Stage C/D completion), not around the cheap + # batch-selection query above - holding a scarce slot while doing nothing but a bounded read + # would only starve other dispatches for no benefit. `slot is None` means every + # `settings.STAGE_E_MAX_CONCURRENT_DISPATCHES` slot is already held elsewhere - PROACTIVE + # throttling, distinct from the envelope's own REACTIVE halted-new-trip below. + with try_acquire_dispatch_slot() as slot: + if slot is None: + logger.info( + "Stage E dispatch throttled - all %s concurrency-cap slots already held", + getattr(settings, "STAGE_E_MAX_CONCURRENT_DISPATCHES", 2), + ) + return DispatchOutcome(status="throttled-concurrency-cap", run_id=run_id) + + dispatch_run_id = run_id or f"stage-e-stream-{timezone.now().strftime('%Y%m%dT%H%M%S%f')}Z" + + # Micro-batch ledger row convention (task brief scope item 6, docs/features/stage-e-operations.md's + # "Phase 2" section): one PilotRunLedger row per micro-batch dispatch, `command= + # "stage_e_streaming_dispatch"`, `dry_run=False` always (PASSIVE mode has no dry-run leg - the + # per-envelope-change dry run §3 decision (5) describes is a one-off owner review of the + # envelope bounds themselves, not a per-batch gate the way BULK mode's forced-dry-run guard is). + ledger = PilotRunLedger.objects.create( + run_id=dispatch_run_id, + command="stage_e_streaming_dispatch", + dry_run=False, + status=PilotRunLedger.Status.RUNNING, + git_sha=get_baked_git_sha(), + counters={"trigger_reason": trigger_reason, "batch_size": len(batch_ids)}, + ) - outcome = DispatchOutcome(status="completed", run_id=dispatch_run_id, card_ids=batch_ids) - batch_start = time.monotonic() + outcome = DispatchOutcome(status="completed", run_id=dispatch_run_id, card_ids=batch_ids) + batch_start = time.monotonic() - try: - lockout_trip = _run_stage_c(batch_ids, dispatch_run_id, outcome) - # Stage D still runs even after a mid-batch lockout trip - "in-flight work drains, nothing - # NEW starts" (docs/features/stage-e-operations.md's HALT semantics) - see _run_stage_d's - # own docstring for why this is always safe to call regardless of how far Stage C got. - _run_stage_d(batch_ids, dispatch_run_id, outcome) - - if lockout_trip is not None: - outcome.status = "completed-with-trip" - outcome.trip_id = lockout_trip.trip_id - - peak_rss_mb = get_process_rss_mb() - ledger.status = PilotRunLedger.Status.COMPLETED - ledger.finished_at = timezone.now() - ledger.counters = merge_counters( - ledger.counters, - { - "elapsed_s": round(time.monotonic() - batch_start, 3), - "stage_c_completed": outcome.stage_c_completed, - "stage_c_fetch_failures": outcome.stage_c_fetch_failures, - "stage_d_join_key_votes": outcome.stage_d_join_key_votes, - "stage_d_join_key_already_voted": outcome.stage_d_join_key_already_voted, - "stage_d_fallback_votes": outcome.stage_d_fallback_votes, - "stage_d_fallback_already_voted": outcome.stage_d_fallback_already_voted, - "stage_d_slow_path_routed": outcome.stage_d_slow_path_routed, - "peak_rss_mb": peak_rss_mb, - "lockout_trip_id": lockout_trip.trip_id if lockout_trip is not None else None, - }, - ) - ledger.save(update_fields=["status", "finished_at", "counters"]) - except Exception as exc: - # Shared FAILED-transition rail (cardpicker.pilot_run_lifecycle.mark_ledger_failed) - a - # no-op if this invocation already reached the COMPLETED save above, otherwise records a - # triage-able counters["failure_reason"] alongside FAILED (docs/proposals/ - # stage-e-streaming.md §3 decision (6)'s "empty-failed-row" gap fix, reused here rather than - # duplicated). A crash mid-Stage-C-loop leaves every already-`persist_evidence`-committed - # card durably written (each card's own persist is its own transaction) - the resume - # contract (docs/features/stage-e-operations.md) holds: a fresh dispatch over the same or an - # overlapping card set skips whatever's already current and picks up the rest, exactly the - # same "truthful ledger, idempotent re-entry" property the batch kill-test already proves. - mark_ledger_failed(ledger, exc) - raise - - return outcome + try: + lockout_trip = _run_stage_c(batch_ids, dispatch_run_id, outcome) + # Stage D still runs even after a mid-batch lockout trip - "in-flight work drains, nothing + # NEW starts" (docs/features/stage-e-operations.md's HALT semantics) - see _run_stage_d's + # own docstring for why this is always safe to call regardless of how far Stage C got. + _run_stage_d(batch_ids, dispatch_run_id, outcome) + + if lockout_trip is not None: + outcome.status = "completed-with-trip" + outcome.trip_id = lockout_trip.trip_id + + peak_rss_mb = get_process_rss_mb() + ledger.status = PilotRunLedger.Status.COMPLETED + ledger.finished_at = timezone.now() + ledger.counters = merge_counters( + ledger.counters, + { + "elapsed_s": round(time.monotonic() - batch_start, 3), + "stage_c_completed": outcome.stage_c_completed, + "stage_c_fetch_failures": outcome.stage_c_fetch_failures, + "stage_d_join_key_votes": outcome.stage_d_join_key_votes, + "stage_d_join_key_already_voted": outcome.stage_d_join_key_already_voted, + "stage_d_fallback_votes": outcome.stage_d_fallback_votes, + "stage_d_fallback_already_voted": outcome.stage_d_fallback_already_voted, + "stage_d_slow_path_routed": outcome.stage_d_slow_path_routed, + "peak_rss_mb": peak_rss_mb, + "lockout_trip_id": lockout_trip.trip_id if lockout_trip is not None else None, + }, + ) + ledger.save(update_fields=["status", "finished_at", "counters"]) + except Exception as exc: + # Shared FAILED-transition rail (cardpicker.pilot_run_lifecycle.mark_ledger_failed) - a + # no-op if this invocation already reached the COMPLETED save above, otherwise records a + # triage-able counters["failure_reason"] alongside FAILED (docs/proposals/ + # stage-e-streaming.md §3 decision (6)'s "empty-failed-row" gap fix, reused here rather than + # duplicated). A crash mid-Stage-C-loop leaves every already-`persist_evidence`-committed + # card durably written (each card's own persist is its own transaction) - the resume + # contract (docs/features/stage-e-operations.md) holds: a fresh dispatch over the same or an + # overlapping card set skips whatever's already current and picks up the rest, exactly the + # same "truthful ledger, idempotent re-entry" property the batch kill-test already proves. + # The concurrency-cap slot is still released (try_acquire_dispatch_slot's own `finally`) + # even though this exception propagates past this `with` block. + mark_ledger_failed(ledger, exc) + raise + + return outcome def dispatch_for_card(card_id: int, reason: str = "event") -> None: diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_concurrency.py b/MPCAutofill/cardpicker/tests/test_stage_e_concurrency.py new file mode 100644 index 000000000..cf61ce185 --- /dev/null +++ b/MPCAutofill/cardpicker/tests/test_stage_e_concurrency.py @@ -0,0 +1,330 @@ +""" +Tests for cardpicker.stage_e_concurrency - the Stage E streaming dispatch concurrency cap +(docs/features/stage-e-operations.md; companion to PR #448's vote-collision fix, Tron gate round 1 +"COMPANION" item). Runs against the real testcontainer Postgres (never sqlite/mocked) since the +whole point of this module is genuine Postgres advisory-lock semantics. + +A DELIBERATE, LOAD-BEARING TEST-DESIGN CONSTRAINT, discovered while writing these tests (not just a +style choice): Postgres SESSION-level advisory locks are RE-ENTRANT within one session - a session +that already holds `pg_try_advisory_lock(ns, 0)` gets an immediate second success (not a move to +slot 1) if it calls the exact same lock again on the SAME connection, incrementing an internal +reference count that then needs a matching number of unlocks. Calling `_try_acquire_slot()` (or +entering `try_acquire_dispatch_slot()`) TWICE on Django's own shared per-thread `connection` WITHOUT +releasing in between therefore does NOT simulate "two independent dispatches" the way it would for +two genuinely separate sessions - it silently re-acquires the SAME slot instead. Every test below +that needs more than one concurrent "dispatcher" therefore uses a genuinely SEPARATE `psycopg2` +connection per dispatcher (`_raw_connection`/`_try_acquire_on_new_connection`) - this is also the +production-faithful choice, since two real concurrent dispatches are always on two separate +django-q worker PROCESSES (separate connections), never the same session calling this module twice. +""" + +import threading +import time +from typing import Any, Optional + +import psycopg2 +import pytest + +from django.db import connection +from django.test import override_settings + +from cardpicker import stage_e_concurrency +from cardpicker.stage_e_concurrency import ( + _LOCK_NAMESPACE, + _release_slot, + _try_acquire_slot, + try_acquire_dispatch_slot, +) + +CAP_2 = override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=2) + + +def _raw_connection() -> "psycopg2.extensions.connection": + """A genuinely independent DB session to the SAME database Django's own test connection is + using - built from `connection.get_connection_params()` (not a guessed host/port/dbname) so + this always matches whatever pytest-django actually connected to, including its own `test_` + database-name prefixing. Forces Django's own connection to actually exist first + (`connection.ensure_connection()`) since `get_connection_params()` alone doesn't establish one. + `autocommit=True` - advisory locks are independent of transactions, and leaving a raw connection + in Postgres's default (non-autocommit) mode would hold an idle transaction open for no reason.""" + connection.ensure_connection() + params = connection.get_connection_params() + raw = psycopg2.connect(**params) + raw.autocommit = True + return raw + + +def _try_acquire_on_connection(conn: "psycopg2.extensions.connection", cap: int) -> Optional[int]: + """The exact same acquire loop `_try_acquire_slot` runs internally, against an explicit, + caller-owned connection rather than Django's shared one - see module docstring for why this + (not a second call to `_try_acquire_slot` itself) is how a second, independent dispatcher is + simulated in these tests.""" + with conn.cursor() as cursor: + for slot in range(cap): + cursor.execute("SELECT pg_try_advisory_lock(%s, %s)", [_LOCK_NAMESPACE, slot]) + (acquired,) = cursor.fetchone() + if acquired: + return slot + return None + + +def _release_on_connection(conn: "psycopg2.extensions.connection", slot: int) -> None: + with conn.cursor() as cursor: + cursor.execute("SELECT pg_advisory_unlock(%s, %s)", [_LOCK_NAMESPACE, slot]) + + +@pytest.fixture(autouse=True) +def _release_any_leaked_locks(db: Any): + """Postgres advisory locks are SESSION-scoped, not transaction-scoped - pytest-django's own + per-test transaction ROLLBACK (the `db` fixture) does NOT release them, unlike ordinary row + writes. Every test below is expected to release everything it acquires, but this fixture is a + safety net: it fully DRAINS every plausible slot's lock count on Django's own test connection + after each test (looping `pg_advisory_unlock` until it reports nothing left, not just once - + a single call would only undo ONE level of the re-entrant reference count the module docstring + describes, silently leaving a still-held lock behind for the next test in the same pytest + session to trip over).""" + yield + with connection.cursor() as cursor: + for slot in range(8): # generous upper bound - real caps in this file never exceed 2 + while True: + cursor.execute("SELECT pg_advisory_unlock(%s, %s)", [_LOCK_NAMESPACE, slot]) + (released,) = cursor.fetchone() + if not released: + break + + +class TestTryAcquireSlot: + @CAP_2 + def test_two_independent_dispatchers_get_distinct_slots(self, db: Any) -> None: + dispatcher_a = _try_acquire_slot() # Django's own connection + raw_b = _raw_connection() + try: + dispatcher_b = _try_acquire_on_connection(raw_b, cap=2) + + assert dispatcher_a == 0 + assert dispatcher_b == 1 + + _release_slot(dispatcher_a) + _release_on_connection(raw_b, dispatcher_b) + finally: + raw_b.close() + + @CAP_2 + def test_a_third_independent_dispatcher_is_refused_once_the_cap_is_exhausted(self, db: Any) -> None: + dispatcher_a = _try_acquire_slot() + raw_b = _raw_connection() + raw_c = _raw_connection() + try: + dispatcher_b = _try_acquire_on_connection(raw_b, cap=2) + dispatcher_c = _try_acquire_on_connection(raw_c, cap=2) + + assert dispatcher_a is not None and dispatcher_b is not None + assert dispatcher_c is None + + _release_slot(dispatcher_a) + _release_on_connection(raw_b, dispatcher_b) + finally: + raw_b.close() + raw_c.close() + + @CAP_2 + def test_releasing_a_slot_makes_it_acquirable_by_a_different_dispatcher(self, db: Any) -> None: + dispatcher_a = _try_acquire_slot() + raw_b = _raw_connection() + raw_c = _raw_connection() + try: + dispatcher_b = _try_acquire_on_connection(raw_b, cap=2) + assert dispatcher_a == 0 and dispatcher_b == 1 + + _release_slot(dispatcher_a) # slot 0 freed + + dispatcher_c = _try_acquire_on_connection(raw_c, cap=2) + assert dispatcher_c == 0 # a THIRD, different dispatcher claims the freed slot + + _release_on_connection(raw_b, dispatcher_b) + _release_on_connection(raw_c, dispatcher_c) + finally: + raw_b.close() + raw_c.close() + + def test_default_cap_is_two(self, db: Any) -> None: + assert stage_e_concurrency._slot_count() == 2 + + +class TestTryAcquireDispatchSlot: + @CAP_2 + def test_context_manager_yields_and_releases_a_slot(self, db: Any) -> None: + with try_acquire_dispatch_slot() as slot: + assert slot == 0 + # still held while inside the block - an independent concurrent dispatcher sees + # exactly the OTHER slot free, and nothing left after that. + raw_b = _raw_connection() + raw_c = _raw_connection() + try: + other_slot = _try_acquire_on_connection(raw_b, cap=2) + nothing_left = _try_acquire_on_connection(raw_c, cap=2) + assert other_slot == 1 + assert nothing_left is None + finally: + _release_on_connection(raw_b, other_slot) + raw_b.close() + raw_c.close() + + # released on clean exit - a fresh independent dispatcher can now claim slot 0 again. + raw_d = _raw_connection() + try: + reacquired = _try_acquire_on_connection(raw_d, cap=2) + assert reacquired == 0 + _release_on_connection(raw_d, reacquired) + finally: + raw_d.close() + + @CAP_2 + def test_slot_is_released_even_when_the_block_raises(self, db: Any) -> None: + with pytest.raises(RuntimeError, match="simulated failure inside the dispatch"): + with try_acquire_dispatch_slot() as slot: + assert slot == 0 + raise RuntimeError("simulated failure inside the dispatch") + + # not leaked - an independent dispatcher can claim slot 0 despite the exception. + raw = _raw_connection() + try: + reacquired = _try_acquire_on_connection(raw, cap=2) + assert reacquired == 0 + _release_on_connection(raw, reacquired) + finally: + raw.close() + + @override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=1) + def test_yields_none_once_the_single_slot_is_already_held(self, db: Any) -> None: + raw = _raw_connection() + try: + held = _try_acquire_on_connection(raw, cap=1) + assert held == 0 + + with try_acquire_dispatch_slot() as slot: + assert slot is None # PROACTIVE throttle - the only slot is already taken + + _release_on_connection(raw, held) + finally: + raw.close() + + +class TestCrossConnectionRace: + """Proves real cross-SESSION safety with independent raw connections standing in for separate + django-q worker PROCESSES - the production shape this module exists to coordinate against.""" + + @CAP_2 + def test_a_second_independent_session_cannot_exceed_the_cap(self, db: Any) -> None: + dispatcher_a = _try_acquire_slot() + raw_b = _raw_connection() + try: + dispatcher_b = _try_acquire_on_connection(raw_b, cap=2) + assert dispatcher_a == 0 and dispatcher_b == 1 + + raw_c = _raw_connection() + try: + with raw_c.cursor() as cursor: + cursor.execute("SELECT pg_try_advisory_lock(%s, %s)", [_LOCK_NAMESPACE, 0]) + (acquired_0,) = cursor.fetchone() + cursor.execute("SELECT pg_try_advisory_lock(%s, %s)", [_LOCK_NAMESPACE, 1]) + (acquired_1,) = cursor.fetchone() + # a genuinely independent third session sees BOTH slots as already held. + assert acquired_0 is False + assert acquired_1 is False + finally: + raw_c.close() + + _release_slot(dispatcher_a) + _release_on_connection(raw_b, dispatcher_b) + finally: + raw_b.close() + + @CAP_2 + def test_killing_the_holding_session_auto_releases_its_slot(self, db: Any) -> None: + """The crash-safety property this module's own docstring cites as the reason advisory + locks were chosen over a DB-row counter: closing (simulating a killed process) the session + that held a slot releases it with no explicit unlock call and no reconciliation code.""" + raw = _raw_connection() + acquired = _try_acquire_on_connection(raw, cap=2) + assert acquired == 0 + + raw.close() # simulates `kill -9` on the process holding this session - no unlock call + + # Django's own connection now claims the same slot successfully - auto-released, not leaked. + reacquired = _try_acquire_slot() + assert reacquired == 0 + _release_slot(reacquired) + + +class TestSimulatedConcurrentDispatchers: + """Exercises the cap the way the task brief asks for explicitly - 'simulated concurrency' - + using real OS threads, each opening its OWN independent DB connection (never Django's shared + per-thread `connection` reused across threads - see module docstring's re-entrancy note), each + HOLDING its slot for a measurable duration (not releasing instantly) so genuine time-overlap + between threads actually happens, and a shared counter proving the cap is never exceeded AT ANY + INSTANT - not just "never exceeded across the whole run's own totals", which a purely + sequential/non-overlapping execution could satisfy trivially even with a broken cap.""" + + @CAP_2 + def test_the_cap_holds_under_genuine_concurrent_contention(self, db: Any) -> None: + thread_count = 6 + hold_seconds = 0.2 + barrier = threading.Barrier(thread_count) + lock = threading.Lock() + state = {"currently_held": 0, "max_observed": 0} + results: list[Optional[int]] = [None] * thread_count + backend_pids: list[int] = [] + + def _worker(index: int) -> None: + raw = _raw_connection() + try: + with lock: + backend_pids.append(raw.get_backend_pid()) + barrier.wait() # maximise real contention - every thread races to acquire at once + slot = _try_acquire_on_connection(raw, cap=2) + results[index] = slot + if slot is None: + return + with lock: + state["currently_held"] += 1 + state["max_observed"] = max(state["max_observed"], state["currently_held"]) + time.sleep(hold_seconds) # hold long enough to force real overlap between threads + with lock: + state["currently_held"] -= 1 + _release_on_connection(raw, slot) + finally: + raw.close() + + threads = [threading.Thread(target=_worker, args=(i,)) for i in range(thread_count)] + for t in threads: + t.start() + for t in threads: + t.join() + + # `raw.close()` above closes each client socket synchronously, but Postgres's own backend + # process can take a moment longer to finish tearing down after that - poll (not a blind + # sleep) until every one of this test's own backend pids has actually disappeared from + # `pg_stat_activity`, so this test session's own final `DROP DATABASE` never races a + # not-quite-gone-yet backend. A correctness requirement of clean test teardown, not of the + # cap itself. + deadline = time.monotonic() + 5.0 + with connection.cursor() as cursor: + while time.monotonic() < deadline: + cursor.execute("SELECT pid FROM pg_stat_activity WHERE pid = ANY(%s)", [backend_pids]) + if not cursor.fetchall(): + break + time.sleep(0.05) + + acquired = [r for r in results if r is not None] + throttled = [r for r in results if r is None] + assert len(acquired) + len(throttled) == thread_count + # the cap was genuinely exercised under real overlap (not a trivially-serial run) AND + # never breached at any instant - both directions matter: `< 2` would mean this test + # failed to create real contention at all (a false-positive pass), `> 2` would mean the + # cap itself is broken. + assert state["max_observed"] == 2 + # with a 0.2s hold and 6 racing threads all starting at once, MORE than 2 of them are + # expected to eventually succeed (slots free up and get reused within the race), but never + # simultaneously - that instant-in-time property is what max_observed pins down above. + assert len(throttled) >= 1 # cap=2 with 6 simultaneous starters must throttle at least one diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index 406f238da..819c1900b 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -183,7 +183,15 @@ entry points (`run_join_key_calculator`/`run_fallback_calculator`/ lacking a full-manifest `ImageEvidence` row — the same shape `run_image_evidence_cohort.py`'s own resume filter uses, imported, not reimplemented). -5. **Stage C** (sequential, per-card, not pooled — a micro-batch is far too +5. **Concurrency-cap slot acquire** (companion change, 2026-07-24 — + `cardpicker.stage_e_concurrency`) — refuses PROACTIVELY + (`status="throttled-concurrency-cap"`, zero DB writes beyond the + advisory-lock check itself) once `settings.STAGE_E_MAX_CONCURRENT_DISPATCHES` + (default 2) dispatches are already running concurrently, anywhere across + this box's django-q2 worker processes. See "Concurrency cap" below for + the full mechanism and the incident that motivated it — distinct from, + and a proactive complement to, the envelope's own reactive host-load bar. +6. **Stage C** (sequential, per-card, not pooled — a micro-batch is far too small for BULK mode's process-pool concurrency to help) — the same `compute_card_evidence`/`persist_evidence` unit `run_image_evidence_cohort.py` drives, one card at a time. A `GoogleFetchLockoutError` stops Stage C for @@ -191,15 +199,17 @@ entry points (`run_join_key_calculator`/`run_fallback_calculator`/ in-flight, already-committed work stays committed; Stage D below still runs against whatever was reached ("in-flight work drains, nothing NEW starts"). -6. **Stage D** — `run_join_key_calculator`/`run_fallback_calculator`/ +7. **Stage D** — `run_join_key_calculator`/`run_fallback_calculator`/ `run_slow_path_calculator`, called AS-IS with the new `card_ids` scope, in the same escalation order every BULK-mode invocation already uses. Each of these already calls `resolve_and_persist_printing` internally for every card it touches — this is what satisfies §3 decision (4)'s "scoped incremental per-touch consensus recompute" with no separate consensus step in the dispatcher at all. -7. **Ledger write** — one `PilotRunLedger` row per micro-batch (see - "Observability" below). +8. **Ledger write, then concurrency-cap slot release** — one `PilotRunLedger` + row per micro-batch (see "Observability" below), then the slot acquired in + step 5 is released (always, including on an exception - see "Concurrency + cap" below). ### Trigger: event-driven, plus a cron backstop (§3 decision (1)) @@ -230,6 +240,51 @@ tail shakedown's own instrumentation (phase 3, not yet run). The default sits inside the brief's own "roughly 10-100 cards per batch" sanity range as a conservative starting point pending that measurement. +### Concurrency cap (companion change, 2026-07-24) + +Caps the number of `dispatch_micro_batch` calls running CONCURRENTLY, across +every django-q2 worker process on the box, to +`settings.STAGE_E_MAX_CONCURRENT_DISPATCHES` (default `2`, env-tunable — +same "placeholder pending real measurement" posture as +`STAGE_E_MICRO_BATCH_SIZE` above). Motivated by the shakedown incident PR +#448 also fixed (the vote-collision half of the same run) — see +`local_calculate_verdicts._split_new_printing_tag_votes`'s own docstring for +the full incident numbers: eight concurrent dispatches, all running +CPU-bound OCR/phash extraction at once, tripped the host-load envelope bar +(`envtrip-20260724T214616-be6e5db9`, 11.85 observed against the 7.0 +ceiling) on a host with only 7 usable compute cores +(`docs/features/catalog-completion-plan.md` L1794/2248/2366's hardware +citation and its own 0.31x-slower-than-sequential CPU-bound-oversubscription +finding). + +**Why a cap in addition to the envelope, not instead of it**: the envelope's +own host-load bar is REACTIVE — `check_envelope` only trips AFTER a fresh +signal sample crosses 7.0, which is necessarily after the load has already +spiked (the incident's own trip landed 0.43s after the damage was done). +The concurrency cap is PROACTIVE — it refuses to even START a dispatch once +its own slots are all held, so the host is never driven past a bounded +concurrency level by Stage E's own dispatches in the first place. Both stay +in place; neither supersedes the other. + +**Mechanism**: Postgres session-scoped advisory locks +(`cardpicker.stage_e_concurrency` — see that module's own docstring for the +full comparison against a cache-based counter and a dedicated django-q +queue, both rejected, and why advisory locks' automatic release-on- +connection-death was the deciding factor over a DB-row counter). No new +migration, no new infrastructure. A throttled dispatch returns +`status="throttled-concurrency-cap"` and writes no ledger row, the same +"halted dispatch never partially starts" convention `halted-open-trip`/ +`halted-new-trip` already established. + +**Runbook implication**: `STAGE_E_MAX_CONCURRENT_DISPATCHES` is the first +tuning knob to raise once real shakedown data shows headroom below the +7-core ceiling — raise it gradually and watch the host-load bar, never +guess a large value up front. Do **not** attempt to defeat a persistent +run of `throttled-concurrency-cap` outcomes by raising this value past what +the envelope's own load bar tolerates — a cap that's too high just moves +the failure back to the reactive host-load trip this change was built to +avoid triggering in the first place. + ### Observability: the streaming-run ledger convention Every micro-batch — from either trigger — writes one `PilotRunLedger` row: diff --git a/docs/upstreaming/extractable-primitives.md b/docs/upstreaming/extractable-primitives.md index 9f651694a..240f9d9ad 100644 --- a/docs/upstreaming/extractable-primitives.md +++ b/docs/upstreaming/extractable-primitives.md @@ -124,6 +124,7 @@ coupling to the vote system is. | Back-face name lookup (issue #199) | `MPCAutofill/cardpicker/printing_metadata_import.py` (`get_back_face_names`, `is_back_face`, `DOUBLE_FACED_LAYOUTS`) | Deterministic name → "is this a known DFC back face" lookup from Scryfall's on-disk `card_faces` bulk data, no network fetch | upstream, proxies-at-home | entangled-with-CanonicalPrinting (colocation) | — | | Self-recording, forced-dry-run-gated command lifecycle (issue #362) | `MPCAutofill/cardpicker/pilot_run_lifecycle.py` (`resilient_terminal_output`, `enforce_dry_run_precondition`, `add_dry_run_guard_arguments`, `scope_hash`, `initial_counters`, `merge_counters`) | Generic pattern for a long-running write management command: a RUNNING→COMPLETED/FAILED audit-row lifecycle with JSON counters, a broken-pipe-safe terminal-output wrapper, and a forced-dry-run precondition gate refusing `--write`/`--apply` without a matching recent dry-run | upstream, proxies-at-home (any Django project with long-running write management commands) | entangled-with-vote-consensus (colocation) - the one model this file depends on, `PilotRunLedger`, lives in `cardpicker/models.py` alongside the vote system, even though this file itself imports nothing from `vote_consensus`/`printing_consensus`/`tag_consensus`/`artist_consensus`/auth directly | — | | Deterministic snapshot-test sequencing (factory-sequence pinning), 2026-07-23 | `MPCAutofill/cardpicker/tests/test_views.py` (`_pin_shared_factory_sequences`) | `factory_boy` `Sequence` counters are process-global for a whole pytest run; a snapshot assertion embedding a sequence-derived value (e.g. an autogenerated `"Artist N"` name) implicitly depends on total call count up to that point in collection order. Rather than every _other_ test module that merely uses the shared factories protecting the one module that asserts on their exact values (the old, repeatedly-forgotten convention — see `docs/troubleshooting.md`'s "5-6 unrelated test snapshots break" entry), the snapshot-owning module pins the shared factories to a fixed baseline (`Factory.reset_sequence(0, force=True)`) before every one of its own tests, making its output self-determined regardless of suite composition, collection order, or how many other tests ran first | upstream, proxies-at-home (any `factory_boy` + snapshot-testing pairing, or any suite with process-global ID/name generators) | CLEAN (pattern-level — the technique itself imports nothing fork-specific; its current call site is `test_views.py`, which does assert some fork-only fields elsewhere in the same file, so this is a pattern to replicate in a fresh file, not a file to lift wholesale — see note) | — | +| Postgres advisory-lock concurrency cap, 2026-07-24 | `MPCAutofill/cardpicker/stage_e_concurrency.py` (`try_acquire_dispatch_slot`, `_try_acquire_slot`, `_release_slot`) | Caps how many callers run a given code path concurrently, across any number of separate OS processes sharing one Postgres database, via `pg_try_advisory_lock`/`pg_advisory_unlock` slot cycling - no new migration, no cache/broker dependency, and crash-safe for free (a killed process's session-scoped lock auto-releases, unlike a DB-row counter) | upstream, proxies-at-home, federation peers (any multi-process Django+Postgres app needing a cross-process concurrency cap) | CLEAN (zero `cardpicker.*` imports at all - only `django.conf.settings`/`django.db.connection` and stdlib; its own module docstring is itself the generic write-up of the technique) | — | ## Docs tooling & federation