fix: durability wave 1 — rewrite-window append overflow, degraded latch, WAL mid-chain tear policy (#452, #54)

CHANGELOG.md

@@ -56,6 +56,45 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - *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).
+- **Durability wave 1 (#452, #54): AOF rewrite-window append drops, degraded-state
+  visibility, escalated reason-DEL backpressure, and a fail-loud WAL mid-chain
+  tear policy.** (1) While a rewrite fold ran, the writer thread was out of its
+  recv loop, so under sustained pipelined writes the bounded (10k) append
+  channel saturated and acked records were dropped — lost even on a clean
+  restart, and default-on since #433 made rewrites automatic. A per-writer
+  `RewriteOverflow` spill buffer (aof_rewrite_buf equivalent; 256 MiB cap,
+  strict ordering, all six fold arms on both runtimes) now buffers the
+  overflow and drains it into the committed incr right after the fold;
+  `INFO persistence` gains `aof_rewrite_overflow_spilled`. Merge-base A/B
+  (tests/recovery_matrix_w1.rs): main lost/error-failed ~5.6k acked writes per
+  hit; fixed is exact across SIGKILL + recovery. (2) Any dropped acked append
+  now latches sticky `aof_last_append_status:err` (INFO) via a single
+  accounting helper. (3) Eviction/expiry reason-DELs get a 100× escalated
+  backpressure bound (500ms) plus a dedicated `aof_reason_del_dropped`
+  counter — a dropped reason-DEL means restart replay resurrects data clients
+  were told was gone. (4) WAL v3 replay now REFUSES to continue past a
+  corrupt record when later segments exist (mid-chain tear = on-disk
+  corruption; applying later segments silently replays operations from after
+  a hole) — `MOON_WAL_SALVAGE=1` is the explicit operator override; a torn
+  FINAL segment stays the benign crash-tail it always was.
+
+  Adversarial-review hardening round (pre-merge): the rewrite fold's
+  exactly-once contract now extends to the overflow buffer — a snapshot
+  **cut** (`mark_cut`, recorded at the fold's atomic snapshot instant)
+  splits spilled entries so a COMMITTED fold discards pre-snapshot spills
+  (their effects are in the new base; replaying them would double-apply
+  INCR/APPEND/LPUSH) while an aborted fold still writes everything.
+  Producer paths are fully gated (`spill_first` on every enqueue leg,
+  including backpressure/AppendSync parks) so mid-fold channel drains can
+  never invert same-key replay order, and the finish drain is two-phase
+  (channel before buffer, disarm under the buffer lock). Arming is
+  unwind-safe: a panicking fold disarms with drop accounting instead of
+  leaving producers spilling into a dead buffer forever. Boot paths now
+  actually ABORT (exit 70) on a fatal mid-chain tear instead of falling
+  back to legacy recovery, reason-DEL backpressure shares ONE bound per
+  eviction/expiry sweep (was one 500ms bound per victim key), and every
+  cap/shutdown drop path routes through the loss-accounting latch.
+
 - **Cluster formation actually converges (pre-existing, both runtimes).**
   A 3-node cluster could never complete its mesh: (1) `CLUSTER MEET`'s
   random-id placeholder was never retired when the peer's handshake arrived

src/command/connection.rs

@@ -301,6 +301,9 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
          aof_backpressure_dropped:{}\r\n\
          aof_last_fsync_status:{}\r\n\
          aof_fsync_failures:{}\r\n\
+         aof_last_append_status:{}\r\n\
+         aof_reason_del_dropped:{}\r\n\
+         aof_rewrite_overflow_spilled:{}\r\n\
          spill_batches_flushed:{}\r\n\
          spill_completions_dropped:{}\r\n\
          spill_failed_reinserted:{}\r\n\
@@ -335,6 +338,14 @@ pub fn info(db: &Database, _args: &[Frame]) -> Frame {
             "err"
         },
         crate::persistence::aof::AOF_FSYNC_FAILURES.load(std::sync::atomic::Ordering::Relaxed),
+        if crate::persistence::aof::AOF_LAST_APPEND_OK.load(std::sync::atomic::Ordering::Relaxed) {
+            "ok"
+        } else {
+            "err"
+        },
+        crate::persistence::aof::AOF_REASON_DEL_DROPPED.load(std::sync::atomic::Ordering::Relaxed),
+        crate::persistence::aof::rewrite_overflow::AOF_REWRITE_OVERFLOW_SPILLED
+            .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(),

src/command/persistence.rs

@@ -256,7 +256,7 @@ pub fn bgrewriteaof_start(pool: &AofWriterPool, db: SharedDatabases) -> Frame {
             b"ERR Background AOF rewrite already in progress",
         ));
     }
-    match pool.try_send_rewrite(AofMessage::Rewrite(db)) {
+    match pool.try_send_rewrite(AofMessage::Rewrite(db, pool.overflow_for(0).clone())) {
         Ok(()) => Frame::SimpleString(Bytes::from_static(
             b"Background append only file rewriting started",
         )),
@@ -324,7 +324,10 @@ pub fn bgrewriteaof_start_sharded(
         }
     }
 
-    match pool.try_send_rewrite(AofMessage::RewriteSharded(shard_databases)) {
+    match pool.try_send_rewrite(AofMessage::RewriteSharded(
+        shard_databases,
+        pool.overflow_for(0).clone(),
+    )) {
         Ok(()) => Frame::SimpleString(Bytes::from_static(
             b"Background append only file rewriting started",
         )),

src/persistence/aof/group_commit.rs

@@ -62,8 +62,8 @@ pub struct GroupCommitBatch {
 fn is_control(msg: &AofMessage) -> bool {
     matches!(
         msg,
-        AofMessage::Rewrite(_)
-            | AofMessage::RewriteSharded(_)
+        AofMessage::Rewrite(..)
+            | AofMessage::RewriteSharded(..)
             | AofMessage::RewritePerShard { .. }
             | AofMessage::Shutdown
     )

src/persistence/aof/mod.rs

@@ -133,6 +133,35 @@ pub fn record_everysec_fsync_result(writer_idx: usize, ok: bool) {
     }
 }
 
+/// Degraded-state latch (#452.4): `false` once ANY acked append has been
+/// dropped at the writer channel since boot. Unlike the rolling counter
+/// above, this is a sticky health bit — operators (and test harnesses) can
+/// alert on `aof_last_append_status:err` in `INFO persistence` without
+/// diffing counters. It never resets: after a drop the AOF is missing an
+/// acked record for the lifetime of this generation, so "everything is fine
+/// again" would be a lie until a successful rewrite folds live state into a
+/// fresh base. (`BGREWRITEAOF` is the operator remediation; wiring the latch
+/// reset into rewrite completion is deliberate follow-up work tracked in
+/// #452 — reset must only happen if NO drop occurred during the fold.)
+pub static AOF_LAST_APPEND_OK: std::sync::atomic::AtomicBool =
+    std::sync::atomic::AtomicBool::new(true);
+
+/// Reason-DELs (eviction/expiry) dropped at the writer channel (#452.4).
+/// Strictly worse than a dropped client write: replay RESURRECTS a key the
+/// server told clients was gone. Kept as a dedicated counter so a non-zero
+/// value can be alerted on independently of generic backpressure drops.
+pub static AOF_REASON_DEL_DROPPED: std::sync::atomic::AtomicU64 =
+    std::sync::atomic::AtomicU64::new(0);
+
+/// Record `n` dropped acked appends: bumps [`AOF_BACKPRESSURE_DROPPED`] and
+/// latches [`AOF_LAST_APPEND_OK`] to `err`. Every drop site MUST go through
+/// this helper so the degraded-state latch can never miss a loss.
+#[inline]
+pub fn record_append_dropped(n: u64) {
+    AOF_BACKPRESSURE_DROPPED.fetch_add(n, std::sync::atomic::Ordering::Relaxed);
+    AOF_LAST_APPEND_OK.store(false, std::sync::atomic::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
@@ -141,6 +170,17 @@ pub fn record_everysec_fsync_result(writer_idx: usize, ok: bool) {
 /// when the writer channel (10k slots) is completely full.
 pub const AOF_SPSC_BACKPRESSURE_BOUND: std::time::Duration = std::time::Duration::from_millis(5);
 
+/// Backpressure bound for reason-DELs (eviction/expiry — #452.4). 100× the
+/// generic SPSC bound: a dropped reason-DEL makes restart replay resurrect
+/// data clients were told is gone, so we trade a longer worst-case shard
+/// stall (only reachable when the writer is >10k appends behind) for a much
+/// smaller loss window. NOT unbounded: a dead writer must not hang the
+/// shard, and a deferred-retry queue would reorder DEL-after-SET on replay —
+/// beyond this bound the drop is counted in [`AOF_REASON_DEL_DROPPED`] and
+/// latches [`AOF_LAST_APPEND_OK`], never silent.
+pub const AOF_REASON_DEL_BACKPRESSURE_BOUND: std::time::Duration =
+    std::time::Duration::from_millis(500);
+
 /// Result of awaiting an `AppendSync` ack under a bounded timeout (F2).
 ///
 /// Distinguishes the three terminal states the `Always` durability path
@@ -235,9 +275,16 @@ pub enum AofMessage {
         ack: crate::runtime::channel::OneshotSender<AofAck>,
     },
     /// Trigger a full AOF rewrite (compaction) using current database state.
-    Rewrite(SharedDatabases),
+    /// The [`rewrite::RewriteOverflow`] is this writer's rewrite-window spill
+    /// buffer (issue #452.1): armed by the writer around the fold, fed by the
+    /// pool's producers when the append channel saturates mid-fold.
+    Rewrite(SharedDatabases, Arc<rewrite::RewriteOverflow>),
     /// Trigger AOF rewrite in sharded mode (all shards' databases).
-    RewriteSharded(Arc<crate::shard::shared_databases::ShardDatabases>),
+    /// Overflow semantics identical to [`AofMessage::Rewrite`].
+    RewriteSharded(
+        Arc<crate::shard::shared_databases::ShardDatabases>,
+        Arc<rewrite::RewriteOverflow>,
+    ),
     /// [F6] Trigger a per-shard AOF rewrite (compaction) in the PerShard
     /// layout. Sent to EVERY per-shard writer at once. Each writer folds its
     /// own shard (drain → AofFold SPSC → snapshot → write new base+incr at
@@ -259,6 +306,11 @@ pub enum AofMessage {
             Arc<parking_lot::Mutex<ringbuf::HeapProd<crate::shard::dispatch::ShardMessage>>>,
         /// Notifier that wakes the shard event loop after an SPSC push.
         fold_notifier: Arc<crate::runtime::channel::Notify>,
+        /// Rewrite-window spill buffer for THIS shard (issue #452.1) — armed
+        /// by the writer around the fold, fed by the pool's producers when
+        /// the append channel saturates mid-fold, drained by the writer into
+        /// the committed incr immediately after the fold.
+        overflow: Arc<rewrite::RewriteOverflow>,
     },
     /// Shut down the AOF writer task gracefully.
     Shutdown,
@@ -580,7 +632,8 @@ pub mod auto_rewrite;
 /// seam (collect/commit) against the public API.
 pub mod group_commit;
 mod pool;
-mod rewrite;
+pub mod rewrite;
+pub mod rewrite_overflow;
 mod writer_task;
 
 pub use pool::AofWriterPool;