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/management/commands/stream_backstop_sweep.py b/MPCAutofill/cardpicker/management/commands/stream_backstop_sweep.py index 020c8db2e..1a5146640 100644 --- a/MPCAutofill/cardpicker/management/commands/stream_backstop_sweep.py +++ b/MPCAutofill/cardpicker/management/commands/stream_backstop_sweep.py @@ -24,8 +24,22 @@ does not fill from this backlog, see its own docstring, so the sweep covers it here instead). Stops when both backlogs come back empty, when the envelope trips ("halted-new-trip"/"halted-open-trip" - the sweep does not retry past a halt; the next scheduled sweep invocation picks up where this one -stopped, exactly like a re-invoked BULK command would), or when `--max-batches` is reached (a safety -bound for a single invocation, not a design limit). +stopped, exactly like a re-invoked BULK command would), when every concurrency-cap slot is already +held ("throttled-concurrency-cap" - same "stop, don't retry, the next scheduled sweep invocation +picks up where this one stopped" posture as a halt, since looping here would just re-sample an +already-saturated cap with no backoff), or when `--max-batches` is reached (a safety bound for a +single invocation, not a design limit). + +THE SWEEP IS THE ONLY RECOVERY PATH FOR A THROTTLED EVENT DISPATCH: `Q_CLUSTER["max_attempts"] = 1` +(`MPCAutofill/MPCAutofill/settings.py`) means an event-driven `async_task` that returns +`DispatchOutcome(status="throttled-concurrency-cap")` is recorded SUCCESSFUL by django-q2 - it never +retries, and the touched card is silently dropped until some future sweep picks it up via the +backlog queries above. That makes it load-bearing that this command itself never treats +"throttled-concurrency-cap" as ordinary work: looping past it burns a `current_trip()` query, an +envelope sample, and (via the Stage C backlog fallback) a `_select_micro_batch` anti-join over the +whole cards table, per iteration, up to `--max-batches` times, precisely when the host is already at +its concurrency ceiling - the worst possible moment for a hot, backoff-free loop. See +`stage_e_concurrency.py`'s own module docstring for why the cap is proactive rather than reactive. IDEMPOTENCE: every batch this command dispatches goes through the exact same `dispatch_micro_batch` conveyor the event trigger uses - Stage C's own resume filter and Stage D's own @@ -49,6 +63,7 @@ DEFAULT_MAX_BATCHES = 1000 _HALT_STATUSES = ("halted-open-trip", "halted-new-trip") +_THROTTLED_STATUS = "throttled-concurrency-cap" def _next_stage_d_backlog_ids(batch_size: int) -> list[int]: @@ -105,6 +120,7 @@ def handle(self, *args: Any, **options: Any) -> None: total_stage_c = 0 total_stage_d_votes = 0 halted_status = None + stopped_reason = None for batch_num in range(max_batches): outcome = dispatch_micro_batch( @@ -117,6 +133,17 @@ def handle(self, *args: Any, **options: Any) -> None: halted_status = outcome.status self.stdout.write(f"Envelope halt ({outcome.status}, trip_id={outcome.trip_id}) - stopping sweep.") break + if outcome.status == _THROTTLED_STATUS: + # PROACTIVE cap saturation, not a halt (stage_e_concurrency.py's own docstring) - a + # STOP condition all the same: looping here would just re-sample an already-saturated + # cap with no backoff (module docstring's own "hot, backoff-free loop" concern), and + # the throttled attempt did no work, so it must not count toward batches_dispatched. + stopped_reason = outcome.status + self.stdout.write( + "sweep stopped: dispatch slots saturated (throttled-concurrency-cap); " + "next scheduled sweep will resume." + ) + break if outcome.status == "empty": # Backlog (a) exhausted for this pass - try backlog (b) before concluding the whole @@ -135,6 +162,13 @@ def handle(self, *args: Any, **options: Any) -> None: halted_status = outcome.status self.stdout.write(f"Envelope halt ({outcome.status}, trip_id={outcome.trip_id}) - stopping sweep.") break + if outcome.status == _THROTTLED_STATUS: + stopped_reason = outcome.status + self.stdout.write( + "sweep stopped: dispatch slots saturated (throttled-concurrency-cap); " + "next scheduled sweep will resume." + ) + break if outcome.status == "empty": self.stdout.write("Backlog exhausted - nothing left to dispatch.") break @@ -147,5 +181,6 @@ def handle(self, *args: Any, **options: Any) -> None: self.stdout.write( f"DONE batches_dispatched={batches_dispatched} stage_c_completed={total_stage_c} " - f"stage_d_votes_or_routes={total_stage_d_votes} halted={halted_status}" + f"stage_d_votes_or_routes={total_stage_d_votes} halted={halted_status} " + f"stopped_reason={stopped_reason}" ) diff --git a/MPCAutofill/cardpicker/stage_e_concurrency.py b/MPCAutofill/cardpicker/stage_e_concurrency.py new file mode 100644 index 000000000..5dfa33a34 --- /dev/null +++ b/MPCAutofill/cardpicker/stage_e_concurrency.py @@ -0,0 +1,165 @@ +""" +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: + """ + Floors at 1 - `settings.STAGE_E_MAX_CONCURRENT_DISPATCHES` set to 0 or negative (a plausible ops + typo, e.g. meant to disable Stage E via a different flag entirely) would otherwise make + `_try_acquire_slot`'s own `range(_slot_count())` empty, so every call returns `None` - dispatches + throttle FOREVER with no error, no envelope trip, and no visible signal beyond an ever-growing + Stage C/D backlog. A `logger.warning` on clamp makes that typo visible instead of silent. + """ + configured = getattr(settings, "STAGE_E_MAX_CONCURRENT_DISPATCHES", 2) + if configured < 1: + logger.warning( + "stage_e_concurrency: STAGE_E_MAX_CONCURRENT_DISPATCHES=%s is below the floor of 1 - " + "clamping to 1 slot instead of throttling every dispatch forever", + configured, + ) + return 1 + return configured + + +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..f0d7953ad --- /dev/null +++ b/MPCAutofill/cardpicker/tests/test_stage_e_concurrency.py @@ -0,0 +1,347 @@ +""" +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 + + @override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=0) + def test_a_zero_configured_cap_floors_to_one_and_warns(self, db: Any, caplog: Any) -> None: + """STAGE_E_MAX_CONCURRENT_DISPATCHES=0 (a plausible ops typo) must not silently throttle + every dispatch forever - `_try_acquire_slot`'s own `range(_slot_count())` would otherwise be + empty, so no slot could ever be acquired, with no error and no envelope trip to surface it. + The floor makes at least one slot available; the warning makes the typo itself visible.""" + with caplog.at_level("WARNING", logger="cardpicker.stage_e_concurrency"): + assert stage_e_concurrency._slot_count() == 1 + assert "STAGE_E_MAX_CONCURRENT_DISPATCHES" in caplog.text + assert "clamping to 1" in caplog.text + + @override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=-3) + def test_a_negative_configured_cap_also_floors_to_one_and_warns(self, db: Any, caplog: Any) -> None: + with caplog.at_level("WARNING", logger="cardpicker.stage_e_concurrency"): + assert stage_e_concurrency._slot_count() == 1 + assert "clamping to 1" in caplog.text + + +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/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py index 3f7f8b5f7..4c7ace4f1 100644 --- a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py @@ -16,10 +16,12 @@ import io from typing import Any +import psycopg2 import pytest from PIL import Image from django.core.management import call_command +from django.db import connection from django.test import override_settings from cardpicker import stage_e_dispatch @@ -43,6 +45,7 @@ check_envelope, current_trip, ) +from cardpicker.stage_e_concurrency import _LOCK_NAMESPACE from cardpicker.stage_e_dispatch import ( _FetchOutcomeWindow, _select_micro_batch, @@ -350,6 +353,71 @@ def _fail_if_called(card, dpi=None): assert vote.printing_id == printing.pk +class TestConcurrencyCapIntegration: + """The concurrency-cap companion change (`cardpicker.stage_e_concurrency`), exercised through + the REAL, post-#448 `dispatch_micro_batch` body - not just the module-level unit coverage in + `test_stage_e_concurrency.py`. A raw, independent `psycopg2` connection holds every available + slot (standing in for another concurrent dispatch on a separate django-q worker process, the + same "genuinely separate session" discipline `test_stage_e_concurrency.py`'s own module + docstring establishes - Postgres session-level advisory locks are re-entrant within one + session, so simulating a second dispatcher via Django's own connection would be a false test).""" + + def _raw_connection_holding_every_slot(self, cap: int) -> "psycopg2.extensions.connection": + connection.ensure_connection() + raw = psycopg2.connect(**connection.get_connection_params()) + raw.autocommit = True + with raw.cursor() as cursor: + for slot in range(cap): + cursor.execute("SELECT pg_try_advisory_lock(%s, %s)", [_LOCK_NAMESPACE, slot]) + (acquired,) = cursor.fetchone() + assert acquired is True, f"test setup failed to claim slot {slot}" + return raw + + @STREAMING_ON + @override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=1) + def test_dispatch_is_throttled_when_the_only_slot_is_already_held( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + card = CardFactory(name="Some Card", content_phash=42) + CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + + def _fail_if_called(card, dpi=None): + raise AssertionError("Stage C must never run once the concurrency cap has throttled") + + _install_stage_c_stub(monkeypatch, fetch_result=_fail_if_called) + + raw = self._raw_connection_holding_every_slot(cap=1) + try: + outcome = dispatch_micro_batch(card_ids=[card.pk]) + finally: + raw.close() # auto-releases the advisory lock, same crash-safety property tested above + + assert outcome.status == "throttled-concurrency-cap" + # a throttled dispatch never partially starts, matching halted-open-trip/halted-new-trip's + # own convention - no ledger row, no evidence, no vote. + assert PilotRunLedger.objects.count() == 0 + assert ImageEvidence.objects.count() == 0 + assert CardPrintingTag.objects.count() == 0 + + @STREAMING_ON + @override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=1) + def test_dispatch_proceeds_normally_once_the_slot_is_released( + self, db: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + card = CardFactory(name="Some Card", content_phash=42) + printing = CanonicalCardFactory(name="Some Card", expansion__code="mom", collector_number="158") + _full_evidence(card, collector_line_set_code="mom", collector_line_collector_number="158") + + raw = self._raw_connection_holding_every_slot(cap=1) + raw.close() # released before dispatching - the cap must not still be considered "held" + + outcome = dispatch_micro_batch(card_ids=[card.pk]) + + assert outcome.status == "completed" + vote = CardPrintingTag.objects.get(card=card, anonymous_id=JOIN_KEY_ANONYMOUS_ID) + assert vote.printing_id == printing.pk + + class TestGoogleLockoutMidBatch: @STREAMING_ON def test_lockout_stops_stage_c_trips_the_envelope_and_refuses_the_next_dispatch( @@ -517,3 +585,44 @@ def test_sweep_stops_on_an_envelope_halt(self, db: Any, monkeypatch: pytest.Monk assert EnvelopeTrip.objects.filter(bar=EnvelopeTrip.Bar.HOST_LOAD).count() == 1 assert PilotRunLedger.objects.count() == 0 # halted before any batch ledger row was written + + @STREAMING_ON + @override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=1) + def test_sweep_stops_on_a_throttled_concurrency_cap_without_looping( + self, db: Any, capsys: pytest.CaptureFixture + ) -> None: + """Tron gate defect fix: pre-fix, "throttled-concurrency-cap" matched neither + `_HALT_STATUSES` nor "empty" and fell through to `batches_dispatched += 1` with an immediate + re-entry into `dispatch_micro_batch` - a hot, backoff-free loop up to `--max-batches`, + precisely when the host is already saturated, reporting a success-shaped + `batches_dispatched=N` for a sweep that did nothing. Mirrors + `TestConcurrencyCapIntegration`'s own "genuinely separate session" discipline (that class's + own docstring) - a raw, independent `psycopg2` connection holds the only slot, standing in + for another concurrent dispatch (a django-q worker, or the event trigger racing this same + sweep) - reusing the SAME session for both would be a false test (re-entrant advisory + locks, this file's own `stage_e_concurrency` test module docstring).""" + CardFactory(name="Some Card", content_phash=42) + + connection.ensure_connection() + raw = psycopg2.connect(**connection.get_connection_params()) + raw.autocommit = True + with raw.cursor() as cursor: + cursor.execute("SELECT pg_try_advisory_lock(%s, %s)", [_LOCK_NAMESPACE, 0]) + (acquired,) = cursor.fetchone() + assert acquired is True, "test setup failed to claim the only slot" + try: + call_command("stream_backstop_sweep", "--max-batches", "5") + finally: + raw.close() # auto-releases the advisory lock, same crash-safety property tested above + + output = capsys.readouterr().out + assert "sweep stopped: dispatch slots saturated (throttled-concurrency-cap)" in output + # exactly one occurrence - proves the loop actually STOPPED on the first throttled outcome + # rather than re-entering dispatch_micro_batch up to --max-batches=5 with no backoff. + assert output.count("sweep stopped: dispatch slots saturated") == 1 + assert "batches_dispatched=0" in output + assert "stopped_reason=throttled-concurrency-cap" in output + # no ledger row - the throttled attempt never started real work, matching a halt's own + # convention (TestConcurrencyCapIntegration's own assertion for dispatch_micro_batch + # directly). + assert PilotRunLedger.objects.count() == 0 diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index 406f238da..cd31cf5dd 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,72 @@ 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. `_slot_count()` floors +`STAGE_E_MAX_CONCURRENT_DISPATCHES` at `1` — a misconfigured `0` or negative +value clamps to `1` slot (with a `logger.warning`) rather than silently +throttling every dispatch forever with no error and no envelope trip to +surface it. + +**Event-dispatch drop semantics**: `Q_CLUSTER["max_attempts"] = 1` +(`MPCAutofill/MPCAutofill/settings.py`) means an event-driven `async_task` that returns +`throttled-concurrency-cap` is recorded SUCCESSFUL by django-q2 — it never +retries, so the touched card is silently deferred to the backstop sweep +(below) rather than lost outright. This is by design, not a gap: the sweep +is the one path that re-tries a card the event trigger dropped this way. + +**Backstop sweep behavior on throttle**: `stream_backstop_sweep` treats +`throttled-concurrency-cap` as a STOP condition, exactly like an envelope +halt — it does not count the throttled attempt toward `batches_dispatched` +and does not loop back into `dispatch_micro_batch` with no backoff; the +next scheduled sweep invocation picks up where this one stopped. The +sweep's summary output reports `stopped_reason=throttled-concurrency-cap` +when this happens, so an operator can see the sweep did nothing this run +rather than reading a `batches_dispatched=0`-with-no-explanation line as +"backlog was just empty." + +**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