fix: 2026-08 deep-review wave — durability, long-uptime arithmetic, cluster election, silent-failure hardening

CHANGELOG.md

@@ -22,6 +22,40 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   MEET + gossip and flags a killed node, per runtime.
 
 ### Fixed
+- **Deep-review wave (2026-08): long-uptime and large-scale correctness.**
+  Eight fix groups from a six-dimension architecture review (durability
+  ordering, long-uptime resource growth, concurrency, cluster correctness,
+  long-horizon arithmetic, silent degradation):
+  - *Eviction/metadata*: `CompactEntry` now stores a full-width u32
+    `last_access` (repurposed padding — zero size cost) and a 24-bit entry
+    version starting at 1. Fixes LFU decay that was effectively random
+    (u16/u32 domain mix truncated `as u8`), LRU inversion for idle gaps
+    beyond ~9.1h, `OBJECT IDLETIME` wrapping at 18.2h, and WATCH/EXEC's
+    8-bit version ABA (wrap now needs 16.7M writes; version 0 reliably
+    means "absent" so WATCH detects creation).
+  - *AOF everysec*: tokio deadline-fsync paths no longer swallow errors and
+    record success; all four sites (both runtimes/layouts) latch
+    `aof_last_fsync_status:ok|err` + `aof_fsync_failures` into INFO.
+  - *Spill*: a failed background spill pwrite re-inserts the already-evicted
+    key into the hot table (payload carried back in the completion) instead
+    of silently serving nil until restart; counted as
+    `spill_failed_reinserted` in INFO.
+  - *AOF rewrite prune*: the old generation is deleted only after a new
+    manifest-sync flush barrier (pending deferred ShardManifest commits
+    made durable — they relied on the old incr as their crash backstop) and
+    an explicit durable dir-fsync of the manifest flip.
+  - *CLIENT TRACKING*: the documented `max_keys` bound is enforced (evict +
+    invalidate instead of unbounded growth).
+  - *Cluster*: inline GET/SET fast path is disabled in cluster mode (was
+    bypassing MOVED/ASK entirely); election acks are now actually received
+    (read back on the request stream), epoch-bound and voter-deduped; the
+    election winner no longer claims an unvoted epoch; graceful `CLUSTER
+    FAILOVER` errors instead of permanently wedging automatic failover;
+    `ClusterState` migrated to `parking_lot::RwLock` (no poisoning, ~25
+    `.unwrap()` sites removed from the per-command path).
+  - *Fail-loud + bounds*: vector background-compaction failures now log at
+    every layer; CDC reaps disconnected subscribers on write-idle shards;
+    `TemporalRegistry` is bounded (262K bindings/shard, oldest evicted).
 - **Cluster formation actually converges (pre-existing, both runtimes).**
   A 3-node cluster could never complete its mesh: (1) `CLUSTER MEET`'s
   random-id placeholder was never retired when the peer's handshake arrived

src/cdc/fanout.rs

@@ -139,8 +139,13 @@ impl CdcSubscriberRegistry {
 
         for (idx, sub) in self.subscribers.iter_mut().enumerate() {
             let mut drained = 0usize;
-            let mut disconnect = false;
-            while drained < MAX_EVENTS_PER_SUBSCRIBER_PER_TICK {
+            // G3 (deep review): reap dead consumers even on write-idle
+            // shards. Without this, a disconnected receiver was only
+            // detected via a failed try_send driven by a NEW record, so
+            // idle shards accumulated dead subscribers (and their tail
+            // readers) indefinitely.
+            let mut disconnect = sub.tx.is_disconnected();
+            while !disconnect && drained < MAX_EVENTS_PER_SUBSCRIBER_PER_TICK {
                 match sub.tail.read_next() {
                     Ok(Some(rec)) => {
                         if rec.lsn < sub.from_lsn {
@@ -248,6 +253,34 @@ mod tests {
         assert_eq!(rx.len(), 5);
     }
 
+    /// G3 (deep review): a subscriber whose consumer disconnected must be
+    /// reaped even when NO new WAL records arrive — previously the dead
+    /// sender was only detected via a failed try_send driven by a fresh
+    /// record, so write-idle shards accumulated dead subscribers (each
+    /// holding a tail reader) indefinitely.
+    #[test]
+    fn test_cdc_fanout_reaps_disconnected_subscriber_when_idle() {
+        let tmp = tempfile::tempdir().unwrap();
+        let wal_dir = tmp.path().join("wal");
+        write_kv_records(&wal_dir, 2);
+
+        let mut reg = CdcSubscriberRegistry::new(0);
+        let (sub, rx) = CdcSubscriber::new(&wal_dir, 1);
+        reg.add(sub);
+        // Drain the backlog so the WAL is fully consumed (idle from now on).
+        reg.fanout_tick(0);
+        assert_eq!(reg.len(), 1);
+
+        // Consumer goes away; no further writes ever happen on this shard.
+        drop(rx);
+        reg.fanout_tick(0);
+        assert_eq!(
+            reg.len(),
+            0,
+            "disconnected subscriber must be reaped without needing a new record"
+        );
+    }
+
     /// C3b-1 — slow-consumer policy: bounded(1) channel fills, the next
     /// `try_send` returns Full, the subscriber is removed from the
     /// registry. The dropped envelopes are NOT redelivered — the

src/cluster/bus.rs

@@ -5,8 +5,9 @@
 //! All I/O is on the listener runtime, never on shard threads.
 #![allow(unused_imports)]
 
+use parking_lot::RwLock;
 use std::net::SocketAddr;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 
 use crate::runtime::cancel::CancellationToken;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -19,9 +20,12 @@ use crate::cluster::gossip::{
 };
 
 /// Shared vote sender: set by gossip ticker when election starts, cleared when election ends.
-/// Bus handler forwards FailoverAuthAck votes through this channel.
+/// Bus handler forwards FailoverAuthAck votes through this channel as
+/// `(voter_node_id, ack_epoch)` — the epoch lets the election loop discard
+/// stale acks from a previous timed-out election, and the voter id lets it
+/// dedup so one master can never be counted twice (deep-review C2).
 pub type SharedVoteTx =
-    Arc<parking_lot::Mutex<Option<crate::runtime::channel::MpscSender<String>>>>;
+    Arc<parking_lot::Mutex<Option<crate::runtime::channel::MpscSender<(String, u64)>>>>;
 
 /// Maximum time to wait for a gossip message BODY after its 4-byte length
 /// prefix has been read (prod-hardening #10). Without this bound, any TCP
@@ -141,12 +145,12 @@ async fn handle_cluster_peer(
             GossipMsgType::Ping | GossipMsgType::Meet => {
                 // Merge their state into ours
                 {
-                    let mut cs = cluster_state.write().unwrap();
+                    let mut cs = cluster_state.write();
                     merge_gossip_into_state(&mut cs, &msg);
                 }
                 // Respond with PONG
                 let pong = {
-                    let cs = cluster_state.read().unwrap();
+                    let cs = cluster_state.read();
                     build_message(&cs, self_addr, GossipMsgType::Pong)
                 };
                 let pong_bytes = serialize_gossip(&pong);
@@ -155,7 +159,7 @@ async fn handle_cluster_peer(
                 stream.write_all(&pong_bytes).await?;
             }
             GossipMsgType::Pong => {
-                let mut cs = cluster_state.write().unwrap();
+                let mut cs = cluster_state.write();
                 merge_gossip_into_state(&mut cs, &msg);
             }
             GossipMsgType::FailoverAuthRequest => {
@@ -165,18 +169,24 @@ async fn handle_cluster_peer(
                     .to_string();
                 let request_epoch = msg.config_epoch;
                 let voted = {
-                    let mut cs = cluster_state.write().unwrap();
+                    let mut cs = cluster_state.write();
                     crate::cluster::failover::handle_failover_auth_request(
                         &mut cs,
                         &sender_id,
                         request_epoch,
                     )
                 };
                 if voted {
-                    // Send FailoverAuthAck back to the requesting replica
+                    // Send FailoverAuthAck back to the requesting replica on
+                    // THIS stream — the requester holds it open and reads the
+                    // reply (deep-review C1). The ack ECHOES the request
+                    // epoch (not this voter's own config epoch) so the
+                    // requester can bind the vote to its election (C2).
                     let ack = {
-                        let cs = cluster_state.read().unwrap();
-                        build_message(&cs, self_addr, GossipMsgType::FailoverAuthAck)
+                        let cs = cluster_state.read();
+                        let mut m = build_message(&cs, self_addr, GossipMsgType::FailoverAuthAck);
+                        m.config_epoch = request_epoch;
+                        m
                     };
                     let ack_bytes = serialize_gossip(&ack);
                     let len = (ack_bytes.len() as u32).to_be_bytes();
@@ -189,9 +199,12 @@ async fn handle_cluster_peer(
                     .unwrap_or("")
                     .trim_end_matches('\0')
                     .to_string();
-                debug!("Received failover ACK from {}", sender_id);
+                debug!(
+                    "Received failover ACK from {} (epoch {})",
+                    sender_id, msg.config_epoch
+                );
                 if let Some(tx) = vote_tx.lock().as_ref() {
-                    let _ = tx.send(sender_id);
+                    let _ = tx.send((sender_id, msg.config_epoch));
                 }
             }
         }