fix(db): keep retained fence proofs when a probe fails (#9200)

## Summary

A failed replica fence probe no longer drops the proofs already
retained, so one failed probe no longer sends every read to the writer.

**Why.** `run_probe` called `fence.close()` on any probe error, which
cleared the ring. Until the next successful probe 500 ms later,
`fence.newest()` was `None` and every routed read went to the writer
with `reason="uninitialized"`. Under load this showed up as a steady
share of `bridge_query` reads labelled `uninitialized` on every pod,
while `heartbeat_age` stayed small. The module already documents that a
retained entry's proof never decays: a failed probe proves nothing new,
but it doesn't make an earlier proof wrong. So clearing the ring only
cost reader capacity.

**What changes.** A failed probe logs and records nothing. The retained
entries stand. A probe that keeps failing still closes the fence through
the existing 30 s staleness gate (`FENCE_STALENESS`), and
`heartbeat_age` remains the liveness signal. Regression handling is
unchanged: an epoch change or a same-epoch token regression still clears
the ring in `record()`. `close()` has no production caller left and
becomes a test hook.

**Likely trigger.** The masked-activity check
(`ProbeError::MaskedActivity`) counts a client backend whose state is
`active` with a `NULL` `xact_start` as masked. A backend that has just
started a statement can briefly show as `active` before its `xact_start`
is set, so busy writers hit this transiently. Under pgbench on Postgres
17, between 0.3% and 1.3% of scans saw such a backend. This PR doesn't
change that classification. See the debt below.

No protocol, schema, metric or client change.

### Testing

- `replica_fence::postgres_tests::failed_probe_keeps_retained_proofs`
(new, no database needed, so it runs in the unit lane like its
neighbours): records an entry, runs a probe tick against a closed writer
pool (the probe fails), then asserts `newest()` and `resolve()` still
return the entry, and that a rewound token recorded under the same epoch
after the outage is a `TokenRegression`. Mutation-checked on Blox:
putting `fence.close()` back on the error path fails it, including when
only the regression assertion is left.
- On Blox: the full `buzz-db` Postgres lane, the `buzz-db` unit tests,
`cargo fmt`, clippy for `buzz-db` and `buzz-relay`, and a `buzz-relay`
build.

### Rollout

There's no flag. Rollback is a revert. After deploy, watch:

- `buzz_db_route_decision{decision="writer",reason="uninitialized"}`
should drop to about 0 once a pod has booted, and the
`decision="reader"` share should rise.
- `buzz_db_replica_heartbeat_age_seconds` should stay small. Staleness
is now the only thing that closes the fence on probe failure, so this is
the liveness signal.
- `reason="stale"` may rise a little. `newest()` returns `None` for an
empty ring and for one older than the staleness gate, and both are
labelled `uninitialized`. Bounded reads can also miss the budget for one
tick when a probe beat is missed and fall back as `stale`. Some of
today's `uninitialized` share is expected to move to `stale` rather than
disappear.

## What got simpler

`ReplicaFence::close()` leaves the production API. A probe error now has
one effect (log and keep proving from what's retained) instead of a
second path that discards valid state. The probe loop is split into a
one-tick function that the test calls directly.

Identified, out of scope:
- `buzz_db` warnings, including this probe's failure warning, are
dropped by the relay's default log filter (`buzz_relay=info`), so
operators can't see why a probe fails.
- `MaskedActivity` misclassifies the transient
`active`-without-`xact_start` state. Treating it as not-open looks
sound, because such a backend's transaction starts after the scan, but
it touches the proof and deserves its own review.
- The probe's exclusive advisory lock waits behind channel writers'
shared locks, so a slow writer delays the probe.
- Probe failures have no metric. With retained proofs a failed probe no
longer shows up as `uninitialized` routes, and its warning is filtered
out of the logs (above), so intermittent failures are now invisible. A
probe-failure counter would fix that.
- An epoch rotation that fails still leaves an opening for a restore
that keeps the epoch. `TokenRegression` clears the ring, so if the
rotation `UPDATE` then errors, the next sample stores the rewound token
under the old epoch against an empty ring. This PR doesn't change that
path. A writer-side high-water token kept outside the ring would close
it.
- `newest()` gives the same `uninitialized` label to an empty ring and a
stale one, which hides a dead probe in route metrics.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Luke Tornquist <tornquist@squareup.com>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

crates/buzz-db/src/runtime/replica_fence.rs

@@ -48,10 +48,14 @@
 //! `RETURNING token` or an epoch change (restore/re-seed) clears the retained
 //! ring, so no stale entry can masquerade as fresh coverage.
 //!
-//! Everything fails **closed**: probe errors, masked `pg_stat_activity`
-//! visibility, an unreadable heartbeat row on the reader session, an epoch
-//! mismatch, or an observed token below every retained entry all route the
-//! request back to the writer — degraded capacity, never holes.
+//! Everything fails **closed**: an unreadable heartbeat row on the reader
+//! session, an epoch mismatch, or an observed token below every retained
+//! entry all route the request back to the writer — degraded capacity, never
+//! holes. A failed probe (a writer error or masked `pg_stat_activity`
+//! visibility) proves nothing new, so it records no entry; the entries
+//! already retained stay valid, because a proof never decays. If the probe
+//! keeps failing, the newest entry ages past [`FENCE_STALENESS`] and the
+//! fence closes.
 //!
 //! Operational bypasses (sessions without the GUC, `session_replication_role
 //! = replica` restores) are outside the proof by design and require holding
@@ -204,8 +208,9 @@ impl ReplicaFence {
         Self::default()
     }
 
-    /// Close the fence: drop all retained proofs; reads route to the writer.
-    pub fn close(&self) {
+    /// Test hook: drop all retained proofs, so reads route to the writer.
+    #[cfg(test)]
+    pub(crate) fn close(&self) {
         let mut inner = self.inner.lock().expect("fence lock poisoned");
         inner.ring.clear();
     }
@@ -538,8 +543,8 @@ struct WriterSample {
     committed_at: Instant,
 }
 
-/// Errors that close the fence. All variants are logged and treated
-/// identically: fail closed.
+/// Errors that fail one probe. All variants are logged and treated
+/// identically: the probe records no entry, and the retained entries stand.
 #[derive(Debug, thiserror::Error)]
 pub enum ProbeError {
     /// A probe query against the writer failed.
@@ -803,20 +808,24 @@ pub struct HeartbeatObservation {
     pub backend: String,
 }
 
-/// Background probe loop: commit a heartbeat token every `PROBE_INTERVAL`;
-/// close the fence on any error. Runs for the life of the process.
+/// Background probe loop: commit a heartbeat token every `PROBE_INTERVAL`.
+/// Runs for the life of the process.
 pub async fn run_probe(writer: PgPool, fence: Arc<ReplicaFence>) {
     let mut interval = tokio::time::interval(PROBE_INTERVAL);
     interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
     loop {
         interval.tick().await;
-        match probe_once(&writer, &fence).await {
-            Ok(_) => {}
-            Err(e) => {
-                fence.close();
-                tracing::warn!(error = %e, "replica fence probe failed; fence closed");
-            }
-        }
+        probe_tick(&writer, &fence).await;
+    }
+}
+
+/// One probe-loop tick. A failure is logged and keeps the retained entries:
+/// they were proved by earlier handshakes and stay valid, so one failed
+/// probe must not route every read to the writer. Only staleness
+/// ([`FENCE_STALENESS`]) closes a fence whose probe keeps failing.
+async fn probe_tick(writer: &PgPool, fence: &ReplicaFence) {
+    if let Err(e) = probe_once(writer, fence).await {
+        tracing::warn!(error = %e, "replica fence probe failed; retained proofs stand");
     }
 }
 
@@ -870,6 +879,44 @@ mod postgres_tests {
         assert!(fence.heartbeat_age().expect("entry retained") > FENCE_STALENESS);
     }
 
+    /// A failed probe proves nothing new but must keep what earlier probes
+    /// proved: the retained entry still answers `newest()` and `resolve()`.
+    /// Closing here routed every read to the writer until the next success.
+    #[tokio::test]
+    async fn failed_probe_keeps_retained_proofs() {
+        let fence = ReplicaFence::new();
+        let epoch = Uuid::new_v4();
+        let wall = Utc::now();
+        fence.record(7, epoch, Instant::now(), wall);
+
+        let writer = sqlx::postgres::PgPoolOptions::new()
+            .connect_lazy(&crate::test_support::database_url())
+            .expect("construct lazy writer pool");
+        writer.close().await;
+        probe_once(&writer, &fence)
+            .await
+            .expect_err("a closed writer pool fails the probe");
+
+        probe_tick(&writer, &fence).await;
+
+        let newest = fence
+            .newest()
+            .expect("retained proof survives a failed probe");
+        assert_eq!((newest.token, newest.fence_wall), (7, wall));
+        assert!(matches!(
+            fence.resolve(7, epoch),
+            ResolveOutcome::Proved(entry) if entry.token == 7
+        ));
+
+        // The retained entry is also what catches an epoch-keeping restore
+        // across the outage: a rewound token is a regression, not a fresh
+        // start. An emptied ring here would accept it under the old epoch.
+        assert_eq!(
+            fence.record(5, epoch, Instant::now(), wall),
+            RecordOutcome::TokenRegression
+        );
+    }
+
     /// Resolve picks the greatest retained entry <= the observed token —
     /// a lagged reader proves from an older wall, never from thin air.
     #[test]