fix: CacheWarmer hardening (#65) (#68)

* docs(plan): warmer hardening plan (#65)

* refactor(warmer): consolidate MAX(tid) query + log/skip on failure (#65)

Add module-level _read_max_tid(conn) helper and a public
PGJsonbStorage.current_max_tid() method that logs on failure and
returns None.  Route _restore_state, poll_invalidations, and the
warmer's load_current_tid_fn through them.  Warmer skips warmup
when current_max_tid returns None instead of installing consensus=0.

Closes I1 and I2 of #65.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor(instance): hoist _read_max_tid import to module top (#65)

Matches the existing import-style block (6 other names from .storage
already imported at module top). Avoids a per-call name lookup in
the poll_invalidations hot path.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* feat(shared-cache): consensus_tid property + set() returns bool (#65)

Both are wanted by the cache warmer's race-detection fix (next
commit).  The existing instance.load / load_multiple call sites
ignore the new return value; no behaviour change there.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(warmer): recover from consensus race; warn on zero accepted writes (#65)

Closes I3 of #65.  After poll_advance, re-read consensus_tid and use
that as the polled_tid for subsequent set() calls — if another
instance advanced consensus ahead of our sampled TID, this ensures
our sets still pass the gate.  Track accepted vs. attempted writes
via the new set() return value; log a WARNING when the entire
warmup was rejected despite non-empty results.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test(shared-cache): pin main-storage _finish → shared.poll_advance (#65)

Coverage gap flagged in #63 final review.  Pins the invariant that
the direct-use write path advances consensus and invalidates
changed zoids in the shared cache, independent of any instance's
tpc_finish.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* docs: CHANGES entry for 1.12.1 warmer hardening (#65)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: jensens

CHANGES.md

@@ -1,5 +1,31 @@
 # Changelog
 
+## 1.12.1
+
+- Harden the `CacheWarmer` lifecycle (#65).
+  - The `MAX(tid)` lookup was duplicated in three places with three
+    different error-handling policies.  Consolidated behind a single
+    `_read_max_tid(conn)` helper and a public
+    `PGJsonbStorage.current_max_tid()` method that logs a warning and
+    returns `None` on failure.  The warmer skips warmup when the
+    method returns `None` instead of installing a fabricated
+    consensus of 0.
+  - `SharedLoadCache.set()` now returns `True` on accept and `False`
+    on rejection.  `SharedLoadCache.consensus_tid` is exposed as a
+    read-only property.  Both are used by the warmer's race-recovery
+    path; existing `instance.load` / `load_multiple` callers ignore
+    the new return value and are unaffected.
+  - The warmer re-reads `shared_cache.consensus_tid` after its own
+    `poll_advance` and uses that value as `polled_tid` for the
+    subsequent `set()` loop — this fixes a startup race where a
+    concurrent instance poll could advance consensus past the
+    warmer's sampled TID and cause every warmup write to be silently
+    rejected.  A WARNING is now logged when the entire warmup was
+    rejected despite a non-empty result set.
+  - Added a regression test that pins `PGJsonbStorage._finish`
+    (direct-use write path) advancing shared consensus and
+    invalidating changed zoids.
+
 ## 1.12.0
 
 - Introduce a process-wide `SharedLoadCache` that replaces per-

docs/superpowers/plans/2026-04-23-warmer-hardening.md

@@ -127,7 +127,16 @@ def warm(self, load_multiple_fn):
 
         Runs in a background daemon thread.  Primes the consensus TID
         on the shared cache to the current PG max_tid so that
-        subsequent ``shared.set`` calls are accepted.
+        subsequent ``shared.set`` calls are accepted.  Re-reads
+        consensus after ``poll_advance`` and uses that as the
+        ``polled_tid`` for set calls — this is the mitigation for the
+        startup race where an instance's poll advances consensus past
+        the warmer's sampled TID before the set loop begins.
+
+        Skips warmup when the TID is unavailable.  Logs a WARNING when
+        every set() was rejected despite a non-empty result set (a
+        likely sign of the race being wider than this mitigation
+        covers).
         """
         from ZODB.utils import p64
         from ZODB.utils import u64
@@ -137,23 +146,58 @@ def warm(self, load_multiple_fn):
             log.info("Cache warmer: no stats yet, skipping warmup")
             return
 
+        current_tid = self._load_current_tid_fn()
+        if current_tid is None:
+            log.warning("Cache warmer: could not read current TID, skipping warmup")
+            return
+
         oids = [p64(z) for z in top_zoids]
         try:
             results = load_multiple_fn(oids)
         except Exception:
             log.warning("Cache warmer: load_multiple failed", exc_info=True)
             return
 
-        # Prime consensus so set() accepts our writes, then populate.
-        current_tid = self._load_current_tid_fn()
+        if not results:
+            log.info("Cache warmer: load_multiple returned no objects")
+            return
+
+        # Prime consensus so set() accepts our writes.  Another instance
+        # may have advanced consensus beyond our sampled current_tid
+        # already — in that case poll_advance is a no-op, and the actual
+        # consensus is higher than current_tid.  Re-read it so our
+        # subsequent set() calls use the effective consensus as their
+        # polled_tid and pass the gate.
         self._shared_cache.poll_advance(new_tid=current_tid, changed_zoids=[])
+        effective_tid = self._shared_cache.consensus_tid
+        if effective_tid is None:  # guarded for paranoia; should not happen
+            log.warning("Cache warmer: consensus still None after poll_advance")
+            return
+
         written = 0
+        attempted = 0
         for oid, (data, tid_bytes) in results.items():
-            self._shared_cache.set(
+            attempted += 1
+            if self._shared_cache.set(
                 zoid=u64(oid),
                 data=data,
                 tid_bytes=tid_bytes,
-                polled_tid=current_tid,
+                polled_tid=effective_tid,
+            ):
+                written += 1
+
+        if written == 0:
+            log.warning(
+                "Cache warmer: all %d set() calls rejected by shared cache "
+                "(consensus=%d, sampled_tid=%d) — likely raced with a "
+                "concurrent instance poll",
+                attempted,
+                effective_tid,
+                current_tid,
+            )
+        else:
+            log.info(
+                "Cache warmer: loaded %d of %d objects into shared cache",
+                written,
+                attempted,
             )
-            written += 1
-        log.info("Cache warmer: loaded %d objects into shared cache", written)

src/zodb_pgjsonb/cache_warmer.py

@@ -17,6 +17,7 @@
 from .storage import _loadBefore_hf
 from .storage import _loadBefore_hp
 from .storage import _NoopSerialCache
+from .storage import _read_max_tid
 from .storage import LoadCache
 from .undo import _compute_undo
 from ZODB.ConflictResolution import ConflictResolvingStorage
@@ -180,10 +181,7 @@ def poll_invalidations(self):
         # (invalidation lookups AND load() calls) see this same state.
         self._begin_read_txn()
 
-        with self._conn.cursor() as cur:
-            cur.execute("SELECT COALESCE(MAX(tid), 0) AS max_tid FROM transaction_log")
-            row = cur.fetchone()
-            new_tid = row["max_tid"]
+        new_tid = _read_max_tid(self._conn)
 
         result = []
         changed_zoids = []

src/zodb_pgjsonb/instance.py

@@ -128,6 +128,18 @@ def __post_init__(self):
 DEFAULT_CACHE_LOCAL_MB = 16
 
 
+def _read_max_tid(conn):
+    """Return the current MAX(tid) from transaction_log, or 0 when empty.
+
+    Raises whatever psycopg raises on a broken connection or closed
+    database.  Callers decide whether to catch.
+    """
+    with conn.cursor() as cur:
+        cur.execute("SELECT COALESCE(MAX(tid), 0) AS max_tid FROM transaction_log")
+        row = cur.fetchone()
+        return row["max_tid"] if row else 0
+
+
 class _NoopSerialCache:
     """Drop-in dict substitute that never stores anything (#62).
 
@@ -249,6 +261,15 @@ def __init__(self, max_mb):
         self.hits = 0
         self.misses = 0
 
+    @property
+    def consensus_tid(self):
+        """The highest TID any instance has polled to (read-only).
+
+        Returns ``None`` before the first ``poll_advance`` call.
+        """
+        with self._lock:
+            return self._consensus_tid
+
     def get(self, zoid, polled_tid):
         """Return (data, tid) for zoid or None.
 
@@ -274,20 +295,19 @@ def get(self, zoid, polled_tid):
     def set(self, zoid, data, tid_bytes, polled_tid):
         """Store (data, tid_bytes) for zoid if the caller is up to date.
 
-        Rejects writes from callers whose snapshot is older than the
-        current consensus (which could be carrying stale pre-
-        invalidation bytes), and never replaces a newer entry with an
-        older one.
+        Returns True when the entry was accepted (possibly replacing an
+        older entry), False when rejected by the consensus gate, the
+        monotonicity gate, or the stale-polled_tid gate.
         """
         with self._lock:
             if self._consensus_tid is None or polled_tid is None:
-                return
+                return False
             if polled_tid < self._consensus_tid:
-                return
+                return False
             tid_int = u64(tid_bytes)
             existing = self._cache.get(zoid)
             if existing is not None and u64(existing[1]) >= tid_int:
-                return
+                return False
             if existing is not None:
                 self._current_bytes -= len(existing[0])
             self._cache[zoid] = (data, tid_bytes)
@@ -296,6 +316,7 @@ def set(self, zoid, data, tid_bytes, polled_tid):
             while self._current_bytes > self._max_bytes and self._cache:
                 _, (evicted_data, _) = self._cache.popitem(last=False)
                 self._current_bytes -= len(evicted_data)
+            return True
 
     def poll_advance(self, new_tid, changed_zoids):
         """Advance consensus_tid and invalidate changed zoids atomically."""
@@ -503,22 +524,11 @@ def __init__(
             estimated_objects = int(cache_per_connection_mb * 1_000_000 / avg_size)
             target = max(1, int(estimated_objects * cache_warm_pct / 100))
 
-            def _current_max_tid():
-                try:
-                    with self._conn.cursor() as cur:
-                        cur.execute(
-                            "SELECT COALESCE(MAX(tid), 0) AS t FROM transaction_log"
-                        )
-                        row = cur.fetchone()
-                        return row["t"] if row else 0
-                except Exception:
-                    return 0
-
             self._warmer = CacheWarmer(
                 self._conn,
                 target_count=target,
                 shared_cache=self._shared_cache,
-                load_current_tid_fn=_current_max_tid,
+                load_current_tid_fn=self.current_max_tid,
                 decay=cache_warm_decay,
             )
             import threading
@@ -579,14 +589,12 @@ def _restore_state(self):
             if max_oid > 0:
                 self._oid = p64(max_oid)
 
-            cur.execute("SELECT COALESCE(MAX(tid), 0) AS max_tid FROM transaction_log")
-            row = cur.fetchone()
-            max_tid = row["max_tid"]
-            if max_tid > 0:
-                self._ltid = p64(max_tid)
-                # Ensure _ts is at least as recent as the last committed TID
-                # so _new_tid() generates monotonically increasing TIDs.
-                self._ts = TimeStamp(self._ltid)
+        max_tid = _read_max_tid(self._conn)
+        if max_tid > 0:
+            self._ltid = p64(max_tid)
+            # Ensure _ts is at least as recent as the last committed TID
+            # so _new_tid() generates monotonically increasing TIDs.
+            self._ts = TimeStamp(self._ltid)
 
     def new_oid(self):
         """Allocate a new OID via PostgreSQL sequence (cross-process safe).
@@ -1055,6 +1063,23 @@ def lastTransaction(self):
         """Return TID of the last committed transaction."""
         return self._ltid
 
+    def current_max_tid(self):
+        """Return the current MAX(tid) in transaction_log, or ``None``.
+
+        On any psycopg / DB error the failure is logged at WARNING and
+        ``None`` is returned.  ``None`` signals callers to skip any
+        work that depends on a fresh TID (e.g. the cache warmer
+        gracefully skips warmup rather than installing a fabricated
+        consensus of 0).
+        """
+        try:
+            return _read_max_tid(self._conn)
+        except Exception:
+            logger.warning(
+                "PGJsonbStorage.current_max_tid: query failed", exc_info=True
+            )
+            return None
+
     def __len__(self):
         """Return approximate number of objects."""
         with self._conn.cursor() as cur:

src/zodb_pgjsonb/storage.py

@@ -1,6 +1,7 @@
 """Tests for CacheWarmer — unit + integration tests."""
 
 from unittest import mock
+from zodb_pgjsonb.cache_warmer import CacheWarmer
 
 import pytest
 
@@ -500,3 +501,107 @@ def load_multiple_fn(oids):
         shared = storage._shared_cache
         assert shared.get(zoid=1, polled_tid=100) == (b"data-" + p64(1), p64(50))
         assert shared.get(zoid=2, polled_tid=100) == (b"data-" + p64(2), p64(50))
+
+
+class TestWarmerRaceRecovery:
+    """#65 I3: warmer tolerates a consensus race and logs silent failures."""
+
+    def test_warm_uses_effective_consensus_when_race_advances_it(self):
+        """If consensus is already ahead, warmer's writes still land."""
+        from ZODB.utils import p64
+        from zodb_pgjsonb.storage import SharedLoadCache
+
+        shared = SharedLoadCache(max_mb=4)
+        # Simulate another instance's poll_advance racing ahead
+        shared.poll_advance(new_tid=1000, changed_zoids=[])
+
+        w = CacheWarmer(
+            conn=_FakeConn(top_oids=[1, 2]),
+            target_count=10,
+            shared_cache=shared,
+            load_current_tid_fn=lambda: 500,  # stale — before the race
+        )
+
+        def loader(oids):
+            return {oid: (b"data-" + oid, p64(500)) for oid in oids}
+
+        w.warm(loader)
+
+        # Entries must be present: the warmer re-read consensus and
+        # used 1000 (the actual value) as polled_tid instead of 500.
+        assert shared.get(zoid=1, polled_tid=1000) == (b"data-" + p64(1), p64(500))
+        assert shared.get(zoid=2, polled_tid=1000) == (b"data-" + p64(2), p64(500))
+
+    def test_warm_logs_warning_when_all_writes_rejected(self, caplog):
+        """If consensus advances past the warmer's effective polled_tid
+        mid-loop, every set() is rejected and the warmer flags it."""
+        from ZODB.utils import p64
+        from zodb_pgjsonb.storage import SharedLoadCache
+
+        import logging
+
+        # A cache subclass whose set() always returns False (simulating
+        # a mid-loop consensus advance that we can't actually time).
+        class _RejectingCache(SharedLoadCache):
+            def set(self, *a, **kw):
+                super().set(*a, **kw)
+                return False
+
+        shared = _RejectingCache(max_mb=4)
+        w = CacheWarmer(
+            conn=_FakeConn(top_oids=[1, 2, 3]),
+            target_count=10,
+            shared_cache=shared,
+            load_current_tid_fn=lambda: 100,
+        )
+
+        def loader(oids):
+            return {oid: (b"x", p64(50)) for oid in oids}
+
+        with caplog.at_level(logging.WARNING, logger="zodb_pgjsonb.cache_warmer"):
+            w.warm(loader)
+
+        warnings = [
+            r
+            for r in caplog.records
+            if r.levelno == logging.WARNING and "rejected" in r.getMessage().lower()
+        ]
+        assert warnings, (
+            f"expected a WARNING about rejected writes, got: "
+            f"{[r.getMessage() for r in caplog.records]}"
+        )
+
+    def test_warm_info_log_on_normal_writes(self, caplog):
+        """Happy path logs at INFO, not WARNING."""
+        from ZODB.utils import p64
+        from zodb_pgjsonb.storage import SharedLoadCache
+
+        import logging
+
+        shared = SharedLoadCache(max_mb=4)
+        w = CacheWarmer(
+            conn=_FakeConn(top_oids=[1, 2]),
+            target_count=10,
+            shared_cache=shared,
+            load_current_tid_fn=lambda: 100,
+        )
+
+        def loader(oids):
+            return {oid: (b"x", p64(50)) for oid in oids}
+
+        with caplog.at_level(logging.INFO, logger="zodb_pgjsonb.cache_warmer"):
+            w.warm(loader)
+
+        warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+        assert not warnings, (
+            f"expected no warnings, got: {[r.getMessage() for r in warnings]}"
+        )
+        infos = [
+            r
+            for r in caplog.records
+            if r.levelno == logging.INFO and "loaded 2 of 2 objects" in r.getMessage()
+        ]
+        assert infos, (
+            f"expected INFO with loaded count, got: "
+            f"{[r.getMessage() for r in caplog.records]}"
+        )