diff --git a/CHANGELOG.md b/CHANGELOG.md
index e5487e070..ffb55a149 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/src/cdc/fanout.rs b/src/cdc/fanout.rs
index be3ffbad2..02a9eec3b 100644
--- a/src/cdc/fanout.rs
+++ b/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
diff --git a/src/cluster/bus.rs b/src/cluster/bus.rs
index 18dd47b9a..b17be221e 100644
--- a/src/cluster/bus.rs
+++ b/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,7 +169,7 @@ 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,
@@ -173,10 +177,16 @@ async fn handle_cluster_peer(
                     )
                 };
                 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));
                 }
             }
         }
diff --git a/src/cluster/command.rs b/src/cluster/command.rs
index e362aa03e..7b8919114 100644
--- a/src/cluster/command.rs
+++ b/src/cluster/command.rs
@@ -3,8 +3,9 @@
 //! All subcommands operate on a shared Arc<RwLock<ClusterState>>.
 //! Called from handle_connection_sharded (intercepted before dispatch, like AUTH/CONFIG).
 
+use parking_lot::RwLock;
 use std::net::SocketAddr;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 
 use bytes::Bytes;
 
@@ -56,7 +57,7 @@ pub fn handle_cluster_command(
 
 /// CLUSTER INFO -- return a bulk string with cluster statistics in Redis format.
 pub fn handle_cluster_info(cs: &Arc<RwLock<ClusterState>>, _self_addr: SocketAddr) -> Frame {
-    let state = cs.read().unwrap();
+    let state = cs.read();
     let cluster_state_str = if state.status == ClusterStatus::Ok {
         "ok"
     } else {
@@ -89,7 +90,7 @@ pub fn handle_cluster_info(cs: &Arc<RwLock<ClusterState>>, _self_addr: SocketAdd
 
 /// CLUSTER MYID -- return this node's 40-char hex node ID.
 pub fn handle_cluster_myid(cs: &Arc<RwLock<ClusterState>>) -> Frame {
-    let state = cs.read().unwrap();
+    let state = cs.read();
     Frame::BulkString(Bytes::from(state.node_id.clone()))
 }
 
@@ -153,7 +154,7 @@ fn format_node_line(node: &ClusterNode, self_node_id: &str) -> String {
 /// CLUSTER NODES -- one line per known node in nodes.conf format:
 /// `<node-id> <ip>:<port>@<bus-port> <flags> <master-id> <ping-sent> <pong-recv> <epoch> <link-state> <slot-ranges>`
 pub fn handle_cluster_nodes(cs: &Arc<RwLock<ClusterState>>, _self_addr: SocketAddr) -> Frame {
-    let state = cs.read().unwrap();
+    let state = cs.read();
     let mut output = String::new();
     for node in state.nodes.values() {
         output.push_str(&format_node_line(node, &state.node_id));
@@ -177,7 +178,7 @@ pub fn handle_cluster_replicas(args: &[Frame], cs: &Arc<RwLock<ClusterState>>) -
     }
     let target_id = extract_string(&args[0]);
 
-    let state = cs.read().unwrap();
+    let state = cs.read();
 
     // ERR if the requested node-id is not in the cluster.
     if !state.nodes.contains_key(&target_id) {
@@ -197,7 +198,7 @@ pub fn handle_cluster_replicas(args: &[Frame], cs: &Arc<RwLock<ClusterState>>) -
 
 /// CLUSTER SLOTS -- return nested array: [start, end, [master-ip, master-port, master-id], [replica...]]
 pub fn handle_cluster_slots(cs: &Arc<RwLock<ClusterState>>) -> Frame {
-    let state = cs.read().unwrap();
+    let state = cs.read();
     let mut result = Vec::new();
     for node in state.nodes.values() {
         if !matches!(node.flags, NodeFlags::Master) {
@@ -274,7 +275,7 @@ pub fn handle_cluster_meet(args: &[Frame], cs: &Arc<RwLock<ClusterState>>) -> Fr
     use crate::replication::state::generate_repl_id;
     let peer_id = generate_repl_id();
 
-    let mut state = cs.write().unwrap();
+    let mut state = cs.write();
     if state.my_node().addr == addr {
         return Frame::Error(Bytes::from_static(b"ERR Can't MEET myself"));
     }
@@ -304,7 +305,7 @@ pub fn handle_cluster_addslots(args: &[Frame], cs: &Arc<RwLock<ClusterState>>) -
         Ok(s) => s,
         Err(e) => return Frame::Error(Bytes::from(e)),
     };
-    let mut state = cs.write().unwrap();
+    let mut state = cs.write();
     for slot in &slots {
         state.my_node_mut().set_slot(*slot);
     }
@@ -322,7 +323,7 @@ pub fn handle_cluster_delslots(args: &[Frame], cs: &Arc<RwLock<ClusterState>>) -
         Ok(s) => s,
         Err(e) => return Frame::Error(Bytes::from(e)),
     };
-    let mut state = cs.write().unwrap();
+    let mut state = cs.write();
     for slot in &slots {
         state.my_node_mut().clear_slot(*slot);
     }
@@ -345,7 +346,7 @@ pub fn handle_cluster_setslot(args: &[Frame], cs: &Arc<RwLock<ClusterState>>) ->
         _ => return Frame::Error(Bytes::from_static(b"ERR invalid subcommand")),
     };
 
-    let mut state = cs.write().unwrap();
+    let mut state = cs.write();
     match subop.as_slice() {
         b"MIGRATING" => {
             if args.len() < 3 {
@@ -418,7 +419,7 @@ pub fn handle_cluster_reset(
     _self_addr: SocketAddr,
 ) -> Frame {
     let hard = args.first().map(|a| matches!(a, Frame::BulkString(b) | Frame::SimpleString(b) if b.eq_ignore_ascii_case(b"hard"))).unwrap_or(false);
-    let mut state = cs.write().unwrap();
+    let mut state = cs.write();
     let my_id = state.node_id.clone();
     // Clear slots on my node
     *state.my_node_mut().slots = [0u8; 2048];
@@ -438,7 +439,7 @@ pub fn handle_cluster_replicate(args: &[Frame], cs: &Arc<RwLock<ClusterState>>)
         return Frame::Error(Bytes::from_static(b"ERR wrong number of arguments"));
     }
     let master_id = extract_string(&args[0]);
-    let mut state = cs.write().unwrap();
+    let mut state = cs.write();
     let my_id = state.node_id.clone();
     if let Some(my_node) = state.nodes.get_mut(&my_id) {
         my_node.flags = NodeFlags::Replica { master_id };
@@ -485,7 +486,7 @@ fn handle_cluster_failover(args: &[Frame], cs: &Arc<RwLock<ClusterState>>) -> Fr
         FailoverMode::Normal
     };
 
-    let mut state = cs.write().unwrap();
+    let mut state = cs.write();
 
     // Must be a replica to failover
     let _master_id = match &state.my_node().flags {
@@ -499,12 +500,18 @@ fn handle_cluster_failover(args: &[Frame], cs: &Arc<RwLock<ClusterState>>) -> Fr
 
     match mode {
         FailoverMode::Normal => {
-            // Set failover_state to trigger election on next gossip tick
-            state.failover_state = crate::cluster::FailoverState::WaitingDelay {
-                start_ms: now_ms(),
-                delay_ms: 0, // gossip ticker will compute actual delay
-            };
-            Frame::SimpleString(Bytes::from_static(b"OK"))
+            // Deep-review R8: this used to park failover_state at
+            // WaitingDelay{delay_ms:0} — which nothing ever consumed (the
+            // ticker's election gate requires FailoverState::None and the
+            // graceful manual-failover protocol is not yet implemented), so
+            // one innocuous admin command permanently disabled AUTOMATIC
+            // failover on the node too. Until the coordinated
+            // (pause-master/offset-sync) protocol lands, fail loudly instead
+            // of wedging.
+            Frame::Error(Bytes::from_static(
+                b"ERR CLUSTER FAILOVER without FORCE or TAKEOVER is not yet supported; \
+                  use CLUSTER FAILOVER FORCE (skip offset sync) or TAKEOVER (skip voting)",
+            ))
         }
         FailoverMode::Force => {
             // Skip voting, promote immediately (master may be unreachable)
@@ -540,7 +547,7 @@ pub fn handle_cluster_count_failure_reports(
     // A report is active when its age is strictly less than 2 * timeout.
     let stale_cutoff = now.saturating_sub(2 * DEFAULT_NODE_TIMEOUT_MS);
 
-    let state = cs.read().unwrap();
+    let state = cs.read();
     let count = match state.nodes.get(&target_id) {
         None => 0i64,
         Some(node) => node
@@ -676,7 +683,7 @@ mod tests {
     fn test_addslots_updates_bitmap() {
         let cs = make_cs();
         {
-            let state = cs.read().unwrap();
+            let state = cs.read();
             assert_eq!(state.assigned_slot_count(), 0);
         }
         let args = vec![
@@ -685,7 +692,7 @@ mod tests {
         ];
         let result = handle_cluster_addslots(&args, &cs);
         assert!(matches!(result, Frame::SimpleString(_)));
-        let state = cs.read().unwrap();
+        let state = cs.read();
         assert_eq!(state.assigned_slot_count(), 2);
         assert!(state.my_node().owns_slot(0));
         assert!(state.my_node().owns_slot(1));
@@ -703,7 +710,7 @@ mod tests {
         ];
         let r = handle_cluster_setslot(&args, &cs);
         assert!(matches!(r, Frame::SimpleString(_)));
-        assert_eq!(cs.read().unwrap().migrating.get(&100), Some(&node_id));
+        assert_eq!(cs.read().migrating.get(&100), Some(&node_id));
     }
 
     /// KEYSLOT: slot_for_key("foo") == 12182 per CRC16-XMODEM.
@@ -719,10 +726,10 @@ mod tests {
         let cs = make_cs();
         // Add slot 5 first
         handle_cluster_addslots(&[Frame::BulkString(bytes::Bytes::from_static(b"5"))], &cs);
-        assert!(cs.read().unwrap().my_node().owns_slot(5));
+        assert!(cs.read().my_node().owns_slot(5));
         // Delete it
         handle_cluster_delslots(&[Frame::BulkString(bytes::Bytes::from_static(b"5"))], &cs);
-        assert!(!cs.read().unwrap().my_node().owns_slot(5));
+        assert!(!cs.read().my_node().owns_slot(5));
     }
 
     /// SETSLOT NODE clears migrating/importing and transfers ownership.
@@ -738,7 +745,7 @@ mod tests {
             Frame::BulkString(bytes::Bytes::from("b".repeat(40))),
         ];
         handle_cluster_setslot(&args, &cs);
-        assert!(cs.read().unwrap().migrating.contains_key(&42));
+        assert!(cs.read().migrating.contains_key(&42));
         // Now SETSLOT 42 NODE <my_id>
         let args2 = vec![
             Frame::BulkString(bytes::Bytes::from_static(b"42")),
@@ -746,7 +753,7 @@ mod tests {
             Frame::BulkString(bytes::Bytes::from(my_id.clone())),
         ];
         handle_cluster_setslot(&args2, &cs);
-        let state = cs.read().unwrap();
+        let state = cs.read();
         assert!(!state.migrating.contains_key(&42));
         assert!(state.my_node().owns_slot(42));
     }
@@ -755,7 +762,7 @@ mod tests {
     #[test]
     fn test_cluster_meet_adds_node() {
         let cs = make_cs();
-        assert_eq!(cs.read().unwrap().nodes.len(), 1);
+        assert_eq!(cs.read().nodes.len(), 1);
         let args = vec![
             Frame::BulkString(bytes::Bytes::from_static(b"MEET")),
             Frame::BulkString(bytes::Bytes::from_static(b"192.168.1.2")),
@@ -763,7 +770,7 @@ mod tests {
         ];
         let result = handle_cluster_command(&args, &cs, "127.0.0.1:6379".parse().unwrap());
         assert!(matches!(result, Frame::SimpleString(_)));
-        assert_eq!(cs.read().unwrap().nodes.len(), 2);
+        assert_eq!(cs.read().nodes.len(), 2);
     }
 
     /// Repeated CLUSTER MEET for one address must not stack placeholders:
@@ -781,11 +788,7 @@ mod tests {
             let result = handle_cluster_command(&args, &cs, "127.0.0.1:6379".parse().unwrap());
             assert!(matches!(result, Frame::SimpleString(_)));
         }
-        assert_eq!(
-            cs.read().unwrap().nodes.len(),
-            2,
-            "one placeholder, not three"
-        );
+        assert_eq!(cs.read().nodes.len(), 2, "one placeholder, not three");
     }
 
     /// MEET-ing our own advertised address is refused.
@@ -802,7 +805,7 @@ mod tests {
             matches!(result, Frame::Error(_)),
             "self-MEET must be an error"
         );
-        assert_eq!(cs.read().unwrap().nodes.len(), 1);
+        assert_eq!(cs.read().nodes.len(), 1);
     }
 
     /// Helper: create a ClusterState where this node is a replica of a FAIL master.
@@ -812,7 +815,7 @@ mod tests {
         let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6379);
         let cs = Arc::new(RwLock::new(ClusterState::new(my_id.clone(), addr)));
         {
-            let mut state = cs.write().unwrap();
+            let mut state = cs.write();
             // Make self a replica
             state.my_node_mut().flags = NodeFlags::Replica {
                 master_id: master_id.clone(),
@@ -857,7 +860,7 @@ mod tests {
         let args = vec![Frame::BulkString(bytes::Bytes::from_static(b"FORCE"))];
         let result = handle_cluster_failover(&args, &cs);
         assert!(matches!(result, Frame::SimpleString(_)));
-        let state = cs.read().unwrap();
+        let state = cs.read();
         assert!(
             matches!(state.my_node().flags, NodeFlags::Master),
             "expected Master after FORCE failover"
@@ -870,11 +873,11 @@ mod tests {
     #[test]
     fn test_failover_takeover_promotes_replica() {
         let cs = make_replica_with_fail_master();
-        let epoch_before = cs.read().unwrap().epoch;
+        let epoch_before = cs.read().epoch;
         let args = vec![Frame::BulkString(bytes::Bytes::from_static(b"TAKEOVER"))];
         let result = handle_cluster_failover(&args, &cs);
         assert!(matches!(result, Frame::SimpleString(_)));
-        let state = cs.read().unwrap();
+        let state = cs.read();
         assert!(
             matches!(state.my_node().flags, NodeFlags::Master),
             "expected Master after TAKEOVER failover"
@@ -908,20 +911,24 @@ mod tests {
         }
     }
 
-    /// CLUSTER FAILOVER (no args) on a replica sets WaitingDelay state.
+    /// R8 (deep review): graceful CLUSTER FAILOVER (no args) used to park
+    /// failover_state at WaitingDelay forever — nothing consumed it, and the
+    /// ticker's election gate requires FailoverState::None, so the command
+    /// permanently disabled automatic failover on the node. Until the
+    /// coordinated protocol exists it must error and leave state untouched.
     #[test]
-    fn test_failover_normal_sets_waiting_delay() {
+    fn test_failover_normal_errors_without_wedging_state() {
         let cs = make_replica_with_fail_master();
         let result = handle_cluster_failover(&[], &cs);
-        assert!(matches!(result, Frame::SimpleString(_)));
-        let state = cs.read().unwrap();
         assert!(
-            matches!(
-                state.failover_state,
-                crate::cluster::FailoverState::WaitingDelay { .. }
-            ),
-            "expected WaitingDelay state, got {:?}",
-            state.failover_state
+            matches!(result, Frame::Error(_)),
+            "graceful failover is unimplemented and must say so, not wedge"
+        );
+        let state = cs.read();
+        assert_eq!(
+            state.failover_state,
+            crate::cluster::FailoverState::None,
+            "failover_state must stay None so automatic failover keeps working"
         );
     }
 
@@ -938,7 +945,7 @@ mod tests {
         let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6379);
         let cs = Arc::new(RwLock::new(ClusterState::new(master_id.clone(), addr)));
         {
-            let mut state = cs.write().unwrap();
+            let mut state = cs.write();
             let r1 = ClusterNode::new(
                 replica1_id.clone(),
                 SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6380),
@@ -965,7 +972,7 @@ mod tests {
     #[test]
     fn cluster_replicas_returns_empty_for_master_with_no_replicas() {
         let cs = make_cs(); // single-node, no replicas
-        let my_id = cs.read().unwrap().node_id.clone();
+        let my_id = cs.read().node_id.clone();
         let args = vec![Frame::BulkString(bytes::Bytes::from(my_id))];
         let result = handle_cluster_replicas(&args, &cs);
         match result {
@@ -1136,7 +1143,7 @@ mod tests {
         let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6379);
         let cs = Arc::new(RwLock::new(ClusterState::new(my_id.clone(), addr)));
         {
-            let mut state = cs.write().unwrap();
+            let mut state = cs.write();
             // Make self a replica of master_id
             state.my_node_mut().flags = NodeFlags::Replica {
                 master_id: master_id.clone(),
@@ -1187,7 +1194,7 @@ mod tests {
     #[test]
     fn cluster_count_failure_reports_returns_zero_for_healthy_node() {
         let cs = make_cs();
-        let my_id = cs.read().unwrap().node_id.clone();
+        let my_id = cs.read().node_id.clone();
         let args = vec![Frame::BulkString(bytes::Bytes::from(my_id))];
         let result = handle_cluster_count_failure_reports(&args, &cs);
         assert_eq!(result, Frame::Integer(0));
@@ -1197,10 +1204,10 @@ mod tests {
     #[test]
     fn cluster_count_failure_reports_counts_active_reports() {
         let cs = make_cs();
-        let my_id = cs.read().unwrap().node_id.clone();
+        let my_id = cs.read().node_id.clone();
         let now = now_ms();
         {
-            let mut state = cs.write().unwrap();
+            let mut state = cs.write();
             let node = state.nodes.get_mut(&my_id).unwrap();
             // Two very recent reports
             node.pfail_reports.insert("reporter1".to_string(), now);
@@ -1215,7 +1222,7 @@ mod tests {
     #[test]
     fn cluster_count_failure_reports_excludes_stale_reports() {
         let cs = make_cs();
-        let my_id = cs.read().unwrap().node_id.clone();
+        let my_id = cs.read().node_id.clone();
         // Use absolute timestamps that are unambiguously on each side of any
         // reasonable stale_cutoff, so the test is not sensitive to clock skew
         // between when we insert and when the handler calls now_ms().
@@ -1226,7 +1233,7 @@ mod tests {
         let stale_ts: u64 = 0;
         let active_ts: u64 = u64::MAX / 2;
         {
-            let mut state = cs.write().unwrap();
+            let mut state = cs.write();
             let node = state.nodes.get_mut(&my_id).unwrap();
             node.pfail_reports
                 .insert("stale_reporter".to_string(), stale_ts);
diff --git a/src/cluster/failover.rs b/src/cluster/failover.rs
index ef24bf255..9456cb445 100644
--- a/src/cluster/failover.rs
+++ b/src/cluster/failover.rs
@@ -13,16 +13,17 @@
 //! 4. Masters vote if: request_epoch > last_vote_epoch.
 //! 5. Replica receiving majority: promotes itself, takes master's slots.
 
+use parking_lot::RwLock;
 use std::net::SocketAddr;
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use rand::RngExt;
-use tokio::io::AsyncWriteExt;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
-use tracing::{info, warn};
+use tracing::{debug, info, warn};
 
-use crate::cluster::gossip::{GossipMsgType, build_message, serialize_gossip};
+use crate::cluster::gossip::{GossipMsgType, build_message, deserialize_gossip, serialize_gossip};
 use crate::cluster::{ClusterState, FailoverState, NodeFlags};
 
 /// Check if we should initiate failover.
@@ -56,14 +57,28 @@ pub fn check_and_initiate_failover(state: &mut ClusterState, my_repl_offset: u64
 
     // Increment epoch for this failover attempt
     state.epoch += 1;
+    promote_self_to_master(state, &master_id);
 
-    // Promote ourselves locally
-    // (Awaiting majority votes before taking over master's slots is handled async)
+    true
+}
+
+/// Promote this node to master AT THE CURRENT config epoch, taking over
+/// `master_id`'s slots.
+///
+/// Deliberately does NOT bump `state.epoch`: the election-win path already
+/// incremented it when requesting votes, and a second bump here (deep-review
+/// C4) made the winner claim an epoch nobody voted for — defeating the
+/// `last_vote_epoch` single-vote-per-epoch guarantee and enabling epoch ties
+/// gossip's strict-`>` merge cannot resolve. Callers that skip voting
+/// (FORCE/TAKEOVER, via `check_and_initiate_failover`) bump the epoch
+/// themselves before calling.
+pub fn promote_self_to_master(state: &mut ClusterState, master_id: &str) {
     state.my_node_mut().flags = NodeFlags::Master;
-    state.my_node_mut().epoch = state.epoch;
+    let epoch = state.epoch;
+    state.my_node_mut().epoch = epoch;
 
     // Transfer master's slots to ourselves
-    let master_slots = state.nodes.get(&master_id).map(|n| n.slots.clone());
+    let master_slots = state.nodes.get(master_id).map(|n| n.slots.clone());
 
     if let Some(slots) = master_slots {
         let my_node = state.my_node_mut();
@@ -75,11 +90,9 @@ pub fn check_and_initiate_failover(state: &mut ClusterState, my_repl_offset: u64
     }
 
     // Remove the failed master from active routing (keep in nodes for history)
-    if let Some(failed) = state.nodes.get_mut(&master_id) {
+    if let Some(failed) = state.nodes.get_mut(master_id) {
         failed.flags = NodeFlags::Fail;
     }
-
-    true
 }
 
 /// Determine if this master should grant a failover vote.
@@ -188,7 +201,8 @@ pub async fn run_election_task(
     cluster_state: Arc<RwLock<ClusterState>>,
     self_addr: SocketAddr,
     _my_repl_offset: u64,
-    vote_rx: crate::runtime::channel::MpscReceiver<String>,
+    vote_rx: crate::runtime::channel::MpscReceiver<(String, u64)>,
+    vote_tx: crate::runtime::channel::MpscSender<(String, u64)>,
 ) {
     // Compute delay (rank 0 for now; multi-replica ranking is future work)
     let replica_rank = 0u32;
@@ -196,7 +210,7 @@ pub async fn run_election_task(
 
     // Set state to WaitingDelay
     {
-        let mut cs = cluster_state.write().unwrap();
+        let mut cs = cluster_state.write();
         cs.failover_state = FailoverState::WaitingDelay {
             start_ms: now_ms(),
             delay_ms: delay,
@@ -207,7 +221,7 @@ pub async fn run_election_task(
 
     // Increment epoch and build FailoverAuthRequest
     let (new_epoch, quorum, master_addrs) = {
-        let mut cs = cluster_state.write().unwrap();
+        let mut cs = cluster_state.write();
         cs.epoch += 1;
         let new_epoch = cs.epoch;
         let quorum = cs.quorum();
@@ -232,22 +246,53 @@ pub async fn run_election_task(
 
     // Build the auth request message
     let auth_msg = {
-        let cs = cluster_state.read().unwrap();
+        let cs = cluster_state.read();
         let mut msg = build_message(&cs, self_addr, GossipMsgType::FailoverAuthRequest);
         msg.config_epoch = new_epoch;
         msg
     };
 
-    // Send to all masters
+    // Send to all masters and READ each ack back on the same stream
+    // (deep-review C1): the previous fire-and-forget dropped the stream
+    // before the master's reply, and no master ever dials our bus to deliver
+    // an ack any other way — every multi-voter election timed out. The bus
+    // inbound FailoverAuthAck arm remains as a secondary path.
     let data = serialize_gossip(&auth_msg);
     for addr in &master_addrs {
         let data = data.clone();
         let addr = *addr;
+        let tx = vote_tx.clone();
         tokio::spawn(async move {
-            if let Ok(mut stream) = TcpStream::connect(addr).await {
-                let len = (data.len() as u32).to_be_bytes();
-                let _ = stream.write_all(&len).await;
-                let _ = stream.write_all(&data).await;
+            let Ok(mut stream) = TcpStream::connect(addr).await else {
+                return;
+            };
+            let len = (data.len() as u32).to_be_bytes();
+            if stream.write_all(&len).await.is_err() || stream.write_all(&data).await.is_err() {
+                return;
+            }
+            let read_ack = async {
+                let mut len_buf = [0u8; 4];
+                stream.read_exact(&mut len_buf).await.ok()?;
+                let body_len = u32::from_be_bytes(len_buf) as usize;
+                if body_len == 0 || body_len > crate::cluster::bus::MAX_GOSSIP_FRAME_LEN {
+                    return None;
+                }
+                let mut buf = vec![0u8; body_len];
+                stream.read_exact(&mut buf).await.ok()?;
+                deserialize_gossip(&buf).ok()
+            };
+            // Bounded: a master that voted NO sends nothing and we must not
+            // hold the task past the election deadline.
+            let Ok(Some(msg)) =
+                tokio::time::timeout(std::time::Duration::from_secs(4), read_ack).await
+            else {
+                return;
+            };
+            if msg.msg_type == GossipMsgType::FailoverAuthAck {
+                let voter = String::from_utf8_lossy(&msg.sender_node_id)
+                    .trim_end_matches('\0')
+                    .to_string();
+                let _ = tx.send((voter, msg.config_epoch));
             }
         });
     }
@@ -259,8 +304,14 @@ pub async fn run_election_task(
         quorum
     );
 
-    // Collect votes with 5-second timeout
+    // Collect votes with 5-second timeout. Dedup by voter id and bind each
+    // ack to THIS election's epoch (deep-review C2): without these, a stale
+    // ack from a previous timed-out election — or repeated acks from one
+    // voter — could fabricate quorum with fewer distinct voters.
     let mut votes: u32 = 1; // self-vote
+    let mut voters_seen: std::collections::HashSet<String> = std::collections::HashSet::new();
+    let my_id = { cluster_state.read().node_id.clone() };
+    voters_seen.insert(my_id);
     let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
 
     loop {
@@ -272,9 +323,19 @@ pub async fn run_election_task(
             break;
         }
         match tokio::time::timeout(remaining, vote_rx.recv_async()).await {
-            Ok(Ok(_voter_id)) => {
+            Ok(Ok((voter_id, ack_epoch))) => {
+                if ack_epoch != new_epoch {
+                    debug!(
+                        "Ignoring stale failover ack from {} (epoch {} != {})",
+                        voter_id, ack_epoch, new_epoch
+                    );
+                    continue;
+                }
+                if !voters_seen.insert(voter_id) {
+                    continue; // duplicate ack from the same voter
+                }
                 votes += 1;
-                let mut cs = cluster_state.write().unwrap();
+                let mut cs = cluster_state.write();
                 if let FailoverState::WaitingVotes {
                     ref mut votes_received,
                     ..
@@ -292,15 +353,24 @@ pub async fn run_election_task(
             "Failover election won: epoch={}, votes={}/{}",
             new_epoch, votes, quorum
         );
-        let mut cs = cluster_state.write().unwrap();
-        check_and_initiate_failover(&mut cs, _my_repl_offset);
+        let mut cs = cluster_state.write();
+        // Promote at the epoch the votes were granted for — NOT via
+        // check_and_initiate_failover, whose extra epoch increment claimed
+        // an unvoted epoch (deep-review C4).
+        let master_id = match &cs.my_node().flags {
+            NodeFlags::Replica { master_id } => Some(master_id.clone()),
+            _ => None,
+        };
+        if let Some(master_id) = master_id {
+            promote_self_to_master(&mut cs, &master_id);
+        }
         cs.failover_state = FailoverState::None;
     } else {
         warn!(
             "Failover election timed out: epoch={}, votes={}/{}",
             new_epoch, votes, quorum
         );
-        let mut cs = cluster_state.write().unwrap();
+        let mut cs = cluster_state.write();
         cs.failover_state = FailoverState::None;
     }
 }
@@ -364,6 +434,43 @@ mod tests {
         assert!(state.my_node().owns_slot(50));
     }
 
+    /// C4 (deep review): the election-win path must promote at the epoch the
+    /// votes were granted for. check_and_initiate_failover's extra increment
+    /// made the winner claim an unvoted epoch, defeating the
+    /// last_vote_epoch single-vote-per-epoch guarantee.
+    #[test]
+    fn test_promote_self_keeps_voted_epoch() {
+        let my_id = "a".repeat(40);
+        let master_id = "b".repeat(40);
+        let mut state = ClusterState::new(my_id.clone(), test_addr(6379));
+        state.my_node_mut().flags = NodeFlags::Replica {
+            master_id: master_id.clone(),
+        };
+        let mut master = crate::cluster::ClusterNode::new(
+            master_id.clone(),
+            test_addr(6380),
+            NodeFlags::Fail,
+            0,
+        );
+        for s in 0u16..=100 {
+            master.set_slot(s);
+        }
+        state.nodes.insert(master_id.clone(), master);
+
+        // Simulate the election task's own increment (the voted epoch).
+        state.epoch += 1;
+        let voted_epoch = state.epoch;
+
+        promote_self_to_master(&mut state, &master_id);
+        assert!(matches!(state.my_node().flags, NodeFlags::Master));
+        assert!(state.my_node().owns_slot(50));
+        assert_eq!(
+            state.epoch, voted_epoch,
+            "promotion must not claim an epoch nobody voted for"
+        );
+        assert_eq!(state.my_node().epoch, voted_epoch);
+    }
+
     /// Pitfall 8: double-vote prevention.
     #[test]
     fn test_failover_vote_epoch_guard() {
diff --git a/src/cluster/gossip.rs b/src/cluster/gossip.rs
index 15ce4eafd..3a099a687 100644
--- a/src/cluster/gossip.rs
+++ b/src/cluster/gossip.rs
@@ -5,8 +5,9 @@
 //! Gossip sections carry rumors about other nodes we know.
 #![allow(unused_imports)]
 
+use parking_lot::RwLock;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
-use std::sync::{Arc, RwLock};
+use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use crate::runtime::cancel::CancellationToken;
@@ -423,7 +424,7 @@ pub async fn run_gossip_ticker(
             _ = tick.tick() => {
                 // Pick a random peer to PING
                 let (target_addr, ping_msg) = {
-                    let mut cs = cluster_state.write().unwrap();
+                    let mut cs = cluster_state.write();
                     check_failure_states(&mut cs, node_timeout_ms);
 
                     // Reset election_spawned when failover state returns to None
@@ -445,15 +446,16 @@ pub async fn run_gossip_ticker(
                                     // Use try_lock to avoid holding std RwLock across await
                                     // We're in a sync context here so blocking_lock is safe
                                     let mut guard = vote_tx.lock();
-                                    *guard = Some(tx);
+                                    *guard = Some(tx.clone());
                                 }
                                 let cs_election = cluster_state.clone();
                                 let sa = self_addr;
                                 let offset = repl_state.read().total_offset();
                                 let vtx = vote_tx.clone();
+                                let tx_direct = tx.clone();
                                 tokio::spawn(async move {
                                     crate::cluster::failover::run_election_task(
-                                        cs_election, sa, offset, rx,
+                                        cs_election, sa, offset, rx, tx_direct,
                                     ).await;
                                     // Clear vote_tx when election ends
                                     *vtx.lock() = None;
@@ -511,7 +513,7 @@ pub async fn run_gossip_ticker(
                                         let mut pong_buf = vec![0u8; pong_len];
                                         if stream.read_exact(&mut pong_buf).await.is_ok() {
                                             if let Ok(pong) = deserialize_gossip(&pong_buf) {
-                                                let mut cs2 = cs.write().unwrap();
+                                                let mut cs2 = cs.write();
                                                 merge_gossip_into_state(&mut cs2, &pong);
                                             }
                                         }
diff --git a/src/cluster/mod.rs b/src/cluster/mod.rs
index a6b710650..37f9599bf 100644
--- a/src/cluster/mod.rs
+++ b/src/cluster/mod.rs
@@ -68,8 +68,20 @@ impl ClusterNode {
         // Redis convention: bus_port = port + 10000. Guard overflow for test
         // servers on ephemeral ports (49152-65535) where port + 10000 > u16::MAX.
         let bus_port = addr.port().checked_add(10000).unwrap_or_else(|| {
-            // Wrap into valid range for test compatibility
-            ((addr.port() as u32 + 10000) % 65536) as u16
+            // Wrap into valid range for test compatibility (ephemeral-port
+            // test clusters). Our OWN port is refused at startup when
+            // port+10000 > 65535 (main.rs), so a wrap here means a PEER
+            // announced such a port — its bus is almost certainly not
+            // reachable at the wrapped value (deep-review A5): say so.
+            let wrapped = ((addr.port() as u32 + 10000) % 65536) as u16;
+            tracing::warn!(
+                "cluster node {} announces client port {} whose bus port wraps to {} — \
+                 gossip to this peer will likely fail (peer ports must be <= 55535)",
+                node_id,
+                addr.port(),
+                wrapped
+            );
+            wrapped
         });
         ClusterNode {
             node_id,
diff --git a/src/command/connection.rs b/src/command/connection.rs
index c716ec31c..35b006a2a 100644
--- a/src/command/connection.rs
+++ b/src/command/connection.rs
@@ -299,8 +299,11 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
          aof_base_size:{}\r\n\
          aof_current_size:{}\r\n\
          aof_backpressure_dropped:{}\r\n\
+         aof_last_fsync_status:{}\r\n\
+         aof_fsync_failures:{}\r\n\
          spill_batches_flushed:{}\r\n\
          spill_completions_dropped:{}\r\n\
+         spill_failed_reinserted:{}\r\n\
          spill_last_heartbeat_ms:{}\r\n",
         if crate::command::persistence::SAVE_IN_PROGRESS.load(std::sync::atomic::Ordering::Relaxed)
         {
@@ -326,8 +329,15 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
         aof_current_size,
         crate::persistence::aof::AOF_BACKPRESSURE_DROPPED
             .load(std::sync::atomic::Ordering::Relaxed),
+        if crate::persistence::aof::aof_last_fsync_ok() {
+            "ok"
+        } else {
+            "err"
+        },
+        crate::persistence::aof::AOF_FSYNC_FAILURES.load(std::sync::atomic::Ordering::Relaxed),
         crate::storage::tiered::spill_thread::spill_batches_flushed_total(),
         crate::storage::tiered::spill_thread::spill_completion_dropped_total(),
+        crate::storage::tiered::spill_thread::spill_failed_reinserted_total(),
         crate::storage::tiered::spill_thread::spill_last_heartbeat_ms(),
     ));
     sections.push_str("\r\n");
diff --git a/src/command/key.rs b/src/command/key.rs
index 1b8816859..c569b6efc 100644
--- a/src/command/key.rs
+++ b/src/command/key.rs
@@ -530,8 +530,9 @@ pub fn object(db: &mut Database, args: &[Frame]) -> Frame {
         match db.get(key) {
             Some(entry) => {
                 let last = entry.last_access();
-                // Wraparound-safe delta in seconds (16-bit)
-                let idle = (now.wrapping_sub(last)) & 0xFFFF;
+                // Full u32 epoch-seconds delta; saturate rather than wrap if
+                // the cached clock lags a concurrent touch.
+                let idle = now.saturating_sub(last);
                 Frame::Integer(idle as i64)
             }
             None => Frame::Error(Bytes::from_static(b"ERR no such key")),
@@ -599,8 +600,9 @@ pub fn object_readonly(db: &Database, args: &[Frame], now_ms: u64) -> Frame {
         match db.get_if_alive(key, now_ms) {
             Some(entry) => {
                 let last = entry.last_access();
-                // Wraparound-safe delta in seconds (16-bit)
-                let idle = (now.wrapping_sub(last)) & 0xFFFF;
+                // Full u32 epoch-seconds delta; saturate rather than wrap if
+                // the cached clock lags a concurrent touch.
+                let idle = now.saturating_sub(last);
                 Frame::Integer(idle as i64)
             }
             None => Frame::Error(Bytes::from_static(b"ERR no such key")),
diff --git a/src/main.rs b/src/main.rs
index e8ee30e76..2f9271141 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1008,7 +1008,7 @@ fn main() -> anyhow::Result<()> {
     moon::admin::metrics_setup::set_global_repl_state(repl_state.clone());
 
     // Cluster mode initialization
-    let cluster_state: Option<std::sync::Arc<std::sync::RwLock<moon::cluster::ClusterState>>> =
+    let cluster_state: Option<std::sync::Arc<parking_lot::RwLock<moon::cluster::ClusterState>>> =
         if config.cluster_enabled {
             // Redis convention: cluster bus port = port + 10000. Refuse ports
             // where that wraps past u16::MAX instead of silently binding the
@@ -1027,11 +1027,8 @@ fn main() -> anyhow::Result<()> {
                 .expect("invalid bind address");
             let node_id = moon::replication::state::generate_repl_id();
             let state = moon::cluster::ClusterState::new(node_id, self_addr);
-            let cs = std::sync::Arc::new(std::sync::RwLock::new(state));
-            info!(
-                "Cluster mode enabled, node ID: {}",
-                cs.read().unwrap().node_id
-            );
+            let cs = std::sync::Arc::new(parking_lot::RwLock::new(state));
+            info!("Cluster mode enabled, node ID: {}", cs.read().node_id);
             Some(cs)
         } else {
             None
@@ -2087,7 +2084,7 @@ fn main() -> anyhow::Result<()> {
 /// it on a dedicated `cluster-ctl` std thread hosting a current-thread tokio
 /// runtime. All I/O here is gossip-rate (100ms ticks), never on shard threads.
 async fn run_cluster_control_plane(
-    cluster_state: std::sync::Arc<std::sync::RwLock<moon::cluster::ClusterState>>,
+    cluster_state: std::sync::Arc<parking_lot::RwLock<moon::cluster::ClusterState>>,
     bind: String,
     port: u16,
     node_timeout: u64,
diff --git a/src/persistence/aof/mod.rs b/src/persistence/aof/mod.rs
index 220bea070..9eadbff22 100644
--- a/src/persistence/aof/mod.rs
+++ b/src/persistence/aof/mod.rs
@@ -92,6 +92,47 @@ pub enum AofAck {
 pub static AOF_BACKPRESSURE_DROPPED: std::sync::atomic::AtomicU64 =
     std::sync::atomic::AtomicU64::new(0);
 
+/// Total everysec deadline-fsync failures across all AOF writers (both
+/// runtimes, both layouts). Monotonic; exposed as `aof_fsync_failures`.
+pub static AOF_FSYNC_FAILURES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+
+/// Per-writer "most recent everysec deadline fsync FAILED" bits (bit =
+/// `min(writer_idx, 63)`; TopLevel layout uses writer 0, PerShard uses the
+/// shard index). Surfaced as `aof_last_fsync_status:ok|err` in INFO
+/// (Redis's `aof_last_write_status` analogue): the status reads "err" while
+/// ANY writer's latest fsync failed. Per-writer bits — not one global bool —
+/// because with N per-shard writers a healthy shard's success within ~1s
+/// would otherwise mask a failing shard's hole (deep-review P2). Kernel
+/// semantics make a *retry* of a failed fsync succeed trivially while the
+/// failed window's pages are already gone (fsyncgate), so each writer's bit
+/// clears only when a fsync that follows a successful write batch completes
+/// on THAT writer — operators must treat any "err" observation as "an AOF
+/// has a hole".
+pub static AOF_FSYNC_ERR_WRITERS: std::sync::atomic::AtomicU64 =
+    std::sync::atomic::AtomicU64::new(0);
+
+/// True while every AOF writer's most recent everysec deadline fsync
+/// succeeded (the INFO `aof_last_fsync_status` predicate).
+pub fn aof_last_fsync_ok() -> bool {
+    AOF_FSYNC_ERR_WRITERS.load(std::sync::atomic::Ordering::Relaxed) == 0
+}
+
+/// Record the outcome of an everysec deadline fsync attempt on `writer_idx`
+/// (0 for the TopLevel writer, shard index for PerShard writers). Failure
+/// bumps [`AOF_FSYNC_FAILURES`] and sets the writer's bit in
+/// [`AOF_FSYNC_ERR_WRITERS`]; success clears only that writer's bit (the
+/// counter keeps the history).
+pub fn record_everysec_fsync_result(writer_idx: usize, ok: bool) {
+    use std::sync::atomic::Ordering;
+    let bit = 1u64 << writer_idx.min(63);
+    if ok {
+        AOF_FSYNC_ERR_WRITERS.fetch_and(!bit, Ordering::Relaxed);
+    } else {
+        AOF_FSYNC_FAILURES.fetch_add(1, Ordering::Relaxed);
+        AOF_FSYNC_ERR_WRITERS.fetch_or(bit, Ordering::Relaxed);
+    }
+}
+
 /// Bound for the SPSC-drain path's blocking AOF backpressure
 /// ([`AofWriterPool::send_append_bounded_blocking`]). Sized to cover the
 /// writer draining one group-commit batch (≤1024 msgs / 8 MiB buffered
@@ -391,13 +432,57 @@ impl PerShardRewriteCoord {
                 crate::command::persistence::AOF_REWRITE_IN_PROGRESS.store(false, Ordering::SeqCst);
                 return;
             }
-            for sid in 0..self.n_shards {
-                m.prune_shard_files(sid as u16, self.old_seq);
+            // Deep-review D1/D4: prune the old generation ONLY once (a) every
+            // manifest-sync agent's pending deferred ShardManifest commit is
+            // durable — the old incr is the AOF-replay backstop for exactly
+            // those spill placements — and (b) the manifest flip's dirent is
+            // durable (a crash booting the OLD manifest against pruned files
+            // replays as a silent fresh init and orphan-sweeps the NEW
+            // generation). On either failure keep both generations: costs
+            // disk, never data.
+            let safe_to_prune = match crate::persistence::manifest_sync::flush_all_agents() {
+                Ok(()) => {
+                    let dirent_durable = m
+                        .manifest_path()
+                        .parent()
+                        .map(crate::persistence::fsync::fsync_directory)
+                        .transpose();
+                    match dirent_durable {
+                        Ok(_) => true,
+                        Err(e) => {
+                            error!(
+                                "F6 rewrite: manifest dirent not durable ({}); keeping old \
+                                 generation seq {} on disk",
+                                e, self.old_seq
+                            );
+                            false
+                        }
+                    }
+                }
+                Err(e) => {
+                    error!(
+                        "F6 rewrite: manifest-sync barrier failed ({}); keeping old \
+                         generation seq {} on disk as the durability backstop",
+                        e, self.old_seq
+                    );
+                    false
+                }
+            };
+            if safe_to_prune {
+                for sid in 0..self.n_shards {
+                    m.prune_shard_files(sid as u16, self.old_seq);
+                }
             }
             drop(m);
             info!(
-                "F6 per-shard rewrite complete: committed seq {} across {} shards, pruned seq {}",
-                self.new_seq, self.n_shards, self.old_seq
+                "F6 per-shard rewrite complete: committed seq {} across {} shards{}",
+                self.new_seq,
+                self.n_shards,
+                if safe_to_prune {
+                    format!(", pruned seq {}", self.old_seq)
+                } else {
+                    format!(" (old seq {} retained)", self.old_seq)
+                }
             );
             // Success: new_seq is committed. Folded writers already reopened onto
             // new_seq in phase 6, so the barrier is a no-op for them — but it must
@@ -1315,4 +1400,30 @@ mod tests {
             std::str::from_utf8(AOF_FSYNC_ERR).unwrap_or("<non-utf8>")
         );
     }
+
+    /// F2/F3 (deep review): everysec fsync failures must latch an "err"
+    /// status and count — the old tokio paths dropped the error and recorded
+    /// success, so an operator had no way to learn the AOF has a hole.
+    /// Per-writer bits (deep-review P2 follow-up): a healthy writer's
+    /// success must NOT clear a failing sibling writer's "err" status.
+    #[test]
+    fn everysec_fsync_failure_latches_err_status() {
+        use std::sync::atomic::Ordering;
+        // Use writer indexes far above any concurrently-running writer-loop
+        // test so parallel tests can't flip these bits.
+        let before = AOF_FSYNC_FAILURES.load(Ordering::Relaxed);
+        record_everysec_fsync_result(61, false);
+        assert!(!aof_last_fsync_ok());
+        assert_eq!(AOF_FSYNC_FAILURES.load(Ordering::Relaxed), before + 1);
+        // A DIFFERENT writer's success must not mask writer 61's hole.
+        record_everysec_fsync_result(62, true);
+        assert!(
+            !aof_last_fsync_ok(),
+            "a healthy writer's fsync success must not clear a failing sibling's err status"
+        );
+        // Success on the failing writer restores the status but never the counter.
+        record_everysec_fsync_result(61, true);
+        assert!(aof_last_fsync_ok());
+        assert_eq!(AOF_FSYNC_FAILURES.load(Ordering::Relaxed), before + 1);
+    }
 }
diff --git a/src/persistence/aof/writer_task.rs b/src/persistence/aof/writer_task.rs
index 76a7355aa..90dd4ed35 100644
--- a/src/persistence/aof/writer_task.rs
+++ b/src/persistence/aof/writer_task.rs
@@ -630,9 +630,13 @@ pub async fn aof_writer_task(
                 let t = Instant::now();
                 if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
                     error!("AOF sync failed (seq {}, everysec): {}", manifest.seq, e);
-                    // Non-fatal for everysec: retry next interval
+                    crate::persistence::aof::record_everysec_fsync_result(0, false);
+                    // Non-fatal for everysec: retry next interval (status
+                    // stays latched "err" in INFO — the failed window's
+                    // pages are already gone even if the retry "succeeds").
                 } else {
                     crate::admin::metrics_setup::record_aof_fsync(t.elapsed().as_micros() as u64);
+                    crate::persistence::aof::record_everysec_fsync_result(0, true);
                     last_fsync = Instant::now();
                     idle_wait.clear_pending();
                 }
@@ -906,10 +910,28 @@ pub async fn aof_writer_task(
                 && !write_error
                 && last_fsync.elapsed() >= std::time::Duration::from_secs(1)
             {
-                let _ = writer.flush().await;
-                let _ = writer.get_ref().sync_data().await;
-                last_fsync = Instant::now();
-                idle_wait.clear_pending();
+                let t = Instant::now();
+                let res = match writer.flush().await {
+                    Ok(()) => writer.get_ref().sync_data().await,
+                    Err(e) => Err(e),
+                };
+                match res {
+                    Err(e) => {
+                        error!("AOF sync failed (everysec, tokio TopLevel): {}", e);
+                        crate::persistence::aof::record_everysec_fsync_result(0, false);
+                        // Keep last_fsync unadvanced so the deadline stays
+                        // armed — a silent success-record here would let the
+                        // failed window's loss self-heal invisibly.
+                    }
+                    Ok(()) => {
+                        crate::admin::metrics_setup::record_aof_fsync(
+                            t.elapsed().as_micros() as u64
+                        );
+                        crate::persistence::aof::record_everysec_fsync_result(0, true);
+                        last_fsync = Instant::now();
+                        idle_wait.clear_pending();
+                    }
+                }
             }
         }
     }
@@ -1335,12 +1357,37 @@ pub async fn per_shard_aof_writer_task(
             // arm under load, leaving >1s of writes buffered in the BufWriter
             // and lost on SIGKILL — the COMPOSE crash-matrix failure.)
             if fsync == FsyncPolicy::EverySec
+                && !write_error
                 && last_fsync.elapsed() >= std::time::Duration::from_secs(1)
             {
-                let _ = writer.flush().await;
-                let _ = writer.get_ref().sync_data().await;
-                last_fsync = Instant::now();
-                idle_wait.clear_pending();
+                let t = Instant::now();
+                let res = match writer.flush().await {
+                    Ok(()) => writer.get_ref().sync_data().await,
+                    Err(e) => Err(e),
+                };
+                match res {
+                    Err(e) => {
+                        error!(
+                            "AOF sync failed shard {} (everysec, tokio PerShard): {}",
+                            shard_id, e
+                        );
+                        crate::persistence::aof::record_everysec_fsync_result(
+                            usize::from(shard_id),
+                            false,
+                        );
+                    }
+                    Ok(()) => {
+                        crate::admin::metrics_setup::record_aof_fsync(
+                            t.elapsed().as_micros() as u64
+                        );
+                        crate::persistence::aof::record_everysec_fsync_result(
+                            usize::from(shard_id),
+                            true,
+                        );
+                        last_fsync = Instant::now();
+                        idle_wait.clear_pending();
+                    }
+                }
             }
         }
     }
@@ -1734,8 +1781,16 @@ pub async fn per_shard_aof_writer_task(
                         "AOF EverySec proactive sync failed shard {} (seq {}): {}",
                         shard_id, manifest.seq, e
                     );
+                    crate::persistence::aof::record_everysec_fsync_result(
+                        usize::from(shard_id),
+                        false,
+                    );
                 } else {
                     crate::admin::metrics_setup::record_aof_fsync(t.elapsed().as_micros() as u64);
+                    crate::persistence::aof::record_everysec_fsync_result(
+                        usize::from(shard_id),
+                        true,
+                    );
                     last_fsync = Instant::now();
                     idle_wait.clear_pending();
                 }
diff --git a/src/persistence/aof_manifest/mod.rs b/src/persistence/aof_manifest/mod.rs
index 9914c23f5..6e0f16faa 100644
--- a/src/persistence/aof_manifest/mod.rs
+++ b/src/persistence/aof_manifest/mod.rs
@@ -1029,17 +1029,57 @@ impl AofManifest {
                 source: e,
             })?;
 
-        // 4. Delete old files (best-effort)
+        // 4. Delete old files — ONLY once it is safe (deep-review D1/D4).
+        //
+        // D4: the manifest rename above fsyncs its parent dir best-effort. If
+        // that dirent is NOT durable and we prune anyway, a crash can boot
+        // with the OLD manifest resolving to already-deleted files — the
+        // replay path then treats it as a fresh init and cleanup_orphans
+        // deletes the new generation too: total dataset loss instead of a
+        // fail-stop. Re-attempt the dir fsync here, propagating: on failure
+        // keep BOTH generations on disk so either manifest resolution finds
+        // complete files.
+        //
+        // D1: the old incr is also the durability backstop for cold-plane
+        // spill placements whose ShardManifest commit is still sitting in a
+        // manifest-sync agent's deferred slot ("AOF replay + orphan sweep
+        // reconstruct anything a lost manifest commit would have recorded").
+        // Barrier those commits durable before deleting the records that
+        // back-stop them; on barrier failure, skip the prune.
+        if let Err(e) = crate::persistence::manifest_sync::flush_all_agents() {
+            warn!(
+                "AOF advance: manifest-sync barrier failed ({}); keeping old \
+                 generation seq {} on disk as the durability backstop",
+                e, old_seq
+            );
+            return Ok(new_incr);
+        }
+        if let Some(parent) = self.manifest_path().parent() {
+            if let Err(e) = fsync_directory(parent) {
+                warn!(
+                    "AOF advance: manifest dirent not durable ({}); keeping old \
+                     generation seq {} on disk (both generations present is \
+                     crash-safe; a pruned old generation with a non-durable \
+                     manifest flip is not)",
+                    e, old_seq
+                );
+                return Ok(new_incr);
+            }
+        }
         let old_base = self.base_path_seq(old_seq);
-        let old_incr = self.incr_path_seq(old_seq);
+        let old_incr_path = self.incr_path_seq(old_seq);
         if old_base.exists() {
             if let Err(e) = std::fs::remove_file(&old_base) {
                 warn!("Failed to delete old base {}: {}", old_base.display(), e);
             }
         }
-        if old_incr.exists() {
-            if let Err(e) = std::fs::remove_file(&old_incr) {
-                warn!("Failed to delete old incr {}: {}", old_incr.display(), e);
+        if old_incr_path.exists() {
+            if let Err(e) = std::fs::remove_file(&old_incr_path) {
+                warn!(
+                    "Failed to delete old incr {}: {}",
+                    old_incr_path.display(),
+                    e
+                );
             }
         }
 
diff --git a/src/persistence/manifest.rs b/src/persistence/manifest.rs
index 7f3cfa8ca..b32169145 100644
--- a/src/persistence/manifest.rs
+++ b/src/persistence/manifest.rs
@@ -305,6 +305,13 @@ pub(crate) static TEST_INJECT_SYNC_DELAY_MS: std::sync::atomic::AtomicU64 =
 pub(crate) static TEST_PERSIST_COUNT: std::sync::atomic::AtomicU64 =
     std::sync::atomic::AtomicU64::new(0);
 
+/// Test-only: make [`ManifestIo::persist`] fail with an injected I/O error
+/// (simulates ENOSPC/EIO on the manifest device) so the sync agent's
+/// failure latch can be exercised.
+#[cfg(test)]
+pub(crate) static TEST_INJECT_PERSIST_ERROR: std::sync::atomic::AtomicBool =
+    std::sync::atomic::AtomicBool::new(false);
+
 /// Serializes tests that use the injected sync delay / persist counter —
 /// `cargo test`'s default parallelism otherwise lets one test's injected
 /// delay leak into an unrelated concurrently-running test (same pattern as
@@ -933,6 +940,9 @@ impl ManifestIo {
             if ms > 0 {
                 std::thread::sleep(std::time::Duration::from_millis(ms));
             }
+            if TEST_INJECT_PERSIST_ERROR.load(std::sync::atomic::Ordering::SeqCst) {
+                return Err(std::io::Error::other("injected persist failure (test)"));
+            }
         }
         // A prior compaction renamed a fresh manifest into place durably but then
         // failed to reopen our handle, so `self.file` still points at the
diff --git a/src/persistence/manifest_sync.rs b/src/persistence/manifest_sync.rs
index c8f94d9e3..f71f519fa 100644
--- a/src/persistence/manifest_sync.rs
+++ b/src/persistence/manifest_sync.rs
@@ -54,6 +54,15 @@ struct SyncShared {
     /// Set once by `shutdown()`; the thread gives back its [`ManifestIo`]
     /// on this and exits after a final flush of the slot.
     shutdown: Option<flume::Sender<ManifestIo>>,
+    /// True while the NEWEST snapshot handed to this agent is known to NOT
+    /// be durable: the last `persist` failed and no newer snapshot has
+    /// succeeded since. An empty-slot round must ack `Err` while this is
+    /// set — the [`flush_all_agents`] barrier's "empty slot ⇒ something at
+    /// least as new already persisted" premise is false after a failed
+    /// persist (the failed root was consumed from the slot), and acking Ok
+    /// would let the AOF rewrite prune the very records that back-stop the
+    /// missing placements. Cleared only by a subsequent successful persist.
+    last_persist_failed: bool,
 }
 
 /// Handle to the per-shard manifest-sync thread. Owned by `ShardManifest`
@@ -69,6 +78,59 @@ pub(crate) struct ManifestSyncAgent {
     join: Option<std::thread::JoinHandle<()>>,
 }
 
+/// Process-global registry of live sync agents, for [`flush_all_agents`].
+/// Entries hold `Weak` shared-state refs plus a wake-sender clone; `spawn`
+/// registers, `shutdown` (and dead-`Weak` pruning) deregisters.
+static AGENT_REGISTRY: Mutex<Vec<(std::sync::Weak<Mutex<SyncShared>>, flume::Sender<()>)>> =
+    Mutex::new(Vec::new());
+
+/// Barrier: block until every live agent's pending deferred snapshot (if any)
+/// is durable (deep-review D1).
+///
+/// The AOF rewrite's commit path deletes the old incr generation — the very
+/// records that back-stop spill placements whose ShardManifest commit is
+/// still sitting in an agent's deferred slot ("AOF replay + orphan sweep
+/// reconstruct anything a lost manifest commit would have recorded"). Pruning
+/// while a deferred commit is pending re-opens the crash window that
+/// justification closed. Callers MUST invoke this before pruning and skip
+/// the prune on error.
+///
+/// Implementation: push an ack-only waiter into each agent (the run loop
+/// fires acks after persisting whatever is in the slot; an empty slot means a
+/// previous round already persisted something at least as new — it acks
+/// immediately with Ok). Bounded wait so a wedged agent fails the barrier
+/// instead of hanging the rewrite.
+pub(crate) fn flush_all_agents() -> io::Result<()> {
+    let mut waiters = Vec::new();
+    {
+        let mut reg = AGENT_REGISTRY.lock();
+        reg.retain(|(shared, _)| shared.strong_count() > 0);
+        for (weak_shared, wake) in reg.iter() {
+            let Some(shared) = weak_shared.upgrade() else {
+                continue;
+            };
+            let (ack_tx, ack_rx) = flume::bounded::<io::Result<()>>(1);
+            shared.lock().acks.push(ack_tx);
+            match wake.try_send(()) {
+                Ok(()) | Err(flume::TrySendError::Full(())) => {}
+                Err(flume::TrySendError::Disconnected(())) => {
+                    return Err(io::Error::other(
+                        "manifest-sync barrier: an agent thread is gone with commits possibly pending",
+                    ));
+                }
+            }
+            waiters.push(ack_rx);
+        }
+    }
+    for rx in waiters {
+        rx.recv_timeout(std::time::Duration::from_secs(30))
+            .map_err(|_| {
+                io::Error::other("manifest-sync barrier: timed out waiting for a persist ack")
+            })??;
+    }
+    Ok(())
+}
+
 impl std::fmt::Debug for ManifestSyncAgent {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("ManifestSyncAgent")
@@ -89,6 +151,7 @@ impl ManifestSyncAgent {
             latest: None,
             acks: Vec::new(),
             shutdown: None,
+            last_persist_failed: false,
         }));
         let (wake, rx) = flume::bounded(1);
         let (io_tx, io_rx) = flume::bounded::<ManifestIo>(1);
@@ -107,11 +170,16 @@ impl ManifestSyncAgent {
                 run(io, thread_shared, rx)
             }) {
             Ok(join) => match io_tx.send(io) {
-                Ok(()) => Ok(Self {
-                    shared,
-                    wake,
-                    join: Some(join),
-                }),
+                Ok(()) => {
+                    AGENT_REGISTRY
+                        .lock()
+                        .push((Arc::downgrade(&shared), wake.clone()));
+                    Ok(Self {
+                        shared,
+                        wake,
+                        join: Some(join),
+                    })
+                }
                 // Thread died before its recv (can only be a panic in thread
                 // start-up glue) — flume returns the unsent io in the error.
                 Err(flume::SendError(io)) => {
@@ -153,9 +221,20 @@ impl ManifestSyncAgent {
             .map_err(|_| io::Error::other("manifest-sync thread died before ack"))?
     }
 
+    /// Remove this agent's entry from [`AGENT_REGISTRY`]. Idempotent.
+    fn deregister(&self) {
+        AGENT_REGISTRY
+            .lock()
+            .retain(|(shared, _)| !std::sync::Weak::ptr_eq(shared, &Arc::downgrade(&self.shared)));
+    }
+
     /// Flush any pending commit and reclaim the file-I/O state. Returns
     /// `None` only if the thread died abnormally (its panic already logged).
     pub(crate) fn shutdown(mut self) -> Option<ManifestIo> {
+        // Deregister BEFORE the final flush: a flush_all_agents barrier
+        // racing this shutdown must not push an ack the exiting thread will
+        // never fire.
+        self.deregister();
         let (gb_tx, gb_rx) = flume::bounded::<ManifestIo>(1);
         self.shared.lock().shutdown = Some(gb_tx);
         let _ = self.notify();
@@ -167,6 +246,20 @@ impl ManifestSyncAgent {
     }
 }
 
+impl Drop for ManifestSyncAgent {
+    /// A handle dropped WITHOUT `shutdown()` (shard-loop panic unwind, early
+    /// teardown) must not leave a zombie: the registry's wake-sender clone
+    /// would otherwise keep the wake channel connected forever, parking the
+    /// sync thread (and its manifest fd) permanently — and the thread's own
+    /// strong `Arc` keeps the `strong_count` pruning from ever collecting
+    /// the entry. Deregistering here drops that last outside sender, so the
+    /// thread's `rx.recv()` disconnects, it flushes the slot, and exits.
+    /// Runs after `shutdown()` too (idempotent retain).
+    fn drop(&mut self) {
+        self.deregister();
+    }
+}
+
 fn run(mut io: ManifestIo, shared: Arc<Mutex<SyncShared>>, rx: flume::Receiver<()>) {
     while rx.recv().is_ok() {
         // Take the whole state atomically. Anything written after this take
@@ -185,14 +278,30 @@ fn run(mut io: ManifestIo, shared: Arc<Mutex<SyncShared>>, rx: flume::Receiver<(
         let res = match latest {
             Some(root) => {
                 let r = io.persist(root);
+                // Latch the outcome BEFORE firing acks so a barrier waiter
+                // pushed mid-persist (its ack lands in the NEXT round) can
+                // never observe a stale "ok" empty-slot answer.
+                shared.lock().last_persist_failed = r.is_err();
                 if let Err(ref e) = r {
                     tracing::warn!(error = %e, "manifest-sync: commit failed (placement metadata only; AOF replay + orphan sweep recover on restart)");
                 }
                 r
             }
             // Empty slot: a previous round already persisted a snapshot at
-            // least as new as anything these (necessarily-raced) wakes saw.
-            None => Ok(()),
+            // least as new as anything these (necessarily-raced) wakes saw —
+            // UNLESS that previous round failed, in which case the newest
+            // snapshot is known non-durable and the ack must say so (the
+            // rewrite-prune barrier skips the prune on this error).
+            None => {
+                if shared.lock().last_persist_failed {
+                    Err(io::Error::other(
+                        "manifest-sync: newest snapshot is not durable (last persist failed \
+                         and nothing newer has superseded it)",
+                    ))
+                } else {
+                    Ok(())
+                }
+            }
         };
         for ack in acks {
             let mirrored = match &res {
@@ -265,6 +374,44 @@ mod tests {
         assert_eq!(reopened.files().len(), 1, "deferred commit must reach disk");
     }
 
+    /// D1 (deep review): the AOF rewrite prunes the old incr — the durability
+    /// backstop for spill placements whose manifest commit is still sitting
+    /// in this agent's deferred slot. The prune path needs a barrier that
+    /// forces every registered agent's pending snapshot durable first.
+    #[test]
+    fn flush_all_agents_forces_pending_deferred_commit_durable() {
+        #[allow(clippy::unwrap_used)] // test-only; poisoning would already be a failed test
+        let _knob = super::super::manifest::TEST_SYNC_KNOB_LOCK.lock().unwrap();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let path = tmp.path().join("shard-9.manifest");
+        let mut m = ShardManifest::create(&path).expect("create");
+        m.enable_deferred_sync(9);
+
+        super::super::manifest::TEST_INJECT_SYNC_DELAY_MS
+            .store(100, std::sync::atomic::Ordering::SeqCst);
+        m.add_file(make_entry(41));
+        m.commit_deferred().expect("deferred send");
+        // Barrier must block until the deferred snapshot (or newer) is durable.
+        super::flush_all_agents().expect("flush barrier");
+        super::super::manifest::TEST_INJECT_SYNC_DELAY_MS
+            .store(0, std::sync::atomic::Ordering::SeqCst);
+
+        // No shutdown flush — the barrier alone must have persisted it.
+        let reopened = ShardManifest::open(&path).expect("reopen");
+        assert_eq!(
+            reopened.files().len(),
+            1,
+            "flush_all_agents must persist the pending deferred snapshot"
+        );
+        m.shutdown_deferred();
+    }
+
+    /// The barrier must not hang (and must succeed) when agents are idle.
+    #[test]
+    fn flush_all_agents_is_noop_when_idle() {
+        super::flush_all_agents().expect("idle barrier must succeed");
+    }
+
     #[test]
     fn durable_commit_blocks_until_persisted() {
         #[allow(clippy::unwrap_used)] // test-only; poisoning would already be a failed test
@@ -387,6 +534,75 @@ mod tests {
         );
     }
 
+    /// Deep-review P1: a FAILED deferred persist must fail the
+    /// `flush_all_agents` barrier (empty-slot rounds included) until a newer
+    /// snapshot persists successfully — otherwise the AOF rewrite prunes the
+    /// old incr while the on-disk manifest is missing spill placements.
+    #[test]
+    fn failed_persist_latches_flush_barrier_until_superseded() {
+        #[allow(clippy::unwrap_used)] // test-only; poisoning would already be a failed test
+        let _knob = super::super::manifest::TEST_SYNC_KNOB_LOCK.lock().unwrap();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let path = tmp.path().join("shard-5.manifest");
+        let mut m = ShardManifest::create(&path).expect("create");
+        m.enable_deferred_sync(5);
+
+        super::super::manifest::TEST_INJECT_PERSIST_ERROR
+            .store(true, std::sync::atomic::Ordering::SeqCst);
+        m.add_file(make_entry(1));
+        m.commit_deferred().expect("deferred send");
+        // Whether the failing round has already consumed the slot or the
+        // barrier's own round persists (and fails) it, the barrier must
+        // report the newest snapshot as non-durable.
+        assert!(
+            super::flush_all_agents().is_err(),
+            "barrier must fail while the newest snapshot is not durable"
+        );
+        // Still latched on a second, empty-slot barrier round.
+        assert!(
+            super::flush_all_agents().is_err(),
+            "empty-slot rounds must stay latched after a failed persist"
+        );
+
+        // A newer snapshot that persists successfully heals the latch.
+        super::super::manifest::TEST_INJECT_PERSIST_ERROR
+            .store(false, std::sync::atomic::Ordering::SeqCst);
+        m.add_file(make_entry(2));
+        m.commit_deferred().expect("deferred send");
+        assert!(
+            super::flush_all_agents().is_ok(),
+            "a successful newer persist must clear the latch"
+        );
+        m.shutdown_deferred();
+    }
+
+    /// Deep-review P2: dropping a deferred-mode manifest WITHOUT
+    /// `shutdown_deferred` (panic unwind, early teardown) must deregister
+    /// the agent and let its thread exit — the registry's wake-sender clone
+    /// must not park the thread (and its manifest fd) forever.
+    #[test]
+    fn dropped_agent_without_shutdown_deregisters() {
+        #[allow(clippy::unwrap_used)] // test-only; poisoning would already be a failed test
+        let _knob = super::super::manifest::TEST_SYNC_KNOB_LOCK.lock().unwrap();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let before = super::AGENT_REGISTRY.lock().len();
+        {
+            let path = tmp.path().join("shard-6.manifest");
+            let mut m = ShardManifest::create(&path).expect("create");
+            m.enable_deferred_sync(6);
+            m.add_file(make_entry(1));
+            m.commit_deferred().expect("deferred send");
+            // No shutdown_deferred: the agent handle drops with the manifest.
+        }
+        assert_eq!(
+            super::AGENT_REGISTRY.lock().len(),
+            before,
+            "agent drop must deregister its registry entry"
+        );
+        // The barrier must not hang or error on the departed agent.
+        assert!(super::flush_all_agents().is_ok());
+    }
+
     #[test]
     fn shutdown_reclaims_io_and_inline_commits_still_work() {
         #[allow(clippy::unwrap_used)] // test-only; poisoning would already be a failed test
diff --git a/src/server/conn/blocking.rs b/src/server/conn/blocking.rs
index 8341e7426..078af5257 100644
--- a/src/server/conn/blocking.rs
+++ b/src/server/conn/blocking.rs
@@ -1976,7 +1976,15 @@ pub(crate) fn try_inline_dispatch(
 
 /// Loop wrapper: call try_inline_dispatch repeatedly until it returns 0.
 /// Returns total number of commands inlined.
+///
+/// `cluster_enabled` must carry `crate::cluster::cluster_enabled()` (a param
+/// for unit-testability): in cluster mode NOTHING may inline — reads and
+/// writes alike must reach the generic dispatch loop's
+/// `try_handle_cluster_routing` so mis-routed keys get MOVED/ASK instead of
+/// being silently served/misplaced against the local DashTable (deep-review
+/// R6 — the classic "three dispatch paths" gap).
 #[cfg(feature = "runtime-monoio")]
+#[allow(clippy::too_many_arguments)]
 pub(crate) fn try_inline_dispatch_loop(
     read_buf: &mut BytesMut,
     write_buf: &mut BytesMut,
@@ -1990,8 +1998,12 @@ pub(crate) fn try_inline_dispatch_loop(
     now_ms: u64,
     num_shards: usize,
     can_inline_writes: bool,
+    cluster_enabled: bool,
     runtime_config: &parking_lot::RwLock<crate::config::RuntimeConfig>,
 ) -> usize {
+    if cluster_enabled {
+        return 0;
+    }
     let mut total = 0;
     loop {
         let n = try_inline_dispatch(
diff --git a/src/server/conn/core.rs b/src/server/conn/core.rs
index a01eb6921..9344eb0b7 100644
--- a/src/server/conn/core.rs
+++ b/src/server/conn/core.rs
@@ -65,7 +65,7 @@ pub(crate) struct ConnectionContext {
     /// dispatch by `try_enforce_readonly` to avoid the per-command RwLock CAS.
     /// `None` when replication is disabled entirely.
     pub is_replica_mirror: Option<Arc<std::sync::atomic::AtomicBool>>,
-    pub cluster_state: Option<Arc<StdRwLock<crate::cluster::ClusterState>>>,
+    pub cluster_state: Option<Arc<parking_lot::RwLock<crate::cluster::ClusterState>>>,
     pub lua: Rc<mlua::Lua>,
     pub script_cache: Rc<RefCell<crate::scripting::ScriptCache>>,
     pub config_port: u16,
@@ -108,7 +108,7 @@ impl ConnectionContext {
         aof_pool: Option<Arc<AofWriterPool>>,
         tracking_table: std::sync::Arc<parking_lot::Mutex<TrackingTable>>,
         repl_state: Option<Arc<parking_lot::RwLock<crate::replication::state::ReplicationState>>>,
-        cluster_state: Option<Arc<StdRwLock<crate::cluster::ClusterState>>>,
+        cluster_state: Option<Arc<parking_lot::RwLock<crate::cluster::ClusterState>>>,
         lua: Rc<mlua::Lua>,
         script_cache: Rc<RefCell<crate::scripting::ScriptCache>>,
         config_port: u16,
diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs
index 7e7fe7f87..571da0b55 100644
--- a/src/server/conn/handler_monoio/dispatch.rs
+++ b/src/server/conn/handler_monoio/dispatch.rs
@@ -260,9 +260,7 @@ pub(super) fn try_handle_cluster_routing(
     let maybe_key = super::extract_primary_key(cmd, cmd_args);
     if let Some(key) = maybe_key {
         let slot = crate::cluster::slots::slot_for_key(key);
-        #[allow(clippy::unwrap_used)]
-        // std RwLock: poison = prior panic = unrecoverable
-        let route = cs.read().unwrap().route_slot(slot, was_asking);
+        let route = cs.read().route_slot(slot, was_asking);
         match route {
             crate::cluster::SlotRoute::Local => {} // proceed
             other => {
diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs
index 4b3f4d00e..2b2c9fc6e 100644
--- a/src/server/conn/handler_monoio/mod.rs
+++ b/src/server/conn/handler_monoio/mod.rs
@@ -1144,6 +1144,9 @@ pub(crate) async fn handle_connection_sharded_monoio<
                 ctx.cached_clock.ms(),
                 ctx.num_shards,
                 can_inline_writes,
+                // R6: cluster mode disables the inline fast path entirely —
+                // GET/SET must reach try_handle_cluster_routing for MOVED/ASK.
+                crate::cluster::cluster_enabled(),
                 &ctx.runtime_config,
             );
             crate::admin::metrics_setup::record_dispatch_local_inline(inlined as u64);
diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs
index 04ea4a5e2..24ef1643b 100644
--- a/src/server/conn/handler_sharded/mod.rs
+++ b/src/server/conn/handler_sharded/mod.rs
@@ -852,8 +852,7 @@ pub(crate) async fn handle_connection_sharded_inner<
                             let maybe_key = extract_primary_key(cmd, cmd_args);
                             if let Some(key) = maybe_key {
                                 let slot = crate::cluster::slots::slot_for_key(key);
-                                #[allow(clippy::unwrap_used)] // std RwLock: poison = prior panic = unrecoverable
-                                let route = cs.read().unwrap().route_slot(slot, was_asking);
+                                let route = cs.read().route_slot(slot, was_asking);
                                 match route {
                                     crate::cluster::SlotRoute::Local => {}
                                     other => {
diff --git a/src/server/conn/tests.rs b/src/server/conn/tests.rs
index 37b120e8c..dc471ac51 100644
--- a/src/server/conn/tests.rs
+++ b/src/server/conn/tests.rs
@@ -298,6 +298,7 @@ fn test_inline_mixed_batch() {
         0,
         1,
         false,
+        false,
         &rt_config,
     );
     assert_eq!(total, 1);
@@ -432,6 +433,7 @@ fn test_inline_multiple_gets() {
         0,
         1,
         false,
+        false,
         &rt_config,
     );
     assert_eq!(total, 2);
@@ -439,6 +441,48 @@ fn test_inline_multiple_gets() {
     assert_eq!(&write_buf[..], b"$1\r\n1\r\n$1\r\n2\r\n");
 }
 
+/// R6 (deep review): in cluster mode the inline fast path must inline
+/// NOTHING — a GET/SET served locally here would bypass MOVED/ASK routing
+/// entirely (misplaced writes, stale reads for slots owned elsewhere).
+#[test]
+fn test_inline_loop_disabled_in_cluster_mode() {
+    let dbs = make_dbs();
+    crate::shard::slice::with_shard_db(0, |db| {
+        db.set(
+            Bytes::from_static(b"foo"),
+            Entry::new_string(Bytes::from_static(b"bar")),
+        );
+    });
+    let mut read_buf = BytesMut::from(&b"*2\r\n$3\r\nGET\r\n$3\r\nfoo\r\n"[..]);
+    let mut write_buf = BytesMut::new();
+    let aof_pool: Option<std::sync::Arc<crate::persistence::aof::AofWriterPool>> = None;
+    let rt_config = make_rt_config();
+
+    let total = try_inline_dispatch_loop(
+        &mut read_buf,
+        &mut write_buf,
+        &dbs,
+        0,
+        0,
+        &aof_pool,
+        &None,
+        0,
+        1,
+        true, // even with writes inlinable...
+        true, // ...cluster mode wins: nothing may inline
+        &rt_config,
+    );
+    assert_eq!(
+        total, 0,
+        "cluster mode must fall through to generic dispatch"
+    );
+    assert!(
+        !read_buf.is_empty(),
+        "command must remain for the generic loop"
+    );
+    assert!(write_buf.is_empty());
+}
+
 /// task #59 (coverage-gap fix, review round 3): plain `GET key` for a key
 /// that only lives in the cold tier is served by THIS inline fast path
 /// first (before the general async `dispatch_read` branch in
diff --git a/src/shard/conn_accept.rs b/src/shard/conn_accept.rs
index d6508f1a1..55993233c 100644
--- a/src/shard/conn_accept.rs
+++ b/src/shard/conn_accept.rs
@@ -180,7 +180,7 @@ pub(crate) fn spawn_tokio_connection(
     all_notifiers: &[Arc<channel::Notify>],
     snapshot_trigger_tx: &channel::WatchSender<u64>,
     repl_state: &Option<Arc<parking_lot::RwLock<ReplicationState>>>,
-    cluster_state: &Option<Arc<StdRwLock<crate::cluster::ClusterState>>>,
+    cluster_state: &Option<Arc<parking_lot::RwLock<crate::cluster::ClusterState>>>,
     cached_clock: &CachedClock,
     remote_subscriber_map: &Arc<parking_lot::RwLock<RemoteSubscriberMap>>,
     all_pubsub_registries: &[Arc<parking_lot::RwLock<PubSubRegistry>>],
@@ -441,7 +441,7 @@ pub(crate) fn spawn_migrated_tokio_connection(
     all_notifiers: &[Arc<channel::Notify>],
     snapshot_trigger_tx: &channel::WatchSender<u64>,
     repl_state: &Option<Arc<parking_lot::RwLock<ReplicationState>>>,
-    cluster_state: &Option<Arc<StdRwLock<crate::cluster::ClusterState>>>,
+    cluster_state: &Option<Arc<parking_lot::RwLock<crate::cluster::ClusterState>>>,
     cached_clock: &CachedClock,
     remote_subscriber_map: &Arc<parking_lot::RwLock<RemoteSubscriberMap>>,
     all_pubsub_registries: &[Arc<parking_lot::RwLock<PubSubRegistry>>],
@@ -625,7 +625,7 @@ pub(crate) fn spawn_monoio_connection(
     all_notifiers: &[Arc<channel::Notify>],
     snapshot_trigger_tx: &channel::WatchSender<u64>,
     repl_state: &Option<Arc<parking_lot::RwLock<ReplicationState>>>,
-    cluster_state: &Option<Arc<StdRwLock<crate::cluster::ClusterState>>>,
+    cluster_state: &Option<Arc<parking_lot::RwLock<crate::cluster::ClusterState>>>,
     cached_clock: &CachedClock,
     remote_subscriber_map: &Arc<parking_lot::RwLock<RemoteSubscriberMap>>,
     all_pubsub_registries: &[Arc<parking_lot::RwLock<PubSubRegistry>>],
@@ -1397,7 +1397,7 @@ pub(crate) fn spawn_migrated_monoio_connection(
     all_notifiers: &[Arc<channel::Notify>],
     snapshot_trigger_tx: &channel::WatchSender<u64>,
     repl_state: &Option<Arc<parking_lot::RwLock<ReplicationState>>>,
-    cluster_state: &Option<Arc<StdRwLock<crate::cluster::ClusterState>>>,
+    cluster_state: &Option<Arc<parking_lot::RwLock<crate::cluster::ClusterState>>>,
     cached_clock: &CachedClock,
     remote_subscriber_map: &Arc<parking_lot::RwLock<RemoteSubscriberMap>>,
     all_pubsub_registries: &[Arc<parking_lot::RwLock<PubSubRegistry>>],
diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs
index 7d01128c8..64b00c1f5 100644
--- a/src/shard/event_loop.rs
+++ b/src/shard/event_loop.rs
@@ -85,7 +85,7 @@ impl super::Shard {
         snapshot_trigger_rx: channel::WatchReceiver<u64>,
         snapshot_trigger_tx: channel::WatchSender<u64>,
         repl_state_ext: Option<Arc<parking_lot::RwLock<ReplicationState>>>,
-        cluster_state: Option<std::sync::Arc<std::sync::RwLock<crate::cluster::ClusterState>>>,
+        cluster_state: Option<std::sync::Arc<parking_lot::RwLock<crate::cluster::ClusterState>>>,
         config_port: u16,
         acl_table: Arc<std::sync::RwLock<crate::acl::AclTable>>,
         runtime_config: Arc<parking_lot::RwLock<RuntimeConfig>>,
diff --git a/src/shard/persistence_tick.rs b/src/shard/persistence_tick.rs
index 138aaba95..80f74b611 100644
--- a/src/shard/persistence_tick.rs
+++ b/src/shard/persistence_tick.rs
@@ -629,10 +629,71 @@ fn apply_completion_vec(
     let mut manifest_dirty = false;
     for c in completions {
         if !c.success {
-            tracing::warn!(
-                file_id = c.file_entry.file_id,
-                "Spill pwrite failed on background thread"
-            );
+            // Deep-review F1: the hot entry was removed at evict time on the
+            // promise this completion would land the key in the cold index.
+            // A failed pwrite breaks that promise — without re-insert the key
+            // exists in NEITHER plane and reads nil until an AOF-replay
+            // restart (and permanently under --appendonly no sync paths).
+            // Fail-closed: put the value back in RAM, mirroring the
+            // enqueue-failure path which retains the hot value. Under a
+            // persistently failing disk this re-arms eviction for the same
+            // key — bounded per tick and loudly counted, which beats silent
+            // wrong answers.
+            if let Some(req) = c.failed_request {
+                crate::shard::slice::with_shard_db(req.db_index, |db| {
+                    // Deep-review P2 stale-shadow guard: if the key was
+                    // re-created and re-evicted while this pwrite was in
+                    // flight, a NEWER spill request supersedes this one —
+                    // re-inserting this older payload would shadow the newer
+                    // cold value in the hot plane (and the next eviction
+                    // would spill the stale copy as authoritative).
+                    let newest = db.spill_inflight_is_newest(&req.key, req.file_id);
+                    db.spill_inflight_clear(&req.key, req.file_id);
+                    if !newest {
+                        tracing::warn!(
+                            file_id = c.file_entry.file_id,
+                            key_len = req.key.len(),
+                            "Spill pwrite failed for a SUPERSEDED request; skipping \
+                             re-insert (a newer spill of this key is in flight or landed)"
+                        );
+                        return;
+                    }
+                    if db.get_version(&req.key) != 0 {
+                        // A newer write recreated the key while the spill was
+                        // in flight; the failed payload is stale — drop it.
+                        return;
+                    }
+                    match crate::storage::eviction::rehydrate_spill_payload(
+                        req.value_type,
+                        &req.value_bytes,
+                        req.ttl_ms,
+                    ) {
+                        Some(entry) => {
+                            db.set(req.key.clone(), entry);
+                            crate::storage::tiered::spill_thread::record_spill_failed_reinserted();
+                            tracing::error!(
+                                file_id = c.file_entry.file_id,
+                                key_len = req.key.len(),
+                                "Spill pwrite failed; evicted key re-inserted into hot table \
+                                 (spill volume is failing writes)"
+                            );
+                        }
+                        None => {
+                            tracing::error!(
+                                file_id = c.file_entry.file_id,
+                                key_len = req.key.len(),
+                                "Spill pwrite failed AND payload does not rehydrate — key lost \
+                                 until AOF-replay restart"
+                            );
+                        }
+                    }
+                });
+            } else {
+                tracing::warn!(
+                    file_id = c.file_entry.file_id,
+                    "Spill pwrite failed on background thread (no payload carried back)"
+                );
+            }
             continue;
         }
 
@@ -660,6 +721,10 @@ fn apply_completion_vec(
                 if let Some(ref mut ci) = db.cold_index {
                     ci.insert(entry.key.clone(), location);
                 }
+                // Retire this request's in-flight record (stale-shadow
+                // guard); a newer request's record is left for its own
+                // completion.
+                db.spill_inflight_clear(&entry.key, entry.req_file_id);
             });
         }
     }
diff --git a/src/storage/db/kv_ops.rs b/src/storage/db/kv_ops.rs
index b179599ad..e29f751a9 100644
--- a/src/storage/db/kv_ops.rs
+++ b/src/storage/db/kv_ops.rs
@@ -294,12 +294,13 @@ impl Database {
                 // Hit path: replace existing entry, bump version.
                 let new_entry = entry_cell.take().expect("update closure called once");
                 old_cost = entry_overhead(&key, existing);
-                let new_version = existing.version() + 1;
+                let new_version = Entry::bump_version(existing.version());
                 *existing = new_entry;
                 existing.set_version(new_version);
             },
             || {
-                // Miss path: insert entry as-is (version defaults to 0).
+                // Miss path: insert entry as-is (constructors start versions
+                // at INITIAL_VERSION=1 so WATCH can detect creation).
                 entry_cell.take().expect("make closure called once on miss")
             },
         );
diff --git a/src/storage/db/mod.rs b/src/storage/db/mod.rs
index 2c0295c67..39a4d742d 100644
--- a/src/storage/db/mod.rs
+++ b/src/storage/db/mod.rs
@@ -198,6 +198,18 @@ pub struct Database {
     pub cold_shard_dir: Option<std::path::PathBuf>,
     /// Hot-key detection sketch, fed by sampled dispatch observations.
     hot_keys: crate::storage::hotkey::HotKeySketch,
+    /// Newest in-flight async-spill request id per key (deep-review P2
+    /// stale-shadow guard). Marked at enqueue (`evict_one_async_spill`),
+    /// consumed when that request's completion applies. The failure
+    /// re-insert arm uses it to detect that a NEWER spill superseded the
+    /// failed one (key re-created and re-evicted while the failed pwrite
+    /// was in flight) — re-inserting the older payload would shadow the
+    /// newer cold value. Touched only on evict/completion paths, never on
+    /// command dispatch. A dropped completion (see
+    /// `spill_completion_dropped_total`) can strand an entry until the key
+    /// is next evicted — bounded and harmless (a stale id only ever
+    /// SUPPRESSES a re-insert of an equally stale payload).
+    spill_inflight: std::collections::HashMap<bytes::Bytes, u64>,
 }
 
 impl Database {
@@ -213,6 +225,7 @@ impl Database {
             cold_index: None,
             cold_shard_dir: None,
             hot_keys: crate::storage::hotkey::HotKeySketch::new(),
+            spill_inflight: std::collections::HashMap::new(),
         }
     }
 
@@ -237,6 +250,28 @@ impl Database {
             cold_index: None,
             cold_shard_dir: None,
             hot_keys: crate::storage::hotkey::HotKeySketch::new(),
+            spill_inflight: std::collections::HashMap::new(),
+        }
+    }
+
+    /// Record `req_id` as the newest in-flight async-spill request for
+    /// `key` (called at enqueue time, before the hot entry is removed).
+    pub fn spill_inflight_mark(&mut self, key: bytes::Bytes, req_id: u64) {
+        self.spill_inflight.insert(key, req_id);
+    }
+
+    /// True when `req_id` is still the newest recorded spill request for
+    /// `key` — i.e. no later eviction re-enqueued the key while this
+    /// request was in flight.
+    pub fn spill_inflight_is_newest(&self, key: &[u8], req_id: u64) -> bool {
+        self.spill_inflight.get(key) == Some(&req_id)
+    }
+
+    /// Consume the in-flight record for `key` if (and only if) it belongs
+    /// to `req_id`; a newer request's record is left for its own completion.
+    pub fn spill_inflight_clear(&mut self, key: &[u8], req_id: u64) {
+        if self.spill_inflight.get(key) == Some(&req_id) {
+            self.spill_inflight.remove(key);
         }
     }
 
@@ -751,22 +786,23 @@ mod tests {
     #[test]
     fn test_version_tracking() {
         let mut db = Database::new();
+        // 0 is reserved for "key absent" so WATCH detects creation.
         assert_eq!(db.get_version(b"key"), 0);
         db.set_string(Bytes::from_static(b"key"), Bytes::from_static(b"v1"));
-        assert_eq!(db.get_version(b"key"), 0); // first set, version 0
+        assert_eq!(db.get_version(b"key"), 1); // first set: INITIAL_VERSION
         db.set_string(Bytes::from_static(b"key"), Bytes::from_static(b"v2"));
-        assert_eq!(db.get_version(b"key"), 1); // second set, version 0+1=1
+        assert_eq!(db.get_version(b"key"), 2); // overwrite bumps
         db.set_string(Bytes::from_static(b"key"), Bytes::from_static(b"v3"));
-        assert_eq!(db.get_version(b"key"), 2);
+        assert_eq!(db.get_version(b"key"), 3);
     }
 
     #[test]
     fn test_increment_version() {
         let mut db = Database::new();
         db.set_string(Bytes::from_static(b"key"), Bytes::from_static(b"v"));
-        assert_eq!(db.get_version(b"key"), 0);
-        db.increment_version(b"key");
         assert_eq!(db.get_version(b"key"), 1);
+        db.increment_version(b"key");
+        assert_eq!(db.get_version(b"key"), 2);
         // non-existent key is a no-op
         db.increment_version(b"missing");
     }
@@ -1149,4 +1185,26 @@ mod tests {
         let now_ms = db.now_ms();
         assert!(!db.exists_if_alive(b"nope", now_ms));
     }
+
+    /// Deep-review P2 stale-shadow guard: a failed spill completion for a
+    /// SUPERSEDED request (key re-created and re-evicted while the failed
+    /// pwrite was in flight) must be detectable, and the superseded
+    /// completion's cleanup must not drop the newer request's record.
+    #[test]
+    fn spill_inflight_supersession_guard() {
+        let mut db = Database::new();
+        let k = bytes::Bytes::from_static(b"k");
+        db.spill_inflight_mark(k.clone(), 7);
+        assert!(db.spill_inflight_is_newest(b"k", 7));
+        // Key re-evicted with a newer request: 7 is now superseded — its
+        // failure arm must NOT re-insert its stale payload.
+        db.spill_inflight_mark(k, 9);
+        assert!(!db.spill_inflight_is_newest(b"k", 7));
+        // The superseded completion's clear leaves the newer record intact.
+        db.spill_inflight_clear(b"k", 7);
+        assert!(db.spill_inflight_is_newest(b"k", 9));
+        // The newest completion consumes its own record.
+        db.spill_inflight_clear(b"k", 9);
+        assert!(!db.spill_inflight_is_newest(b"k", 9));
+    }
 }
diff --git a/src/storage/entry.rs b/src/storage/entry.rs
index c7bb6dbc6..dd702974a 100644
--- a/src/storage/entry.rs
+++ b/src/storage/entry.rs
@@ -309,22 +309,33 @@ pub fn lfu_log_incr(counter: u8, lfu_log_factor: u8) -> u8 {
 }
 
 /// Time-based decay for LFU counter.
+///
+/// `last_access` is the full u32 epoch-seconds clock (same domain as
+/// `current_secs()`). `saturating_sub` rather than `wrapping_sub`: around the
+/// year-2106 clock wrap (or transient cross-thread cache skew) a "future"
+/// last_access must yield zero decay, not a near-2^32 elapsed time that would
+/// zero every counter.
 pub fn lfu_decay(counter: u8, last_access: u32, lfu_decay_time: u64) -> u8 {
     if lfu_decay_time == 0 {
         return counter;
     }
     let now = current_secs();
-    let elapsed_secs = now.wrapping_sub(last_access) as u64;
+    let elapsed_secs = now.saturating_sub(last_access) as u64;
     let elapsed_min = elapsed_secs / 60;
-    let decay = (elapsed_min / lfu_decay_time) as u8;
+    let decay = (elapsed_min / lfu_decay_time).min(u8::MAX as u64) as u8;
     counter.saturating_sub(decay)
 }
 
-/// Pack last_access (16 bits), version (8 bits), and access_counter (8 bits) into a u32.
-/// Layout: [last_access:16 | version:8 | access_counter:8]
+/// First version assigned to a newly created entry. Never 0: `get_version()`
+/// returns 0 for a missing key, so WATCH distinguishes "key created between
+/// WATCH and EXEC" from "key still absent" only if live entries never report 0.
+pub const INITIAL_VERSION: u32 = 1;
+
+/// Pack version (24 bits) and access_counter (8 bits) into a u32.
+/// Layout: [version:24 | access_counter:8]
 #[inline]
-fn pack_metadata_u32(last_access: u16, version: u8, access_counter: u8) -> u32 {
-    ((last_access as u32) << 16) | ((version as u32) << 8) | (access_counter as u32)
+fn pack_metadata_u32(version: u32, access_counter: u8) -> u32 {
+    ((version & 0xFF_FFFF) << 8) | (access_counter as u32)
 }
 
 /// A compact 32-byte entry in the database, wrapping a CompactValue with TTL and packed metadata.
@@ -336,8 +347,12 @@ fn pack_metadata_u32(last_access: u16, version: u8, access_counter: u8) -> u32 {
 ///   overflow fix, then W3 used the extra range to store milliseconds instead of seconds —
 ///   PEXPIRE/PEXPIREAT precision previously truncated to whole seconds, expiring keys up to
 ///   999ms EARLY and misreporting PTTL).
-/// - `metadata: u32` (4 bytes, offset 24) -- packed [last_access:16 | version:8 | counter:8]
-/// - _pad: u32 (4 bytes, offset 28) -- alignment padding
+/// - `metadata: u32` (4 bytes, offset 24) -- packed [version:24 | counter:8]
+/// - `last_access_secs: u32` (4 bytes, offset 28) -- full epoch-seconds LRU clock
+///   (formerly alignment padding; repurposed so LRU/LFU/IDLETIME are exact
+///   beyond the old 16-bit field's 18.2h wrap, at zero size cost). The freed
+///   16 metadata bits widened `version` 8→24 bits, pushing the WATCH/EXEC ABA
+///   window from 256 to 16.7M intervening writes.
 ///
 /// NOTE: the size changed from 24 → 32 bytes when the TTL field was widened from u32 to u64.
 /// Memory overhead per key increases by 8 bytes (1/3 overhead on a 100M-key dataset = ~800 MB).
@@ -349,9 +364,10 @@ pub struct CompactEntry {
     pub value: CompactValue,
     /// Absolute expiry time in Unix milliseconds. 0 = no expiry.
     pub ttl_ms: u64,
-    /// Packed metadata: [last_access:16 | version:8 | access_counter:8]
+    /// Packed metadata: [version:24 | access_counter:8]
     pub metadata: u32,
-    _pad: u32,
+    /// Last access time, full epoch seconds (`current_secs()` domain).
+    last_access_secs: u32,
 }
 
 const _: () = assert!(std::mem::size_of::<CompactEntry>() == 32);
@@ -362,16 +378,16 @@ pub type Entry = CompactEntry;
 impl CompactEntry {
     // --- Accessor methods for packed metadata ---
 
-    /// Get the version (8-bit, wraps at 0xFF).
+    /// Get the version (24-bit, wraps to [`INITIAL_VERSION`], never 0).
     #[inline]
     pub fn version(&self) -> u32 {
-        ((self.metadata >> 8) & 0xFF) as u32
+        (self.metadata >> 8) & 0xFF_FFFF
     }
 
-    /// Get the last access time (16-bit relative seconds).
+    /// Get the last access time (full u32 epoch seconds).
     #[inline]
     pub fn last_access(&self) -> u32 {
-        (self.metadata >> 16) as u32
+        self.last_access_secs
     }
 
     /// Get the LFU access counter (8-bit Morris counter).
@@ -380,18 +396,16 @@ impl CompactEntry {
         (self.metadata & 0xFF) as u8
     }
 
-    /// Set the version (8-bit, truncated to lower 8 bits).
+    /// Set the version (24-bit, truncated to lower 24 bits).
     #[inline]
     pub fn set_version(&mut self, v: u32) {
-        let v8 = (v & 0xFF) as u8;
-        self.metadata = (self.metadata & !(0xFF << 8)) | ((v8 as u32) << 8);
+        self.metadata = (self.metadata & 0xFF) | ((v & 0xFF_FFFF) << 8);
     }
 
-    /// Set the last access time (truncated to 16 bits).
+    /// Set the last access time (full u32 epoch seconds).
     #[inline]
     pub fn set_last_access(&mut self, t: u32) {
-        let t16 = (t & 0xFFFF) as u16;
-        self.metadata = (self.metadata & 0xFFFF) | ((t16 as u32) << 16);
+        self.last_access_secs = t;
     }
 
     /// Set the LFU access counter.
@@ -400,11 +414,18 @@ impl CompactEntry {
         self.metadata = (self.metadata & !0xFF) | (c as u32);
     }
 
-    /// Increment version, wrapping at 0xFF.
+    /// Next version after `v`: 24-bit wrap that skips 0 (the WATCH
+    /// "key absent" sentinel — see [`INITIAL_VERSION`]).
+    #[inline]
+    pub fn bump_version(v: u32) -> u32 {
+        let n = (v + 1) & 0xFF_FFFF;
+        if n == 0 { INITIAL_VERSION } else { n }
+    }
+
+    /// Increment version (24-bit wrap, skips 0).
     #[inline]
     pub fn increment_version(&mut self) {
-        let v = ((self.version() + 1) & 0xFF) as u32;
-        self.set_version(v);
+        self.set_version(Self::bump_version(self.version()));
     }
 
     // --- Expiry helpers ---
@@ -465,8 +486,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::String(value)),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -475,8 +496,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::String(value)),
             ttl_ms: expires_at_ms,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -485,8 +506,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::Hash(HashMap::new())),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -495,8 +516,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::List(VecDeque::new())),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -505,8 +526,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::Set(HashSet::new())),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -518,8 +539,8 @@ impl CompactEntry {
                 scores: BTreeMap::new(),
             }),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -528,8 +549,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::HashListpack(Listpack::new())),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -538,8 +559,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::ListListpack(Listpack::new())),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -548,8 +569,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::SetListpack(Listpack::new())),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -558,8 +579,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::SetIntset(Intset::new())),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -571,8 +592,8 @@ impl CompactEntry {
                 members: HashMap::new(),
             }),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -581,8 +602,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::SortedSetListpack(Listpack::new())),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -591,8 +612,8 @@ impl CompactEntry {
         CompactEntry {
             value: CompactValue::from_redis_value(RedisValue::Stream(Box::new(StreamData::new()))),
             ttl_ms: 0,
-            metadata: pack_metadata_u32(current_secs() as u16, 0, LFU_INIT_VAL),
-            _pad: 0,
+            metadata: pack_metadata_u32(INITIAL_VERSION, LFU_INIT_VAL),
+            last_access_secs: current_secs(),
         }
     }
 
@@ -619,7 +640,7 @@ mod tests {
         assert!(!entry.has_expiry());
         assert_eq!(entry.ttl_ms, 0);
         assert_eq!(entry.value.type_name(), "string");
-        assert_eq!(entry.version(), 0);
+        assert_eq!(entry.version(), INITIAL_VERSION);
         assert_eq!(entry.access_counter(), LFU_INIT_VAL);
     }
 
@@ -633,7 +654,7 @@ mod tests {
         let recovered_ms = entry.expires_at_ms();
         // Allow 1 second tolerance due to integer division
         assert!((recovered_ms as i64 - exp_ms as i64).unsigned_abs() < 1000);
-        assert_eq!(entry.version(), 0);
+        assert_eq!(entry.version(), INITIAL_VERSION);
         assert_eq!(entry.access_counter(), LFU_INIT_VAL);
     }
 
@@ -665,7 +686,7 @@ mod tests {
         let entry = Entry::new_hash();
         assert!(!entry.has_expiry());
         assert_eq!(entry.value.type_name(), "hash");
-        assert_eq!(entry.version(), 0);
+        assert_eq!(entry.version(), INITIAL_VERSION);
     }
 
     #[test]
@@ -673,7 +694,7 @@ mod tests {
         let entry = Entry::new_list();
         assert!(!entry.has_expiry());
         assert_eq!(entry.value.type_name(), "list");
-        assert_eq!(entry.version(), 0);
+        assert_eq!(entry.version(), INITIAL_VERSION);
     }
 
     #[test]
@@ -681,7 +702,7 @@ mod tests {
         let entry = Entry::new_set();
         assert!(!entry.has_expiry());
         assert_eq!(entry.value.type_name(), "set");
-        assert_eq!(entry.version(), 0);
+        assert_eq!(entry.version(), INITIAL_VERSION);
     }
 
     #[test]
@@ -689,7 +710,7 @@ mod tests {
         let entry = Entry::new_sorted_set();
         assert!(!entry.has_expiry());
         assert_eq!(entry.value.type_name(), "zset");
-        assert_eq!(entry.version(), 0);
+        assert_eq!(entry.version(), INITIAL_VERSION);
     }
 
     #[test]
@@ -760,30 +781,63 @@ mod tests {
     #[test]
     fn test_metadata_packing_roundtrip() {
         let mut entry = Entry::new_string(Bytes::from_static(b"test"));
-        // Test version (8-bit: max 255)
-        entry.set_version(123);
-        assert_eq!(entry.version(), 123);
-        // Test last_access (16-bit: max 65535)
-        entry.set_last_access(54321);
-        // last_access truncates to u16
-        assert_eq!(entry.last_access(), 54321);
-        // version should be preserved
-        assert_eq!(entry.version(), 123);
+        // Version is 24-bit: values beyond the old 8-bit range must survive.
+        entry.set_version(123_456);
+        assert_eq!(entry.version(), 123_456);
+        // last_access is full u32 seconds — no 16-bit truncation.
+        entry.set_last_access(1_754_000_000);
+        assert_eq!(entry.last_access(), 1_754_000_000);
+        // version preserved across last_access writes
+        assert_eq!(entry.version(), 123_456);
         // Test access_counter
         entry.set_access_counter(42);
         assert_eq!(entry.access_counter(), 42);
         // Other fields preserved
-        assert_eq!(entry.version(), 123);
-        assert_eq!(entry.last_access(), 54321);
+        assert_eq!(entry.version(), 123_456);
+        assert_eq!(entry.last_access(), 1_754_000_000);
     }
 
     #[test]
-    fn test_increment_version_wraps_at_8bit() {
+    fn test_increment_version_wraps_24bit_skipping_zero() {
         let mut entry = Entry::new_string(Bytes::from_static(b"test"));
-        entry.set_version(0xFF);
-        assert_eq!(entry.version(), 0xFF);
+        entry.set_version(0xFF_FFFF);
+        assert_eq!(entry.version(), 0xFF_FFFF);
+        // Wrap must skip 0: version 0 is the WATCH "key absent" sentinel, so a
+        // live entry may never report it.
         entry.increment_version();
-        assert_eq!(entry.version(), 0); // wraps
+        assert_eq!(entry.version(), 1);
+    }
+
+    #[test]
+    fn test_new_entries_start_at_version_one() {
+        // WATCH creation-detection: get_version() returns 0 for a missing key,
+        // so a freshly created entry must NOT also report 0.
+        let entry = Entry::new_string(Bytes::from_static(b"test"));
+        assert_ne!(entry.version(), 0);
+    }
+
+    #[test]
+    fn test_lfu_decay_full_domain_via_entry() {
+        // Regression: lfu_decay used to receive the 16-bit-truncated
+        // last_access and subtract it from the full u32 clock, making the
+        // decay value effectively random. With the full-width field, a key
+        // idle 2h decays by exactly 120 (minutes) at lfu_decay_time=1.
+        let now = current_secs();
+        let mut entry = Entry::new_string(Bytes::from_static(b"test"));
+        entry.set_last_access(now - 7200);
+        assert_eq!(lfu_decay(200, entry.last_access(), 1), 200 - 120);
+        // A key idle long enough saturates to 0 rather than wrapping.
+        entry.set_last_access(now - 40_000);
+        assert_eq!(lfu_decay(200, entry.last_access(), 1), 0);
+    }
+
+    #[test]
+    fn test_last_access_survives_long_idle() {
+        // Regression: an 18.2h+ idle time must not wrap (OBJECT IDLETIME).
+        let now = current_secs();
+        let mut entry = Entry::new_string(Bytes::from_static(b"test"));
+        entry.set_last_access(now - 100_000); // ~27.8h
+        assert_eq!(now - entry.last_access(), 100_000);
     }
 
     #[test]
diff --git a/src/storage/eviction.rs b/src/storage/eviction.rs
index 84ff4dadd..babe3e61f 100644
--- a/src/storage/eviction.rs
+++ b/src/storage/eviction.rs
@@ -12,7 +12,7 @@ use crate::protocol::Frame;
 use crate::storage::Database;
 use crate::storage::compact_key::CompactKey;
 use crate::storage::compact_value::RedisValueRef;
-use crate::storage::entry::lfu_decay;
+use crate::storage::entry::{RedisValue, lfu_decay};
 use crate::storage::tiered::kv_serde;
 use crate::storage::tiered::kv_spill;
 use crate::storage::tiered::spill_thread::SpillRequest;
@@ -182,18 +182,18 @@ fn sample_random_keys(
     out
 }
 
-/// Compare two LRU timestamps with u16 wraparound handling.
-/// Uses signed-distance comparison: treats the 16-bit clock as circular.
+/// Compare two LRU timestamps with u32 wraparound handling.
+/// Uses signed-distance comparison: treats the 32-bit epoch-seconds clock as
+/// circular, so verdicts are correct for access-time gaps up to ±68 years —
+/// including across the year-2106 clock wrap. (The old 16-bit variant
+/// inverted verdicts once two keys' access times differed by more than
+/// ~9.1h, making LRU evict hot keys under diurnal workloads.)
 /// Returns true if `a` is considered older (less recent) than `b`.
 #[inline]
 pub fn lru_is_older(a: u32, b: u32) -> bool {
-    let a16 = a as i16;
-    let b16 = b as i16;
     // Signed difference handles wraparound: if a was accessed before b,
-    // (a16 - b16) is negative (a < b in time), meaning a is older.
-    // Wraparound case: a=65400 (pre-wrap), b=100 (post-wrap):
-    //   a16=-136, b16=100, -136-100=-236 < 0 → correctly identifies a as older.
-    a16.wrapping_sub(b16) < 0
+    // the circular distance a-b is negative (a < b in time) → a is older.
+    (a.wrapping_sub(b) as i32) < 0
 }
 
 /// Sum used_memory across all databases for aggregate eviction decisions.
@@ -695,6 +695,37 @@ fn build_spill_payload(
     Some((value_type, value_bytes, flags, ttl_ms))
 }
 
+/// Inverse of [`build_spill_payload`]: rebuild a hot [`Entry`] from a spill
+/// payload that never reached disk (deep-review F1).
+///
+/// `evict_one_async_spill` removes the hot entry as soon as the request is
+/// queued, on the promise that the background thread's completion lands the
+/// key in the cold index. When the pwrite FAILS, that promise is broken: the
+/// key is in neither plane and every read returns nil until an AOF-replay
+/// restart. The completion handler re-inserts via this helper instead
+/// (fail-closed, mirroring the enqueue-failure path which retains the hot
+/// value).
+///
+/// Returns `None` only if the payload does not deserialize — impossible for
+/// bytes produced by `build_spill_payload` unless memory was corrupted in
+/// transit; the caller logs that loudly.
+pub(crate) fn rehydrate_spill_payload(
+    value_type: ValueType,
+    value_bytes: &Bytes,
+    ttl_ms: Option<u64>,
+) -> Option<crate::storage::Entry> {
+    let redis_value = match value_type {
+        ValueType::String => RedisValue::String(value_bytes.clone()),
+        _ => kv_serde::deserialize_collection(value_bytes, value_type)?,
+    };
+    let mut entry = crate::storage::Entry::new_string(Bytes::new());
+    entry.value = crate::storage::compact_value::CompactValue::from_redis_value(redis_value);
+    if let Some(ttl) = ttl_ms {
+        entry.set_expires_at_ms(ttl);
+    }
+    Some(entry)
+}
+
 /// Batch-durable synchronous spill (W2: the ONE sync spill implementation).
 ///
 /// Originally built for the no-AOF-backstop case (`--appendonly no`); since
@@ -894,6 +925,11 @@ fn evict_one_async_spill(
         if sender.try_send(req).is_err() {
             return false;
         }
+        // Deep-review P2 stale-shadow guard: record this request as the
+        // newest in-flight spill for the key BEFORE freeing RAM, so a later
+        // failed-pwrite re-insert of an OLDER superseded request for the
+        // same key can be detected and suppressed.
+        db.spill_inflight_mark(Bytes::copy_from_slice(key.as_bytes()), file_id);
 
         // Now safe to free RAM. The bg thread holds the SpillRequest and will
         // produce a SpillCompletion that updates cold_index for this db_index.
@@ -1102,6 +1138,53 @@ mod tests {
     use crate::persistence::manifest::ShardManifest;
     use crate::storage::entry::{Entry, current_secs, current_time_ms};
 
+    #[test]
+    fn spill_payload_round_trips_through_rehydrate() {
+        // F1 (deep review): a failed background spill must be able to
+        // resurrect the exact hot entry from its request payload.
+        use std::collections::HashMap;
+        let mut map = HashMap::new();
+        map.insert(Bytes::from_static(b"f1"), Bytes::from_static(b"v1"));
+        map.insert(Bytes::from_static(b"f2"), Bytes::from_static(b"v2"));
+        let mut entry = Entry::new_string(Bytes::new());
+        entry.value = crate::storage::compact_value::CompactValue::from_redis_value(
+            crate::storage::entry::RedisValue::Hash(map.clone()),
+        );
+        entry.set_expires_at_ms(current_time_ms() + 60_000);
+
+        let (vt, bytes, _flags, ttl) = build_spill_payload(&entry).expect("serializable");
+        let restored = rehydrate_spill_payload(vt, &bytes, ttl).expect("round-trip");
+        match restored.as_redis_value() {
+            RedisValueRef::Hash(h) => assert_eq!(*h, map),
+            _ => panic!("expected hash after rehydrate"),
+        }
+        assert_eq!(restored.expires_at_ms(), entry.expires_at_ms());
+
+        // String payloads round-trip too.
+        let s = Entry::new_string(Bytes::from_static(b"hello"));
+        let (vt, bytes, _f, ttl) = build_spill_payload(&s).expect("serializable");
+        let restored = rehydrate_spill_payload(vt, &bytes, ttl).expect("round-trip");
+        match restored.as_redis_value() {
+            RedisValueRef::String(v) => assert_eq!(v, b"hello"),
+            _ => panic!("expected string after rehydrate"),
+        }
+    }
+
+    #[test]
+    fn lru_is_older_valid_beyond_nine_hours() {
+        // Regression: the old i16 comparison window (±32768s ≈ ±9.1h)
+        // inverted verdicts for diurnal idle gaps — a key idle 10h compared
+        // as NEWER than one touched a minute ago, so LRU evicted the hot key.
+        let now: u32 = 1_754_000_000;
+        let idle_10h = now - 36_000;
+        let idle_1m = now - 60;
+        assert!(lru_is_older(idle_10h, idle_1m));
+        assert!(!lru_is_older(idle_1m, idle_10h));
+        // Still correct at multi-day idle.
+        let idle_3d = now - 259_200;
+        assert!(lru_is_older(idle_3d, idle_10h));
+    }
+
     // -----------------------------------------------------------------
     // compute_elastic_budget (GAP-1)
     // -----------------------------------------------------------------
diff --git a/src/storage/tiered/spill_thread.rs b/src/storage/tiered/spill_thread.rs
index c47e3f261..49564e6ff 100644
--- a/src/storage/tiered/spill_thread.rs
+++ b/src/storage/tiered/spill_thread.rs
@@ -151,6 +151,26 @@ pub fn spill_batches_flushed_total() -> u64 {
     SPILL_BATCHES_FLUSHED.load(Ordering::Relaxed)
 }
 
+/// Cumulative keys re-inserted into the hot table after a FAILED spill write
+/// (deep-review F1). Every increment means a pwrite failed (ENOSPC/EIO) after
+/// the key was already evicted from RAM; without the re-insert the key would
+/// read as nil until an AOF-replay restart. Exposed as
+/// `spill_failed_reinserted` in INFO — a nonzero value is an operator signal
+/// that the spill volume is failing writes.
+static SPILL_FAILED_REINSERTED: AtomicU64 = AtomicU64::new(0);
+
+/// Cumulative failed-spill hot re-inserts. Exposed for INFO / metrics.
+#[inline]
+pub fn spill_failed_reinserted_total() -> u64 {
+    SPILL_FAILED_REINSERTED.load(Ordering::Relaxed)
+}
+
+/// Record one failed-spill hot re-insert.
+#[inline]
+pub fn record_spill_failed_reinserted() {
+    SPILL_FAILED_REINSERTED.fetch_add(1, Ordering::Relaxed);
+}
+
 use bytes::Bytes;
 use tracing::warn;
 
@@ -166,6 +186,10 @@ use crate::storage::tiered::kv_spill::{
 ///
 /// Contains all data needed for pwrite -- no references to shard state.
 /// `Bytes` fields are reference-counted (cheap clone on event loop side).
+/// `Clone` exists so a FAILED write can carry the request back to the event
+/// loop for hot re-insert (deep-review F1) — cheap: both payload fields are
+/// refcounted `Bytes`.
+#[derive(Clone)]
 pub struct SpillRequest {
     pub key: Bytes,
     /// Logical database index the key was evicted from. Used by completion
@@ -208,6 +232,11 @@ pub struct SpillCompletionEntry {
     /// event loop can populate `ColdLocation::value_type` (#364: SCAN TYPE
     /// filter over cold keys) without re-reading the just-written file.
     pub value_type: ValueType,
+    /// The originating request's `file_id` (each request gets a unique
+    /// monotonic id even when batching flushes many requests into one
+    /// file). Lets the completion handler retire the per-key in-flight
+    /// spill record for exactly this request (stale-shadow guard).
+    pub req_file_id: u64,
 }
 
 /// Completion sent from background thread back to event loop.
@@ -221,6 +250,11 @@ pub struct SpillCompletion {
     pub entries: Vec<SpillCompletionEntry>,
     /// Whether the pwrite succeeded. If false, no entries should be indexed.
     pub success: bool,
+    /// On failure: the original request, so the event loop can re-insert the
+    /// already-evicted key into the hot table (deep-review F1 — without this
+    /// the key exists in neither plane and reads nil until restart).
+    /// Always `None` on success.
+    pub failed_request: Option<Box<SpillRequest>>,
 }
 
 /// Build a `FileEntry` skeleton for a spill file (fields not tracked by Moon are zero).
@@ -339,12 +373,14 @@ pub(crate) fn flush_buffer(buffer: &mut Vec<SpillRequest>) -> Vec<SpillCompletio
                                 slot_idx,
                                 ttl_ms: req.ttl_ms,
                                 value_type: req.value_type,
+                                req_file_id: req.file_id,
                             })
                             .collect();
                         completions.push(SpillCompletion {
                             file_entry: make_file_entry(file_id, total_pages, byte_size, chunk_db),
                             entries,
                             success: true,
+                            failed_request: None,
                         });
                     }
                     Err(e) => {
@@ -408,8 +444,10 @@ fn spill_single_entry(req: &SpillRequest, file_id: u64) -> SpillCompletion {
                     slot_idx: 0,
                     ttl_ms: req.ttl_ms,
                     value_type: req.value_type,
+                    req_file_id: req.file_id,
                 }],
                 success: true,
+                failed_request: None,
             },
             Err(e) => {
                 warn!(
@@ -422,6 +460,7 @@ fn spill_single_entry(req: &SpillRequest, file_id: u64) -> SpillCompletion {
                     file_entry: make_file_entry(file_id, 0, 0, req.db_index),
                     entries: Vec::new(),
                     success: false,
+                    failed_request: Some(Box::new(req.clone())),
                 }
             }
         },
@@ -436,6 +475,7 @@ fn spill_single_entry(req: &SpillRequest, file_id: u64) -> SpillCompletion {
                 file_entry: make_file_entry(file_id, 0, 0, req.db_index),
                 entries: Vec::new(),
                 success: false,
+                failed_request: Some(Box::new(req.clone())),
             }
         }
     }
@@ -708,6 +748,36 @@ mod tests {
         completions
     }
 
+    /// F1 (deep review): a failed spill write must carry the original request
+    /// back so the event loop can re-insert the already-evicted key.
+    #[test]
+    fn failed_spill_write_carries_request_back() {
+        let tmp = tempfile::tempdir().unwrap();
+        // Make shard_dir an existing FILE so the spill write cannot succeed.
+        let bogus_dir = tmp.path().join("not-a-dir");
+        std::fs::write(&bogus_dir, b"occupied").unwrap();
+
+        let req = SpillRequest {
+            key: Bytes::from_static(b"lost-key"),
+            db_index: 3,
+            value_bytes: Bytes::from_static(b"payload"),
+            value_type: ValueType::String,
+            flags: 0,
+            ttl_ms: Some(12345),
+            file_id: 42,
+            shard_dir: bogus_dir,
+        };
+        let completion = spill_single_entry(&req, req.file_id);
+        assert!(!completion.success);
+        let carried = completion
+            .failed_request
+            .expect("failure completion must carry the request payload back");
+        assert_eq!(carried.key, req.key);
+        assert_eq!(carried.db_index, 3);
+        assert_eq!(carried.value_bytes, req.value_bytes);
+        assert_eq!(carried.ttl_ms, Some(12345));
+    }
+
     #[test]
     fn pace_is_zero_when_no_readers_waiting() {
         assert_eq!(
@@ -888,6 +958,7 @@ mod tests {
             file_entry: make_file_entry(7, 1, PAGE_4K as u64, 0),
             entries: Vec::new(),
             success: true,
+            failed_request: None,
         };
 
         // Saturate the single slot so the next send must wait for a free slot.
diff --git a/src/temporal/mod.rs b/src/temporal/mod.rs
index 3a6e09d40..95e90a4f8 100644
--- a/src/temporal/mod.rs
+++ b/src/temporal/mod.rs
@@ -27,6 +27,14 @@ pub struct TemporalRegistry {
     entries: BTreeMap<i64, u64>,
 }
 
+/// Upper bound on retained wall-clock->LSN bindings per shard (deep-review
+/// G4: the registry was insert-only, growing one entry per
+/// `TEMPORAL.SNAPSHOT_AT` call for the life of the process). At 16 bytes a
+/// binding this caps the registry at ~4 MB/shard; when full, the OLDEST
+/// binding is evicted — bounding how far back `AS_OF` resolves rather than
+/// growing without limit.
+const MAX_TEMPORAL_BINDINGS: usize = 262_144;
+
 impl TemporalRegistry {
     /// Create an empty registry.
     pub fn new() -> Self {
@@ -37,8 +45,13 @@ impl TemporalRegistry {
     ///
     /// Both `wall_ms` and `lsn` MUST be captured by the caller at the
     /// handler level before calling this method.
+    ///
+    /// Bounded: at [`MAX_TEMPORAL_BINDINGS`] the oldest binding is evicted.
     pub fn record(&mut self, wall_ms: i64, lsn: u64) {
         self.entries.insert(wall_ms, lsn);
+        while self.entries.len() > MAX_TEMPORAL_BINDINGS {
+            self.entries.pop_first();
+        }
     }
 
     /// Find the LSN that was active at wall-clock time T.
@@ -129,6 +142,24 @@ mod tests {
         assert!(reg.lsn_at(1000).is_none());
     }
 
+    /// G4 (deep review): the registry must stay bounded — one binding per
+    /// TEMPORAL.SNAPSHOT_AT for the life of the process was unbounded
+    /// growth. Overflow evicts the OLDEST binding; recent bindings resolve.
+    #[test]
+    fn test_registry_bounded_evicts_oldest() {
+        let mut reg = TemporalRegistry::new();
+        let overflow = 10;
+        for i in 0..(MAX_TEMPORAL_BINDINGS + overflow) {
+            reg.record(i as i64, i as u64);
+        }
+        assert_eq!(reg.len(), MAX_TEMPORAL_BINDINGS, "cap must hold");
+        // The oldest `overflow` bindings are gone...
+        assert!(reg.lsn_at(overflow as i64 - 1).is_none());
+        // ...and the newest still resolves exactly.
+        let last = (MAX_TEMPORAL_BINDINGS + overflow - 1) as i64;
+        assert_eq!(reg.lsn_at(last), Some(last as u64));
+    }
+
     #[test]
     fn test_registry_exact_match() {
         let mut reg = TemporalRegistry::new();
diff --git a/src/tracking/invalidation.rs b/src/tracking/invalidation.rs
index e6c7b5bfb..570327d00 100644
--- a/src/tracking/invalidation.rs
+++ b/src/tracking/invalidation.rs
@@ -117,7 +117,15 @@ pub fn track_read_keys(
     }
     let mut table = table.lock();
     for key in &keys {
-        table.track_key(client_id, key, noloop);
+        if let Some((evicted_key, senders)) = table.track_key(client_id, key, noloop) {
+            // Cap eviction (G1): tell the evicted key's trackers to drop
+            // their cached copy — silently forgetting the tracking entry
+            // would leave client-side caches permanently stale.
+            let push = invalidation_push(std::slice::from_ref(&evicted_key));
+            for tx in senders {
+                let _ = tx.try_send(push.clone());
+            }
+        }
     }
 }
 
diff --git a/src/tracking/mod.rs b/src/tracking/mod.rs
index 678561855..989fbe032 100644
--- a/src/tracking/mod.rs
+++ b/src/tracking/mod.rs
@@ -87,18 +87,22 @@ pub struct TrackingTable {
     /// Redirect map: source_client_id -> target_client_id
     redirects: HashMap<u64, u64>,
     /// Maximum keys tracked (bounded table)
-    #[allow(dead_code)]
     max_keys: usize,
 }
 
 impl TrackingTable {
     pub fn new() -> Self {
+        Self::with_max_keys(1_000_000)
+    }
+
+    /// Construct with an explicit key cap (tests; production uses `new`).
+    pub fn with_max_keys(max_keys: usize) -> Self {
         Self {
             key_clients: HashMap::new(),
             bcast_clients: Vec::new(),
             client_channels: HashMap::new(),
             redirects: HashMap::new(),
-            max_keys: 1_000_000,
+            max_keys,
         }
     }
 
@@ -120,11 +124,51 @@ impl TrackingTable {
     }
 
     /// Track that a client has read a key (normal mode).
-    pub fn track_key(&mut self, client_id: u64, key: &Bytes, noloop: bool) {
-        let clients = self.key_clients.entry(key.clone()).or_insert_with(Vec::new);
-        if !clients.iter().any(|(id, _)| *id == client_id) {
-            clients.push((client_id, noloop));
+    ///
+    /// Enforces the `max_keys` bound (deep-review G1: the documented cap was
+    /// dead code, so a long-lived tracking client reading many distinct
+    /// never-written keys grew this table without limit). When tracking a NEW
+    /// key would exceed the cap, an arbitrary existing entry is evicted and
+    /// its `(key, senders)` returned — the caller must push an invalidation
+    /// for it so the evicted key's clients drop their cached copy (Redis's
+    /// "fake invalidation" on tracking-table eviction). Returns `None` when
+    /// no eviction occurred.
+    pub fn track_key(
+        &mut self,
+        client_id: u64,
+        key: &Bytes,
+        noloop: bool,
+    ) -> Option<(Bytes, Vec<channel::MpscSender<Frame>>)> {
+        if let Some(clients) = self.key_clients.get_mut(key) {
+            if !clients.iter().any(|(id, _)| *id == client_id) {
+                clients.push((client_id, noloop));
+            }
+            return None;
         }
+
+        let evicted = if self.key_clients.len() >= self.max_keys.max(1) {
+            // Evict an arbitrary entry (HashMap has no age order; correctness
+            // needs only that the evicted key's trackers are told to drop it).
+            #[allow(clippy::unwrap_used)] // len >= 1 guaranteed by the branch
+            let victim = self.key_clients.keys().next().unwrap().clone();
+            let clients = self.key_clients.remove(&victim).unwrap_or_default();
+            let mut senders = Vec::new();
+            for (cid, _noloop) in clients {
+                // No noloop skip: cap eviction is not a self-write — every
+                // tracker of the victim key must drop its cached copy.
+                let target_id = self.redirects.get(&cid).copied().unwrap_or(cid);
+                if let Some(tx) = self.client_channels.get(&target_id) {
+                    senders.push(tx.clone());
+                }
+            }
+            Some((victim, senders))
+        } else {
+            None
+        };
+
+        self.key_clients
+            .insert(key.clone(), vec![(client_id, noloop)]);
+        evicted
     }
 
     /// Get the list of client IDs tracking a given key (for testing).
@@ -242,6 +286,45 @@ mod tests {
         assert_eq!(clients, vec![1, 2]);
     }
 
+    /// G1 (deep review): the documented max_keys bound was dead code — a
+    /// long-lived tracking client reading many distinct never-written keys
+    /// grew key_clients without limit. The cap must evict an existing entry
+    /// (with invalidation senders so the evicted key's clients drop their
+    /// cached copy) instead of growing.
+    #[test]
+    fn test_track_key_enforces_max_keys_bound() {
+        let mut table = TrackingTable::with_max_keys(2);
+        let (tx, rx) = channel::mpsc_bounded::<Frame>(16);
+        table.register_client(1, tx);
+        assert!(
+            table
+                .track_key(1, &Bytes::from_static(b"k1"), false)
+                .is_none()
+        );
+        assert!(
+            table
+                .track_key(1, &Bytes::from_static(b"k2"), false)
+                .is_none()
+        );
+        // Re-tracking an existing key never evicts.
+        assert!(
+            table
+                .track_key(1, &Bytes::from_static(b"k2"), false)
+                .is_none()
+        );
+
+        // Third distinct key: one existing entry must be evicted, with the
+        // evicted key's client senders returned for invalidation.
+        let evicted = table
+            .track_key(1, &Bytes::from_static(b"k3"), false)
+            .expect("cap reached: eviction expected");
+        assert!(evicted.0 == Bytes::from_static(b"k1") || evicted.0 == Bytes::from_static(b"k2"));
+        assert_eq!(evicted.1.len(), 1, "evicted key's tracker must be notified");
+        assert_eq!(table.key_clients.len(), 2, "table must stay at the cap");
+        assert_eq!(table.tracked_clients(&Bytes::from_static(b"k3")), vec![1]);
+        drop(rx);
+    }
+
     #[test]
     fn test_invalidate_key_returns_senders_and_removes() {
         let mut table = TrackingTable::new();
diff --git a/src/vector/store.rs b/src/vector/store.rs
index af60db413..62005c07f 100644
--- a/src/vector/store.rs
+++ b/src/vector/store.rs
@@ -923,7 +923,17 @@ impl VectorIndex {
                 });
                 true
             }
-            Err(_) => false, // worker queue full — retry next tick
+            Err(e) => {
+                // F4: leave a correlatable trail for a submit loop that never
+                // drains (WorkersBusy also masks a dead pool). Debug level —
+                // benign momentary saturation would spam warn per tick.
+                tracing::debug!(
+                    error = ?e,
+                    mutable_len,
+                    "vector background-compact submit deferred; retry next tick"
+                );
+                false
+            }
         }
     }
 
@@ -991,6 +1001,15 @@ impl VectorIndex {
             Err(flume::TryRecvError::Empty) => return false,
             Err(flume::TryRecvError::Disconnected) => {
                 // Worker panicked or dropped — clear inflight and give up.
+                // F4 (deep review): this used to be silent, so a dead worker
+                // pool meant a permanent compact stall (mutable segment
+                // growing past COMPACT_THRESHOLD, searches degrading to
+                // brute force) with nothing for the operator to correlate.
+                tracing::error!(
+                    "vector background-compact worker died (panic or pool teardown); \
+                     compaction will be resubmitted but may stall permanently — \
+                     mutable segment growth and search latency will degrade"
+                );
                 self.bg_compact_inflight = None;
                 return false;
             }
@@ -1001,7 +1020,17 @@ impl VectorIndex {
 
         let mut immutable = match result {
             Ok(imm) => imm,
-            Err(_) => return false, // compaction failed — drop, retry later
+            Err(e) => {
+                // F4: compaction failure (e.g. the B2 disk persist of the
+                // built segment on a full disk) retries next tick — say so
+                // instead of looping silently forever.
+                tracing::warn!(
+                    error = %e,
+                    frozen_len = inflight.frozen_len,
+                    "vector background compaction failed; will retry next tick"
+                );
+                return false;
+            }
         };
 
         // ── Reconciliation ────────────────────────────────────────────────────
diff --git a/tests/integration.rs b/tests/integration.rs
index b3812d59f..9a145b128 100644
--- a/tests/integration.rs
+++ b/tests/integration.rs
@@ -4003,7 +4003,7 @@ async fn start_cluster_server() -> (u16, CancellationToken) {
         let self_addr: std::net::SocketAddr = format!("127.0.0.1:{}", config.port).parse().unwrap();
         let node_id = moon::replication::state::generate_repl_id();
         let state = moon::cluster::ClusterState::new(node_id, self_addr);
-        let cluster_state = Some(std::sync::Arc::new(std::sync::RwLock::new(state)));
+        let cluster_state = Some(std::sync::Arc::new(parking_lot::RwLock::new(state)));
 
         let all_notifiers = mesh.all_notifiers();
         let all_pubsub_registries: Vec<