fix(cache): keep an aborted invalidation bump queued

app/core/cache/invalidation.py

@@ -284,8 +284,29 @@ async def _flush_pending_bumps(self) -> None:
             # later bump instead of being coalesced into the version already
             # being written.
             self._pending_bumps.discard(namespace)
-            if not await self.bump(namespace):
+            try:
+                if not await self.bump(namespace):
+                    self._pending_bumps.add(namespace)
+            except asyncio.CancelledError:
+                # The marker is cleared before the write, so an aborted write
+                # would otherwise leave the namespace neither written nor
+                # pending, breaking the required retry. Restored even when the
+                # abort is ambiguous — a redundant bump only re-runs peers'
+                # idempotent callbacks.
+                self._pending_bumps.add(namespace)
+                raise
+            except Exception:
+                # ``bump()`` reports normal failure by returning False, so a
+                # raise is abnormal — but re-raising would abort the flush and,
+                # since the loop is sorted, a persistently raising namespace
+                # would starve every namespace sorting after it on every cycle.
+                # Restore it and keep flushing the rest.
                 self._pending_bumps.add(namespace)
+                logger.warning(
+                    "cache_invalidation flush bump raised for namespace %s; kept pending",
+                    _NAMESPACE_LOG_LABELS.get(namespace, "unknown"),
+                    exc_info=True,
+                )
 
     async def _poll_once(self) -> bool:
         """Flush pending bumps and reconcile observed versions once.

openspec/changes/keep-aborted-invalidation-bumps-pending/proposal.md

@@ -0,0 +1,27 @@
+## Why
+
+The invalidation-bus spec already requires that "coalesced (`request_bump`) namespaces MUST remain pending and be retried on subsequent poll cycles until a bump succeeds". The implementation violated it for one case.
+
+`_flush_pending_bumps` clears each namespace's pending marker before awaiting its write — deliberately, so a `request_bump()` arriving mid-write re-queues instead of being coalesced into the version already being written. But it restored the marker only when `bump()` returned `False`. A write that was **cancelled** or **raised** left the namespace neither written nor pending, with nothing logged and no retry holding it. Since `_run` swallows poll exceptions and keeps cycling, a raising write silently lost its namespace during ordinary operation.
+
+## What Changes
+
+- Restore the pending marker when the bump write aborts, so the required retry actually happens. The two abort kinds are handled differently: `CancelledError` restores and re-raises (task teardown must abort the flush), while an ordinary `Exception` — abnormal, since `bump()` reports failure by returning `False` — restores, logs at warning, and continues, so a persistently raising namespace cannot starve the namespaces sorting after it.
+
+The restore is unconditional even when the abort's outcome is ambiguous (cancellation or a driver error arriving after the database accepted the commit): a redundant bump only re-runs peers' idempotent invalidation callbacks, while dropping an unconfirmed write leaves them stale until the fallback TTL. The bus already tolerates extra version increments — `request_bump` arriving mid-flush deliberately produces one.
+
+Process shutdown is deliberately out of scope: `stop()` cancels the polling task, so a bump queued at that moment has no cycle left to drain it. That is already the documented contract — "a lost bump still converges within the fallback TTL" — and guaranteeing delivery against an unresponsive database at shutdown is a separate concern with its own bounding and task-ownership design.
+
+## Why the ambiguous case still restores
+
+A cancellation or driver error can arrive after the database accepted the commit, so the restore can produce a redundant bump. That is the deliberate trade: a redundant bump only re-runs peers' idempotent invalidation callbacks, while dropping an unconfirmed write leaves them stale until the fallback TTL. The bus already tolerates extra increments — a `request_bump` arriving mid-flush produces one by design.
+
+## Capabilities
+
+### New Capabilities
+
+None.
+
+### Modified Capabilities
+
+- `query-caching`: state explicitly that an aborted (not merely failed) write keeps its namespace queued.

openspec/changes/keep-aborted-invalidation-bumps-pending/specs/query-caching/spec.md

@@ -0,0 +1,57 @@
+## 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, including when the write aborts rather than merely failing: an aborted write MUST restore the pending marker regardless of whether the database had already accepted its commit. A write that raises MUST NOT prevent the remaining pending namespaces from flushing in the same cycle. 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`.
+
+#### 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: An aborted bump write keeps its namespace queued
+
+- **GIVEN** a coalesced flush has cleared a namespace's pending marker and is awaiting its bump write
+- **WHEN** that write aborts — cancelled or raised — before the database accepts its commit
+- **THEN** the namespace is restored to the pending set for a later cycle, and no version is written
+
+#### Scenario: A raising namespace does not starve the others
+
+- **GIVEN** two pending namespaces where the first (in sort order) raises on every bump attempt
+- **WHEN** a flush cycle runs
+- **THEN** the raising namespace stays pending with no version written
+- **AND** the other namespace is bumped in that same cycle
+
+#### Scenario: An abort after the commit was accepted still restores the namespace
+
+- **GIVEN** a bump write aborts — cancelled, or the driver raises — after the database accepted its commit but before completion is reported
+- **WHEN** the abort is handled
+- **THEN** the namespace is restored to the pending set and bumped on a later cycle
+- **AND** the resulting duplicate version increment is accepted

openspec/changes/keep-aborted-invalidation-bumps-pending/tasks.md

@@ -0,0 +1,13 @@
+## 1. Fix
+
+- [x] 1.1 Restore the pending namespace in `_flush_pending_bumps` when the bump write aborts: cancellation restores and re-raises; an abnormal raise restores, logs, and continues with the remaining namespaces so a persistently raising namespace cannot starve the ones sorting after it
+
+## 2. Tests
+
+- [x] 2.1 A cancelled write restores the marker, and the marker is cleared before the write (locking in the intended coalescing)
+- [x] 2.2 A raising write restores the marker and does not block later namespaces from flushing
+- [x] 2.3 End-to-end: the running poller retries the aborted namespace and writes its version
+
+## 3. Spec
+
+- [x] 3.1 Make "remains pending" explicitly cover an aborted write, not only a failed one, keeping the requirement normative and the rationale in the proposal

tests/integration/test_cache_invalidation_bus.py

@@ -386,6 +386,66 @@ def test_namespace_log_labels_cover_all_namespaces() -> None:
     }
 
 
+@pytest.mark.asyncio
+async def test_pending_bump_survives_a_cancelled_flush(db_setup, monkeypatch) -> None:
+    """The marker is cleared before the write is awaited, so a cancelled write
+    must restore it — otherwise the namespace is neither written nor pending
+    and no later cycle can retry it. (At process stop no cycle remains either
+    way; shutdown delivery is explicitly out of scope, and the restore there
+    only keeps the pending set honest.)"""
+    namespace = "test_flush_cancelled"
+    started = asyncio.Event()
+
+    async def never_finishes(ns: str) -> bool:
+        started.set()
+        await asyncio.Event().wait()
+        return True
+
+    poller = CacheInvalidationPoller(SessionLocal)
+    monkeypatch.setattr(poller, "bump", never_finishes)
+    poller.request_bump(namespace)
+
+    flush_task = asyncio.create_task(poller._flush_pending_bumps())
+    await asyncio.wait_for(started.wait(), timeout=2.0)
+    assert namespace not in poller._pending_bumps, "marker is cleared before the write, by design"
+
+    flush_task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await flush_task
+
+    assert namespace in poller._pending_bumps
+    assert await _namespace_version(namespace) is None
+
+
+@pytest.mark.asyncio
+async def test_pending_bump_survives_a_raising_flush_and_does_not_starve_others(db_setup, monkeypatch) -> None:
+    """A raise is abnormal (bump() reports failure by returning False), so it
+    must not abort the flush: the loop is sorted, and a persistently raising
+    namespace sorting first would otherwise starve every namespace after it
+    on every cycle. The raiser stays pending; the rest still land."""
+    raising_namespace = "test_flush_raised_a"
+    healthy_namespace = "test_flush_raised_b"
+    real_bump = CacheInvalidationPoller.bump
+
+    poller = CacheInvalidationPoller(SessionLocal)
+
+    async def raising_for_one(ns: str) -> bool:
+        if ns == raising_namespace:
+            raise RuntimeError("driver exploded")
+        return await real_bump(poller, ns)
+
+    monkeypatch.setattr(poller, "bump", raising_for_one)
+    poller.request_bump(raising_namespace)
+    poller.request_bump(healthy_namespace)
+
+    await poller._flush_pending_bumps()
+
+    assert raising_namespace in poller._pending_bumps
+    assert await _namespace_version(raising_namespace) is None
+    assert healthy_namespace not in poller._pending_bumps
+    assert await _namespace_version(healthy_namespace) == 1
+
+
 @pytest.mark.asyncio
 async def test_pending_coalesced_bump_flushes_after_recovery(db_setup) -> None:
     namespace = "test_pending_flush"
@@ -793,3 +853,41 @@ async def test_inflight_poll_does_not_clobber_concurrent_local_bump(db_setup) ->
     await source._poll_once()
     assert source_calls == []
     assert source._known_versions.get(NAMESPACE_RESET_CREDITS) == 2
+
+
+@pytest.mark.asyncio
+async def test_aborted_bump_is_retried_by_the_running_poller(db_setup, monkeypatch) -> None:
+    """End-to-end: the point of restoring the marker is that the background
+    poller actually retries. Without the restore the first raise loses the
+    namespace and no later cycle ever writes its version."""
+    namespace = "test_abort_retried_by_poller"
+    attempts = 0
+    real_bump = CacheInvalidationPoller.bump
+
+    poller = CacheInvalidationPoller(SessionLocal, poll_interval_seconds=0.01)
+
+    retry_settled = asyncio.Event()
+
+    async def failing_then_real(ns: str) -> bool:
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise RuntimeError("driver exploded")
+        result = await real_bump(poller, ns)
+        # Signal only after bump() fully returns: the committed row can become
+        # visible while the shielded session cleanup is still running, and
+        # stopping the poller at that instant would cancel the retry mid-flight.
+        retry_settled.set()
+        return result
+
+    monkeypatch.setattr(poller, "bump", failing_then_real)
+    poller.request_bump(namespace)
+    await poller.start()
+    try:
+        await asyncio.wait_for(retry_settled.wait(), timeout=5.0)
+    finally:
+        await poller.stop()
+
+    assert attempts >= 2, "the poller must retry the aborted namespace"
+    assert await _namespace_version(namespace) == 1
+    assert namespace not in poller._pending_bumps