From defd3ef033b8868b869c24828572e11a89b6410c Mon Sep 17 00:00:00 2001 From: wilfordgrimley <2397930+WilfordGrimley@users.noreply.github.com> Date: Thu, 30 Jul 2026 10:10:19 +0000 Subject: [PATCH] Make the 7/s fetch ceiling global instead of per process Owner clarification, 2026-07-30: "to be clear: the 7 fetches per second cap is a global cap, it shouldn't be per process or per core". `harvest_fetch_limiter._DestinationLimiter` kept its pacing state in one process's memory, so PR #644's `rate_per_sec = 7.0` bought 7/s PER PROCESS. That is a real ceiling for the pooled runner (one process, one thread pool) and no ceiling at all for the conveyor: django-q2's workers are separate OS processes, so N concurrent dispatches fetched at N x 7/s - 14/s at the shipped STAGE_E_MAX_CONCURRENT_DISPATCHES = 2, rising with that cap. New `cardpicker/harvest_rate_coordinator.py` moves the pacer's own `max(now, next_allowed) + interval` arithmetic into a single atomic `INSERT ... ON CONFLICT DO UPDATE` over one cursor row every fetching process shares. One round trip per fetch, Postgres's own clock, no advisory lock, no explicit transaction. No new migration: the cursor lives in the existing `shared_cache` table (0092), under a key Django's own `make_key` cannot produce. The migration graph is contended and a one-row table is not worth a leaf. Losing the row is benign - the next reservation re-inserts at "now". Degradation is divided, not open and not closed: if Postgres is unreachable, each process falls back to its own pacer at rate / (STAGE_E_MAX_CONCURRENT_DISPATCHES + 1), so the aggregate ceiling still holds with every process fetching. Failing open would restore the exact defect; failing closed would halt an unattended 230,753-card pass over a blip. A degraded reservation waits, never raises, never trips. Purely additive to #644: the throttle-not-halt conversion, the 429/503 classification, the backoff decay and the envelope bar classification are untouched. Backoff stays per-process and reaches the shared cursor as a widened interval, which can only slow the aggregate. Proven with real forked OS processes, not one limiter: three processes at a 20/s ceiling deliver 20/s together. Against the pre-fix pacer the same test measures 63/s. Coordination costs 1.1-1.3 ms per reservation (0.8-0.9% of the 143 ms interval a 7/s ceiling already imposes); eight acquisitions at the real 7.0/s ceiling take 1.003s against 1.000s for the ceiling alone, +0.3%. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013NhYmT1PxCcyemA16dFDxN --- .../cardpicker/harvest_fetch_limiter.py | 68 ++- .../cardpicker/harvest_rate_coordinator.py | 357 ++++++++++++++ .../tests/test_harvest_fetch_limiter.py | 17 + .../tests/test_harvest_rate_coordinator.py | 446 ++++++++++++++++++ docs/features/stage-e-operations.md | 50 ++ 5 files changed, 931 insertions(+), 7 deletions(-) create mode 100644 MPCAutofill/cardpicker/harvest_rate_coordinator.py create mode 100644 MPCAutofill/cardpicker/tests/test_harvest_rate_coordinator.py diff --git a/MPCAutofill/cardpicker/harvest_fetch_limiter.py b/MPCAutofill/cardpicker/harvest_fetch_limiter.py index b0b3c1dd0..3bbcb619e 100644 --- a/MPCAutofill/cardpicker/harvest_fetch_limiter.py +++ b/MPCAutofill/cardpicker/harvest_fetch_limiter.py @@ -66,6 +66,20 @@ envelope breach (host load, RSS, non-throttle fetch failures, a 403 lockout) still halts, unchanged. See `operating_envelope.py` and `docs/features/stage-e-operations.md`. +THE CEILING IS GLOBAL, NOT PER PROCESS (owner clarification, 2026-07-30: "to be clear: the 7 +fetches per second cap is a global cap, it shouldn't be per process or per core"). Everything in +`_DestinationLimiter` below - `_next_allowed`, its `threading.Lock`, its `threading.Semaphore` - +is PER-PROCESS state, and that was the whole ceiling until now. It held for the pooled runner +(`run_image_evidence_cohort`: one process, one thread pool) and it did NOT hold for the conveyor +(`stage_e_dispatch` under django-q2, whose workers are separate OS PROCESSES): N concurrent +dispatches each paced themselves to 7/s independently, for N x 7/s at the destination, scaling with +`STAGE_E_MAX_CONCURRENT_DISPATCHES`. `acquire()` now takes its pacing decision from +`harvest_rate_coordinator` - one atomic Postgres statement over a cursor shared by every fetching +process - so the aggregate is the ceiling regardless of how many processes fetch. The per-process +arithmetic survives, divided by the process count, only as the degraded fallback when the +coordination store is unreachable. See that module's docstring for the mechanism, the rejected +alternatives, and the fail-open/fail-closed reasoning. + BACKOFF IS NO LONGER STICKY-FOREVER (same ruling). It was: "the multiplier only grows, never resets", on the reasoning that recovering the fast rate mid-run risks re-tripping the same undocumented ceiling. That reasoning holds for a SHORT run and fails for the one this project @@ -86,6 +100,8 @@ import requests +from cardpicker import harvest_rate_coordinator + logger = logging.getLogger(__name__) @@ -293,16 +309,50 @@ def note_clean_response(self) -> None: multiplier, ) - def acquire(self) -> "_LimiterSlot": - if self._locked_out: - raise GoogleFetchLockoutError(f"{self._config.name} is locked out (403) - refusing further requests") - self._semaphore.acquire() + def _reserve_locally(self, interval: float) -> float: + """The ORIGINAL per-process minimum-interval arithmetic, now reached only when + cross-process coordination is unavailable (`harvest_rate_coordinator.reserve` returned + `None`). Kept verbatim - it is a correct pacer, it was only ever wrong about its SCOPE. The + caller widens `interval` by `degraded_divisor()` before calling this, so N processes each + running this fallback still sum to no more than the configured ceiling.""" with self._lock: now = time.monotonic() - interval = self._interval * self._backoff_multiplier wait_time = max(0.0, self._next_allowed - now) self._next_allowed = max(now, self._next_allowed) + interval - self._request_count += 1 + return wait_time + + def acquire(self) -> "_LimiterSlot": + """Blocks the calling thread until this destination's GLOBAL budget clears it to fetch, then + returns a context manager holding the per-process concurrency semaphore for the fetch's + duration. + + The pacing decision is made by `harvest_rate_coordinator.reserve` - a single atomic Postgres + statement over a cursor every fetching PROCESS shares - not by this object's own + `_next_allowed`. That is the whole point: this class's state is per-process, and django-q2's + workers are separate OS processes, so a purely local pacer delivered N x the configured rate + (see `harvest_rate_coordinator`'s own module docstring for the full defect writeup). The + local pacer survives as the degraded fallback only. + + Backoff stays local and is passed INTO the reservation as an already-widened interval, so PR + #644's throttle-not-halt behaviour is unchanged: a process under rate pressure contributes a + larger gap to the shared cursor, which can only ever slow the aggregate, never raise it.""" + if self._locked_out: + raise GoogleFetchLockoutError(f"{self._config.name} is locked out (403) - refusing further requests") + self._semaphore.acquire() + try: + with self._lock: + interval = self._interval * self._backoff_multiplier + self._request_count += 1 + wait_time = harvest_rate_coordinator.reserve(self._config.name, interval) + if wait_time is None: + wait_time = self._reserve_locally(interval * harvest_rate_coordinator.degraded_divisor()) + except BaseException: + # Nothing between the semaphore acquire and the sleep is expected to raise - `reserve` + # swallows its own failures by contract. If something does anyway, the semaphore must + # not leak, or this destination silently loses a concurrency slot for the life of the + # process. + self._semaphore.release() + raise if wait_time > 0: time.sleep(wait_time) return _LimiterSlot(self._semaphore) @@ -342,9 +392,13 @@ def get_limiter(config: DestinationLimiterConfig) -> _DestinationLimiter: def reset_limiters() -> None: """Test-only: drops every registered limiter so each test starts with fresh pacing/trip - state instead of leaking across tests via the module-level registry.""" + state instead of leaking across tests via the module-level registry, and drops the rate + coordinator's dedicated connection with them (the cross-process cursor itself is per + destination NAME, so tests that want a fresh cursor must call + `harvest_rate_coordinator.clear_cursor` for their own destination - see that module).""" with _REGISTRY_LOCK: _LIMITERS.clear() + harvest_rate_coordinator.reset_connection() def rate_limited_get(config: DestinationLimiterConfig, url: str, **kwargs: Any) -> "requests.Response": diff --git a/MPCAutofill/cardpicker/harvest_rate_coordinator.py b/MPCAutofill/cardpicker/harvest_rate_coordinator.py new file mode 100644 index 000000000..6e488622c --- /dev/null +++ b/MPCAutofill/cardpicker/harvest_rate_coordinator.py @@ -0,0 +1,357 @@ +""" +CROSS-PROCESS rate ceiling for `harvest_fetch_limiter` - the globality half of the owner's +2026-07-30 rate ruling ("to be clear: the 7 fetches per second cap is a global cap, it shouldn't +be per process or per core"). + +THE DEFECT THIS FIXES. `harvest_fetch_limiter._DestinationLimiter` is a strict minimum-interval +pacer whose entire state - `_next_allowed`, the `threading.Lock` guarding it, the +`threading.Semaphore` bounding concurrency - lives in ONE Python process's memory. That is a +genuine ceiling for the pooled runner (`run_image_evidence_cohort`, a single process with a thread +pool), and it is NOT a ceiling for the conveyor: `stage_e_dispatch.dispatch_micro_batch` runs under +django-q2, whose workers are separate OS PROCESSES (`multiprocessing`, not threads - +`Q_CLUSTER["workers"]`, `MPCAutofill/settings.py`). Each such process imports this module fresh, +builds its own `_DestinationLimiter`, and paces itself in isolation. N concurrent dispatches +therefore fetch at N x `rate_per_sec`. At the production `STAGE_E_MAX_CONCURRENT_DISPATCHES = 2` +that is 14/s against a ratified 7/s, and it SCALES WITH THE CAP - raising the dispatch cap silently +raises the destination rate. + +This is the third appearance of the same per-process trap in this subsystem (the `threading +.Semaphore(config.max_concurrency)` concurrency bound has it too; `stage_e_concurrency`'s own module +docstring records the 2026-07-24 incident where a per-process assumption produced eight concurrent +dispatches). PR #644 concluded that "the rate limit, not the concurrency limit, is what protects the +destination" - which makes the rate limit's GLOBALITY the load-bearing property, and it was missing. + +WHY NOT THE ADVISORY-LOCK PRIMITIVE AS-IS. `stage_e_concurrency` caps CONCURRENT DISPATCHES across +processes with Postgres session-scoped advisory locks (`pg_try_advisory_lock`, `_slot_count()`), and +that is this repo's established cross-process coordination mechanism. It was read first and it does +NOT carry a rate on its own, for a reason worth stating rather than working around: an advisory lock +is a pure MUTUAL-EXCLUSION primitive. It has no payload and no time dimension - it can answer "how +many are in flight right now" and nothing else. A rate is a value (a cursor) plus a clock, so it +needs somewhere to keep the value. The two shapes that CAN be built from advisory locks alone were +both considered and rejected: + + * "K slots, each held for K/rate seconds" turns the counting primitive into a rate correctly on + paper, but the holder must BLOCK for the full hold time, so throughput is bounded by the + fetching thread count rather than by the ceiling (6 fetch threads x a 1s hold = 6/s under a 7/s + ceiling - it under-delivers), and it needs a live Postgres session per in-flight fetch. + * An advisory lock used as a mutex around shared state still needs the shared state. That is the + design below, minus the mutex: Postgres gives the mutual exclusion for free inside a single + statement, so the lock buys nothing here. + +WHAT THIS MODULE DOES INSTEAD - one atomic statement over a shared cursor. The per-process pacer's +own arithmetic (`_next_allowed = max(now, _next_allowed) + interval`) is exactly a cursor update. +Moving that ONE expression into Postgres makes it global, with no change to its semantics: + + INSERT ... ON CONFLICT (cache_key) DO UPDATE + SET value = (GREATEST(now, stored) + interval) + RETURNING GREATEST(0, value - interval - now) + +`INSERT ... ON CONFLICT DO UPDATE` takes a row lock for the duration of that one statement, so the +read-modify-write is atomic against every other process and every other thread with zero extra +round trips - no advisory lock, no explicit transaction, no `SELECT ... FOR UPDATE` pair. The +statement returns the caller's own wait in seconds, computed entirely against POSTGRES's clock +(`clock_timestamp()`) so no two processes are comparing different monotonic epochs. The caller then +sleeps locally. One network round trip per fetch, total. + +WHERE THE CURSOR LIVES - the existing `shared_cache` table, and deliberately NOT a new migration. +`settings.SHARED_CACHE_TABLE` (migration `0092_shared_cache_table`) already exists in production, in +CI and in every developer database, is created by the `migrate` that `docker/django/entrypoint.sh` +already runs, and its own migration docstring frames it as exactly this: cross-process state backed +by the existing Postgres, needing no new service. Adding a dedicated one-row table instead would +mean a new migration, and this repo's migration graph is actively contended (PR #611's CI guard +fails a graph that forks as merged with the base) - a new leaf is a real, avoidable merge hazard for +a single numeric value. + +The cursor is written as a RAW numeric string under a key (`_CURSOR_KEY_PREFIX`) that Django's own +`BaseCache.make_key` can never produce - every Django cache key in that table is prefixed +`"::"`. We therefore never collide with, and never have to unpickle, anything the `"shared"` +cache alias itself wrote; we are using the table as a generic Postgres key/value row, not as a +Django cache. The two ways Django could remove our row - `cache.clear()` (TRUNCATE) and +`DatabaseCache._cull` (only above `MAX_ENTRIES = 1000`, and this table holds a handful of entries) - +are both BENIGN: a missing row is re-INSERTed by the next reservation at "now", which costs at most +one extra immediately-allowed fetch, never a burst. + +DEDICATED CONNECTION, NOT `django.db.connection`. Same reasoning `stage_e_concurrency` already +established, for a different consequence. This module must never run its statement inside a caller's +open transaction: `INSERT ... ON CONFLICT DO UPDATE` holds the cursor ROW LOCK until commit, so a +long-lived enclosing `atomic()` block anywhere on the fetch path would stall EVERY other fetching +process for the length of that transaction - a global rate ceiling that occasionally becomes a +global stop. Owning one autocommit `psycopg2` connection per process makes that structurally +impossible. It is one connection per process (not per thread) held for the process's life, guarded +by a `threading.Lock`: the extra Postgres connection count stays at 1 per fetching process rather +than 1 per fetch thread, and the serialisation the lock adds costs nothing, because every thread is +contending for the same cursor row inside Postgres anyway. Connection parameters come from Django's +own `connection.get_connection_params()`, so this follows test-database prefixing automatically. + +FAILURE MODE - DEGRADE, NEVER FAIL OPEN AND NEVER FAIL CLOSED. If the coordination store is +unreachable mid-run, `reserve()` returns `None` and the caller falls back to its per-process pacer +running at `rate / _degraded_divisor()` - i.e. the configured ceiling divided by the maximum number +of processes that can be fetching at once. Rationale: + + * FAIL OPEN (per-process pacing at the full rate) is precisely the defect this module exists to + remove - it would restore N x 7/s at the exact moment nothing is watching. + * FAIL CLOSED (refuse to fetch) kills a 230,753-card unattended pass over a transient blip, and + contradicts the ruling's own "throttle, do not shut it down". + * DIVIDING is the only option that keeps the aggregate ceiling intact WITHOUT a halt: worst case + every process is fetching, and N processes x rate/N = rate. The cost is that the run fetches + more slowly than it strictly could while coordination is down - which is a cost paid exactly + when the database is already unhealthy, i.e. when Stage C cannot persist evidence anyway, so + slowing the fetch side is not the binding problem. A degraded reservation still WAITS; it never + raises, never halts, and never feeds the operating envelope's fetch-failure window + (requirement: rate pressure is throttled, not halted - PR #644). + +The divisor is DERIVED, not a new knob (owner: "default the default things, disable them with +flags"): `STAGE_E_MAX_CONCURRENT_DISPATCHES` bounds the conveyor's concurrent dispatch processes, +plus one for a pooled/manual runner (`run_image_evidence_cohort`, `stream_full_catalog`) which takes +no dispatch slot and can legitimately run alongside. + +Failures are latched for `_DEGRADED_COOLDOWN_SECONDS` so a sustained outage costs one failed +connection attempt per cooldown rather than one per fetch, and are logged once per cooldown with a +running count rather than once per fetch. +""" + +import logging +import os +import re +import threading +import time +from typing import Optional + +import psycopg2 + +from django.conf import settings +from django.db import connection + +logger = logging.getLogger(__name__) + +# Key namespace for the per-destination cursor rows. Django's own `BaseCache.make_key` always emits +# `"::"`, so a key that starts with a letter and contains no leading colon +# can never be produced by the `"shared"` cache alias itself - we cannot collide with it, and it +# cannot hand our raw numeric string to `pickle.loads`. +_CURSOR_KEY_PREFIX = "harvest-rate-cursor:" + +# How long a cursor row is kept alive. Only meaningful against `DatabaseCache._base_set`'s +# expired-row sweep (it deletes `WHERE expires < now` whenever anything else writes to this table); +# far enough out that a live run's cursor is never swept, and short enough that an abandoned +# destination's row does not outlive the code that wrote it forever. +_CURSOR_TTL_DAYS = 30 + +# After a coordination failure, skip the Postgres round trip entirely for this long and pace +# locally at the divided rate. Bounds both the cost and the log volume of a sustained outage; short +# enough that a transient blip costs at most this much degraded throughput. +_DEGRADED_COOLDOWN_SECONDS = 5.0 + +# Guards the identifier interpolated into the SQL below. `settings.SHARED_CACHE_TABLE` is a +# hardcoded literal today, but it is a setting, and a setting is interpolated into SQL here because +# a table name cannot be a bound parameter. Asserting the shape is cheaper than trusting it. +_SAFE_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + +# The whole primitive, in one statement. +# +# * The VALUES row carries `now + gap`, so `EXCLUDED.value - gap` IS Postgres's own reading of "now" +# for this call - available inside the conflict branch without evaluating `clock_timestamp()` a +# second time, and without a CTE that `ON CONFLICT DO UPDATE` may not reference. +# * The SET expression is `GREATEST(now, stored) + gap` - byte-for-byte the pacer arithmetic +# `_next_allowed = max(now, _next_allowed) + interval`, just evaluated where every process can see +# it. +# * RETURNING yields THIS caller's wait: the cursor value it was handed, minus its own gap (that is +# the instant it is cleared to fetch), minus now. Floored at zero. The `clock_timestamp()` in +# RETURNING is evaluated a few microseconds after the one in VALUES, which can only shorten the +# reported wait by that same few microseconds - it can never overstate the caller's budget. +# * `ON CONFLICT DO UPDATE` takes the row lock for the statement's duration, which is what makes the +# read-modify-write atomic across processes without any explicit locking of our own. +_RESERVE_SQL = """ +INSERT INTO {table} (cache_key, value, expires) +VALUES ( + %(key)s, + (EXTRACT(EPOCH FROM clock_timestamp())::numeric + %(gap)s::numeric)::text, + clock_timestamp() + %(ttl)s::interval +) +ON CONFLICT (cache_key) DO UPDATE SET + value = ( + GREATEST( + EXCLUDED.value::numeric - %(gap)s::numeric, + {table}.value::numeric + ) + %(gap)s::numeric + )::text, + expires = EXCLUDED.expires +RETURNING GREATEST( + 0::numeric, + {table}.value::numeric - %(gap)s::numeric - EXTRACT(EPOCH FROM clock_timestamp())::numeric +)::float8 +""" + + +_connection: Optional["psycopg2.extensions.connection"] = None +_connection_lock = threading.Lock() +_degraded_until = 0.0 +_degraded_failures = 0 + + +def _cursor_table() -> str: + table = getattr(settings, "SHARED_CACHE_TABLE", "shared_cache") + if not _SAFE_IDENTIFIER.match(table): + raise ValueError(f"SHARED_CACHE_TABLE is not a bare SQL identifier: {table!r}") + return table + + +def degraded_divisor() -> int: + """The number the configured rate is divided by while coordination is unavailable - see the + module docstring's "FAILURE MODE" section. Derived from `STAGE_E_MAX_CONCURRENT_DISPATCHES` + (the conveyor's own cross-process cap on concurrently-executing dispatches) plus one for a + pooled or manual runner, which holds no dispatch slot and can run alongside them. Never a new + setting of its own: a second knob for the same fact would just be a way for the two to disagree. + """ + dispatches = getattr(settings, "STAGE_E_MAX_CONCURRENT_DISPATCHES", 2) + return max(1, dispatches) + 1 + + +def _open_connection() -> "psycopg2.extensions.connection": + """Opens the process's own autocommit `psycopg2` connection - see the module docstring's + "DEDICATED CONNECTION" section for why this is never `django.db.connection`. Parameters are + read from Django's own connection (`ensure_connection()` first, since `get_connection_params()` + alone establishes nothing), so this always targets the database Django is configured against, + including pytest-django's prefixed test database.""" + connection.ensure_connection() + raw = psycopg2.connect(**connection.get_connection_params()) + raw.autocommit = True + return raw + + +def _note_failure(exc: BaseException) -> None: + global _degraded_until, _degraded_failures + now = time.monotonic() + _degraded_failures += 1 + if now >= _degraded_until: + logger.warning( + "harvest_rate_coordinator: cross-process rate coordination unavailable (%s: %s) - " + "degrading to per-process pacing at 1/%d of the configured ceiling for the next %.0fs " + "(%d failure(s) so far). The aggregate ceiling still holds; the run continues.", + type(exc).__name__, + exc, + degraded_divisor(), + _DEGRADED_COOLDOWN_SECONDS, + _degraded_failures, + ) + _degraded_until = now + _DEGRADED_COOLDOWN_SECONDS + + +def reserve(destination: str, interval_seconds: float) -> Optional[float]: + """Reserves this caller's turn on the GLOBAL cursor for `destination` and returns how long it + must wait, in seconds, before fetching (0.0 = go now). + + Returns `None` - never raises - when coordination is unavailable, which the caller must treat as + "pace yourself locally at the divided rate" (see `degraded_divisor()`), NOT as a failure, a halt, + or a fetch outcome. This is the only return value that is not a wait, and it is deliberately + distinguishable from `0.0`. + + `interval_seconds` is the caller's CURRENT pacing gap - the configured `1 / rate_per_sec` already + multiplied by whatever backoff the caller has accumulated. Backoff therefore stays per-process, + exactly as PR #644 built it: a process that has seen a 429 widens its OWN contribution to the + shared cursor, which can only ever slow the aggregate down, never speed it past the ceiling. + """ + global _connection + + if time.monotonic() < _degraded_until: + return None + + key = f"{_CURSOR_KEY_PREFIX}{destination}" + params = {"key": key, "gap": interval_seconds, "ttl": f"{_CURSOR_TTL_DAYS} days"} + sql = _RESERVE_SQL.format(table=_cursor_table()) + + try: + with _connection_lock: + if _connection is None or _connection.closed: + _connection = _open_connection() + try: + with _connection.cursor() as cursor: + cursor.execute(sql, params) + (wait_seconds,) = cursor.fetchone() + except psycopg2.Error: + # A connection that died between calls (Postgres restart, idle timeout, network + # blip) is the ordinary case here, and it is recoverable within this same call: + # drop it, open a fresh one, run the statement once more. Anything that fails twice + # falls through to the degraded path below. + try: + _connection.close() + except Exception: + pass + _connection = _open_connection() + with _connection.cursor() as cursor: + cursor.execute(sql, params) + (wait_seconds,) = cursor.fetchone() + except Exception as exc: # noqa: BLE001 - deliberately broad; see the module docstring + _note_failure(exc) + return None + + return float(wait_seconds) + + +def clear_cursor(destination: str) -> None: + """Test/ops helper: forgets the shared cursor for one destination, so the next reservation + starts from "now". Never raises - a coordination store that is unreachable has, from this + function's point of view, already forgotten it.""" + global _connection + + try: + with _connection_lock: + if _connection is None or _connection.closed: + _connection = _open_connection() + with _connection.cursor() as cursor: + cursor.execute( + f"DELETE FROM {_cursor_table()} WHERE cache_key = %s", + [f"{_CURSOR_KEY_PREFIX}{destination}"], + ) + except Exception: + logger.debug("harvest_rate_coordinator: clear_cursor(%s) could not reach the store", destination) + + +def reset_connection() -> None: + """Drops this process's dedicated connection (closing it properly) and clears the degraded + latch. For tests between cases, and for any caller that wants a guaranteed-fresh session. + + NOT for a forked child - see `_forget_connection_after_fork` below for why closing an INHERITED + connection is actively harmful.""" + global _connection, _degraded_until, _degraded_failures + with _connection_lock: + if _connection is not None: + try: + _connection.close() + except Exception: + pass + _connection = None + _degraded_until = 0.0 + _degraded_failures = 0 + + +def _forget_connection_after_fork() -> None: + """Registered as an `os.register_at_fork(after_in_child=...)` handler, because this subsystem's + two real deployments BOTH fork: django-q2's `Cluster` forks its worker processes, and the pooled + runner's own pool is a `multiprocessing` context away from doing the same. + + A forked child inherits the parent's connection OBJECT, pointing at a socket the PARENT still + owns. Two processes writing down one libpq socket interleaves the wire protocol and corrupts + both. The child must therefore ABANDON the object - and specifically must NOT `close()` it, the + way `reset_connection` does: `PQfinish` sends a Terminate message to the server before closing + the fd, which would kill the session the PARENT is still using. Dropping the reference and + letting the child's copy of the fd close on exit is the correct disposal; the parent's session + is untouched, and the child opens its own on its next reservation. + + Deliberately automatic rather than a documented obligation on callers: the failure it prevents + is silent, intermittent, and looks like unrelated Postgres protocol errors, so "every fork site + must remember" is not a contract worth relying on. + + `_connection_lock` is REPLACED, not reused: `fork()` clones only the calling thread, so a lock + another thread happened to hold at fork time is inherited already-locked with no owner left + alive to release it - the child would deadlock on its first reservation.""" + global _connection, _connection_lock, _degraded_until, _degraded_failures + _connection = None + _connection_lock = threading.Lock() + _degraded_until = 0.0 + _degraded_failures = 0 + + +os.register_at_fork(after_in_child=_forget_connection_after_fork) + + +__all__ = ["clear_cursor", "degraded_divisor", "reserve", "reset_connection"] diff --git a/MPCAutofill/cardpicker/tests/test_harvest_fetch_limiter.py b/MPCAutofill/cardpicker/tests/test_harvest_fetch_limiter.py index a453bcfaf..210673bbb 100644 --- a/MPCAutofill/cardpicker/tests/test_harvest_fetch_limiter.py +++ b/MPCAutofill/cardpicker/tests/test_harvest_fetch_limiter.py @@ -4,6 +4,23 @@ per-test (via `get_limiter(config).session`), matching `rate_limited_get`'s own call site (one Session per destination limiter, reused across calls - 2026-07-24 IO audit finding 2) rather than reaching into `image_cdn_fetch`/`local_phash`'s call sites. + +NO DATABASE HERE, AND WHAT THAT MEANS SINCE THE CEILING WENT GLOBAL. As of the 2026-07-30 owner +clarification ("the 7 fetches per second cap is a global cap, it shouldn't be per process or per +core"), `_DestinationLimiter.acquire()` takes its pacing decision from `harvest_rate_coordinator` - +a shared cursor in Postgres. No test in THIS file carries a `django_db` marker, so coordination is +unavailable to them by construction, and every acquisition here therefore runs the DOCUMENTED +DEGRADED FALLBACK: the original per-process pacer, widened by +`harvest_rate_coordinator.degraded_divisor()`. That is deliberate and it is useful - this whole +file is standing evidence that the fallback keeps working and never raises - but it also means THE +TIMINGS BELOW ARE THE DEGRADED ONES (slower than configured, by the divisor), and that nothing in +this file can see the global ceiling at all. Every timing assertion here is a LOWER bound on +elapsed time, so a wider interval does not invalidate any of them. + +The global ceiling has its own file, `test_harvest_rate_coordinator.py`, because proving it needs +MORE THAN ONE limiter running at once - a single-limiter test cannot tell a global ceiling from a +per-process one, which is exactly how the per-process defect survived in this file for as long as +it did. """ import threading diff --git a/MPCAutofill/cardpicker/tests/test_harvest_rate_coordinator.py b/MPCAutofill/cardpicker/tests/test_harvest_rate_coordinator.py new file mode 100644 index 000000000..50da4fab9 --- /dev/null +++ b/MPCAutofill/cardpicker/tests/test_harvest_rate_coordinator.py @@ -0,0 +1,446 @@ +""" +Cross-process rate-ceiling tests for `harvest_rate_coordinator` + `harvest_fetch_limiter`. + +WHY THESE TESTS LOOK LIKE THIS. The defect they exist to catch - a rate limiter whose state lives +in one process's memory, so N processes fetch at N x the configured rate - is INVISIBLE to a +single-instance test. `test_harvest_fetch_limiter.py`'s existing pacing tests all drive ONE +`_DestinationLimiter` and they all passed throughout the defect's life, because one limiter really +does hold its own interval. Every rate assertion below therefore drives MORE THAN ONE independent +pacer at once and asserts the COMBINED rate: + + * `TestAggregateCeilingAcrossProcesses` forks real OS processes - the exact shape django-q2 uses + for its workers, which is where the defect actually lived. + * `TestAggregateCeilingAcrossInstances` drives several independent limiter OBJECTS from threads - + the same discriminator, fast enough to run on every change. + +Both assert an UPPER bound on the aggregate rate AND a LOWER bound on the elapsed span, because +those are the two halves of the same claim: the destination sees no more than the ceiling, and the +work genuinely took as long as a real ceiling would make it take. A per-process-only pacer fails +the elapsed-span assertion by a factor of the process count, which is what makes these tests +non-vacuous. +""" + +import multiprocessing +import time +from typing import Any + +import pytest + +from django.test import override_settings + +from cardpicker import harvest_rate_coordinator +from cardpicker.harvest_fetch_limiter import ( + GOOGLE_IMAGE, + DestinationLimiterConfig, + get_limiter, + reset_limiters, +) + + +@pytest.fixture(autouse=True) +def _reset_limiter_registry(): + reset_limiters() + yield + reset_limiters() + + +def _timed_acquires(config: DestinationLimiterConfig, count: int) -> list[float]: + """Acquires `count` times against `config`'s limiter, returning the wall-clock instant each + acquisition was cleared. `time.time()` (not `time.monotonic()`) deliberately: these timestamps + are merged ACROSS PROCESSES, and only a wall clock is comparable between them.""" + limiter = get_limiter(config) + stamps = [] + for _ in range(count): + with limiter.acquire(): + stamps.append(time.time()) + return stamps + + +def _aggregate_rate(stamps: list[float]) -> float: + """Requests per second across the merged timeline. `len - 1` over the span, not `len` over the + span: the first acquisition consumes no interval (it is the one the ceiling clears + immediately), so N acquisitions at a ceiling of R take (N-1)/R seconds, not N/R.""" + ordered = sorted(stamps) + span = ordered[-1] - ordered[0] + assert span > 0, "every acquisition landed on the same instant - the pacer did nothing at all" + return (len(ordered) - 1) / span + + +def _peak_windowed_rate(stamps: list[float], window: float = 1.0) -> float: + """The worst sliding-`window` burst in the merged timeline, in requests per second. The + aggregate average can hide a burst; a destination experiences the burst.""" + ordered = sorted(stamps) + peak = 0 + for start_index, start in enumerate(ordered): + count = 0 + for stamp in ordered[start_index:]: + if stamp - start > window: + break + count += 1 + peak = max(peak, count) + return peak / window + + +class TestAggregateCeilingAcrossProcesses: + """The proof that matters: separate OS PROCESSES, which is what django-q2's workers are and + what the per-process pacer could never bound.""" + + RATE = 20.0 + PROCESSES = 3 + ACQUIRES_EACH = 10 + + @staticmethod + def _child(name: str, rate: float, count: int, sink: Any) -> None: + # A forked child inherits the parent's limiter registry and its coordinator connection + # object. `harvest_rate_coordinator` drops the latter automatically via its + # `os.register_at_fork` handler; the registry is cleared here so this child's pacer state is + # genuinely its own, exactly as a freshly-spawned django-q2 worker's would be. + # + # Django's own inherited connections are ABANDONED, not `close_all()`d, for exactly the + # reason `harvest_rate_coordinator._forget_connection_after_fork` documents: closing an + # inherited libpq handle sends Terminate down a socket the PARENT is still using, which here + # kills the pytest process's own database session mid-test. Nulling `.connection` leaves the + # parent's session alone and makes the child open its own on first use. + from django.db import connections + + for alias in list(connections): + connections[alias].connection = None + reset_limiters() + config = DestinationLimiterConfig(name=name, rate_per_sec=rate, max_concurrency=10) + sink.put(_timed_acquires(config, count)) + + @pytest.mark.django_db(transaction=True) + def test_three_processes_share_one_ceiling(self) -> None: + """Three processes, each running its own limiter at 20/s, must together deliver 20/s - not + 60/s. With the pre-fix per-process pacer this test finishes in roughly + `ACQUIRES_EACH / RATE` seconds instead of `(PROCESSES * ACQUIRES_EACH - 1) / RATE`, so the + elapsed-span assertion below fails by a factor of `PROCESSES`.""" + name = "test-xproc-ceiling" + harvest_rate_coordinator.clear_cursor(name) + context = multiprocessing.get_context("fork") + sink = context.Queue() + + children = [ + context.Process(target=self._child, args=(name, self.RATE, self.ACQUIRES_EACH, sink)) + for _ in range(self.PROCESSES) + ] + stamps: list[float] = [] + try: + for child in children: + child.start() + for _ in children: + stamps.extend(sink.get(timeout=60)) + finally: + for child in children: + child.join(timeout=60) + if child.exitcode is None: + child.terminate() + + total = self.PROCESSES * self.ACQUIRES_EACH + assert len(stamps) == total + span = max(stamps) - min(stamps) + rate = _aggregate_rate(stamps) + + # Upper bound: the destination never saw more than the ceiling (plus scheduling slop). + assert rate <= self.RATE * 1.25, f"aggregate {rate:.2f}/s exceeded the {self.RATE}/s ceiling" + assert _peak_windowed_rate(stamps) <= self.RATE * 1.25 + # Lower bound: the work actually took as long as a real ceiling makes it take. This is the + # assertion a per-process pacer fails - it would finish ~PROCESSES times sooner. + assert span >= (total - 1) / self.RATE * 0.75, f"finished in {span:.2f}s - too fast to have been paced globally" + + +class TestAggregateCeilingAcrossInstances: + """Same discriminator, independent limiter OBJECTS in threads rather than forked processes. + Independent objects means independent `_next_allowed`/lock/semaphore state - i.e. exactly what + separate processes have - so this catches the same defect in a fraction of the time.""" + + @pytest.mark.django_db(transaction=True) + def test_four_independent_limiters_share_one_ceiling(self) -> None: + import threading + + name = "test-xinstance-ceiling" + rate = 25.0 + instances = 4 + acquires_each = 8 + harvest_rate_coordinator.clear_cursor(name) + + # Deliberately NOT `get_limiter` - that returns the process-wide singleton, which would + # collapse this into the single-instance test that cannot see the defect. + from cardpicker.harvest_fetch_limiter import _DestinationLimiter + + limiters = [ + _DestinationLimiter(DestinationLimiterConfig(name=name, rate_per_sec=rate, max_concurrency=10)) + for _ in range(instances) + ] + stamps: list[float] = [] + stamps_lock = threading.Lock() + + def _drive(limiter: "_DestinationLimiter") -> None: + mine = [] + for _ in range(acquires_each): + with limiter.acquire(): + mine.append(time.time()) + with stamps_lock: + stamps.extend(mine) + + threads = [threading.Thread(target=_drive, args=(limiter,)) for limiter in limiters] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=60) + + total = instances * acquires_each + assert len(stamps) == total + span = max(stamps) - min(stamps) + assert _aggregate_rate(stamps) <= rate * 1.25 + assert _peak_windowed_rate(stamps) <= rate * 1.25 + assert span >= (total - 1) / rate * 0.75, f"finished in {span:.2f}s - four limiters were not sharing a ceiling" + + +class TestSharedCursor: + @pytest.mark.django_db(transaction=True) + def test_reserve_returns_zero_on_a_fresh_cursor_then_a_real_wait(self) -> None: + name = "test-cursor-fresh" + harvest_rate_coordinator.clear_cursor(name) + + first = harvest_rate_coordinator.reserve(name, 0.5) + second = harvest_rate_coordinator.reserve(name, 0.5) + + assert first == pytest.approx(0.0, abs=0.05) + assert 0.4 <= (second or 0.0) <= 0.55 + + @pytest.mark.django_db(transaction=True) + def test_destinations_do_not_share_a_cursor(self) -> None: + for name in ("test-cursor-sep-a", "test-cursor-sep-b"): + harvest_rate_coordinator.clear_cursor(name) + + harvest_rate_coordinator.reserve("test-cursor-sep-a", 0.5) + harvest_rate_coordinator.reserve("test-cursor-sep-a", 0.5) + + # b's budget must be untouched by a's traffic. + assert harvest_rate_coordinator.reserve("test-cursor-sep-b", 0.5) == pytest.approx(0.0, abs=0.05) + + @pytest.mark.django_db(transaction=True) + def test_clear_cursor_forgets_the_budget(self) -> None: + name = "test-cursor-clear" + harvest_rate_coordinator.clear_cursor(name) + harvest_rate_coordinator.reserve(name, 2.0) + assert (harvest_rate_coordinator.reserve(name, 2.0) or 0.0) > 1.0 + + harvest_rate_coordinator.clear_cursor(name) + + assert harvest_rate_coordinator.reserve(name, 2.0) == pytest.approx(0.0, abs=0.05) + + @pytest.mark.django_db(transaction=True) + def test_a_widened_interval_is_honoured(self) -> None: + """PR #644's backoff stays per-process and reaches the shared cursor as a bigger gap. The + cursor must honour it - a global ceiling that ignored a backing-off process's widened + interval would silently undo throttling for everyone else. + + Note which caller pays: a reservation's own gap delays the NEXT caller, not itself (that is + the pre-existing pacer semantic - `wait = next_allowed - now` is read before + `next_allowed += interval`). So the backing-off process is not punished twice; what its + backoff buys is a wider gap in front of whoever fetches after it, which is exactly the + aggregate slowdown the destination asked for.""" + name = "test-cursor-backoff" + harvest_rate_coordinator.clear_cursor(name) + + harvest_rate_coordinator.reserve(name, 0.1) # a normal process: contributes a 0.1s gap + after_normal = harvest_rate_coordinator.reserve(name, 0.8) or 0.0 # backing off; waits 0.1s + after_widened = harvest_rate_coordinator.reserve(name, 0.1) or 0.0 # pays the widened gap + + assert 0.05 <= after_normal <= 0.15 + # These reservations are issued back-to-back with no sleeping, so the third caller's wait is + # the whole accumulated backlog. The BACKOFF's contribution is the difference between the + # two waits, which is the 0.8s gap the widened reservation put into the shared cursor - + # observed by a DIFFERENT caller, which is the property being proven. + assert 0.75 <= after_widened - after_normal <= 0.85 + + +class TestLimiterWiring: + def test_acquire_takes_its_pacing_from_the_shared_cursor_not_the_local_one(self, monkeypatch) -> None: + """Direct proof of the wiring, independent of any timing. `acquire()` must consult the + coordinator on EVERY acquisition and must pass its own current interval (post-backoff), so + a future refactor cannot quietly fall back to `_next_allowed` while the timing tests still + pass on a fast machine.""" + seen: list[tuple[str, float]] = [] + + def _record(destination: str, interval: float) -> float: + seen.append((destination, interval)) + return 0.0 + + monkeypatch.setattr(harvest_rate_coordinator, "reserve", _record) + config = DestinationLimiterConfig(name="test-wiring", rate_per_sec=50.0, max_concurrency=4) + limiter = get_limiter(config) + + with limiter.acquire(): + pass + limiter.backoff() # x2 - the widened interval must reach the coordinator, not stay local + with limiter.acquire(): + pass + + assert [destination for destination, _ in seen] == ["test-wiring", "test-wiring"] + assert [interval for _, interval in seen] == [pytest.approx(0.02), pytest.approx(0.04)] + + def test_the_semaphore_is_not_leaked_if_pacing_blows_up(self, monkeypatch) -> None: + """`acquire()` takes the concurrency semaphore BEFORE it reserves budget. If anything + between the two raised without releasing, the destination would permanently lose a + concurrency slot per occurrence and eventually deadlock the whole fetch pool.""" + + def _explode(destination: str, interval: float) -> float: + raise RuntimeError("coordination exploded in a way reserve() does not cover") + + monkeypatch.setattr(harvest_rate_coordinator, "reserve", _explode) + config = DestinationLimiterConfig(name="test-semaphore-leak", rate_per_sec=1000.0, max_concurrency=1) + limiter = get_limiter(config) + + for _ in range(3): + with pytest.raises(RuntimeError): + limiter.acquire() + + # The single slot is still free - i.e. all three failed acquisitions gave it back. + assert limiter._semaphore.acquire(blocking=False) is True + limiter._semaphore.release() + + +class TestGracefulDegradation: + """Requirement: a process that cannot obtain rate budget WAITS. It does not fail, does not + halt, and does not feed the operating envelope's fetch-failure window.""" + + def test_reserve_returns_none_instead_of_raising_when_the_store_is_unreachable(self, monkeypatch) -> None: + harvest_rate_coordinator.reset_connection() + monkeypatch.setattr( + harvest_rate_coordinator, + "_open_connection", + lambda: (_ for _ in ()).throw(RuntimeError("postgres is gone")), + ) + + assert harvest_rate_coordinator.reserve("test-degrade-none", 0.01) is None + + harvest_rate_coordinator.reset_connection() + + def test_acquire_still_paces_and_never_raises_when_coordination_is_down(self, monkeypatch) -> None: + """The fallback is the ORIGINAL per-process pacer widened by `degraded_divisor()`, so the + aggregate ceiling still holds with every process fetching. Proven here by the elapsed time: + four acquisitions at 20/s cost 3 divided intervals, not 3 full ones.""" + monkeypatch.setattr(harvest_rate_coordinator, "reserve", lambda destination, interval: None) + config = DestinationLimiterConfig(name="test-degrade-pacing", rate_per_sec=20.0, max_concurrency=10) + limiter = get_limiter(config) + divisor = harvest_rate_coordinator.degraded_divisor() + + start = time.monotonic() + for _ in range(4): + with limiter.acquire(): + pass + elapsed = time.monotonic() - start + + assert elapsed >= 3 * 0.05 * divisor * 0.8 + + def test_a_sustained_outage_is_latched_rather_than_retried_every_fetch(self, monkeypatch) -> None: + """A coordination outage must cost one connection attempt per cooldown, not one per fetch - + otherwise a dead Postgres turns every fetch into a connect timeout and the run stalls + harder than the outage itself warrants.""" + harvest_rate_coordinator.reset_connection() + attempts = {"count": 0} + + def _boom() -> Any: + attempts["count"] += 1 + raise RuntimeError("postgres is gone") + + monkeypatch.setattr(harvest_rate_coordinator, "_open_connection", _boom) + + for _ in range(25): + assert harvest_rate_coordinator.reserve("test-degrade-latch", 0.001) is None + + assert attempts["count"] == 1 + harvest_rate_coordinator.reset_connection() + + def test_the_degraded_divisor_covers_every_process_that_can_fetch(self) -> None: + """Derived from the conveyor's own cross-process dispatch cap plus one for a pooled or + manual runner, which holds no dispatch slot. Not a new setting - see the module docstring.""" + with override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=2): + assert harvest_rate_coordinator.degraded_divisor() == 3 + with override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=8): + assert harvest_rate_coordinator.degraded_divisor() == 9 + with override_settings(STAGE_E_MAX_CONCURRENT_DISPATCHES=0): + assert harvest_rate_coordinator.degraded_divisor() == 2 # floored, never a divide by zero + + +class TestCoordinationCost: + """Requirement: do not regress the pooled runner. The pooled path is the monolith's Stage C and + its throughput is why it was chosen, so the per-request Postgres round trip this change adds has + to be MEASURED against the interval it is amortised into, not assumed to be cheap.""" + + @pytest.mark.django_db(transaction=True) + def test_a_reservation_costs_a_small_fraction_of_the_google_interval(self) -> None: + name = "test-cost" + harvest_rate_coordinator.clear_cursor(name) + # A gap of zero makes every reservation return 0.0, so this times the COORDINATION round + # trip alone with no pacing sleep mixed in. + harvest_rate_coordinator.reserve(name, 0.0) # warm the connection; not measured + + samples = 200 + start = time.monotonic() + for _ in range(samples): + assert harvest_rate_coordinator.reserve(name, 0.0) == pytest.approx(0.0, abs=0.05) + per_call = (time.monotonic() - start) / samples + + google_interval = 1.0 / 7.0 + print( + f"\ncoordination cost: {per_call * 1000:.3f} ms/reservation, " + f"{per_call / google_interval * 100:.2f}% of the 7/s interval" + ) + # Loose on purpose - this is a regression guard against the round trip becoming structurally + # expensive (a transaction pair, an advisory-lock round trip, a retry storm), not a + # benchmark. Measured locally at ~0.3ms against a container Postgres; 10ms is 30x that and + # still only 7% of the 143ms interval a 7/s ceiling already imposes. + assert per_call < 0.010, f"a reservation cost {per_call * 1000:.1f} ms - that is no longer amortisable" + + @pytest.mark.django_db(transaction=True) + def test_the_round_trip_does_not_become_the_binding_term_at_the_google_ceiling(self) -> None: + """The cost question that actually matters for the pooled runner: does coordination slow + fetching DOWN, or is it amortised into a pacing interval that already exists? Driven at the + real shipped ceiling (7/s), 8 acquisitions must still take the 7 intervals the ceiling + alone imposes - if the round trip were the binding term this would run measurably long.""" + name = "test-cost-throughput" + harvest_rate_coordinator.clear_cursor(name) + config = DestinationLimiterConfig(name=name, rate_per_sec=GOOGLE_IMAGE.rate_per_sec, max_concurrency=6) + limiter = get_limiter(config) + + start = time.monotonic() + for _ in range(8): + with limiter.acquire(): + pass + elapsed = time.monotonic() - start + + ceiling_only = 7 / GOOGLE_IMAGE.rate_per_sec + print( + f"\n8 acquisitions at {GOOGLE_IMAGE.rate_per_sec}/s: {elapsed:.3f}s " + f"vs {ceiling_only:.3f}s for the ceiling alone ({(elapsed / ceiling_only - 1) * 100:+.1f}%)" + ) + assert elapsed <= ceiling_only * 1.10 + + +class TestForkSafety: + def test_a_forked_child_does_not_inherit_the_parents_connection_object(self) -> None: + """`os.register_at_fork` disposal, checked directly. A child that kept the parent's + connection object would interleave two processes' statements down one libpq socket; a child + that `close()`d it would terminate the parent's session. It must simply forget it.""" + harvest_rate_coordinator._connection = object() # a stand-in; never used, only forgotten + + harvest_rate_coordinator._forget_connection_after_fork() + + assert harvest_rate_coordinator._connection is None + + def test_reset_connection_clears_the_degraded_latch(self, monkeypatch) -> None: + monkeypatch.setattr( + harvest_rate_coordinator, + "_open_connection", + lambda: (_ for _ in ()).throw(RuntimeError("postgres is gone")), + ) + assert harvest_rate_coordinator.reserve("test-latch-clear", 0.001) is None + assert harvest_rate_coordinator._degraded_until > 0.0 + + harvest_rate_coordinator.reset_connection() + + assert harvest_rate_coordinator._degraded_until == 0.0 diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index e89e5bab2..05da3a715 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -133,6 +133,56 @@ fabricated ceiling PR #589 **removed** from `stage_e_batch_sizing`. The honest hardware-vs-destination signal remains #589's own `HostProfile.fetch_overcommitted`. +**7/sec is a GLOBAL cap, not a per-process or per-core one.** This is the part +two separate readers have now got wrong, so it is stated flatly: + +> "to be clear: the 7 fetches per second cap is a global cap, it shouldn't be +> per process or per core" — owner, 2026-07-30 + +The number is a budget for **everything this deployment fetches from Google, +added together**. It is _not_ 7/sec for each django-q2 worker, _not_ 7/sec for +each dispatch, _not_ 7/sec per core, and _not_ 7/sec for the pooled runner +plus another 7/sec for the conveyor running beside it. Google sees one IP; the +budget is that IP's. + +Why that needed saying, and what it cost: `harvest_fetch_limiter ._DestinationLimiter` keeps its pacing state — `_next_allowed`, its lock, its +concurrency semaphore — in **one Python process's memory**. That is a genuine +7/sec for the **pooled runner** (`run_image_evidence_cohort`: one process, one +thread pool, one limiter). It was **never** a ceiling for the **conveyor**: +`stage_e_dispatch` runs under django-q2, whose workers are separate **OS +processes**, each importing the module fresh and pacing itself in isolation. N +concurrent dispatches therefore fetched at **N × 7/sec** — 14/sec at the +shipped `STAGE_E_MAX_CONCURRENT_DISPATCHES = 2`, and rising with that cap, so +tuning the conveyor's concurrency silently retuned the destination's rate. + +The fix is `cardpicker/harvest_rate_coordinator.py`: every process reserves its +turn from **one cursor row in Postgres**, updated by a single atomic +`INSERT ... ON CONFLICT DO UPDATE` that runs the same +`max(now, next_allowed) + interval` arithmetic the in-memory pacer always ran — +just somewhere every process can see it. One round trip per fetch, no new +table, no new migration, no new service (the row lives in the existing +`shared_cache` table). It is on by default with no flag. + +**If you are reasoning about the rate, reason about the total.** To ask what +the destination is actually seeing, add up every fetching process; do not read +one worker's `current_rate()` log line and multiply nothing by it. Concurrency +knobs (`STAGE_E_MAX_CONCURRENT_DISPATCHES`, fetch thread counts) now change +only **how the 7/sec is shared out**, never how much of it there is. + +**When coordination is unreachable, the pass degrades — it does not stop, and +it does not go uncapped.** If Postgres cannot be reached for a reservation, +each process falls back to its own in-memory pacer running at **7/sec ÷ (the +maximum number of processes that can fetch at once)**, derived from +`STAGE_E_MAX_CONCURRENT_DISPATCHES` plus one for a pooled or manual runner. +Worst case — every process fetching — the total is still 7/sec. Failing _open_ +(per-process pacing at the full rate) was rejected because it restores exactly +the N × 7/sec defect at the moment nobody is watching; failing _closed_ +(refusing to fetch) was rejected because it hard-stops an unattended +230,753-card pass over a transient blip, which is the opposite of this ruling. +A degraded pass therefore fetches **more slowly than it could**, logs a warning +once per 5-second window, and keeps going. It raises nothing, trips nothing, +and adds nothing to the fetch-failure window. + **Backoff now decays.** It used to be sticky for the life of the process — the multiplier only ever grew. On a one-shot multi-hour pass that meant a single early 429 pinned the whole run at half speed with no way back. It is now