diff --git a/MPCAutofill/cardpicker/harvest_fetch_limiter.py b/MPCAutofill/cardpicker/harvest_fetch_limiter.py index b0b3c1dd0..8e9281d86 100644 --- a/MPCAutofill/cardpicker/harvest_fetch_limiter.py +++ b/MPCAutofill/cardpicker/harvest_fetch_limiter.py @@ -43,6 +43,16 @@ OWNER RATE RULING (2026-07-30): "the limit needs to be on the amount we are fetching from google api 7/s or hardware whichever comes first. and the limit needs to throttle not shut it down." + * THE CEILING IS GLOBAL, NOT PER PROCESS (owner clarification, 2026-07-30: "the 7 fetches per + second cap is a global cap, it shouldn't be per process or per core"). `_DestinationLimiter` + paces with a `threading.Lock` and a process-local `_next_allowed`, so `rate_per_sec = 7.0` + ORIGINALLY bought 7/s PER PROCESS - correct by accident for the single-process pooled runner, + and wrong for the conveyor, whose django-q2 workers are separate OS PROCESSES each holding + their own limiter (N concurrent dispatches = N x 7/s; 14/s at the production + `STAGE_E_MAX_CONCURRENT_DISPATCHES = 2`, scaling with the cap). The reservation is therefore + made against ONE shared Postgres row - see `cardpicker.harvest_rate_budget`, which owns the + mechanism and the reasoning. Note the asymmetry with `max_concurrency` below, which REMAINS + per-process on purpose: it is a local resource bound, not the destination-protecting ceiling. * THE CEILING IS 7.0/s. `GOOGLE_IMAGE.rate_per_sec` drops 8.0 -> 7.0. The probe-derived 8.0 was the highest measured-clean step; 7.0 is the owner's own ratified number and is strictly under it, so nothing the probe established is contradicted - only tightened. @@ -86,6 +96,13 @@ import requests +from cardpicker.harvest_rate_budget import ( + RateBudgetUnavailable, + record_backoff, + record_clean_response, + reserve_slot, +) + logger = logging.getLogger(__name__) @@ -260,11 +277,20 @@ def lock_out(self) -> None: logger.error("%s destination locked out (403) - this is a hard stop, not a pacing change", self._config.name) def backoff(self) -> None: + """Escalate pacing after a 429/503. The multiplier is advanced in the SHARED row first, so + one process observing rate pressure slows EVERY process (2026-07-30: the interval that + advances the shared `next_allowed_at` has to be a single agreed value, or processes holding + different multipliers write conflicting paces into the same row - see `GlobalFetchPace`'s + own docstring). The local copy is kept in step so the degraded fallback path + (`_local_wait`) stays correct if the row later becomes unreachable.""" + shared = record_backoff(self._config.name, 2.0, self._MAX_BACKOFF_MULTIPLIER) with self._lock: self._backoff_multiplier = min(self._backoff_multiplier * 2.0, self._MAX_BACKOFF_MULTIPLIER) self._clean_streak = 0 - multiplier = self._backoff_multiplier - logger.warning("%s destination backing off (429) - pacing interval now x%.1f", self._config.name, multiplier) + multiplier = shared if shared is not None else self._backoff_multiplier + logger.warning( + "%s destination backing off (429/503) - GLOBAL pacing interval now x%.1f", self._config.name, multiplier + ) def note_clean_response(self) -> None: """One response that was neither a lockout nor a backoff code - the decay half of the @@ -275,7 +301,15 @@ def note_clean_response(self) -> None: back never accumulates enough clean responses to recover. A no-op while the multiplier is already 1.0, which is the overwhelmingly common case - a - run that never sees a 429 never touches the streak counter at all.""" + run that never sees a 429 never touches the streak counter at all. + + BOTH the streak and the multiplier live in the SHARED row (2026-07-30), for the same reason + the backoff does and one more: N processes each counting their own streak would reach the + decay threshold N times over and recover N times faster than the single agreed schedule + intends - quietly accelerating recovery in exactly the situation (many workers, sustained + pressure) where it is supposed to be slowest. The local copies below are maintained only to + keep `_local_wait`'s degraded fallback honest.""" + record_clean_response(self._config.name, self._CLEAN_RESPONSES_BEFORE_DECAY, self._MIN_BACKOFF_MULTIPLIER) with self._lock: if self._backoff_multiplier <= self._MIN_BACKOFF_MULTIPLIER: self._clean_streak = 0 @@ -294,19 +328,59 @@ def note_clean_response(self) -> None: ) def acquire(self) -> "_LimiterSlot": + """Claim this caller's turn, then block until it arrives. + + THE PACING RESERVATION IS GLOBAL (owner clarification, 2026-07-30: "the 7 fetches per + second cap is a global cap, it shouldn't be per process or per core"). The wait comes from + `harvest_rate_budget.reserve_slot`, a single atomic UPDATE against one shared Postgres row, + so every fetching process in the deployment draws from ONE request sequence. The + process-local `_next_allowed` arithmetic this used to do is now the FALLBACK only - see + `_local_wait` - because it silently meant `rate_per_sec` PER PROCESS, i.e. N x 7/s across N + django-q2 workers. + + The sleep itself happens HERE, after the reservation returns and outside any database lock + or transaction: the row is advanced and released in one statement, and this thread then + waits on its own. A slow caller therefore delays only itself, never the shared sequence. + + The concurrency semaphore is unchanged and remains deliberately per-process - it is a local + resource bound (how many of THIS process's threads may be in flight at once), not the + destination-protecting ceiling. The rate limit is that, which is exactly why it is the one + that had to become global.""" if self._locked_out: raise GoogleFetchLockoutError(f"{self._config.name} is locked out (403) - refusing further requests") self._semaphore.acquire() + try: + wait_time = reserve_slot(self._config.name, self._interval) + except RateBudgetUnavailable as exc: + # DEGRADE, never fail (module docstring's own posture, and `harvest_rate_budget`'s + # "FAIL-OPEN TO LOCAL PACING" section): the ceiling reverts to per-process, which is + # where this repo already was before the shared row existed, and the run keeps going. + logger.error( + "%s: global rate budget unavailable (%s) - DEGRADED to per-process pacing at %.1f/s " + "for this request; the aggregate ceiling is not being enforced while this persists", + self._config.name, + exc, + self._config.rate_per_sec, + ) + wait_time = self._local_wait() 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 if wait_time > 0: time.sleep(wait_time) return _LimiterSlot(self._semaphore) + def _local_wait(self) -> float: + """The ORIGINAL per-process pacing arithmetic, kept verbatim as the degraded fallback for + when the shared row is unreachable. Uses `time.monotonic()`, whose epoch is per-process - + which is precisely why this cannot be the primary path and why the shared row uses the + DATABASE's `clock_timestamp()` instead (see `harvest_rate_budget`'s own docstring).""" + 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 + return wait_time + class _LimiterSlot: """Context manager returned by `_DestinationLimiter.acquire()` - releases the concurrency diff --git a/MPCAutofill/cardpicker/harvest_rate_budget.py b/MPCAutofill/cardpicker/harvest_rate_budget.py new file mode 100644 index 000000000..bad33057d --- /dev/null +++ b/MPCAutofill/cardpicker/harvest_rate_budget.py @@ -0,0 +1,400 @@ +""" +THE GLOBAL (cross-process) fetch-rate budget - the enforcement half of the owner clarification of +2026-07-30: "the 7 fetches per second cap is a global cap, it shouldn't be per process or per +core". + +WHAT WAS WRONG. `harvest_fetch_limiter._DestinationLimiter` paces with a `threading.Lock` and a +process-local `_next_allowed`, so `GOOGLE_IMAGE.rate_per_sec = 7.0` bought 7/s PER PROCESS: + + * the pooled runner (`run_image_evidence_cohort`) is single-process, so it was correct by + accident - one limiter, one process, a genuine 7/s; + * the conveyor is not. django-q2 workers are separate OS PROCESSES (`multiprocessing`, not + threads - `stage_e_concurrency`'s own docstring establishes this for the concurrency cap), so + each worker constructs its OWN limiter from the module-level registry in ITS OWN address space + and paces independently. N concurrent dispatches therefore issued N x 7/s. At the production + `settings.STAGE_E_MAX_CONCURRENT_DISPATCHES = 2` that is 14/s against a 7/s ruling, and it + scales with the cap. + +This is the same per-process trap as `threading.Semaphore(config.max_concurrency)` - which this +project has now been bitten by twice, and which PR #589 deleted a false ceiling term from +`stage_e_batch_sizing` over. PR #644 established that "the rate limit, not the concurrency limit, +is what actually protects the destination", which is exactly what makes the rate limit's +GLOBALITY the load-bearing property rather than a refinement. + +MECHANISM - one Postgres row per destination (`cardpicker.models.GlobalFetchPace`), advanced by a +single atomic `UPDATE ... RETURNING`: + + UPDATE cardpicker_globalfetchpace + SET next_allowed_at = GREATEST(clock_timestamp(), next_allowed_at) + + WHERE destination = %s + RETURNING next_allowed_at, clock_timestamp(), backoff_multiplier + +That is the SAME arithmetic `_DestinationLimiter.acquire()` already did in Python +(`next_allowed = max(now, next_allowed) + interval`), lifted verbatim into the one place every +process can see. The statement returns the slot this caller claimed; the caller then sleeps until +it LOCALLY, holding no lock and no transaction while it waits. + +WHY THIS IS ATOMIC ACROSS PROCESSES, which is the whole correctness claim: two concurrent UPDATEs +against the same row serialise on Postgres's own row lock, and under READ COMMITTED (Django's +default) the blocked statement RE-EVALUATES its SET expression against the committed new row +version once it acquires the lock. So the second writer reads the first writer's advanced +`next_allowed_at`, not the stale one it originally saw, and adds its own interval on top. The +reservations therefore form a single strictly-increasing sequence spaced by the interval, no +matter how many processes are competing. Nothing here is advisory or best-effort. + +WHY A ROW RATHER THAN `stage_e_concurrency`'s ADVISORY LOCKS - the alternative the owner +explicitly asked to be evaluated first, since that module is this repo's established cross-process +primitive. It cannot carry a rate, and its own reasoning inverts here: + + * An advisory lock is a BINARY held/not-held token. It expresses "how many at once" + (concurrency); it cannot express "how many per unit time", because a rate needs a remembered + TIMESTAMP and a lock stores no value. The only way to fake it - hold each of N locks for + exactly 1/R seconds - turns every fetching process into a sleeper occupying a Postgres session + for the duration, and makes the rate a function of how long a lock is HELD rather than how + often it is ACQUIRED. That is a worse primitive in every dimension. + * `stage_e_concurrency` rejected a DB row for CRASH SAFETY: a `kill -9`'d worker leaves a + row-based slot claimed forever, because nothing ties a row to a process's lifetime, whereas a + session-scoped advisory lock is auto-released when the connection dies. That objection is + about a CLAIM. This row holds no claim - only a timestamp. A process killed mid-reservation + leaves `next_allowed_at` at most one interval in the future, self-healing in ~143ms at 7/s + with zero reconciliation code. The exact property that made a row unsafe for a slot makes it + correct for a pace. + +Redis was not considered a real option: `docker/docker-compose.prod.yml` runs django, worker, +nginx, postgres and elasticsearch and NO Redis, and Django's cache here is the default per-PROCESS +`LocMemCache` (the same fact `stage_e_concurrency` and `cardpicker.review_clusters` both already +establish). Postgres is the one piece of shared, process-visible state this pipeline guarantees. + +THE DATABASE'S CLOCK, NEVER PYTHON'S. Every timestamp in this module comes from +`clock_timestamp()`. This is load-bearing, not stylistic: `_DestinationLimiter` uses +`time.monotonic()`, whose epoch is PER PROCESS and meaningless to compare across processes. +Persisting one process's monotonic reading for another to read would silently reintroduce exactly +the defect this module exists to remove, while LOOKING coordinated. `clock_timestamp()` (not +`now()`/`CURRENT_TIMESTAMP`, which are transaction-start times and would hand every statement in a +transaction the same instant) is the one clock every process shares. + +A DEDICATED, THREAD-CACHED CONNECTION. Reservations run on a `psycopg2` connection this module +opens for itself, cached per thread for the life of the process - NOT `django.db.connection`. +Two reasons, and the first is a correctness bug rather than a preference: + + 1. TRANSACTION ISOLATION. The row lock a reservation takes is released when its transaction + commits. On `django.db.connection`, a reservation made INSIDE a caller's `atomic()` block + would hold that lock until the caller's whole transaction committed - serialising every other + fetching process in the deployment behind one long transaction, converting a 143ms pace into + an unbounded stall. On a dedicated `autocommit=True` connection the lock is released the + instant the statement returns, always, regardless of what the calling code is doing. + 2. COST. Opening a connection per request would be absurd at 7/s x N processes; opening one per + FETCH THREAD, once, is not. The cache is a `threading.local`, so each fetch thread (one for + the conveyor, the pool size for `run_image_evidence_cohort`) pays a single connect for the + life of the process. See `_thread_connection`. + +FAIL-OPEN TO LOCAL PACING, NEVER FAIL-STOPPED. If the reservation cannot be made (database +unreachable, connection dies mid-statement), this module raises `RateBudgetUnavailable` and +`_DestinationLimiter` falls back to its own in-process pacer. That is a DEGRADATION - the ceiling +becomes per-process again, which is where this repo already was - and it is logged as an error, +but it does not fail or halt the run. This matches the owner's own standing requirement that a +process which cannot get rate budget WAITS rather than dying, and it matches PR #644's whole +posture: rate machinery degrades, it does not shut things down. Note this is the OPPOSITE choice +from `stage_e_concurrency`'s deliberate fail-CLOSED, and for a reason: an unavailable CONCURRENCY +cap means an unbounded number of dispatches can pile onto the host, so refusing is cheaper than +proceeding; an unavailable RATE budget still leaves every process paced at the configured ceiling +by its own local limiter, so the downside of proceeding is bounded and known, while refusing would +stop the pipeline outright over a transient database blip. +""" + +import logging +import threading +from typing import Any, Optional + +import psycopg2 + +from django.db import connection as django_connection + +logger = logging.getLogger(__name__) + + +class RateBudgetUnavailable(Exception): + """Raised when a reservation could not be made against the shared row - the caller + (`harvest_fetch_limiter._DestinationLimiter.acquire`) is expected to fall back to its own + in-process pacing and continue, never to fail or halt. See the module docstring's + "FAIL-OPEN TO LOCAL PACING" section for why this degrades rather than stops.""" + + +_TABLE = "cardpicker_globalfetchpace" + +# One dedicated psycopg2 connection per THREAD, reused for that thread's whole life (module +# docstring's "A DEDICATED, THREAD-CACHED CONNECTION"). Not a pool and not per-request: the set of +# threads that ever fetch is small and long-lived (one fetch-ahead thread per conveyor dispatch, +# the fetch pool in `run_image_evidence_cohort`), so a `threading.local` gives one connect per +# thread rather than one per request, with no pool bookkeeping of its own. +_local = threading.local() + +# Guards `_ensure_row`'s own "has this process already created the row" memo. The row creation is +# idempotent in SQL (`ON CONFLICT DO NOTHING`), so this is purely to skip a redundant round trip on +# every reservation after the first, not a correctness lock. +_ensured_lock = threading.Lock() +_ensured: set[str] = set() + +# Every connection `_thread_connection` has ever opened, so `reset_for_tests` can close ALL of +# them, not just the calling thread's. Without this a test that drove reservations from worker +# threads left those threads' connections open after the threads themselves had exited, and +# Postgres refused to drop the test database ("is being accessed by other users") - a teardown +# failure in one test file that would surface as a confusing error in whichever file ran next. +# Production never needs this (fetch threads live as long as the process), but a registry that +# makes cleanup POSSIBLE costs one list append per thread. +_all_connections_lock = threading.Lock() +_all_connections: list[Any] = [] + +# Connection parameters, resolved ONCE per process and reused by every thread. Deliberately not +# re-read per thread: `django.db.connection` is itself thread-local, so touching it from a fetch +# thread OPENS A DJANGO CONNECTION on that thread which nothing will ever close (Django closes the +# request/task thread's connection, not an arbitrary worker's). In production those threads live as +# long as the process so the leak is bounded and invisible; in the test suite it left orphaned +# sessions behind and Postgres refused to drop the test database. Resolving the params once and +# never consulting Django's connection again from a fetch thread avoids the whole class of problem. +_params_lock = threading.Lock() +_params: Optional[dict[str, Any]] = None + + +def _thread_connection() -> "psycopg2.extensions.connection": + """This thread's dedicated autocommit connection, opened on first use and reused thereafter. + + Connection PARAMETERS come from Django's own connection (`get_connection_params()`, after + `ensure_connection()` since the params alone don't establish one) rather than separately- + guessed settings, so this always targets whatever database Django is actually configured + against - including a test run's prefixed database name. Same discipline + `stage_e_concurrency._open_dedicated_connection` already established. + + `autocommit=True` is the load-bearing setting, not a tidiness one: it is what guarantees the + row lock a reservation takes is released the moment the statement returns, rather than being + held for the lifetime of whatever transaction the calling code happens to be inside (module + docstring, "TRANSACTION ISOLATION"). + + A connection that has been closed underneath us (database restart, network drop, idle + timeout) is detected via `conn.closed` and transparently replaced - a fetch thread that has + been alive for hours must not start failing every reservation because of one blip. + """ + conn: Optional[Any] = getattr(_local, "conn", None) + if conn is not None and not conn.closed: + return conn + raw = psycopg2.connect(**_connection_params()) + raw.autocommit = True + _local.conn = raw + with _all_connections_lock: + _all_connections.append(raw) + return raw + + +def _connection_params() -> dict[str, Any]: + """Django's own connection parameters, resolved once per process and cached. + + Read from `django.db.connection.get_connection_params()` rather than separately-guessed + settings so this always targets whatever database Django is actually configured against - + including a test run's prefixed database name, which is written into `settings_dict` by the + test runner before any of this executes. Same discipline + `stage_e_concurrency._open_dedicated_connection` established, with one deliberate difference: + NO `ensure_connection()` call. That would open a Django connection on whichever thread happened + to ask first, and Django connections are thread-local with nothing to close a fetch thread's + (see `_params`' own comment). `get_connection_params()` reads `settings_dict` and needs no live + connection of its own.""" + global _params + if _params is not None: + return _params + with _params_lock: + if _params is None: + _params = django_connection.get_connection_params() + return _params + + +def _discard_thread_connection() -> None: + """Drops this thread's cached connection after an error, so the NEXT reservation opens a fresh + one instead of retrying forever against a connection that is already broken.""" + conn = getattr(_local, "conn", None) + _local.conn = None + if conn is not None: + try: + conn.close() + except Exception: # noqa: BLE001 - closing a already-broken connection must never mask the + # original failure that got us here. + pass + + +def _ensure_row(conn: "psycopg2.extensions.connection", destination: str) -> None: + """Creates this destination's row if it does not exist yet, idempotently and racelessly + (`ON CONFLICT DO NOTHING` - two processes starting at the same instant both issue this and + exactly one wins, with neither raising). `next_allowed_at` seeds to `clock_timestamp()`, i.e. + "the next request may go immediately", so a fresh deployment does not pay a phantom wait. + + Memoised per process so this costs one extra round trip on the first reservation only, never + on the 230,752 after it.""" + if destination in _ensured: + return + with conn.cursor() as cursor: + cursor.execute( + f"INSERT INTO {_TABLE} (destination, next_allowed_at, backoff_multiplier, clean_streak) " + "VALUES (%s, clock_timestamp(), 1.0, 0) ON CONFLICT (destination) DO NOTHING", + [destination], + ) + with _ensured_lock: + _ensured.add(destination) + + +def reserve_slot(destination: str, base_interval_seconds: float) -> float: + """ + Claim this caller's place in the GLOBAL request sequence for `destination`, and return how + many seconds it must wait before issuing its request. The caller sleeps that long itself, + holding nothing. + + The reservation is one atomic statement (module docstring): it advances the shared + `next_allowed_at` by one interval and returns the slot it just claimed. Concurrent callers in + ANY process serialise on the row lock and each get a distinct, correctly-spaced slot. + + `base_interval_seconds` is `1 / rate_per_sec` WITHOUT the backoff multiplier - the multiplier + lives in the row and is applied in SQL, so every process paces by the same agreed value (see + `GlobalFetchPace`'s own docstring for why a global rate ceiling requires a global backoff + term). A returned wait is never negative: a caller that has fallen behind the shared sequence + proceeds immediately rather than being handed a nonsensical negative sleep. + + Raises `RateBudgetUnavailable` if the reservation could not be made - see the module + docstring's "FAIL-OPEN TO LOCAL PACING" section; the caller degrades to local pacing and keeps + going, it never fails or halts. + """ + try: + conn = _thread_connection() + _ensure_row(conn, destination) + with conn.cursor() as cursor: + cursor.execute( + f"UPDATE {_TABLE} " + " SET next_allowed_at = GREATEST(clock_timestamp(), next_allowed_at) " + " + (%s * backoff_multiplier) * INTERVAL '1 second' " + " WHERE destination = %s " + "RETURNING EXTRACT(EPOCH FROM (next_allowed_at - clock_timestamp())) " + " - (%s * backoff_multiplier)", + [base_interval_seconds, destination, base_interval_seconds], + ) + row = cursor.fetchone() + except Exception as exc: # noqa: BLE001 - every failure mode degrades identically, see docstring + _discard_thread_connection() + raise RateBudgetUnavailable(f"could not reserve a global rate slot for {destination}: {exc}") from exc + + if row is None: + # The row vanished between `_ensure_row` and the UPDATE (someone truncated the table). Drop + # the memo so the next call recreates it, and degrade this one request to local pacing. + with _ensured_lock: + _ensured.discard(destination) + raise RateBudgetUnavailable(f"no {_TABLE} row for {destination} - it was created and then removed") + + # `next_allowed_at` in RETURNING is the POST-update value (this caller's slot PLUS one + # interval), so subtracting the interval back off yields the wait until this caller's OWN slot. + return max(0.0, float(row[0])) + + +def record_backoff(destination: str, factor: float, ceiling: float) -> Optional[float]: + """Multiply the SHARED backoff multiplier (capped at `ceiling`) and reset the shared clean + streak - the global half of `_DestinationLimiter.backoff()`. A 429 observed by ONE process + must slow every process, or the shared `next_allowed_at` is being advanced by intervals the + processes disagree about and the effective rate is whatever the least-backed-off process + believes. + + Returns the new multiplier, or `None` if the update could not be made (the caller keeps its + own local multiplier and continues - same degradation posture as `reserve_slot`).""" + return _update_multiplier( + destination, + "SET backoff_multiplier = LEAST(backoff_multiplier * %s, %s), clean_streak = 0", + [factor, ceiling], + ) + + +def record_clean_response(destination: str, streak_target: int, floor: float) -> Optional[float]: + """Advance the SHARED clean streak and, once it reaches `streak_target`, halve the shared + multiplier (floored at `floor`) and reset the streak - the global half of + `_DestinationLimiter.note_clean_response()`. + + The streak MUST be shared, not per-process, for the same reason the multiplier is: N processes + each counting their own streak would reach the target N times over and decay the shared + multiplier N times faster than the single agreed schedule intends, quietly accelerating + recovery in exactly the situation (many workers, sustained pressure) where it should be + slowest. Done in one atomic statement so two processes crossing the threshold together cannot + both halve. + + Returns the current multiplier, or `None` if the update could not be made.""" + return _update_multiplier( + destination, + "SET clean_streak = CASE WHEN backoff_multiplier <= %s THEN 0 " + " WHEN clean_streak + 1 >= %s THEN 0 " + " ELSE clean_streak + 1 END, " + " backoff_multiplier = CASE WHEN backoff_multiplier > %s AND clean_streak + 1 >= %s " + " THEN GREATEST(backoff_multiplier / 2.0, %s) " + " ELSE backoff_multiplier END", + [floor, streak_target, floor, streak_target, floor], + ) + + +def _update_multiplier(destination: str, set_clause: str, params: list[Any]) -> Optional[float]: + """Shared plumbing for the two mutators above - one atomic UPDATE returning the resulting + multiplier, degrading to `None` (never raising at the caller) on any failure.""" + try: + conn = _thread_connection() + _ensure_row(conn, destination) + with conn.cursor() as cursor: + cursor.execute( + f"UPDATE {_TABLE} {set_clause} WHERE destination = %s RETURNING backoff_multiplier", + [*params, destination], + ) + row = cursor.fetchone() + except Exception: # noqa: BLE001 - degrade, never raise into the fetch path + _discard_thread_connection() + logger.exception("harvest_rate_budget: could not update the shared backoff state for %s", destination) + return None + return None if row is None else float(row[0]) + + +def current_state(destination: str) -> Optional[tuple[float, int]]: + """`(backoff_multiplier, clean_streak)` for this destination, or `None` if there is no row yet. + Read-only observability for tests and on-call inspection - nothing in the fetch path calls + this.""" + try: + conn = _thread_connection() + with conn.cursor() as cursor: + cursor.execute( + f"SELECT backoff_multiplier, clean_streak FROM {_TABLE} WHERE destination = %s", [destination] + ) + row = cursor.fetchone() + except Exception: # noqa: BLE001 + _discard_thread_connection() + return None + return None if row is None else (float(row[0]), int(row[1])) + + +def reset_for_tests() -> None: + """Test-only: drops this process's row-existence memo and closes EVERY connection this module + has opened on any thread, so a test that truncated the table (or swapped databases) starts + clean and Postgres can actually drop the test database afterwards. Mirrors + `harvest_fetch_limiter.reset_limiters`' own purpose for the in-process registry. + + Closing all of them, not just the calling thread's, is the point: a test that drove + reservations from worker threads leaves those threads' connections open after the threads have + exited, and the test-database teardown then fails with "is being accessed by other users".""" + global _params + with _ensured_lock: + _ensured.clear() + with _params_lock: + _params = None + _local.conn = None + with _all_connections_lock: + connections, _all_connections[:] = list(_all_connections), [] + for conn in connections: + try: + conn.close() + except Exception: # noqa: BLE001 - a connection that is already gone is already clean + pass + + +__all__ = [ + "RateBudgetUnavailable", + "current_state", + "record_backoff", + "record_clean_response", + "reserve_slot", + "reset_for_tests", +] diff --git a/MPCAutofill/cardpicker/migrations/0102_global_fetch_pace.py b/MPCAutofill/cardpicker/migrations/0102_global_fetch_pace.py new file mode 100644 index 000000000..ad572cd82 --- /dev/null +++ b/MPCAutofill/cardpicker/migrations/0102_global_fetch_pace.py @@ -0,0 +1,26 @@ +# Generated by Django 4.2.30 on 2026-07-30 09:48 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("cardpicker", "0101_delete_printingtagvote"), + ] + + operations = [ + migrations.CreateModel( + name="GlobalFetchPace", + fields=[ + ("destination", models.CharField(max_length=64, primary_key=True, serialize=False)), + ("next_allowed_at", models.DateTimeField()), + ("backoff_multiplier", models.FloatField(default=1.0)), + ("clean_streak", models.IntegerField(default=0)), + ], + options={ + "verbose_name": "global fetch pace", + "verbose_name_plural": "global fetch paces", + }, + ), + ] diff --git a/MPCAutofill/cardpicker/models.py b/MPCAutofill/cardpicker/models.py index d7413bbe6..166af2eb4 100755 --- a/MPCAutofill/cardpicker/models.py +++ b/MPCAutofill/cardpicker/models.py @@ -2648,6 +2648,75 @@ def __str__(self) -> str: return f"anonymous_id={self.anonymous_id} pool={self.pool} question_type={self.question_type}" +class GlobalFetchPace(models.Model): + """ + THE CROSS-PROCESS PACING STATE for one outbound fetch destination - the shared row that makes + `harvest_fetch_limiter`'s req/s ceiling a GLOBAL cap rather than a per-process one. + + WHY THIS EXISTS (owner clarification, 2026-07-30: "the 7 fetches per second cap is a global + cap, it shouldn't be per process or per core"). `_DestinationLimiter` paces with a + `threading.Lock` and a process-local `_next_allowed`, so `rate_per_sec = 7.0` meant 7/s PER + PROCESS. The pooled runner (`run_image_evidence_cohort`) is single-process, so it was correct + by accident; the conveyor is not - django-q2 workers are separate OS PROCESSES + (`multiprocessing`, `Q_CLUSTER["workers"]`), each constructing its own limiter and pacing + independently, so N concurrent dispatches produced N x 7/s. At the production + `STAGE_E_MAX_CONCURRENT_DISPATCHES = 2` that is 14/s against a 7/s ruling, scaling with the + cap. This is the same per-process trap as `threading.Semaphore(max_concurrency)` - the one + `stage_e_batch_sizing` (PR #589) removed a false ceiling term over, and the one + `stage_e_concurrency` was built to escape for slot counts. + + WHY A ROW AND NOT AN ADVISORY LOCK - the question `stage_e_concurrency`'s own docstring + forces, since that module rejected a DB row for slot counting and chose advisory locks + instead. Both halves of that reasoning INVERT for a rate: + + 1. An advisory lock is a BINARY held/not-held token. It can express "how many at once" + (concurrency); it cannot express "how many per unit time" (a rate), because a rate needs + a remembered TIMESTAMP and a lock stores no value. Enforcing 7/s with N binary locks + would mean holding each for exactly 1/7s - turning every fetching process into a sleeper + occupying a Postgres session, and making the rate a function of how long a lock is held + rather than how often it is acquired. + 2. The crash-safety objection that made a DB row WRONG for slots does not apply here. That + objection was that a `kill -9`'d worker leaves a claimed slot claimed forever, since + nothing ties a row to a process's lifetime. This row holds no CLAIM - only a timestamp. A + process killed mid-reservation leaves `next_allowed_at` at most one interval in the + future, which self-heals in ~143ms at 7/s with zero reconciliation code. The very + property that made a row unsafe for a claim makes it exactly right for a pace. + + ONE ROW PER DESTINATION, created on demand (`harvest_rate_budget._ensure_row`), never deleted. + `destination` matches `DestinationLimiterConfig.name` - the same key the in-process limiter + registry uses, so the two are always talking about the same destination. + + `next_allowed_at` IS WRITTEN AND READ ONLY IN THE DATABASE'S OWN CLOCK (`clock_timestamp()`), + never a Python-side timestamp. This is load-bearing, not stylistic: the in-process limiter + uses `time.monotonic()`, whose epoch is PER PROCESS and meaningless to compare across + processes - persisting one process's monotonic reading for another to read would silently + reintroduce exactly the per-process defect this row exists to fix. + + `backoff_multiplier`/`clean_streak` are here for the same reason, not as an extra feature: the + interval that advances `next_allowed_at` must be a SINGLE agreed value across every process, + or two processes holding different multipliers write conflicting paces into the same row and + the effective rate becomes whatever the least-backed-off process believes. A global rate + ceiling requires a global backoff term. The semantics are `harvest_fetch_limiter`'s own, + unchanged (doubling per 429/503, capped; halving per `_CLEAN_RESPONSES_BEFORE_DECAY` + consecutive clean responses; floored at 1.0) - only the storage moved. + """ + + destination = models.CharField(max_length=64, primary_key=True) + # The instant the NEXT request to this destination may be issued, in the DATABASE's clock. A + # reservation reads it, advances it by one interval, and returns the slot it claimed - all in + # one atomic UPDATE (see `harvest_rate_budget.reserve_slot`). + next_allowed_at = models.DateTimeField() + backoff_multiplier = models.FloatField(default=1.0) + clean_streak = models.IntegerField(default=0) + + class Meta: + verbose_name = "global fetch pace" + verbose_name_plural = "global fetch paces" + + def __str__(self) -> str: + return f"{self.destination} next_allowed_at={self.next_allowed_at.isoformat()} x{self.backoff_multiplier:.1f}" + + __all__ = [ "Faces", "CardTypes", @@ -2673,4 +2742,5 @@ def __str__(self) -> str: "ImageEvidence", "QuestionFeedServedPool", "QuestionFeedServedLog", + "GlobalFetchPace", ] diff --git a/MPCAutofill/cardpicker/tests/test_harvest_rate_budget.py b/MPCAutofill/cardpicker/tests/test_harvest_rate_budget.py new file mode 100644 index 000000000..31ea81805 --- /dev/null +++ b/MPCAutofill/cardpicker/tests/test_harvest_rate_budget.py @@ -0,0 +1,261 @@ +""" +Tests for the GLOBAL (cross-process) fetch-rate budget - `cardpicker.harvest_rate_budget` and the +`_DestinationLimiter` wiring that consumes it. + +THE POINT OF THIS FILE, stated plainly because a weaker version of it is what let the defect +through: every meaningful test here runs MORE THAN ONE limiter instance concurrently, each with its +own pacing state and its own database connection. A single-instance test CANNOT detect a +per-process rate cap - `test_harvest_fetch_limiter.py`'s existing pacing tests all passed against a +limiter that gave N x the configured rate across N processes, because each of them only ever +exercised one. The multi-instance shape is the test, not an implementation detail of it. + +Independent `_DestinationLimiter` objects on separate threads are a faithful stand-in for separate +OS processes here: each has its own `_next_allowed`/`_backoff_multiplier` (exactly what a forked +django-q2 worker gets), and `harvest_rate_budget`'s connection cache is a `threading.local`, so +each thread also reserves on its own database connection. The coordination surface under test - one +shared row, N independent reservers - is identical. +""" + +import threading +import time +from typing import Any, List + +import pytest + +from cardpicker import harvest_rate_budget +from cardpicker.harvest_fetch_limiter import ( + DestinationLimiterConfig, + _DestinationLimiter, + reset_limiters, +) +from cardpicker.harvest_rate_budget import ( + current_state, + record_backoff, + record_clean_response, + reserve_slot, +) + + +@pytest.fixture(autouse=True) +def _reset_shared_state(db: Any): + from cardpicker.models import GlobalFetchPace + + reset_limiters() + harvest_rate_budget.reset_for_tests() + GlobalFetchPace.objects.all().delete() + yield + reset_limiters() + harvest_rate_budget.reset_for_tests() + + +def _drive(limiters: List[_DestinationLimiter], requests_each: int) -> float: + """Run every limiter concurrently, `requests_each` acquisitions apiece, and return the wall + time the whole thing took. Each limiter runs on its own thread, so each reserves on its own + database connection - the multi-process shape this file exists to exercise.""" + barrier = threading.Barrier(len(limiters)) + errors: List[BaseException] = [] + + def _run(limiter: _DestinationLimiter) -> None: + try: + barrier.wait() # all threads start together, so the measured window is the real one + for _ in range(requests_each): + with limiter.acquire(): + pass + except BaseException as exc: # noqa: BLE001 - surfaced in the main thread below + errors.append(exc) + + threads = [threading.Thread(target=_run, args=(limiter,)) for limiter in limiters] + started = time.monotonic() + for thread in threads: + thread.start() + for thread in threads: + thread.join() + elapsed = time.monotonic() - started + if errors: + raise errors[0] + return elapsed + + +def _independent_limiters(count: int, name: str, rate_per_sec: float) -> List[_DestinationLimiter]: + """`count` limiter instances for the SAME destination, each constructed separately so none of + them shares pacing state with any other - i.e. what N separate django-q2 worker processes + actually get from the module-level registry in their own address spaces. Deliberately NOT + `get_limiter()`, which would hand back one shared instance and quietly turn this into a + single-process test.""" + config = DestinationLimiterConfig(name=name, rate_per_sec=rate_per_sec, max_concurrency=10) + return [_DestinationLimiter(config) for _ in range(count)] + + +class TestTheAggregateRateIsCapped: + """Requirement 1: the aggregate across all fetching processes must not exceed the ceiling.""" + + @pytest.mark.django_db(transaction=True) + def test_four_independent_limiters_share_one_budget(self) -> None: + """THE TEST THIS CHANGE EXISTS FOR. Four independent limiters (= four worker processes) at + a configured 50/s must together take at least (40-1)/50 = 0.78s to issue 40 requests. + + Under the per-process pacing this replaces, each limiter would pace itself independently + and the whole thing would finish in about (10-1)/50 = 0.18s at an aggregate ~220/s - which + is exactly the 4x overshoot the owner's clarification identified. The margin between the + two outcomes is 4x, far outside any timing noise this assertion tolerates.""" + rate = 50.0 + processes, each = 4, 10 + limiters = _independent_limiters(processes, "test-global-aggregate", rate) + + elapsed = _drive(limiters, each) + + total = processes * each + floor = (total - 1) / rate + assert elapsed >= floor * 0.9, ( + f"{total} requests across {processes} independent limiters took {elapsed:.3f}s; a global " + f"{rate}/s cap requires at least {floor:.3f}s. A per-process cap would finish in " + f"~{(each - 1) / rate:.3f}s - that is the defect this asserts against." + ) + achieved = total / elapsed + assert achieved <= rate * 1.2, f"aggregate {achieved:.1f}/s exceeded the {rate}/s ceiling" + + @pytest.mark.django_db(transaction=True) + def test_a_single_limiter_still_paces_correctly(self) -> None: + """The one-process case must not regress: the same ceiling, reached the same way.""" + rate = 50.0 + limiters = _independent_limiters(1, "test-global-single", rate) + + elapsed = _drive(limiters, 10) + + assert elapsed >= (10 - 1) / rate * 0.9 + + @pytest.mark.django_db(transaction=True) + def test_reservations_are_a_single_strictly_increasing_sequence(self) -> None: + """The mechanism, asserted directly rather than through wall time: concurrent reservers + each get a DISTINCT slot spaced by the interval, because the blocked UPDATE re-evaluates + against the committed row under READ COMMITTED. If two reservers could read the same + `next_allowed_at` and both add to it, waits would collide and the sum would be short.""" + interval = 0.05 + slots: List[float] = [] + slots_lock = threading.Lock() + barrier = threading.Barrier(8) + + def _reserve() -> None: + barrier.wait() + wait = reserve_slot("test-global-sequence", interval) + # The ABSOLUTE instant this reserver was granted, not the relative wait it was handed. + # The two differ by however long the OS delayed this thread between the reservation + # returning and this line, and asserting on relative waits would fold that scheduling + # jitter straight into the measurement - a thread descheduled for 50ms reports a wait + # 50ms shorter than the slot it actually holds, which makes correctly-spaced slots look + # bunched. Absolute instants are immune to that: a late thread records a late `now` and + # a correspondingly smaller `wait`, and their sum is the same slot either way. + granted = time.monotonic() + wait + with slots_lock: + slots.append(granted) + + threads = [threading.Thread(target=_reserve) for _ in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + ordered = sorted(slots) + assert len(ordered) == 8 + # Eight reservations at a 50ms interval span at least 7 intervals end to end. + assert ordered[-1] - ordered[0] >= interval * 7 * 0.9 + # And no two reservers were handed the same slot - the property that fails if two + # concurrent UPDATEs could both read the same pre-update `next_allowed_at`. + for earlier, later in zip(ordered, ordered[1:]): + assert later - earlier >= interval * 0.9 + + +class TestDegradationNeverFails: + """Requirement 4: a process that cannot get rate budget WAITS, it does not fail or halt.""" + + @pytest.mark.django_db(transaction=True) + def test_an_unavailable_budget_degrades_to_local_pacing_without_raising( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """With the shared row unreachable, `acquire()` must still pace (locally) and still + return - never raise, never halt. The ceiling degrades from global to per-process, which + is where this repo already was; that is a documented, logged degradation, not a failure.""" + import cardpicker.harvest_fetch_limiter as limiter_module + + def _unavailable(destination: str, interval: float) -> float: + raise harvest_rate_budget.RateBudgetUnavailable("simulated outage") + + monkeypatch.setattr(limiter_module, "reserve_slot", _unavailable) + limiter = _independent_limiters(1, "test-global-degraded", 50.0)[0] + + started = time.monotonic() + for _ in range(5): + with limiter.acquire(): + pass + elapsed = time.monotonic() - started + + # Still paced - the fallback is the original per-process arithmetic, not "no limit". + assert elapsed >= (5 - 1) / 50.0 * 0.9 + + @pytest.mark.django_db(transaction=True) + def test_a_missing_row_is_recreated_rather_than_failing_forever(self) -> None: + """Someone truncating the table mid-run must not wedge every fetcher permanently.""" + from cardpicker.models import GlobalFetchPace + + reserve_slot("test-global-recreate", 0.01) + GlobalFetchPace.objects.all().delete() + harvest_rate_budget.reset_for_tests() + + assert reserve_slot("test-global-recreate", 0.01) >= 0.0 + assert GlobalFetchPace.objects.filter(destination="test-global-recreate").exists() + + +class TestSharedBackoffState: + """PR #644's throttle/decay semantics, preserved but now agreed across processes - a global + rate ceiling requires a global backoff term, or processes write conflicting paces into the + same row.""" + + @pytest.mark.django_db(transaction=True) + def test_backoff_from_one_process_is_visible_to_another(self) -> None: + record_backoff("test-global-backoff", 2.0, 16.0) + assert current_state("test-global-backoff") == (2.0, 0) + + record_backoff("test-global-backoff", 2.0, 16.0) + assert current_state("test-global-backoff") == (4.0, 0) + + @pytest.mark.django_db(transaction=True) + def test_backoff_is_capped(self) -> None: + for _ in range(10): + record_backoff("test-global-cap", 2.0, 16.0) + multiplier, _ = current_state("test-global-cap") + assert multiplier == 16.0 + + @pytest.mark.django_db(transaction=True) + def test_the_clean_streak_is_shared_so_decay_is_not_n_times_faster(self) -> None: + """The reason the STREAK had to move into the row too, not just the multiplier: N processes + each counting their own streak would cross the decay threshold N times over and recover N + times faster than the single agreed schedule intends.""" + record_backoff("test-global-decay", 2.0, 16.0) + record_backoff("test-global-decay", 2.0, 16.0) + assert current_state("test-global-decay")[0] == 4.0 + + target = 5 + for _ in range(target - 1): + record_clean_response("test-global-decay", target, 1.0) + assert current_state("test-global-decay")[0] == 4.0 # not yet + + record_clean_response("test-global-decay", target, 1.0) + assert current_state("test-global-decay") == (2.0, 0) + + @pytest.mark.django_db(transaction=True) + def test_decay_never_overshoots_the_configured_ceiling(self) -> None: + for _ in range(50): + record_clean_response("test-global-floor", 2, 1.0) + assert current_state("test-global-floor")[0] == 1.0 + + @pytest.mark.django_db(transaction=True) + def test_a_shared_backoff_actually_slows_the_shared_sequence(self) -> None: + """The multiplier is not just recorded, it is applied - by the SQL that advances the row, + so a backoff recorded by one process paces every other process.""" + interval = 0.02 + record_backoff("test-global-applied", 2.0, 16.0) # x2 + + reserve_slot("test-global-applied", interval) # claim the current slot + wait = reserve_slot("test-global-applied", interval) + + assert wait >= interval * 2 * 0.9, "the shared multiplier was not applied to the shared pace" diff --git a/docs/features/stage-e-operations.md b/docs/features/stage-e-operations.md index e89e5bab2..042e8d8ff 100644 --- a/docs/features/stage-e-operations.md +++ b/docs/features/stage-e-operations.md @@ -133,6 +133,84 @@ fabricated ceiling PR #589 **removed** from `stage_e_batch_sizing`. The honest hardware-vs-destination signal remains #589's own `HostProfile.fetch_overcommitted`. +#### The 7/s cap is GLOBAL, not per process and not per core + +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" + +**This is the distinction two people have now misread, so it is stated +explicitly.** `_DestinationLimiter` paces with a `threading.Lock` and a +process-local `next_allowed` timestamp. Setting `rate_per_sec = 7.0` therefore +bought **7/s per process**, which is not the same thing as 7/s: + +| fetcher | processes | what the old per-process cap actually gave | +| ------------------------------------------- | ---------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| Pooled runner (`run_image_evidence_cohort`) | **one** — a single process with an internal fetch thread pool | a genuine 7/s. Correct, but **by accident**: one process, one limiter. | +| Conveyor (`stage_e_dispatch` via django-q2) | **many** — workers are separate OS processes (`multiprocessing`) | **N × 7/s.** Each worker builds its own limiter in its own address space and paces independently. At the production `STAGE_E_MAX_CONCURRENT_DISPATCHES = 2` that is **14/s against a 7/s ruling**, and it scales with the cap. | + +This is the same per-process trap as `threading.Semaphore(max_concurrency)` — +the one PR #589 deleted a false ceiling term from `stage_e_batch_sizing` over. +Because the rate limit (not the concurrency limit) is what actually protects +the destination, the rate limit's **globality** is load-bearing, not a +refinement. + +**The mechanism: one Postgres row per destination** +(`cardpicker.models.GlobalFetchPace`, module `cardpicker.harvest_rate_budget`). +Every fetching process reserves its turn with a single atomic +`UPDATE … SET next_allowed_at = GREATEST(clock_timestamp(), next_allowed_at) + interval … RETURNING`, then sleeps until its slot **locally**, holding no lock +and no transaction. Concurrent reservers serialise on Postgres's own row lock, +and under READ COMMITTED the blocked statement re-evaluates against the +committed row — so the reservations form **one** strictly-increasing sequence +no matter how many processes compete. + +Three properties worth knowing on call: + +- **Timestamps are the database's clock** (`clock_timestamp()`), never a + process's `time.monotonic()`. Monotonic epochs are per-process and not + comparable across processes; persisting one for another to read would + reintroduce the exact defect while _looking_ coordinated. +- **The backoff multiplier and clean streak live in the same row.** A global + rate ceiling requires a global backoff term — two processes holding different + multipliers would write conflicting paces into the same row, and the effective + rate would be whatever the least-backed-off process believed. The 429/503 + semantics and the decay schedule from the section above are unchanged; only + their storage moved. +- **`max_concurrency` remains per-process, deliberately.** It is a local + resource bound (how many of _this_ process's threads may be in flight), not + the destination-protecting ceiling. + +**Why a row rather than the advisory locks `stage_e_concurrency` already uses.** +That module is this repo's established cross-process primitive, so it was the +first candidate. It cannot carry a rate, and both halves of its own reasoning +invert here: an advisory lock is a **binary** held/not-held token that expresses +"how many at once", not "how many per unit time" — a rate needs a remembered +timestamp and a lock stores no value. And the crash-safety objection that made a +DB row _wrong_ for slot counting (a `kill -9`'d worker leaves a claimed slot +claimed forever) does not apply: this row holds **no claim, only a timestamp**. A +process killed mid-reservation leaves `next_allowed_at` at most one interval in +the future, self-healing in ~143 ms at 7/s with zero reconciliation. Redis was +not an option — `docker/docker-compose.prod.yml` runs django, worker, nginx, +postgres and elasticsearch, and no Redis. + +**Cost.** A reservation is one local round trip: **1.5 ms mean, 1.3 ms p50, +2.3 ms p95** measured over 300 calls against the containerised Postgres the test +suite uses. Against a ~300 ms image fetch that is **0.78 % at p95**, and against +the 143 ms pacing interval the reservation itself schedules it is ~1 %. The +pooled runner is not measurably slowed; its fetch threads touch the row only 7 +times a second in total, so the row lock runs at roughly a 1 % duty cycle. + +**If the database is unreachable, pacing degrades to per-process and the run +continues** — logged at ERROR, never fatal. That is a real loss (the aggregate +ceiling is not enforced while it persists) but a bounded one: every process is +still paced at 7/s by its own local limiter. This is deliberately the **opposite** +of `stage_e_concurrency`'s fail-closed posture, because the downsides differ — an +unavailable _concurrency_ cap lets unbounded dispatches pile onto the host, while +an unavailable _rate_ budget still leaves every process individually paced. It +also matches the standing rule that rate machinery degrades rather than shutting +things down. + **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