diff --git a/MPCAutofill/cardpicker/local_calculate_verdicts.py b/MPCAutofill/cardpicker/local_calculate_verdicts.py index 841dd9d07..5dfb6beac 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,108 @@ 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 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 + 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`. 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 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 + `_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 +1157,21 @@ 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. 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, 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) - 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 +1347,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 +1475,19 @@ 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). 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, 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) - result.votes_written = len(votes_batch) + result.votes_written = len(new_votes) return result @@ -1553,6 +1681,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..969c413c7 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,40 @@ 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, 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 + 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 +467,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..8286b2b28 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,97 @@ 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 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) + + 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 +699,69 @@ 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 + + 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 @@ -1548,6 +1703,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..3f7f8b5f7 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,57 @@ def _fetch_crashes_on_second_card(card, dpi=None): assert resumed_ledger.counters["stage_c_completed"] == 2 +class TestConcurrentDispatchVoteCollision: + """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, 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( + 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..406f238da 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,81 @@ 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: **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 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)