From 0540597838f7010faa3a8bcec68acf846a812d17 Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Sat, 25 Jul 2026 00:57:23 +0000 Subject: [PATCH] Fix Stage E concurrency cap: hold advisory lock on dedicated connection, not django.db.connection --- MPCAutofill/cardpicker/admin.py | 13 ++ .../migrations/0081_stageethrottlecounter.py | 22 ++ MPCAutofill/cardpicker/models.py | 47 ++++ MPCAutofill/cardpicker/stage_e_concurrency.py | 168 +++++++++++--- MPCAutofill/cardpicker/stage_e_dispatch.py | 27 ++- .../tests/test_stage_e_concurrency.py | 219 ++++++++++++++---- .../cardpicker/tests/test_stage_e_dispatch.py | 6 + docs/features/stage-e-operations.md | 94 ++++++-- docs/lessons.md | 23 ++ docs/troubleshooting.md | 54 +++++ docs/upstreaming/extractable-primitives.md | 32 +-- 11 files changed, 591 insertions(+), 114 deletions(-) create mode 100644 MPCAutofill/cardpicker/migrations/0081_stageethrottlecounter.py diff --git a/MPCAutofill/cardpicker/admin.py b/MPCAutofill/cardpicker/admin.py index b8b6950f9..b57623809 100644 --- a/MPCAutofill/cardpicker/admin.py +++ b/MPCAutofill/cardpicker/admin.py @@ -26,6 +26,7 @@ ProjectMember, SavedDeck, Source, + StageEThrottleCounter, Tag, TagAliasSuggestion, TagSuggestionStatus, @@ -301,6 +302,18 @@ class AdminEnvelopeTrip(admin.ModelAdmin[EnvelopeTrip]): ) +@admin.register(StageEThrottleCounter) +class AdminStageEThrottleCounter(admin.ModelAdmin[StageEThrottleCounter]): + # Observability signal for `cardpicker.stage_e_concurrency`'s "throttled-concurrency-cap" + # outcome (StageEThrottleCounter's own docstring for the full "why a singleton counter, not a + # per-event row" reasoning) - the runbook's "tune STAGE_E_MAX_CONCURRENT_DISPATCHES against + # the observed throttle rate" instruction (docs/features/stage-e-operations.md) points an + # operator here. Always exactly one row. Read-only, same rationale as AdminEnvelopeTrip above + # - `StageEThrottleCounter.record()` is the only code path permitted to advance `count`. + list_display = ("singleton_key", "count", "last_throttled_at") + readonly_fields = ("singleton_key", "count", "last_throttled_at") + + @admin.register(LandsAmbiguousResidue) class AdminLandsAmbiguousResidue(admin.ModelAdmin[LandsAmbiguousResidue]): list_display = ("card", "artist_name", "run_id", "candidate_pks", "created_at") diff --git a/MPCAutofill/cardpicker/migrations/0081_stageethrottlecounter.py b/MPCAutofill/cardpicker/migrations/0081_stageethrottlecounter.py new file mode 100644 index 000000000..1cdbf0ef3 --- /dev/null +++ b/MPCAutofill/cardpicker/migrations/0081_stageethrottlecounter.py @@ -0,0 +1,22 @@ +# Generated by Django 4.2.30 on 2026-07-25 00:37 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("cardpicker", "0080_questionfeedservedlog"), + ] + + operations = [ + migrations.CreateModel( + name="StageEThrottleCounter", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("singleton_key", models.PositiveSmallIntegerField(default=1, unique=True)), + ("count", models.PositiveIntegerField(default=0)), + ("last_throttled_at", models.DateTimeField(blank=True, null=True)), + ], + ), + ] diff --git a/MPCAutofill/cardpicker/models.py b/MPCAutofill/cardpicker/models.py index 94729f5b0..491fbeeba 100755 --- a/MPCAutofill/cardpicker/models.py +++ b/MPCAutofill/cardpicker/models.py @@ -1308,6 +1308,53 @@ def __str__(self) -> str: return f"[{state}] {self.bar} trip_id={self.trip_id}" +class StageEThrottleCounter(models.Model): + """ + Stage E Phase 2 companion - a SINGLETON, always-exactly-one-row atomic counter for + `cardpicker.stage_e_concurrency`'s "throttled-concurrency-cap" outcome (Tron gate round-1 + "COMPANION" review, observability anomaly 4, 2026-07-25: a throttled dispatch wrote no ledger + row and emitted only a `logger.info` line, so the runbook's own "tune + STAGE_E_MAX_CONCURRENT_DISPATCHES against the observed throttle rate" + (docs/features/stage-e-operations.md) instruction had nothing queryable to check against). + + Deliberately NOT a `PilotRunLedger` row and NOT `EnvelopeTrip`-shaped (one row per event) - + see `EnvelopeTrip`'s own docstring for the same "different shape needs a different table" + reasoning this mirrors. A per-throttle-event row would WRITE-AMPLIFY under exactly the + failure shape this whole feature exists to guard against: a burst of concurrent dispatches + hitting an exhausted cap can throttle far more often than any dispatch ever completes - + unlike `PilotRunLedger`'s one-row-per-invocation cadence or `EnvelopeTrip`'s one-row-per- + breach cadence, both of which stay bounded by how often real work actually runs. + + Exactly one row, ever - `singleton_key` is `unique=True` so a first-ever-throttle race + between two worker processes resolves to a single winning row via `record()`'s own + `get_or_create` fallback (Django/Postgres serialize the losing INSERT into an + `IntegrityError`, which `get_or_create` already retries as a fetch). `count` is only ever + advanced via an atomic `F("count") + 1` UPDATE - race-safe under Postgres row-level locking + even with many worker processes throttling at once, never a Python-side read-modify-write + (which would silently lose increments under that exact concurrency). + """ + + singleton_key = models.PositiveSmallIntegerField(default=1, unique=True) + count = models.PositiveIntegerField(default=0) + last_throttled_at = models.DateTimeField(null=True, blank=True) + + def __str__(self) -> str: + return f"Stage E throttle count={self.count} (last {self.last_throttled_at})" + + @classmethod + def record(cls) -> None: + """Called once per `"throttled-concurrency-cap"` dispatch outcome + (`cardpicker.stage_e_dispatch.dispatch_micro_batch`). Prefers the atomic `UPDATE` path + (the common case, after the singleton row exists); only falls back to `get_or_create` the + first time this counter is ever touched on a given deployment.""" + from django.db.models import F + from django.utils import timezone + + updated = cls.objects.filter(singleton_key=1).update(count=F("count") + 1, last_throttled_at=timezone.now()) + if not updated: + cls.objects.get_or_create(singleton_key=1, defaults={"count": 1, "last_throttled_at": timezone.now()}) + + class CardScanLog(models.Model): """ Persists ABSTENTION evidence exactly like `AbstractWeightedVote` subclasses persist assent diff --git a/MPCAutofill/cardpicker/stage_e_concurrency.py b/MPCAutofill/cardpicker/stage_e_concurrency.py index 5dfa33a34..95055fb58 100644 --- a/MPCAutofill/cardpicker/stage_e_concurrency.py +++ b/MPCAutofill/cardpicker/stage_e_concurrency.py @@ -55,23 +55,72 @@ `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. +CONNECTION-LIFECYCLE CONTRACT (2026-07-25 REWRITE - the ORIGINAL version of this section was WRONG, +see below): 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. This module therefore holds its lock on a +DEDICATED `psycopg2` connection that IT ALONE owns for the duration of one `try_acquire_dispatch_ +slot()` call - never `django.db.connection` (Django's shared per-thread connection proxy). + +WHAT WENT WRONG IN PRODUCTION (2026-07-25T00:25Z shakedown, `envtrip-20260725T002504-73e1eb6d`, +`{'ceiling': 7.0, 'load_avg': 11.4013671875}`): the FIRST version of this module acquired and +released its advisory lock on `django.db.connection`, on the strength of a claim - made by direct +inspection of `django_q.worker` - that a single `dispatch_micro_batch` call always runs as one +uninterrupted segment on one connection. That claim was checked against the wrong code path. +`django_q.worker.close_old_django_connections()` (called only between tasks) was never the risk; +`django_q.brokers.orm.ORM.get_connection()` was. That method calls `django.db.close_old_connections()` +UNCONDITIONALLY whenever it's invoked outside an atomic block (`if transaction.get_autocommit(...)`), +and this project's `DATABASES["default"]` has no `CONN_MAX_AGE` override, so Django's own default (0) +applies - `BaseDatabaseWrapper.close_if_unusable_or_obsolete` treats `close_at = now + 0` as already +expired, so `close_old_connections()` closes `django.db.connection`'s underlying connection THE FIRST +TIME anything calls it, unconditionally, not just after some elapsed age. And something DOES call it +from squarely inside this module's own locked region: `cardpicker.stage_e_signals`'s `post_save` +receivers fire during Stage C's `persist_evidence` (INSIDE `dispatch_micro_batch`'s `with +try_acquire_dispatch_slot()` block) and call `django_q.tasks.async_task(...)`, which synchronously +calls `broker.enqueue(pack)` on the SAME thread/process - for the installed ORM broker, `enqueue` +calls `get_connection()` first, which closes the connection right then. A closed connection +auto-releases every advisory lock its session held (the same crash-safety property this module +relies on deliberately for a genuinely killed process), so the lock this module thought it was still +holding was gone mid-dispatch - EVERY worker then found EVERY slot "free", explaining the shakedown's +zero `throttled-concurrency-cap` outcomes despite eight concurrent dispatches, and the eight +`pg_advisory_unlock reported slot N was not held by this connection` warnings this module's own +defensive guard (`_release_slot`) logged when the final unlock ran against a since-reconnected +`django.db.connection` (Django transparently reopens a closed connection on next use, but that is a +NEW Postgres backend session - the unlock call executes there, not on the session that held the +lock). + +THE FIX: hold the lock on a connection `stage_e_concurrency` opens for itself +(`psycopg2.connect(**connection.get_connection_params())`, `autocommit=True` so it's never left +idle-in-transaction and lock lifetime is tied to the session rather than any transaction this module +never starts) and NOTHING ELSE ever touches - not django-q's broker, not `close_old_connections`, not +any other code in the process. That connection lives for exactly one `try_acquire_dispatch_slot()` +call: opened before the acquire, held across the whole `yield` (i.e. across all of Stage C/D, exactly +where the lock needs to survive), explicitly `pg_advisory_unlock`'d AND `close()`'d in a `finally` - +the explicit unlock is the fast/clean path, the `close()` is the crash-safety backstop that fires even +if the explicit unlock itself somehow fails (Postgres auto-releases every advisory lock a closing +session held, the exact mechanism this module's own crash-safety design already leans on for a +genuinely killed process - see the DB-row-counter rejection above). The "not held by this connection" +warning guard (`_release_slot`) is KEPT UNCHANGED, not removed - it is what caught this bug in the +first place (the 8 log lines in the incident evidence), and it should now never fire again; a report +of it firing again is a signal something ELSE has broken this contract. + +FAIL-CLOSED ON CONNECTION-CREATION FAILURE: if opening the dedicated connection itself raises (DB +unreachable, connection pool/limit exhausted, etc.), `try_acquire_dispatch_slot()` yields `None` - +the same "throttled, do no work this call" signal an exhausted cap already produces - rather than +letting the dispatch proceed uncapped. An uncapped dispatch is exactly the failure this module exists +to prevent; a connection-creation failure is precisely the moment this module is LEAST able to +guarantee the cap holds, so it is also the moment to be most conservative. The cost is a spuriously +throttled micro-batch (picked up again by the next event or the backstop sweep, matching the existing +"throttled dispatch defers to the sweep" convention already documented in +docs/features/stage-e-operations.md) - a strictly cheaper failure mode than a repeat of this +incident. """ import logging from contextlib import contextmanager from typing import Iterator, Optional +import psycopg2 + from django.conf import settings from django.db import connection @@ -105,17 +154,42 @@ def _slot_count() -> int: return configured -def _try_acquire_slot() -> Optional[int]: +def _open_dedicated_connection() -> "psycopg2.extensions.connection": + """ + Opens a NEW `psycopg2` connection this module alone owns for the lifetime of one + `try_acquire_dispatch_slot()` call - see the module docstring's "CONNECTION-LIFECYCLE CONTRACT" + section for the full incident this replaces `django.db.connection` to fix (django-q2's ORM + broker calls `django.db.close_old_connections()` - which, with `CONN_MAX_AGE` unset, closes the + connection UNCONDITIONALLY - from squarely inside this module's own locked region whenever a + follow-on dispatch is enqueued via `async_task`). + + Connection PARAMETERS are read from Django's own `django.db.connection` + (`connection.get_connection_params()`, calling `connection.ensure_connection()` first since + `get_connection_params()` alone doesn't establish one) - not a separately-guessed host/port/ + dbname - so this always targets whatever database Django itself is actually configured against, + including test-run database-name prefixing. `autocommit=True`: advisory locks are independent of + transactions, and leaving this connection in Postgres's default (non-autocommit) mode would hold + an idle-in-transaction session open for the whole dispatch for no reason, and would tie this + lock's release semantics to a commit/rollback this module never issues. + """ + connection.ensure_connection() + params = connection.get_connection_params() + raw = psycopg2.connect(**params) + raw.autocommit = True + return raw + + +def _try_acquire_slot(conn: "psycopg2.extensions.connection") -> 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 + Tries every slot index in `[0, _slot_count())` in ascending order, on the given (dedicated) + connection, 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: + with conn.cursor() as cursor: for slot in range(_slot_count()): cursor.execute("SELECT pg_try_advisory_lock(%s, %s)", [_LOCK_NAMESPACE, slot]) (acquired,) = cursor.fetchone() @@ -124,16 +198,17 @@ def _try_acquire_slot() -> Optional[int]: return None -def _release_slot(slot: int) -> None: +def _release_slot(conn: "psycopg2.extensions.connection", 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 + SAME dedicated 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: + wasn't held (`pg_advisory_unlock` returns `false`, never an error, for that case) - KEPT + UNCHANGED from the pre-fix version deliberately: this is the exact guard that caught the + production incident (2026-07-25T00:25Z shakedown, 8 occurrences) this rewrite fixes, and it + should now never fire again - a fresh report of it firing is a signal that something else has + broken the dedicated-connection contract, not something to remove because "the incident is + fixed now".""" + with conn.cursor() as cursor: cursor.execute("SELECT pg_advisory_unlock(%s, %s)", [_LOCK_NAMESPACE, slot]) (released,) = cursor.fetchone() if not released: @@ -149,17 +224,36 @@ 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). + held (or the dedicated connection itself couldn't be opened - see "FAIL-CLOSED ON + CONNECTION-CREATION FAILURE" in the module docstring) - 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. + + Opens a DEDICATED connection for this call (`_open_dedicated_connection`) and ALWAYS closes it on + exit, in a `finally`, whether the `with` block raises or not, and whether or not a slot was ever + acquired - a dispatch that crashes mid-batch must not leak either the slot or the connection. + Explicitly `pg_advisory_unlock`s before closing (the fast/clean path, and what makes + `_release_slot`'s "not held" warning guard meaningful); the `close()` itself is the crash-safety + backstop - Postgres auto-releases every advisory lock a closing session held, so even if the + explicit unlock somehow failed to run, closing the connection still frees the slot. """ - slot = _try_acquire_slot() try: + conn = _open_dedicated_connection() + except Exception: + logger.exception( + "stage_e_concurrency: failed to open the dedicated advisory-lock connection - " + "failing CLOSED (treating this dispatch as throttled) rather than proceeding uncapped" + ) + yield None + return + + slot: Optional[int] = None + try: + slot = _try_acquire_slot(conn) yield slot finally: - if slot is not None: - _release_slot(slot) + try: + if slot is not None: + _release_slot(conn, slot) + finally: + conn.close() diff --git a/MPCAutofill/cardpicker/stage_e_dispatch.py b/MPCAutofill/cardpicker/stage_e_dispatch.py index a0dbba4ce..934ea8963 100644 --- a/MPCAutofill/cardpicker/stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/stage_e_dispatch.py @@ -82,7 +82,13 @@ run_join_key_calculator, run_slow_path_calculator, ) -from cardpicker.models import Card, EnvelopeTrip, ImageEvidence, PilotRunLedger +from cardpicker.models import ( + Card, + EnvelopeTrip, + ImageEvidence, + PilotRunLedger, + StageEThrottleCounter, +) from cardpicker.operating_envelope import ( FETCH_FAILURE_WINDOW, EnvelopeSignals, @@ -164,7 +170,12 @@ class DispatchOutcome: `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. + exist). No `PilotRunLedger` row is written, matching the other halted statuses - but + (2026-07-25, Tron gate observability anomaly 4) `StageEThrottleCounter.record()` DOES + advance a singleton, always-exactly-one-row counter (`cardpicker.models. + StageEThrottleCounter`'s own docstring has the full "why a counter, not a per-event row" + reasoning) so the runbook's own "tune STAGE_E_MAX_CONCURRENT_DISPATCHES against the + observed throttle rate" instruction has something queryable to check. - "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 @@ -396,8 +407,9 @@ def dispatch_micro_batch( 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. + concurrency-cap check's own advisory-lock round trip, plus - 2026-07-25 - a throttled outcome's + `StageEThrottleCounter.record()` call, a single-row atomic counter update, never a growing + 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) @@ -448,6 +460,13 @@ def dispatch_micro_batch( "Stage E dispatch throttled - all %s concurrency-cap slots already held", getattr(settings, "STAGE_E_MAX_CONCURRENT_DISPATCHES", 2), ) + # Observability signal (Tron gate anomaly 4, 2026-07-25): a throttled dispatch writes + # no PilotRunLedger row (see the comment above this `with` block for why), so this + # singleton counter (StageEThrottleCounter's own docstring has the full "why a + # counter, not a per-event row" reasoning) is the ONLY durable, queryable record that + # throttling happened - the runbook's "tune STAGE_E_MAX_CONCURRENT_DISPATCHES against + # the observed throttle rate" instruction has nothing else to check against. + StageEThrottleCounter.record() 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" diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_concurrency.py b/MPCAutofill/cardpicker/tests/test_stage_e_concurrency.py index f0d7953ad..02d9c2e15 100644 --- a/MPCAutofill/cardpicker/tests/test_stage_e_concurrency.py +++ b/MPCAutofill/cardpicker/tests/test_stage_e_concurrency.py @@ -8,14 +8,18 @@ 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. +reference count that then needs a matching number of unlocks. Calling this module's own +`_try_acquire_slot`/`_release_slot` TWICE on the SAME 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 gives each one its OWN genuinely SEPARATE `psycopg2` +connection (`_raw_connection`, `_try_acquire_on_connection`/`_release_on_connection`) - this is also +the production-faithful choice, since two real concurrent dispatches are always on two separate +django-q worker PROCESSES (separate connections), and (2026-07-25 rewrite) the module itself now +opens its OWN dedicated connection per call rather than ever touching Django's shared one - see +`stage_e_concurrency`'s own module docstring, "CONNECTION-LIFECYCLE CONTRACT" section, for the +production incident that made "a dedicated connection, not django.db.connection" the whole point of +this rewrite. """ import threading @@ -25,7 +29,7 @@ import psycopg2 import pytest -from django.db import connection +from django.db import close_old_connections, connection from django.test import override_settings from cardpicker import stage_e_concurrency @@ -46,7 +50,10 @@ def _raw_connection() -> "psycopg2.extensions.connection": 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.""" + in Postgres's default (non-autocommit) mode would hold an idle transaction open for no reason. + This is also EXACTLY the approach `stage_e_concurrency._open_dedicated_connection` itself now + uses in production (2026-07-25 rewrite) - this test helper predates that rewrite and is the + reason the task that produced it pointed the fix here.""" connection.ensure_connection() params = connection.get_connection_params() raw = psycopg2.connect(**params) @@ -56,9 +63,9 @@ def _raw_connection() -> "psycopg2.extensions.connection": 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.""" + caller-owned connection - kept as an independent re-implementation (not just a call to + `stage_e_concurrency._try_acquire_slot`) deliberately, so these tests aren't purely tautological + against the module's own acquire loop.""" with conn.cursor() as cursor: for slot in range(cap): cursor.execute("SELECT pg_try_advisory_lock(%s, %s)", [_LOCK_NAMESPACE, slot]) @@ -73,16 +80,34 @@ def _release_on_connection(conn: "psycopg2.extensions.connection", slot: int) -> cursor.execute("SELECT pg_advisory_unlock(%s, %s)", [_LOCK_NAMESPACE, slot]) +def _slot_0_is_held_by_someone_else() -> bool: + """From a genuinely separate, freshly-opened session: `True` if slot 0 is currently held by + ANY other session (this function's own connection is closed again immediately either way, so + it never itself holds the lock afterwards).""" + raw = _raw_connection() + try: + with raw.cursor() as cursor: + cursor.execute("SELECT pg_try_advisory_lock(%s, %s)", [_LOCK_NAMESPACE, 0]) + (acquired,) = cursor.fetchone() + if acquired: + cursor.execute("SELECT pg_advisory_unlock(%s, %s)", [_LOCK_NAMESPACE, 0]) + return not acquired + finally: + raw.close() + + @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).""" + writes. Every test below is expected to release everything it acquires (raw connections are + always `close()`d in a `finally`, which auto-releases anything still held), but this fixture is + a defensive safety net for Django's OWN long-lived connection specifically - the one thing nnot + covered by a raw connection's own teardown: 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 @@ -96,54 +121,60 @@ def _release_any_leaked_locks(db: Any): 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_a = _raw_connection() raw_b = _raw_connection() try: - dispatcher_b = _try_acquire_on_connection(raw_b, cap=2) + dispatcher_a = _try_acquire_slot(raw_a) + dispatcher_b = _try_acquire_slot(raw_b) assert dispatcher_a == 0 assert dispatcher_b == 1 - _release_slot(dispatcher_a) - _release_on_connection(raw_b, dispatcher_b) + _release_slot(raw_a, dispatcher_a) + _release_slot(raw_b, dispatcher_b) finally: + raw_a.close() 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_a = _raw_connection() 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) + dispatcher_a = _try_acquire_slot(raw_a) + dispatcher_b = _try_acquire_slot(raw_b) + dispatcher_c = _try_acquire_slot(raw_c) 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) + _release_slot(raw_a, dispatcher_a) + _release_slot(raw_b, dispatcher_b) finally: + raw_a.close() 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_a = _raw_connection() raw_b = _raw_connection() raw_c = _raw_connection() try: - dispatcher_b = _try_acquire_on_connection(raw_b, cap=2) + dispatcher_a = _try_acquire_slot(raw_a) + dispatcher_b = _try_acquire_slot(raw_b) assert dispatcher_a == 0 and dispatcher_b == 1 - _release_slot(dispatcher_a) # slot 0 freed + _release_slot(raw_a, dispatcher_a) # slot 0 freed - dispatcher_c = _try_acquire_on_connection(raw_c, cap=2) + dispatcher_c = _try_acquire_slot(raw_c) 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) + _release_slot(raw_b, dispatcher_b) + _release_slot(raw_c, dispatcher_c) finally: + raw_a.close() raw_b.close() raw_c.close() @@ -226,6 +257,34 @@ def test_yields_none_once_the_single_slot_is_already_held(self, db: Any) -> None finally: raw.close() + @CAP_2 + def test_does_not_use_djangos_shared_connection_at_all(self, db: Any) -> None: + """Companion assertion to the regression tests below, at the unit level: acquiring and + releasing a slot must not touch `django.db.connection`'s own cursor - if it did, the + pre-fix bug (lock held on a connection django-q's broker can close mid-dispatch) would be + back. Patches `django.db.connection.cursor` to explode if called, for the duration of one + acquire/release cycle only - restored via an explicit `finally`, NOT `pytest`'s + `monkeypatch` fixture, because this test's own teardown (the autouse leaked-lock drain + fixture, and pytest-django's own `_post_teardown` -> `check_constraints`) both call + `connection.cursor()` themselves AFTER the test body returns but BEFORE a + function-scoped `monkeypatch` fixture would have restored it, which made the patch leak + into (and fail) teardown machinery that has nothing to do with this test's own + assertion.""" + original_cursor = connection.cursor + + def _boom(*args: Any, **kwargs: Any) -> Any: + raise AssertionError( + "stage_e_concurrency touched django.db.connection.cursor() - it must only ever " + "use its own dedicated connection (see module docstring)" + ) + + connection.cursor = _boom # type: ignore[method-assign] + try: + with try_acquire_dispatch_slot() as slot: + assert slot == 0 + finally: + connection.cursor = original_cursor # type: ignore[method-assign] + class TestCrossConnectionRace: """Proves real cross-SESSION safety with independent raw connections standing in for separate @@ -233,10 +292,11 @@ class TestCrossConnectionRace: @CAP_2 def test_a_second_independent_session_cannot_exceed_the_cap(self, db: Any) -> None: - dispatcher_a = _try_acquire_slot() + raw_a = _raw_connection() raw_b = _raw_connection() try: - dispatcher_b = _try_acquire_on_connection(raw_b, cap=2) + dispatcher_a = _try_acquire_slot(raw_a) + dispatcher_b = _try_acquire_slot(raw_b) assert dispatcher_a == 0 and dispatcher_b == 1 raw_c = _raw_connection() @@ -252,9 +312,10 @@ def test_a_second_independent_session_cannot_exceed_the_cap(self, db: Any) -> No finally: raw_c.close() - _release_slot(dispatcher_a) - _release_on_connection(raw_b, dispatcher_b) + _release_slot(raw_a, dispatcher_a) + _release_slot(raw_b, dispatcher_b) finally: + raw_a.close() raw_b.close() @CAP_2 @@ -268,10 +329,16 @@ def test_killing_the_holding_session_auto_releases_its_slot(self, db: Any) -> No 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) + # A fresh independent session now claims the same slot successfully - auto-released, not + # leaked (production-faithful: this module never holds a lock on Django's own connection, + # so a fresh dedicated connection standing in for a fresh worker is the right check here). + raw2 = _raw_connection() + try: + reacquired = _try_acquire_slot(raw2) + assert reacquired == 0 + _release_slot(raw2, reacquired) + finally: + raw2.close() class TestSimulatedConcurrentDispatchers: @@ -345,3 +412,71 @@ def _worker(index: int) -> None: # 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 + + +class TestRegressionDedicatedConnectionSurvivesFollowOnEnqueue: + """ + Regression tests for the 2026-07-25T00:25Z production incident + (`envtrip-20260725T002504-73e1eb6d`, `{'ceiling': 7.0, 'load_avg': 11.4013671875}`) - zero + `throttled-concurrency-cap` outcomes despite 8 concurrent django-q workers, plus 8 occurrences of + this module's own `pg_advisory_unlock reported slot N was not held by this connection` warning. + See `stage_e_concurrency`'s own module docstring, "WHAT WENT WRONG IN PRODUCTION" section, for + the full root-cause writeup this test reproduces: `cardpicker.stage_e_signals`'s `post_save` + receivers call `django_q.tasks.async_task(...)` from INSIDE `dispatch_micro_batch`'s locked + region (during Stage C's `persist_evidence`); `async_task` synchronously calls the installed ORM + broker's `enqueue`, which calls `django.db.close_old_connections()` whenever not inside an atomic + block - and with `CONN_MAX_AGE` unset (this project's `DATABASES["default"]` has no override, + Django's own default is `0`), that call closes the connection UNCONDITIONALLY, not just past some + age. A closed connection auto-releases every Postgres advisory lock its session held. + + PROVEN TO CATCH THE REGRESSION (verified by hand against the pre-fix module, not committed - see + this change's own PR description): pointing `try_acquire_dispatch_slot` back at + `django.db.connection` instead of a dedicated connection makes both tests below FAIL - the + "genuinely separate session" check finds slot 0 free (silently reacquirable), instead of still + held. + + `transactional_db`, not `db`: the `db` fixture wraps every test in an outer atomic block + (`get_autocommit()` is `False` throughout the test), which is not the connection state a real + django-q worker process is in when a signal receiver fires mid-dispatch - `ORM.get_connection`'s + own `if transaction.get_autocommit(...)` check is exactly why the production trigger only fires + OUTSIDE an atomic block, and calling `close_old_connections()` directly while genuinely inside + the `db` fixture's own outer atomic block would corrupt that fixture's own transactional + isolation for the rest of the test (Django's "didn't restore autocommit, drop the connection" + rule in `close_if_unusable_or_obsolete` fires regardless of which code calls it). + `transactional_db` gives this test a real, autocommit=True connection - the production-faithful + state, and the same fixture this codebase's own `TestConcurrency`-style tests + (`test_local_identify_printing_tags.py`) already use for the identical reason. + """ + + @CAP_2 + def test_slot_survives_a_follow_on_async_task_enqueue_inside_the_locked_region(self, transactional_db: Any) -> None: + """The exact production trigger, reproduced end to end: a real `django_q.tasks.async_task` + call, made from inside the locked region, exactly where `stage_e_signals` makes it.""" + from django_q.tasks import async_task + + with try_acquire_dispatch_slot() as slot: + assert slot == 0 + + async_task("cardpicker.stage_e_dispatch.dispatch_for_card", 1, "evidence-change") + + assert _slot_0_is_held_by_someone_else() is True + + # released cleanly on exit despite the mid-block connection churn. + assert _slot_0_is_held_by_someone_else() is False + + @CAP_2 + def test_slot_survives_close_old_connections_called_directly_inside_the_locked_region( + self, transactional_db: Any + ) -> None: + """A more minimal, django-q-version-independent reproduction of the identical root cause: + directly calling `django.db.close_old_connections()` (what the ORM broker calls internally) + from inside the locked region must not affect this module's own lock, because the lock is + held on a connection `close_old_connections()` never touches.""" + with try_acquire_dispatch_slot() as slot: + assert slot == 0 + + close_old_connections() + + assert _slot_0_is_held_by_someone_else() is True + + assert _slot_0_is_held_by_someone_else() is False diff --git a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py index 4c7ace4f1..bfcda6b03 100644 --- a/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py +++ b/MPCAutofill/cardpicker/tests/test_stage_e_dispatch.py @@ -37,6 +37,7 @@ ImageEvidence, PilotRunLedger, PrintingTagStatus, + StageEThrottleCounter, VoteSource, ) from cardpicker.operating_envelope import ( @@ -398,6 +399,11 @@ def _fail_if_called(card, dpi=None): assert PilotRunLedger.objects.count() == 0 assert ImageEvidence.objects.count() == 0 assert CardPrintingTag.objects.count() == 0 + # observability signal (Tron gate anomaly 4, 2026-07-25) - the ONE durable, queryable + # record that this throttle happened, since no ledger row was written above. + counter = StageEThrottleCounter.objects.get() + assert counter.count == 1 + assert counter.last_throttled_at is not None @STREAMING_ON @override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=1) diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index cd31cf5dd..2fc84eabc 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -1,4 +1,4 @@ -As of: 2026-07-24 +As of: 2026-07-25 What this is: the admin-facing operational truth for Stage E's envelope enforcement primitive (Phase 1) and streaming dispatch loop (Phase 2), both implementing [`docs/proposals/stage-e-streaming.md`](../proposals/stage-e-streaming.md) @@ -240,7 +240,7 @@ 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) +### Concurrency cap (companion change, 2026-07-24; connection-lifecycle fix + throttle-observability counter, 2026-07-25) Caps the number of `dispatch_micro_batch` calls running CONCURRENTLY, across every django-q2 worker process on the box, to @@ -271,14 +271,73 @@ in place; neither supersedes the other. 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. +migration for the lock mechanism itself, no new infrastructure. A throttled +dispatch returns `status="throttled-concurrency-cap"` and writes no +`PilotRunLedger` row, the same "halted dispatch never partially starts" +convention `halted-open-trip`/`halted-new-trip` already established — but it +DOES advance a small observability counter, see "Throttle observability" +below. `_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. + +**The lock rides a DEDICATED connection, not Django's own +(2026-07-25 fix, PRODUCTION INCIDENT)**: the first live shakedown +(`envtrip-20260725T002504-73e1eb6d`, `{'ceiling': 7.0, 'load_avg': 11.4013671875}`) found this cap did NOT bind — zero `throttled-concurrency- cap` outcomes despite 8 concurrent django-q workers, plus 8 occurrences of +this module's own `pg_advisory_unlock reported slot N was not held by this connection` warning. Root cause: the original version of this module +acquired and released its advisory lock on `django.db.connection` (Django's +shared per-thread connection), on the strength of a claim — checked against +the wrong code path — that a single `dispatch_micro_batch` call always runs +as one uninterrupted segment on one connection. The REAL trigger: +`cardpicker.stage_e_signals`'s `post_save` receivers fire during Stage C's +`persist_evidence` (squarely inside the locked region) and call +`django_q.tasks.async_task(...)`, which synchronously calls the installed +ORM broker's `enqueue` — and `django_q.brokers.orm.ORM.get_connection()` +calls `django.db.close_old_connections()` unconditionally whenever it's not +inside an atomic block. This project's `DATABASES["default"]` has no +`CONN_MAX_AGE` override, so Django's own default (`0`) applies — the +connection is treated as already-expired the moment anything asks, not just +after some elapsed age — so `close_old_connections()` closed +`django.db.connection` the first time it was called, mid-dispatch. A closed +connection auto-releases every advisory lock its session held, so every +worker then found every slot "free". **The fix**: this module now opens a +DEDICATED `psycopg2` connection (`autocommit=True`) it alone owns for the +duration of one `try_acquire_dispatch_slot()` call — never +`django.db.connection`, never touched by django-q's broker or any other +code in the process. The connection is explicitly `pg_advisory_unlock`'d +AND `close()`'d in a `finally`, so the lock is released even if the +explicit unlock itself somehow fails (closing a session is Postgres's own +backstop release mechanism — the same crash-safety property this module's +own DB-row-counter rejection already leans on). The "not held by this +connection" warning guard is KEPT, unchanged — it is what caught this +incident, and should now never fire again; a fresh report of it firing is a +signal that something else has broken the dedicated-connection contract. +Connection-CREATION failure (the dedicated connection itself can't be +opened) fails **CLOSED**: the dispatch is treated as throttled +(`status="throttled-concurrency-cap"`) rather than proceeding uncapped — an +uncapped dispatch is exactly the failure this incident was. See +`cardpicker/stage_e_concurrency.py`'s own module docstring for the full +incident writeup, and `cardpicker/tests/test_stage_e_concurrency.py`'s +`TestRegressionDedicatedConnectionSurvivesFollowOnEnqueue` for the +regression tests (proven, by hand, to fail against the pre-fix module). + +**Throttle observability (Tron gate anomaly 4, 2026-07-25)**: since a +throttled dispatch writes no `PilotRunLedger` row, and this runbook's own +"raise `STAGE_E_MAX_CONCURRENT_DISPATCHES` once real shakedown data shows +headroom" instruction below needs SOMETHING queryable to check against, a +throttled outcome now also calls `StageEThrottleCounter.record()` +(`cardpicker/models.py`) — a SINGLETON, always-exactly-one-row counter +(`count`, `last_throttled_at`), visible in the Django admin. Deliberately +NOT a per-throttle-event row (the `PilotRunLedger`/`EnvelopeTrip` pattern): +a burst of concurrent dispatches hitting an exhausted cap can throttle far +more often than any dispatch ever completes, so a per-event row would +WRITE-AMPLIFY under exactly the failure shape this whole feature exists to +guard against. `count` is only ever advanced via an atomic `F("count") + 1` +UPDATE, race-safe under Postgres row-level locking even with many worker +processes throttling at once. New migration `0081_stageethrottlecounter` +(one small table, no relation to any other model) — this is the one piece +of "new infrastructure" this change adds, scoped deliberately narrowly to +observability, not the lock mechanism itself. **Event-dispatch drop semantics**: `Q_CLUSTER["max_attempts"] = 1` (`MPCAutofill/MPCAutofill/settings.py`) means an event-driven `async_task` that returns @@ -300,11 +359,16 @@ rather than reading a `batches_dispatched=0`-with-no-explanation line as **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. +guess a large value up front. Check the observed throttle rate via +`StageEThrottleCounter` (Django admin, or `StageEThrottleCounter.objects. get().count`/`.last_throttled_at`) before deciding it's worth raising at +all — before the 2026-07-25 fix above, this number was always zero +regardless of real load, which is exactly what made the cap's own +non-binding failure invisible; a persistently zero count after the fix is +a genuine "cap has headroom" signal, not a repeat of that gap. 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 diff --git a/docs/lessons.md b/docs/lessons.md index 87a5e329e..91ff9044b 100644 --- a/docs/lessons.md +++ b/docs/lessons.md @@ -915,3 +915,26 @@ old code touches OUTSIDE the per-task function's own locals (closures, module-le already-open resources) and re-derive fork-safety for each one explicitly - don't assume "the tests still pass" proves this, since a small/mocked test run may never exercise the actual shared state that breaks at real concurrency. + +## "No connection pool exists, so the connection is stable" is a claim that needs a test exercising the region's own side effects, not a static read + +The Stage E concurrency cap (`cardpicker.stage_e_concurrency`, PR #450) shipped its lock on +`django.db.connection` on the strength of a claim - made by directly reading `django_q.worker` - +that a single `dispatch_micro_batch` call always runs as one uninterrupted segment on one +connection. That claim was checked against the wrong code path: `django_q.worker`'s own +connection-recycling only happens BETWEEN tasks, never mid-task, which is true and irrelevant - +the actual risk was `django_q.brokers.orm.ORM.get_connection()`, reached from INSIDE the locked +region via a `post_save` signal receiver (`cardpicker.stage_e_signals`) calling +`django_q.tasks.async_task(...)`, which calls `django.db.close_old_connections()` +unconditionally whenever not inside an atomic block - a completely different module than the one +the static review inspected. The tests passed (they exercised the lock's own acquire/release +logic correctly) while production failed, because no test ever exercised the SIDE EFFECT that +actually threatened the connection: a follow-on `async_task` enqueue happening from inside the +held lock's own critical section. **The general check**: a static review that concludes "no +connection pool/recycling path exists here" only rules out connection instability from the code +PATHS it actually read - it does not prove the connection survives everything the region's own +code (including anything a signal handler triggers) might call into. Prove connection stability +with a test that exercises the region's real side effects (here: actually calling the enqueue +path, or the specific `close_old_connections()` primitive it bottoms out in, from inside the +locked block and asserting the lock survives from a genuinely separate session) - never by static +inspection of "what obviously touches the connection" alone. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 40c5c2bf2..96a577e58 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -1743,3 +1743,57 @@ unless `--allow-missing-scryfall-cache` is passed explicitly. This catches the failure mode structurally even if the volume mount is ever missed again (e.g. a fresh box rebuild that skips the compose file, or a manual `docker run` bypassing compose entirely). + +## Stage E concurrency cap configured but zero `throttled-concurrency-cap` outcomes, host load trips anyway + +**Symptom**: `settings.STAGE_E_MAX_CONCURRENT_DISPATCHES` is set (e.g. +the default `2`), several django-q2 workers are dispatching Stage E +micro-batches concurrently, and yet: (a) `DispatchOutcome.status` is +never `"throttled-concurrency-cap"` no matter how much concurrent load +there is, (b) the worker log carries the literal warning +`pg_advisory_unlock reported slot N was not held by this connection - possible connection recycling mid-dispatch` (`cardpicker. stage_e_concurrency`'s own defensive guard), one occurrence per +dispatch, and (c) the host load average envelope bar +(`operating_envelope`, ceiling `7.0`) trips anyway, from a burst of +CPU-bound Stage C work that the concurrency cap should have prevented +from ever starting. + +**Cause**: this exact signature is the 2026-07-25T00:25Z production +shakedown (`envtrip-20260725T002504-73e1eb6d`, +`{'ceiling': 7.0, 'load_avg': 11.4013671875}`) — the cap's advisory lock +was held on `django.db.connection` (Django's shared per-thread +connection) instead of a dedicated connection. `cardpicker. stage_e_signals`'s `post_save` receivers fire during Stage C's +`persist_evidence`, squarely inside the cap's own locked region, and +call `django_q.tasks.async_task(...)`, which synchronously calls the +installed ORM broker's `enqueue` — `django_q.brokers.orm.ORM. get_connection()` calls `django.db.close_old_connections()` +unconditionally whenever not inside an atomic block. With `CONN_MAX_AGE` +unset (this project's `DATABASES["default"]` has no override, so +Django's default `0` applies), that call closes the connection the +first time anything asks, not just after some elapsed age. A closed +connection auto-releases every Postgres advisory lock its session held, +so the cap's own lock died mid-dispatch and every subsequent worker +found every slot "free" — the warning above is Postgres reporting that +the FINAL unlock call (which runs on a transparently-reconnected, and +therefore DIFFERENT, backend session) found nothing to release. + +**Fix** (shipped 2026-07-25, same day): `cardpicker/stage_e_concurrency.py` +now opens a DEDICATED `psycopg2` connection (`autocommit=True`) it alone +owns for the lifetime of one `try_acquire_dispatch_slot()` call — never +`django.db.connection`. See that module's own docstring ("WHAT WENT +WRONG IN PRODUCTION" section) and +[`docs/features/stage-e-operations.md`](features/stage-e-operations.md)'s +"Concurrency cap" section for the full writeup, and +`cardpicker/tests/test_stage_e_concurrency.py`'s +`TestRegressionDedicatedConnectionSurvivesFollowOnEnqueue` for the +regression tests that reproduce this exact failure (proven, by hand, to +fail against the pre-fix module). + +**How to confirm it's this** (if the warning above recurs after the +fix): the warning guard was deliberately KEPT, not removed, specifically +so a regression here stays visible — a fresh occurrence means something +else has broken the "dedicated connection, never `django.db.connection`" +contract, not that the original bug is back verbatim. Check +`StageEThrottleCounter.objects.get().count`/`.last_throttled_at` +(Django admin, or `cardpicker.models.StageEThrottleCounter`) — introduced +in the same fix as the one durable, queryable signal for whether +throttling is happening at all (throttled dispatches write no +`PilotRunLedger` row). diff --git a/docs/upstreaming/extractable-primitives.md b/docs/upstreaming/extractable-primitives.md index 240f9d9ad..04ae8eff1 100644 --- a/docs/upstreaming/extractable-primitives.md +++ b/docs/upstreaming/extractable-primitives.md @@ -109,22 +109,22 @@ coupling to the vote system is. ## Backend -| Primitive | File(s) | Problem solved | Candidate consumers | Entanglement | License note | -| ----------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------ | -| Outbound rate limiter ("lh4 rate limiter") | `MPCAutofill/cardpicker/local_phash.py` (`_RateLimiter`, `run_content_phash_backfill`) | Paces a threaded worker pool to a strict `<= N req/sec` ceiling against Google's `lh4.googleusercontent.com` image-resize endpoint | upstream, proxies-at-home (as a copy-paste, not an import — see note) | entangled-with-CanonicalPrinting (colocation) | — | -| Perceptual-hash storage utility | `MPCAutofill/cardpicker/local_phash.py` (`_hash_to_int`, `_int_to_hash`, `compute_card_art_hash`, `find_best_match`) | Encodes/decodes an `imagehash.ImageHash` as a signed 64-bit DB int; threshold+margin best-match selection | upstream, proxies-at-home | entangled-with-CanonicalPrinting (colocation) | — | -| Search-query sanitisation | `MPCAutofill/cardpicker/search/sanitisation.py` | Normalizes free-text queries/names (lowercase, strip bracketed text/punctuation/digits, collapse whitespace) for consistent matching | upstream, proxies-at-home | CLEAN | — | -| Scryfall-style search-operator parser | `MPCAutofill/cardpicker/search/operator_parser.py` (`parse_query`) | Parses a raw query string into residual free text + structured `operator:value`/`-operator:value` tokens (quoted values, case-insensitive operator names, unknown-operator errors) - pure string-in/structure-out, no knowledge of what an operator name maps to downstream | upstream, proxies-at-home | CLEAN | — | -| OCR crop/preprocessing helpers | `MPCAutofill/cardpicker/local_ocr.py` (`crop_collector_line`, `preprocess_variants`, `preprocess_fallback_variants`, `run_tesseract`, `run_tesseract_text_and_words`, `parse_collector_line`, `parse_legal_line`, `_normalize_collector_number`, `_median_from_histogram`) | Fractional-bbox crop, grayscale/upscale/threshold-both-polarities preprocessing (plus a heavier-upscale/sharpen + percentile-threshold fallback tier, issue #259), regex parse of an OCR'd collector-number line, and regex-based "not for sale"/proxy-marker/playtest/copyright-year detection over an OCR'd legal line (2026-07-23, PR #384's marker-detection expansion - widened to catch maker-brand-glued forms like "JestaProxy") | upstream, proxies-at-home | CLEAN | — | -| Image color/quality-signal math | `MPCAutofill/cardpicker/local_image_quality.py` (`is_image_truncated`, `compute_blur_variance`, `compute_entropy`, `compute_color_profile`) | Truncation check, Laplacian-kernel blur variance, grayscale entropy, and per-channel RGB mean/stddev, all pure `PIL.ImageStat`/`ImageFilter` calls against an already-fetched image | upstream, proxies-at-home, federation peers | CLEAN (zero `cardpicker.*` imports at all — only `PIL`) | — | -| Bleed/border geometry helpers | `MPCAutofill/cardpicker/local_fallback.py` (`normalize_crop_box`, `classify_bleed_edge`) | Pure crop-box remapping (bleed vs. trim) and aspect-ratio-based border classification | upstream, proxies-at-home (as a copy-paste, not an import — see note) | entangled-with-vote-consensus (colocation) | — | -| Generic backend utilities | `MPCAutofill/cardpicker/utils.py` (`get_json_endpoint_rate_limited`, `twos_complement`, `section_timer`, `time_to_hours_minutes_seconds`, `log_hours_minutes_seconds_elapsed`) | Rate-limited JSON GET wrapper, signed-int bit-twiddling, timing decorator/formatter | upstream, proxies-at-home | CLEAN | — | -| Batch-flush checkpoint pattern | `MPCAutofill/cardpicker/local_phash.py` (`run_content_phash_backfill`), `deductive_backfill.py` (`run_backfill`), `local_identify_printing_tags.py` (`run_pilot`) | Sliding-window worker pool + periodic bulk-flush + NULL-filter-as-checkpoint for resumable backfill jobs | upstream, proxies-at-home (needs generalizing first — see note) | entangled — no clean instance exists yet | — | -| Elasticsearch connection helpers | `MPCAutofill/cardpicker/search/search_functions.py` (`get_elasticsearch_connection`, `ping_elasticsearch`, `elastic_connection`, `SearchExceptions`) | Thread-local ES client + a decorator translating raw ES connection errors into app exceptions | upstream, proxies-at-home | entangled-with-consensus (colocation) | — | -| 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) | — | +| Primitive | File(s) | Problem solved | Candidate consumers | Entanglement | License note | +| ---------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------ | +| Outbound rate limiter ("lh4 rate limiter") | `MPCAutofill/cardpicker/local_phash.py` (`_RateLimiter`, `run_content_phash_backfill`) | Paces a threaded worker pool to a strict `<= N req/sec` ceiling against Google's `lh4.googleusercontent.com` image-resize endpoint | upstream, proxies-at-home (as a copy-paste, not an import — see note) | entangled-with-CanonicalPrinting (colocation) | — | +| Perceptual-hash storage utility | `MPCAutofill/cardpicker/local_phash.py` (`_hash_to_int`, `_int_to_hash`, `compute_card_art_hash`, `find_best_match`) | Encodes/decodes an `imagehash.ImageHash` as a signed 64-bit DB int; threshold+margin best-match selection | upstream, proxies-at-home | entangled-with-CanonicalPrinting (colocation) | — | +| Search-query sanitisation | `MPCAutofill/cardpicker/search/sanitisation.py` | Normalizes free-text queries/names (lowercase, strip bracketed text/punctuation/digits, collapse whitespace) for consistent matching | upstream, proxies-at-home | CLEAN | — | +| Scryfall-style search-operator parser | `MPCAutofill/cardpicker/search/operator_parser.py` (`parse_query`) | Parses a raw query string into residual free text + structured `operator:value`/`-operator:value` tokens (quoted values, case-insensitive operator names, unknown-operator errors) - pure string-in/structure-out, no knowledge of what an operator name maps to downstream | upstream, proxies-at-home | CLEAN | — | +| OCR crop/preprocessing helpers | `MPCAutofill/cardpicker/local_ocr.py` (`crop_collector_line`, `preprocess_variants`, `preprocess_fallback_variants`, `run_tesseract`, `run_tesseract_text_and_words`, `parse_collector_line`, `parse_legal_line`, `_normalize_collector_number`, `_median_from_histogram`) | Fractional-bbox crop, grayscale/upscale/threshold-both-polarities preprocessing (plus a heavier-upscale/sharpen + percentile-threshold fallback tier, issue #259), regex parse of an OCR'd collector-number line, and regex-based "not for sale"/proxy-marker/playtest/copyright-year detection over an OCR'd legal line (2026-07-23, PR #384's marker-detection expansion - widened to catch maker-brand-glued forms like "JestaProxy") | upstream, proxies-at-home | CLEAN | — | +| Image color/quality-signal math | `MPCAutofill/cardpicker/local_image_quality.py` (`is_image_truncated`, `compute_blur_variance`, `compute_entropy`, `compute_color_profile`) | Truncation check, Laplacian-kernel blur variance, grayscale entropy, and per-channel RGB mean/stddev, all pure `PIL.ImageStat`/`ImageFilter` calls against an already-fetched image | upstream, proxies-at-home, federation peers | CLEAN (zero `cardpicker.*` imports at all — only `PIL`) | — | +| Bleed/border geometry helpers | `MPCAutofill/cardpicker/local_fallback.py` (`normalize_crop_box`, `classify_bleed_edge`) | Pure crop-box remapping (bleed vs. trim) and aspect-ratio-based border classification | upstream, proxies-at-home (as a copy-paste, not an import — see note) | entangled-with-vote-consensus (colocation) | — | +| Generic backend utilities | `MPCAutofill/cardpicker/utils.py` (`get_json_endpoint_rate_limited`, `twos_complement`, `section_timer`, `time_to_hours_minutes_seconds`, `log_hours_minutes_seconds_elapsed`) | Rate-limited JSON GET wrapper, signed-int bit-twiddling, timing decorator/formatter | upstream, proxies-at-home | CLEAN | — | +| Batch-flush checkpoint pattern | `MPCAutofill/cardpicker/local_phash.py` (`run_content_phash_backfill`), `deductive_backfill.py` (`run_backfill`), `local_identify_printing_tags.py` (`run_pilot`) | Sliding-window worker pool + periodic bulk-flush + NULL-filter-as-checkpoint for resumable backfill jobs | upstream, proxies-at-home (needs generalizing first — see note) | entangled — no clean instance exists yet | — | +| Elasticsearch connection helpers | `MPCAutofill/cardpicker/search/search_functions.py` (`get_elasticsearch_connection`, `ping_elasticsearch`, `elastic_connection`, `SearchExceptions`) | Thread-local ES client + a decorator translating raw ES connection errors into app exceptions | upstream, proxies-at-home | entangled-with-consensus (colocation) | — | +| 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 (connection-lifecycle fix 2026-07-25) | `MPCAutofill/cardpicker/stage_e_concurrency.py` (`try_acquire_dispatch_slot`, `_try_acquire_slot`, `_release_slot`, `_open_dedicated_connection`) | 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 on a DEDICATED `psycopg2` connection the module opens and closes for itself each call (2026-07-25: the original version held its lock on `django.db.connection` and lost it in production when django-q2's ORM broker closed that shared connection mid-dispatch - see `docs/troubleshooting.md`'s Stage E entry) - no new migration for the lock mechanism itself, 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` (connection PARAMETERS only, never held on it), `psycopg2`, and stdlib; its own module docstring is itself the generic write-up of the technique and the incident that shaped it) | — | ## Docs tooling & federation