From 39283463e86bbbbb5e8f0581b8348863beb6ae49 Mon Sep 17 00:00:00 2001 From: mastertyko <11311479+mastertyko@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:50:25 +0200 Subject: [PATCH] fix(cache): recover invalidation after failed prime --- app/core/cache/invalidation.py | 18 +++- app/main.py | 12 +-- .../.openspec.yaml | 2 + .../design.md | 60 ++++++++++++ .../proposal.md | 28 ++++++ .../specs/model-catalog-compat/spec.md | 96 +++++++++++++++++++ .../specs/query-caching/spec.md | 46 +++++++++ .../tasks.md | 15 +++ .../test_cache_invalidation_bus.py | 78 +++++++++++++++ .../test_upstream_route_cache_invalidation.py | 55 ++++++++++- tests/unit/test_cache_invalidation_poller.py | 64 +++++++++++++ 11 files changed, 464 insertions(+), 10 deletions(-) create mode 100644 openspec/changes/recover-cache-invalidation-after-prime-failure/.openspec.yaml create mode 100644 openspec/changes/recover-cache-invalidation-after-prime-failure/design.md create mode 100644 openspec/changes/recover-cache-invalidation-after-prime-failure/proposal.md create mode 100644 openspec/changes/recover-cache-invalidation-after-prime-failure/specs/model-catalog-compat/spec.md create mode 100644 openspec/changes/recover-cache-invalidation-after-prime-failure/specs/query-caching/spec.md create mode 100644 openspec/changes/recover-cache-invalidation-after-prime-failure/tasks.md create mode 100644 tests/unit/test_cache_invalidation_poller.py diff --git a/app/core/cache/invalidation.py b/app/core/cache/invalidation.py index 101cba3413..587108ea29 100644 --- a/app/core/cache/invalidation.py +++ b/app/core/cache/invalidation.py @@ -89,8 +89,10 @@ async def initialize(self) -> None: On success ``_poll_initialized`` is set to ``True``. If the read fails the method raises with state unchanged (baseline empty, ``_poll_initialized`` - still ``False``), so the caller can degrade to the first-poll-baselines - behavior. + still ``False``), so the caller can retry before background polling. If + the caller continues, ``start()`` arms conservative callback delivery for + the first successfully observed versions instead of accepting them as a + callback-less baseline. """ session = self._session_factory() try: @@ -118,7 +120,10 @@ async def prime(self) -> None: Mirrors ``initialize``'s error contract: if the baseline read fails the poller stays uninitialized (``_poll_initialized`` still ``False``) and - this method raises so the caller can retry or explicitly degrade. + this method raises so the caller can retry before background polling. + Starting without a successful retry makes the first recovered poll + reconcile positive versions through their callbacks before acknowledging + them. ``_poll_once`` swallows the read error, so a silent success here would let the first *background* poll absorb a peer bump as the initial baseline, voiding the delivery guarantee priming exists to provide. @@ -129,6 +134,13 @@ async def prime(self) -> None: async def start(self) -> None: if self._task and not self._task.done(): return + # Callback-less baseline acquisition is safe only before process-local + # state can be served. Once background polling starts, a missing baseline + # means the observed versions are uncertain: treat each positive first + # observation as a change and reconcile it through the normal callback / + # acknowledgement path. A successful prime already set this flag after + # recording exact versions, so normal startup remains unchanged. + self._poll_initialized = True self._stop.clear() self._task = asyncio.create_task(self._run()) diff --git a/app/main.py b/app/main.py index df28262a46..eab8576e2c 100644 --- a/app/main.py +++ b/app/main.py @@ -358,12 +358,12 @@ async def lifespan(app: FastAPI): try: await cache_poller.prime() except Exception: - # prime() raises when the baseline version read fails; degrade to - # first-poll-baselines (matching initialize()'s contract) rather than - # continuing as if the seed succeeded. A peer bump landing before the - # first background poll may then be absorbed as the initial baseline - # and only converge on the fallback TTL / next bump, but the failure is - # surfaced here instead of silently voiding the delivery guarantee. + # prime() raises when the baseline version read fails, leaving the poller + # uninitialized so an explicit retry would remain baseline-only. Startup + # continues, but start() arms conservative recovery: the first successful + # background read invokes callbacks for positive versions before + # acknowledging them, so a peer bump cannot become a callback-less + # baseline after local caches are warm. logger.warning("cache invalidation baseline prime failed", exc_info=True) try: await routing_availability_cache.refresh_from_db() diff --git a/openspec/changes/recover-cache-invalidation-after-prime-failure/.openspec.yaml b/openspec/changes/recover-cache-invalidation-after-prime-failure/.openspec.yaml new file mode 100644 index 0000000000..ab39675458 --- /dev/null +++ b/openspec/changes/recover-cache-invalidation-after-prime-failure/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-07-30 diff --git a/openspec/changes/recover-cache-invalidation-after-prime-failure/design.md b/openspec/changes/recover-cache-invalidation-after-prime-failure/design.md new file mode 100644 index 0000000000..0edf091caa --- /dev/null +++ b/openspec/changes/recover-cache-invalidation-after-prime-failure/design.md @@ -0,0 +1,60 @@ +## Context + +`CacheInvalidationPoller.prime()` normally records every namespace version before startup warms process-local caches. If that read fails, startup logs the failure, warms the routing and model-registry state, and starts background polling with no known versions. The first successful poll currently treats every observed row as a callback-less baseline, even if a peer advanced it after the failed read. + +The poller therefore has two distinct lifecycle phases: explicit baseline acquisition before process-local state is served, and background reconciliation after that state may be warm. Only the former may accept an observed version without invoking its callbacks. + +## Goals / Non-Goals + +**Goals:** + +- Recover conservatively after a failed startup baseline read. +- Run registered callbacks before acknowledging positive versions first observed during background polling. +- Preserve callback retry and monotonic version-acknowledgement behavior. +- Preserve baseline-only behavior when `prime()` is explicitly retried before background polling. +- Prove the fix at both the warmed upstream-route resolver cache and the account-routing / bridge-session reuse seam. + +**Non-Goals:** + +- Failing startup or adding startup retries. +- Changing namespace versions, bump ordering, callback registration, poll intervals, or cache TTLs. +- Adding configuration, schema, migrations, or cross-replica payloads. +- Changing normal startup behavior when baseline priming succeeds. + +## Decisions + +### Background start ends callback-less baseline acquisition + +`start()` will transition an uninitialized poller into conservative background-reconciliation mode before it creates the polling task. A successful prime has already made this transition with recorded versions. After a failed prime, the transition makes a positive version with no known baseline satisfy the existing change predicate, so the namespace callback runs before acknowledgement. + +This uses the existing `_poll_initialized` state rather than adding a second failure latch. The relevant distinction is not whether one particular read failed, but whether the caller is still explicitly acquiring a pre-service baseline or has started background polling after caches may be warm. + +Alternatives considered: + +- Mark the poller initialized inside the `prime()` failure path: rejected because an explicit `prime()` retry could then invoke callbacks even though its documented purpose is baseline-only acquisition. +- Retry or fail startup: rejected because it changes availability policy and is broader than the stale-cache defect. +- Clear routing caches directly in `app.main` after the exception: rejected because it duplicates callback wiring, does not cover later first-observed namespace rows, and leaves other poller consumers inconsistent. +- Add a separate recovery latch: workable, but redundant with the existing lifecycle boundary and adds state combinations without improving the guarantee. + +### Recovery uses normal callback acknowledgement rules + +Recovery will use `_run_callbacks()` and the existing per-namespace acknowledgement path. A callback failure leaves that namespace unacknowledged and the next poll retries it. Existing monotonic handling for concurrent `bump_local()` acknowledgements remains unchanged. + +### The integration proofs keep the real routing gates + +The regressions will use two pollers sharing the integration database. One keeps a resolved upstream-route outcome warm across a failed prime and a peer `upstream_route` bump. The other keeps a routing snapshot seeded `ACTIVE` across a peer status change plus `account_routing` bump and checks the actual bridge-session reuse predicate. Together they prove that recovery changes externally relevant routing decisions, not merely a callback counter. + +## Risks / Trade-offs + +- [A failed prime followed by background start may replay callbacks for versions that predate startup] → Callbacks are designed to be idempotent local reconciliation; this conservative replay happens only on the exceptional no-baseline path and is safer than serving unknown cache state. +- [A recovery callback can fail] → The existing unacknowledged-version retry path remains authoritative and is covered by focused tests. +- [The background task could poll before recovery mode is active] → Transition lifecycle state synchronously in `start()` before creating the task. +- [A namespace has no row yet] → No callback runs until a later bump creates a positive version; that first observed version is then treated as changed. + +## Migration Plan + +No data or configuration migration is required. Deploy as an application-only change. Rollback restores the previous callback-less first-poll fallback after a failed prime; no persisted state requires reversal. + +## Open Questions + +None. diff --git a/openspec/changes/recover-cache-invalidation-after-prime-failure/proposal.md b/openspec/changes/recover-cache-invalidation-after-prime-failure/proposal.md new file mode 100644 index 0000000000..c403ef2b15 --- /dev/null +++ b/openspec/changes/recover-cache-invalidation-after-prime-failure/proposal.md @@ -0,0 +1,28 @@ +## Why + +A transient failure while priming cache-invalidation versions leaves the poller without a baseline, but startup continues and warms routing state. A peer mutation that lands afterward can then be accepted by the first successful background poll as a callback-less baseline, leaving a stale routing decision in service until another bump or restart. + +## What Changes + +- Make background polling recover fail-safe when no version baseline was recorded: observed positive namespace versions are delivered through their callbacks before they are acknowledged. +- Preserve baseline-only semantics for an explicit `prime()` retry before background polling begins. +- Cover the production-sensitive upstream-route cache and account-routing bridge-reuse paths with two-replica regressions that prove warmed decisions are invalidated after the version read recovers. +- Replace the model-catalog contract's documented callback-less degradation with the same conservative recovery behavior. + +## Capabilities + +### New Capabilities + +None. + +### Modified Capabilities + +- `query-caching`: Require the first successful background poll after a failed startup baseline read to reconcile observed namespaces before acknowledging their versions. +- `model-catalog-compat`: Keep surfacing a failed baseline prime while requiring background recovery to invoke the model-registry callback instead of accepting a callback-less first-poll baseline. + +## Impact + +- `app/core/cache/invalidation.py`: background-poller lifecycle state. +- `app/main.py`: startup failure-semantics documentation. +- Cache-invalidation integration coverage for the warmed account-routing / bridge-reuse path and existing model-registry startup behavior. +- No API, configuration, dependency, database-schema, or migration change. diff --git a/openspec/changes/recover-cache-invalidation-after-prime-failure/specs/model-catalog-compat/spec.md b/openspec/changes/recover-cache-invalidation-after-prime-failure/specs/model-catalog-compat/spec.md new file mode 100644 index 0000000000..2f3a744854 --- /dev/null +++ b/openspec/changes/recover-cache-invalidation-after-prime-failure/specs/model-catalog-compat/spec.md @@ -0,0 +1,96 @@ +## MODIFIED Requirements + +### Requirement: Refreshed model catalog is replica-coherent + +The leader refresh cycle SHALL persist the complete registry state (models, plan maps, per-account tier maps, suppression set, authoritative flags, metadata retention state, and the refresh wall-clock timestamp) to the single-row `model_registry_snapshot` table and SHALL bump the `model_registry` cache-invalidation namespace only after the persist commits (write-then-bump). The payload write and the bump SHALL be skipped when the serialized content hash is unchanged from the last persisted state AND the stored row was still within `model_registry_snapshot_max_age_seconds`; the stored `refreshed_at` timestamp SHALL still be advanced so snapshot age reflects the leader's latest successful refresh. When the content hash is unchanged but the stored row had already aged past `model_registry_snapshot_max_age_seconds` before this refresh revived it, the leader SHALL still bump the `model_registry` namespace (only the payload rewrite stays skipped): an expired row causes followers to clear their local registry and reset their applied-content-hash marker, so an unchanged-content revival still requires a bump for them to re-apply within the cache-invalidation poll bound instead of waiting for the non-leader scheduler backstop. Every replica MUST apply a newly persisted snapshot within the cache-invalidation poll bound and MUST invalidate its local account-selection cache on apply; that account-selection invalidation MUST be local-only (non-propagating), because reconcile only applies a change the leader already published (which bumped `model_registry` to reach every replica) and each replica clears its own selection cache on apply, so a propagating clear would make every follower durably re-bump `account_selection` and amplify bus traffic with no peer-visible effect. When the reconcile is driven by the `model_registry` invalidation callback and the snapshot load fails (transient DB read error or malformed payload), the callback MUST surface the failure to the invalidation poller so the poller leaves the `model_registry` version unacknowledged and retries on the next poll cycle (matching the `account_routing` refresh callback), rather than acknowledging the bump and stranding the replica on the stale catalog until the non-leader scheduler backstop; the startup one-shot reconcile and the refresh-tick backstop instead swallow such a load failure (keeping the current in-memory state) so they never fail startup or the scheduler loop. Payload decode MUST treat a set-backed or mapping-backed catalog field whose persisted value has the wrong type — for example a `model_plans`/`plan_models`/`model_accounts`/per-account tier entry persisted as a scalar or object where a list of slugs is expected, or a model entry that is not an object — as a malformed payload and raise, rather than silently dropping the offending entry and applying a partial catalog; a genuinely-absent or empty container (an absent key, an empty map, or an empty list) is not malformed and MUST decode successfully. After apply, `/v1/models`, plan gating (`plan_types_for_model`), suppression (`is_suppressed_model`), and per-account service-tier routing on a non-leader MUST be identical to the leader. A non-leader refresh tick MUST NOT fetch the upstream catalog and SHALL instead reconcile from the persisted snapshot when the stored snapshot header differs from the last applied one (backstop for a lost invalidation bump). A leader catalog clear SHALL persist an explicit cleared marker and bump, so followers revert to the bootstrap floor rather than serving a withdrawn catalog. Every replica SHALL install its `model_registry` cache-invalidation callback (the global invalidation poller) before starting the model refresh scheduler, so a first leader tick that persists a changed snapshot cannot silently drop its bump. Every replica SHALL record the invalidation-poller version baseline before running its one-shot startup reconcile, so a leader bump that lands in the window between that reconcile's snapshot read and the poller's first background tick is delivered as an invalidation callback (within the poll bound) rather than absorbed as the poller's initial callback-less baseline (which would defer convergence to the non-leader scheduler backstop). The baseline-priming read SHALL surface a failure to its caller and leave the poller without a recorded baseline. If startup continues, the first successful background poll MUST conservatively treat a positive `model_registry` version observed without a baseline as changed, invoke the reconcile callback, and acknowledge it only after that callback succeeds. This MAY replay a pre-startup version, but MUST NOT absorb a peer bump as a callback-less baseline. + +#### Scenario: Follower serves the refreshed catalog on /v1/models + +- **GIVEN** replica A (leader) completes a registry refresh whose catalog adds a new slug and withdraws a bootstrap slug +- **AND** replica A persists the snapshot and bumps the `model_registry` namespace +- **WHEN** replica B's cache-invalidation poller observes the version change +- **THEN** replica B applies the snapshot to its in-memory registry +- **AND** `GET /v1/models` served by replica B lists the new slug and omits the withdrawn slug + +#### Scenario: Follower enforces suppression of a withdrawn slug + +- **GIVEN** the leader's refreshed snapshot marks a previously served slug as suppressed +- **WHEN** a follower applies the persisted snapshot +- **THEN** `is_suppressed_model` returns true for that slug on the follower + +#### Scenario: Follower enforces plan gating for a newly gated slug + +- **GIVEN** the leader's refreshed snapshot maps a slug to exactly one plan type +- **WHEN** a follower applies the persisted snapshot +- **THEN** `plan_types_for_model` on the follower returns exactly that plan set instead of no filtering + +#### Scenario: Catalog clear propagates to followers + +- **GIVEN** the leader clears the registry because no active accounts remain +- **WHEN** the leader persists the cleared marker and bumps, and a follower applies it +- **THEN** the follower reverts to the bootstrap catalog floor + +#### Scenario: Lost bump converges via the refresh-tick backstop + +- **GIVEN** a snapshot was persisted but the invalidation bump was lost +- **WHEN** a non-leader replica's next refresh tick runs +- **THEN** the replica detects the header mismatch, applies the persisted snapshot, and converges within one refresh interval + +#### Scenario: Transient load failure in the callback is retried, not acknowledged + +- **GIVEN** the leader persisted a changed snapshot and bumped the `model_registry` namespace +- **AND** a follower's snapshot load transiently fails on the invalidation callback (e.g. a DB read error or a momentarily unreadable payload) +- **WHEN** the follower's poll cycle runs the callback and it fails +- **THEN** the poller does not acknowledge the observed `model_registry` version and retries the callback on the next poll cycle +- **AND** once the transient failure clears, the retry applies the persisted snapshot within the poll bound without requiring a new leader bump + +#### Scenario: Malformed set-backed field is rejected, not silently dropped + +- **GIVEN** the leader bumped the `model_registry` namespace and the persisted payload is valid JSON but a set-backed field is wrong-typed (e.g. `model_plans` maps a slug to `{"gpt-x": "pro"}` instead of a list of plan slugs) +- **WHEN** a follower's invalidation callback loads and decodes the payload +- **THEN** the decode raises rather than dropping the offending entry +- **AND** the poller leaves the `model_registry` version unacknowledged and no partial catalog is applied (the follower keeps its prior in-memory state and retries on the next poll) + +#### Scenario: Empty set-backed maps decode successfully + +- **GIVEN** a persisted snapshot whose set-backed fields are genuinely empty (empty maps, or a slug mapped to an empty list) +- **WHEN** a replica decodes the payload +- **THEN** the decode succeeds and the corresponding sets are empty (empty is not treated as malformed) + +#### Scenario: Applying a snapshot does not re-bump account_selection + +- **GIVEN** the leader persisted a changed snapshot and bumped `model_registry` +- **WHEN** a follower applies the snapshot and invalidates its local account-selection cache +- **THEN** the follower does not enqueue or write an `account_selection` cache-invalidation bump + +#### Scenario: Non-leader tick performs no upstream fetch + +- **WHEN** a non-leader replica's refresh tick runs +- **THEN** it performs no upstream model-catalog fetch, regardless of whether it reconciled from the store + +#### Scenario: First leader bump is not dropped at startup + +- **GIVEN** a replica is starting up +- **WHEN** the model refresh scheduler starts +- **THEN** the global cache-invalidation poller with the `model_registry` callback is already installed, so an immediate leader persist-and-bump reaches followers within the poll bound + +#### Scenario: Bump during the startup reconcile window is not dropped + +- **GIVEN** a replica is starting up and has recorded the invalidation-poller version baseline +- **AND** a leader persists a changed snapshot and bumps the `model_registry` namespace in the window between the replica's one-shot startup reconcile and the poller's first background tick +- **WHEN** the poller's first background tick runs +- **THEN** it observes the version advanced past the recorded baseline and invokes the reconcile callback, so the replica applies the new snapshot within the poll bound rather than waiting for the non-leader scheduler backstop + +#### Scenario: Reviving an expired unchanged snapshot bumps the bus + +- **GIVEN** a snapshot was persisted with content hash H and its stored row then aged past `model_registry_snapshot_max_age_seconds`, so followers dropped to the bootstrap floor and reset their applied-content-hash marker +- **WHEN** the leader's next refresh succeeds with the same catalog bytes (content hash H again) +- **THEN** the leader advances `refreshed_at` without rewriting the payload but still bumps the `model_registry` namespace +- **AND** the followers observe the version change and re-apply the revived snapshot within the poll bound rather than waiting for the non-leader scheduler backstop + +#### Scenario: Failed startup baseline prime recovers through reconciliation + +- **GIVEN** a replica's baseline-priming read fails transiently and no `model_registry` version baseline is recorded +- **WHEN** its first successful background poll observes a positive `model_registry` version +- **THEN** the poller MUST invoke the model-registry reconcile callback before acknowledging that version +- **AND** the replica MUST NOT defer convergence to the scheduler backstop merely because startup baseline priming failed diff --git a/openspec/changes/recover-cache-invalidation-after-prime-failure/specs/query-caching/spec.md b/openspec/changes/recover-cache-invalidation-after-prime-failure/specs/query-caching/spec.md new file mode 100644 index 0000000000..f3d1c106ad --- /dev/null +++ b/openspec/changes/recover-cache-invalidation-after-prime-failure/specs/query-caching/spec.md @@ -0,0 +1,46 @@ +## MODIFIED Requirements + +### Requirement: Cache invalidation bumps and polling are resilient and observable + +`bump()` MUST retry transient write failures (including SQLite "database is locked") with a short backoff; on final failure it MUST log at ERROR with the namespace, increment `codex_lb_cache_invalidation_bump_failures_total{namespace}`, and MUST NOT fail the originating mutation. Coalesced (`request_bump`) namespaces MUST remain pending and be retried on subsequent poll cycles until a bump succeeds, and a `request_bump` arriving while a flush for the same namespace is already awaiting its bump MUST be preserved and produce a later bump. When any invalidation callback for a namespace fails, the poller MUST NOT acknowledge the observed version and MUST re-run that namespace's callbacks on subsequent poll cycles until they succeed. The poller MUST escalate consecutive poll failures above debug level after a bounded count (WARNING after 3, ERROR after 10) and increment `codex_lb_cache_invalidation_poll_failures_total`. After a startup baseline read fails, a process that continues without a recorded baseline MUST treat each positive version first observed for a registered namespace by the next successful background poll as changed, run that namespace's registered callbacks, and acknowledge the version only after those callbacks succeed. This recovery MAY cause a redundant invalidation for a version that predates startup; it MUST NOT silently absorb a peer bump into a callback-less baseline. + +#### Scenario: Bump failure under database lock is observable and does not fail the mutation + +- **GIVEN** the database rejects cache-invalidation writes with a lock error for longer than the retry budget +- **WHEN** a mutation attempts a durable namespace bump +- **THEN** the mutation itself still succeeds +- **AND** an ERROR log naming the namespace is emitted and the bump-failure counter increments + +#### Scenario: Pending coalesced namespace flushes on the next successful cycle + +- **GIVEN** a coalesced `request_bump` namespace failed to flush during a poll cycle +- **WHEN** the database becomes writable again +- **THEN** the next poll cycle flushes the pending namespace and increments its version + +#### Scenario: Bump requested during an in-flight flush produces a later bump + +- **GIVEN** a coalesced flush is awaiting the bump write for a namespace +- **WHEN** another mutation commits and requests a bump for the same namespace before the flush completes +- **THEN** the namespace is re-queued and flushed again on a subsequent cycle, incrementing the version beyond the in-flight bump + +#### Scenario: Failed invalidation callback keeps the version unacknowledged and is retried + +- **GIVEN** a replica observes an `account_routing` version bump +- **AND** its routing snapshot refresh fails with a transient database error +- **WHEN** the poll cycle completes +- **THEN** the replica does not record the new version as seen +- **AND** the refresh is retried on subsequent poll cycles until it succeeds + +#### Scenario: Consecutive poll failures escalate above debug + +- **GIVEN** a replica's poller cannot read the `cache_invalidation` table +- **WHEN** three consecutive polls fail +- **THEN** a WARNING is logged and the poll-failure counter increments + +#### Scenario: Failed startup prime cannot absorb a route-cache bump + +- **GIVEN** replica B's startup cache-invalidation baseline read fails and no `upstream_route` version is recorded +- **AND** replica B continues serving traffic and warms an upstream-route resolution cache entry +- **WHEN** replica A commits a route-input mutation and advances `upstream_route` before replica B's first successful version read +- **THEN** replica B's first successful background poll MUST run the registered `upstream_route` invalidation callback before acknowledging the observed version +- **AND** the warmed route entry MUST be cleared in that poll instead of remaining stale until its TTL or a later bump diff --git a/openspec/changes/recover-cache-invalidation-after-prime-failure/tasks.md b/openspec/changes/recover-cache-invalidation-after-prime-failure/tasks.md new file mode 100644 index 0000000000..6ff3b9379a --- /dev/null +++ b/openspec/changes/recover-cache-invalidation-after-prime-failure/tasks.md @@ -0,0 +1,15 @@ +## 1. Regression Coverage + +- [x] 1.1 Add deterministic two-replica integration regressions proving a failed baseline prime cannot absorb later `upstream_route` or `account_routing` bumps while stale routing decisions remain warm. +- [x] 1.2 Add a hermetic unit regression for failed prime, background start, callback delivery, and acknowledgement. + +## 2. Poller Recovery + +- [x] 2.1 Transition an uninitialized poller to conservative callback delivery before background polling starts, while preserving explicit `prime()` retry semantics. +- [x] 2.2 Update startup lifecycle documentation to describe callback-based recovery after a failed prime. + +## 3. Verification + +- [x] 3.1 Run the focused cache-invalidation, route-cache, model-registry startup, and bridge-reuse integration checks required for Sensitive routing/cache work. +- [x] 3.2 Run scoped lint/format checks and strict validation for the change plus affected main specs. +- [x] 3.3 Review the final diff and worktree status; record any untested or blocked checks. diff --git a/tests/integration/test_cache_invalidation_bus.py b/tests/integration/test_cache_invalidation_bus.py index 4438a00de6..fd38a4998f 100644 --- a/tests/integration/test_cache_invalidation_bus.py +++ b/tests/integration/test_cache_invalidation_bus.py @@ -8,6 +8,7 @@ from __future__ import annotations +import asyncio import logging from datetime import datetime, timezone from types import SimpleNamespace @@ -162,6 +163,60 @@ async def test_remote_pause_stops_stale_bridge_session_reuse(db_setup, poller_sl assert _http_bridge_session_account_active(stale_session) is False +@pytest.mark.asyncio +async def test_failed_prime_recovery_stops_stale_bridge_session_reuse(db_setup, poller_slot, monkeypatch) -> None: + """A failed startup version read must not let the first recovered poll + acknowledge a later peer pause without refreshing the warmed routing state.""" + account_id = "acct-bus-prime-recovery" + await _insert_account(account_id) + + # A namespace row already exists before this replica starts, but its + # baseline read fails transiently. + remote_poller = CacheInvalidationPoller(SessionLocal) + assert await remote_poller.bump(NAMESPACE_ACCOUNT_ROUTING) is True + flaky_versions = _FlakySessionFactory(failures=1) + local_poller = CacheInvalidationPoller(flaky_versions) + routing_cache = RoutingAvailabilityCache(SessionLocal) + monkeypatch.setattr("app.modules.proxy.account_cache._routing_availability_cache", routing_cache) + set_cache_invalidation_poller(local_poller) + + callback_calls = 0 + + async def refresh_routing_snapshot() -> None: + nonlocal callback_calls + callback_calls += 1 + await routing_cache.refresh_from_db() + + local_poller.on_invalidation(NAMESPACE_ACCOUNT_ROUTING, refresh_routing_snapshot) + + with pytest.raises(RuntimeError, match="baseline version read did not complete"): + await local_poller.prime() + + # Startup continues and warms an ACTIVE snapshot after the failed prime. + await routing_cache.refresh_from_db() + stale_session = _fake_bridge_session(_make_account(account_id, AccountStatus.ACTIVE)) + assert _http_bridge_session_account_active(stale_session) is True + + # A peer pauses the account and advances the version before this replica's + # first successful background read. + await _set_account_status(account_id, AccountStatus.PAUSED) + assert await remote_poller.bump(NAMESPACE_ACCOUNT_ROUTING) is True + + parked = asyncio.Event() + + async def park_background_loop() -> None: + await parked.wait() + + monkeypatch.setattr(local_poller, "_run", park_background_loop) + await local_poller.start() + try: + await local_poller._poll_once() + assert callback_calls == 1 + assert _http_bridge_session_account_active(stale_session) is False + finally: + await local_poller.stop() + + @pytest.mark.asyncio async def test_reauth_on_peer_clears_local_routing_marker(db_setup, poller_slot) -> None: """A routing-unavailable marker set locally is cleared when another replica @@ -608,6 +663,29 @@ async def test_initialize_failure_leaves_poller_uninitialized(db_setup) -> None: assert poller._known_versions == {} +@pytest.mark.asyncio +async def test_prime_retry_after_failure_remains_baseline_only(db_setup) -> None: + """Retrying prime before background start records a baseline without + callbacks; conservative recovery begins only when background polling starts.""" + remote_poller = CacheInvalidationPoller(SessionLocal) + assert await remote_poller.bump(NAMESPACE_ACCOUNT_ROUTING) is True + expected_version = await _namespace_version(NAMESPACE_ACCOUNT_ROUTING) + assert expected_version is not None + + calls: list[str] = [] + poller = CacheInvalidationPoller(_FlakySessionFactory(failures=1)) + poller.on_invalidation(NAMESPACE_ACCOUNT_ROUTING, lambda: calls.append("routing")) + + with pytest.raises(RuntimeError, match="baseline version read did not complete"): + await poller.prime() + await poller.prime() + + assert calls == [] + assert await remote_poller.bump(NAMESPACE_ACCOUNT_ROUTING) is True + await poller._poll_once() + assert calls == ["routing"] + + @pytest.mark.asyncio async def test_bump_local_suppresses_source_callback_but_peer_still_fires(db_setup) -> None: """A replica that has already invalidated locally uses ``bump_local`` so its diff --git a/tests/integration/test_upstream_route_cache_invalidation.py b/tests/integration/test_upstream_route_cache_invalidation.py index c5fdb21f8c..6406f1dd5e 100644 --- a/tests/integration/test_upstream_route_cache_invalidation.py +++ b/tests/integration/test_upstream_route_cache_invalidation.py @@ -6,15 +6,18 @@ import pytest from sqlalchemy import select +from sqlalchemy.exc import OperationalError +from sqlalchemy.ext.asyncio import AsyncSession from app.core.auth import generate_unique_account_id from app.core.cache.invalidation import ( NAMESPACE_SETTINGS, NAMESPACE_UPSTREAM_ROUTE, + CacheInvalidationPoller, get_cache_invalidation_poller, ) from app.core.config.settings import get_settings -from app.core.upstream_proxy.cache import get_upstream_route_cache +from app.core.upstream_proxy.cache import UpstreamRouteCache, get_upstream_route_cache from app.db.models import CacheInvalidation from app.db.session import SessionLocal @@ -84,6 +87,56 @@ def _seed_dummy_entry() -> None: assert cache.get("seeded-account") is not None +class _FailFirstVersionSessionFactory: + def __init__(self) -> None: + self._failed = False + + def __call__(self) -> AsyncSession: + if not self._failed: + self._failed = True + raise OperationalError("stmt", {}, Exception("peer-version read failed")) + return SessionLocal() + + +async def test_failed_prime_recovery_clears_warm_route_cache(db_setup, route_cache_ttl, monkeypatch) -> None: + """A route-cache bump landing after a failed startup prime must clear a warm + resolver outcome on the first successful background version read.""" + remote_poller = CacheInvalidationPoller(SessionLocal) + assert await remote_poller.bump(NAMESPACE_UPSTREAM_ROUTE) is True + + route_cache = UpstreamRouteCache() + callback_calls = 0 + + def clear_route_cache() -> None: + nonlocal callback_calls + callback_calls += 1 + route_cache.clear() + + local_poller = CacheInvalidationPoller(_FailFirstVersionSessionFactory()) + local_poller.on_invalidation(NAMESPACE_UPSTREAM_ROUTE, clear_route_cache) + + with pytest.raises(RuntimeError, match="baseline version read did not complete"): + await local_poller.prime() + route_cache.store_route("seeded-account", None, generation=route_cache.generation) + assert route_cache.get("seeded-account") is not None + + assert await remote_poller.bump(NAMESPACE_UPSTREAM_ROUTE) is True + + parked = asyncio.Event() + + async def park_background_loop() -> None: + await parked.wait() + + monkeypatch.setattr(local_poller, "_run", park_background_loop) + await local_poller.start() + try: + await local_poller._poll_once() + assert callback_calls == 1 + assert route_cache.get("seeded-account") is None + finally: + await local_poller.stop() + + async def test_binding_upsert_clears_cache_and_bumps_namespace(async_client, route_cache_ttl) -> None: account_id = await _import_account(async_client, "acc-route-cache-binding", "route-cache-binding@example.com") pool_id = await _create_pool_with_endpoint(async_client) diff --git a/tests/unit/test_cache_invalidation_poller.py b/tests/unit/test_cache_invalidation_poller.py new file mode 100644 index 0000000000..9898c756ee --- /dev/null +++ b/tests/unit/test_cache_invalidation_poller.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import asyncio +from typing import cast + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.cache.invalidation import NAMESPACE_UPSTREAM_ROUTE, CacheInvalidationPoller + + +class _VersionRows: + @staticmethod + def all() -> list[tuple[str, int]]: + return [(NAMESPACE_UPSTREAM_ROUTE, 2)] + + +class _ReadableSession: + def in_transaction(self) -> bool: + return False + + async def execute(self, *_args: object, **_kwargs: object) -> _VersionRows: + return _VersionRows() + + async def close(self) -> None: + return None + + +class _FailFirstVersionSessionFactory: + def __init__(self) -> None: + self._failed = False + + def __call__(self) -> AsyncSession: + if not self._failed: + self._failed = True + raise RuntimeError("peer-version read failed") + return cast(AsyncSession, _ReadableSession()) + + +@pytest.mark.asyncio +async def test_background_start_reconciles_first_version_after_failed_prime(monkeypatch) -> None: + calls: list[str] = [] + poller = CacheInvalidationPoller(_FailFirstVersionSessionFactory()) + poller.on_invalidation(NAMESPACE_UPSTREAM_ROUTE, lambda: calls.append("clear")) + + with pytest.raises(RuntimeError, match="baseline version read did not complete"): + await poller.prime() + + parked = asyncio.Event() + + async def park_background_loop() -> None: + await parked.wait() + + monkeypatch.setattr(poller, "_run", park_background_loop) + await poller.start() + try: + await poller._poll_once() + await poller._poll_once() + finally: + await poller.stop() + + # The first successful poll reconciles and acknowledges version 2; the + # unchanged second observation must not invoke the callback again. + assert calls == ["clear"]