Add Stage E streaming dispatch concurrency cap

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)

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}"
         )

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)