diff --git a/CHANGELOG.md b/CHANGELOG.md
index 02df02f9..174c025f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **AOF+WAL double-write eliminated for the common config (PR #211)** — new
+  `--wal-kv-log auto|on|off` (default `auto`). At `--appendonly yes` every
+  SPSC-executed KV write was logged to BOTH the per-shard AOF and the per-shard WAL
+  (measured 2.7× file-byte / 4.1× device write amplification at `--shards 4`), yet
+  startup recovery wipes WAL-replayed state and replays the AOF — the WAL copy was
+  pure disk wear. `auto` skips WAL KV records while the AOF is the recovery
+  authority and no CDC subscriber is attached (logging re-engages dynamically when
+  one attaches); `on` restores the pre-0.6 always-log behavior (needed for PITR /
+  full CDC history alongside AOF); `off` never logs KV records. FPI/checkpoint/
+  feature records are unaffected.
+- **everysec/no AOF appends are no longer silently dropped under backpressure
+  (PR #211)** — a full writer channel used to `warn!` + drop the record while the
+  client still received `+OK` (client-acked write loss + AOF/memory divergence on
+  replay). Durable handler paths now await enqueue under `--aof-fsync-timeout-ms`
+  and surface `ChannelFull`/`WriteFailed` as an error frame; the synchronous
+  SPSC-drain and monoio inline-SET paths apply a bounded blocking send (ONE 5ms
+  budget shared across a whole pipeline/MULTI batch) and, on loss, replace the
+  success frame with `MOONERR AOF backpressure` instead of acking — every drop is
+  `error!`-logged and counted. `--wal-kv-log` also rejects unknown values at parse
+  time. Watch `aof_backpressure_dropped` in `INFO`.
+
 - **Docker image build** — the multi-stage `Dockerfile` now copies the vendored
   `vendor/monoio` source into the cargo-chef planner and cook stages. The v0.5.0
   release introduced a `[patch.crates-io] monoio = { path = "vendor/monoio" }`
diff --git a/docs/configuration.md b/docs/configuration.md
index ec0fd26d..a6149668 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -24,7 +24,8 @@ All options are available as command-line flags. Run `moon --help` for the full
 |------|---------|-------------|
 | `--appendonly` | `yes` | Enable AOF persistence (`yes`/`no`) — Moon is durable by default |
 | `--appendfsync` | `everysec` | AOF fsync policy (`always`/`everysec`/`no`) |
-| `--aof-fsync-timeout-ms` | `2000` | Bound on a write's wait for its fsync barrier under `always` (0 = unbounded) |
+| `--aof-fsync-timeout-ms` | `2000` | Bound on a write's wait for durability — the fsync ack under `always`, writer-queue backpressure under `everysec` (0 = unbounded) |
+| `--wal-kv-log` | `auto` | KV logging into the per-shard WAL. `auto`: skipped while the AOF is the recovery authority and no CDC subscriber is attached (halves write volume at `--shards >= 2`); `on`: always log (needed for PITR / full CDC history with AOF on); `off`: never |
 | `--appendfilename` | `appendonly.aof` | AOF filename |
 | `--save` | *(none)* | RDB auto-save rules (e.g., `"3600 1 300 100"`) |
 | `--dir` | `.` | Directory for persistence files |
diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md
index 9f2f8f68..d61626ee 100644
--- a/docs/guides/configuration.md
+++ b/docs/guides/configuration.md
@@ -26,12 +26,25 @@ Moon is configured through command-line flags and an optional Redis-style config
 | `--requirepass` | *(none)* | Require clients to authenticate with this password |
 | `--check-config` | `false` | Validate configuration and exit without starting |
 
+### Defaults are tuned for the single-shard case
+
+Most deployments should run the defaults as-is: `--shards 1` gives the best
+per-op latency for low-concurrency and non-pipelined traffic (a cross-shard hop
+costs ~10µs), keeps every key co-located (no `{hash-tag}` planning), and — with
+`--appendonly yes` + `--wal-kv-log auto` — writes each KV record to disk exactly
+**once** (the AOF). Reach for more shards only with 8+ concurrent connections or
+deep pipelines (measured 1.3–2.5× Redis at 8–64 conns on 4 shards), and see
+[Tuning](tuning.md) for `--io-busy-poll-us` (a p=1 latency win **only** on
+pinned/dedicated cores).
+
 ## Persistence
 
 | Flag | Default | Description |
 |------|---------|-------------|
 | `--appendonly` | `yes` | Enable append-only file persistence (`yes`/`no`) — Moon is durable by default |
 | `--appendfsync` | `everysec` | AOF fsync policy: `always`, `everysec`, or `no` |
+| `--aof-fsync-timeout-ms` | `2000` | Max time a write may block awaiting durability (fsync ack under `always`, writer-queue backpressure under `everysec`) before it fails loudly instead of parking the connection. `0` = unbounded |
+| `--wal-kv-log` | `auto` | KV command logging into the per-shard WAL. `auto`: skipped while the AOF is the recovery authority (`--appendonly yes`) and no CDC subscriber is attached — halves write volume at `--shards >= 2`; re-engages automatically when a CDC subscriber attaches. `on`: always log (pre-0.6 behavior; needed for [PITR](pitr.md) or full [CDC](cdc.md) history alongside AOF). `off`: never log KV records |
 | `--appendfilename` | `appendonly.aof` | AOF filename |
 | `--save` | *(none)* | RDB auto-save rules (e.g., `"3600 1 300 100"`) |
 | `--dbfilename` | `dump.rdb` | RDB snapshot filename |
diff --git a/docs/guides/persistence.md b/docs/guides/persistence.md
index 72c188f1..f0ef2fc3 100644
--- a/docs/guides/persistence.md
+++ b/docs/guides/persistence.md
@@ -44,6 +44,25 @@ Moon uses WAL v2 with:
 
 The hot-path cost of WAL append is ~5ns (`buf.extend_from_slice()`), with batch `write_all` every 1ms tick and fsync on the configured schedule.
 
+### One KV log at a time (`--wal-kv-log`)
+
+With `--appendonly yes` the **AOF is the crash-recovery authority**: startup
+replays the AOF over a wiped keyspace, discarding whatever the WAL replayed
+first. Moon therefore skips the WAL copy of each KV command by default
+(`--wal-kv-log auto`) — one durable KV log instead of two, halving on-disk
+write volume at `--shards >= 2` with zero recovery loss. The WAL still carries
+checkpoint/FPI and feature records, and KV logging re-engages automatically
+when a CDC subscriber attaches. Set `--wal-kv-log on` if you need
+point-in-time recovery or full CDC history alongside the AOF.
+
+> **⚠ The WAL is not a standalone durability log.** With `--appendonly no`,
+> only cross-shard (SPSC-dispatched) writes reach the WAL — writes local to a
+> connection's own shard are not logged anywhere, so crash recovery loses
+> roughly `1/num_shards` of writes at `--shards >= 2` (measured: 79% recovered
+> at 4 shards) and **everything** at `--shards 1`. Keep `--appendonly yes`
+> (the default) whenever you need KV durability; the WAL's KV stream exists
+> for CDC, PITR, and disk-offload — not as an AOF replacement.
+
 ## RDB snapshots
 
 RDB creates point-in-time snapshots of the entire dataset.
diff --git a/scripts/gcloud-wal-aof-ab.sh b/scripts/gcloud-wal-aof-ab.sh
new file mode 100755
index 00000000..39fd2e5b
--- /dev/null
+++ b/scripts/gcloud-wal-aof-ab.sh
@@ -0,0 +1,222 @@
+#!/usr/bin/env bash
+# gcloud-wal-aof-ab.sh — same-instance A/B proof of the write-path durability fixes
+# (--wal-kv-log gate + everysec bounded backpressure) on GCloud real SSD.
+#
+# Proves, per config {--wal-kv-log on (old behavior) | auto (fix) | walonly
+# (--appendonly no + default disk-offload: WAL-v3 is the sole KV log)} at
+# shards={1,4}, appendonly=yes everysec (defaults) except walonly:
+#   1. VOLUME     — on-disk bytes attributed AOF vs WAL (+ /proc/<pid>/io write_bytes).
+#                   Expect: auto ==> WAL header-only at shards=4 (~44% total cut); shards=1 unchanged.
+#   2. THROUGHPUT — SET RPS, P1 & P64, c50, median of 3 reps (fresh server per rep).
+#                   Expect: no regression (OrbStack showed the tax is flat; real SSD may show a win).
+#   3. RECOVERY   — auto @ shards=4: write N keys, kill -9, restart, DBSIZE must equal N
+#                   (AOF alone recovers everything with the WAL gate active).
+#   4. BACKPRESSURE — INFO aof_backpressure_dropped must stay 0 in every run
+#                   (bounded-backpressure path never engages on a healthy disk).
+#
+# Measure-only: never edits src. Patterned on gcloud-kv-scale-bench.sh (provision/
+# scp/measure/teardown). The instance builds the LOCAL source snapshot (scp'd
+# tarball via `git archive`), NOT the GitHub main — the fix is unpushed.
+#
+# Subcommands:
+#   --measure   INNER: runs ON the instance (expects ~/moon-src + toolchain)
+#   --one       OUTER (default): provision $GCE_MACHINE, run, teardown
+set -uo pipefail
+
+GCE_MACHINE="${GCE_MACHINE:-c3-standard-4}"
+GCE_NAME="${GCE_NAME:-moon-wal-ab}"
+GCE_ZONE="${GCE_ZONE:-us-central1-a}"
+REQUESTS="${REQUESTS:-500000}"
+KEYSPACE="${KEYSPACE:-1000000}"
+REPS="${REPS:-3}"
+PORT="${PORT:-6399}"
+
+log(){ printf '%s\n' "$*" >&2; }
+die(){ log "FATAL: $*"; exit 1; }
+
+# ============================================================================ INNER
+wait_port(){ # host port tries — every probe timeout-bounded (a half-dead server
+  # that accepts but never replies must not hang the harness)
+  local t=0; until timeout 2 redis-cli -p "$2" ping 2>/dev/null | grep -q PONG; do
+    t=$((t+1)); [[ $t -ge 100 ]] && return 1; sleep 0.2
+  done
+}
+
+stop_moon(){ # $1=pid — SIGKILL + pattern backstop + wait-for-port-free.
+  # Moon handles SIGTERM gracefully and can linger; SO_REUSEPORT then lets the
+  # NEXT server bind the same port alongside the old one (two servers, split
+  # traffic, hung pings). kill -9 is the house convention (see CLAUDE.md).
+  kill -9 "$1" 2>/dev/null; wait "$1" 2>/dev/null
+  pkill -9 -f "release/moon --port $PORT" 2>/dev/null
+  local t=0
+  while pgrep -f "release/moon --port $PORT" >/dev/null 2>&1; do
+    t=$((t+1)); [[ $t -ge 50 ]] && { log "WARN: moon refused to die"; return 1; }; sleep 0.2
+  done
+  return 0
+}
+
+rps_of(){ # extract last "requests per second" (redis-benchmark 8.x \r progress trap)
+  tr '\r' '\n' | grep -i "requests per second" | tail -1 | grep -oE '[0-9.]+' | head -1
+}
+
+dir_bytes(){ # $1=dir  $2=glob-ere  -> total bytes of matching files
+  find "$1" -type f 2>/dev/null | grep -E "$2" | xargs -r stat -c '%s' | awk '{s+=$1} END{print s+0}'
+}
+
+start_moon(){ # $1=shards $2=walkv $3=dir -> pid
+  if [[ "$2" == "walonly" ]]; then
+    # WAL-only posture: AOF off; disk-offload stays at its default (enable) so
+    # WAL-v3 is the sole KV log (wal-kv-log auto => keep logging, no AOF to
+    # defer to). NOTE: conn-LOCAL writes never enter the SPSC path and are not
+    # WAL-logged — the recovery check below sizes that gap empirically.
+    ./moon-src/target/release/moon --port "$PORT" --shards "$1" \
+      --appendonly no --dir "$3" >"$3/server.log" 2>&1 &
+  else
+    ./moon-src/target/release/moon --port "$PORT" --shards "$1" \
+      --appendonly yes --appendfsync everysec --wal-kv-log "$2" \
+      --dir "$3" >"$3/server.log" 2>&1 &
+  fi
+  echo $!
+}
+
+measure_one(){ # $1=shards $2=walkv  -> prints a report block
+  local shards="$1" walkv="$2" dir pid rps_p1=() rps_p64=() r
+  echo "### shards=$shards wal-kv-log=$walkv"
+
+  # -- volume run (single, fixed workload) --
+  dir=$(mktemp -d /ssd-bench/walab.XXXX)
+  pid=$(start_moon "$shards" "$walkv" "$dir")
+  wait_port localhost "$PORT" || { cat "$dir/server.log"; die "moon never accepted (volume run)"; }
+  redis-benchmark -p "$PORT" -t set -n "$REQUESTS" -r "$KEYSPACE" -d 64 -c 50 -P 1 -q >/dev/null 2>&1
+  timeout 5 redis-cli -p "$PORT" set MARKERKEY_ZZZ9 MARKERVALUE_QQQ8 >/dev/null
+  sleep 3   # let 1ms-tick flush + everysec fsync settle
+  local aof_b wal_b pio drops
+  aof_b=$(dir_bytes "$dir" 'appendonly.*\.aof$|appendonlydir/.*\.aof$')
+  wal_b=$(dir_bytes "$dir" '\.wal$')
+  pio=$(awk '/^write_bytes/{print $2}' "/proc/$pid/io" 2>/dev/null || echo 0)
+  drops=$(timeout 5 redis-cli -p "$PORT" info persistence 2>/dev/null | tr -d '\r' | awk -F: '/aof_backpressure_dropped/{print $2}')
+  local marker_wal=0
+  if find "$dir" -name '*.wal' | head -5 | xargs -r strings 2>/dev/null | grep -q MARKERVALUE_QQQ8; then marker_wal=1; fi
+  echo "volume: aof_bytes=$aof_b wal_bytes=$wal_b proc_write_bytes=$pio backpressure_dropped=${drops:-NA} marker_in_wal=$marker_wal"
+
+  # -- recovery check (shards=4: auto proves AOF-alone recovery; walonly sizes
+  #    WAL-alone recovery, incl. the conn-local-write gap) --
+  if [[ "$shards" == "4" && ( "$walkv" == "auto" || "$walkv" == "walonly" ) ]]; then
+    local nkeys; nkeys=$(timeout 5 redis-cli -p "$PORT" dbsize | grep -oE '[0-9]+')
+    stop_moon "$pid"
+    pid=$(start_moon "$shards" "$walkv" "$dir")
+    wait_port localhost "$PORT" || die "moon never accepted (recovery run)"
+    local nkeys2 marker2
+    nkeys2=$(timeout 5 redis-cli -p "$PORT" dbsize | grep -oE '[0-9]+')
+    marker2=$(timeout 5 redis-cli -p "$PORT" get MARKERKEY_ZZZ9)
+    echo "recovery: keys_before_kill9=$nkeys keys_after_restart=$nkeys2 marker=$marker2"
+  fi
+  stop_moon "$pid"
+  rm -rf "$dir"
+
+  # -- throughput reps (fresh server per rep; phantom-regression rule) --
+  for r in $(seq 1 "$REPS"); do
+    dir=$(mktemp -d /ssd-bench/walab.XXXX)
+    pid=$(start_moon "$shards" "$walkv" "$dir")
+    wait_port localhost "$PORT" || die "moon never accepted (rep $r)"
+    rps_p1+=("$(redis-benchmark -p "$PORT" -t set -n "$REQUESTS" -r "$KEYSPACE" -d 64 -c 50 -P 1 2>/dev/null | rps_of)")
+    rps_p64+=("$(redis-benchmark -p "$PORT" -t set -n "$REQUESTS" -r "$KEYSPACE" -d 64 -c 50 -P 64 2>/dev/null | rps_of)")
+    drops=$(timeout 5 redis-cli -p "$PORT" info persistence 2>/dev/null | tr -d '\r' | awk -F: '/aof_backpressure_dropped/{print $2}')
+    stop_moon "$pid"
+    rm -rf "$dir"
+    echo "rep$r: P1=${rps_p1[-1]:-NA} P64=${rps_p64[-1]:-NA} drops=${drops:-NA}"
+  done
+  echo "throughput: P1=[${rps_p1[*]}] P64=[${rps_p64[*]}]"
+  echo
+}
+
+measure(){
+  set -e
+  sudo mkdir -p /ssd-bench && sudo chmod 777 /ssd-bench
+  pkill -9 -f "release/moon --port $PORT" 2>/dev/null; rm -rf /ssd-bench/walab.* 2>/dev/null
+  command -v redis-benchmark >/dev/null || { sudo apt-get update -qq && sudo apt-get install -y -qq redis-tools; }
+  # steal gate (dedicated vCPU expected ~0)
+  awk '/^cpu /{print "cpu-steal-jiffies:", $9}' /proc/stat
+  echo "# ===== WAL/AOF A/B ($(uname -m), $(date -u +%FT%TZ)) ====="
+  echo "# moon: $(./moon-src/target/release/moon --version 2>/dev/null || echo unknown)"
+  set +e
+  for shards in 1 4; do
+    for walkv in on auto walonly; do
+      measure_one "$shards" "$walkv"
+    done
+  done
+  echo "# ===== END ====="
+}
+
+# ============================================================================ OUTER
+run_one(){
+  command -v gcloud >/dev/null || die "gcloud CLI not found"
+  local imgfam
+  case "$GCE_MACHINE" in c4a-*|t2a-*) imgfam=ubuntu-2404-lts-arm64;; *) imgfam=ubuntu-2404-lts-amd64;; esac
+  local GSSH=(gcloud compute ssh "$GCE_NAME" --zone="$GCE_ZONE" --quiet
+        --ssh-flag=-oStrictHostKeyChecking=no --ssh-flag=-oConnectTimeout=15)
+
+  log "=== packing LOCAL source (HEAD commit) ==="
+  git -C "$(dirname "$0")/.." archive --format=tar.gz -o /tmp/moon-src.tar.gz --prefix=moon-src/ HEAD \
+    || die "git archive failed (commit your changes first — the A/B ships HEAD)"
+
+  log "=== provisioning $GCE_NAME ($GCE_MACHINE, $GCE_ZONE, $imgfam, pd-ssd) ==="
+  gcloud compute instances create "$GCE_NAME" \
+    --machine-type="$GCE_MACHINE" --zone="$GCE_ZONE" \
+    --image-family="$imgfam" --image-project=ubuntu-os-cloud \
+    --boot-disk-size=100GB --boot-disk-type=pd-ssd --quiet || die "instance create failed"
+  local _inst="$GCE_NAME" _zone="$GCE_ZONE"
+  trap "log '=== tearing down $_inst ==='; gcloud compute instances delete '$_inst' --zone='$_zone' -q 2>/dev/null || true" EXIT INT TERM
+
+  log "=== waiting for SSH ==="
+  local tries=0
+  until "${GSSH[@]}" --command='echo up' >/dev/null 2>&1; do
+    tries=$((tries+1)); [[ $tries -ge 40 ]] && die "SSH never came up"; sleep 5
+  done
+
+  log "=== toolchain ==="
+  "${GSSH[@]}" --command='
+    set -e
+    sudo apt-get update -qq
+    sudo apt-get install -y -qq build-essential pkg-config libssl-dev git curl ca-certificates redis-tools
+    command -v cargo >/dev/null || curl --proto "=https" --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --default-toolchain 1.94.1
+  ' || die "toolchain provisioning failed"
+
+  log "=== pushing source + harness ==="
+  gcloud compute scp /tmp/moon-src.tar.gz "$(dirname "$0")/gcloud-wal-aof-ab.sh" \
+    "$GCE_NAME":~/ --zone="$GCE_ZONE" --quiet --scp-flag=-oStrictHostKeyChecking=no || die "scp failed"
+
+  log "=== building release on instance (the long part) ==="
+  "${GSSH[@]}" --command='
+    set -e; source $HOME/.cargo/env
+    tar xzf moon-src.tar.gz
+    cd moon-src && cargo build --release 2>&1 | tail -3
+    file target/release/moon | grep -q ELF || { echo "NOT AN ELF"; exit 1; }
+  ' || die "build failed"
+
+  local rawfile="tmp/wal-ab-${GCE_MACHINE}.txt"; mkdir -p tmp
+  log "=== running --measure -> $rawfile ==="
+  "${GSSH[@]}" --command="
+    source \$HOME/.cargo/env
+    REQUESTS='$REQUESTS' KEYSPACE='$KEYSPACE' REPS='$REPS' PORT='$PORT' \
+      bash ~/gcloud-wal-aof-ab.sh --measure
+  " | tee "$rawfile"
+  log "=== results in $rawfile; teardown follows (trap) ==="
+}
+
+measure_single(){ # $1=shards $2=walkv — one config only (chunked outer driving)
+  set +e
+  sudo mkdir -p /ssd-bench && sudo chmod 777 /ssd-bench
+  pkill -9 -f "release/moon --port $PORT" 2>/dev/null; rm -rf /ssd-bench/walab.* 2>/dev/null
+  command -v redis-benchmark >/dev/null || { sudo apt-get update -qq && sudo apt-get install -y -qq redis-tools; }
+  awk '/^cpu /{print "cpu-steal-jiffies:", $9}' /proc/stat
+  measure_one "$1" "$2"
+}
+
+case "${1:---one}" in
+  --measure) measure ;;
+  --measure-one) measure_single "${2:?usage: --measure-one <shards> <on|auto|walonly>}" \
+                                "${3:?usage: --measure-one <shards> <on|auto|walonly>}" ;;
+  --one)     run_one ;;
+  *) die "unknown subcommand: $1 (use --measure | --one)" ;;
+esac
diff --git a/src/config.rs b/src/config.rs
index 0bca8fda..7c7fea87 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -336,6 +336,20 @@ pub struct ServerConfig {
     #[arg(long = "wal-max-checkpoint-lag-ms", default_value_t = 10_000)]
     pub wal_max_checkpoint_lag_ms: u64,
 
+    /// KV command logging into the per-shard WAL: auto | on | off.
+    ///
+    /// `auto` (default): KV records are skipped while the AOF is the crash-
+    /// recovery authority (`--appendonly yes`) and no CDC subscriber is
+    /// attached — startup recovery wipes WAL-replayed state and replays the
+    /// AOF, so the WAL copy is pure write amplification (~2× disk writes at
+    /// shards>=2). Logging resumes dynamically when a CDC subscriber attaches.
+    /// `on`: always log (pre-0.6 behavior; required for point-in-time
+    /// recovery or full CDC history alongside `--appendonly yes`).
+    /// `off`: never log KV records (FPI/checkpoint/feature records still
+    /// written). With `--appendonly no` this leaves NO KV durability log.
+    #[arg(long = "wal-kv-log", default_value = "auto", value_parser = ["auto", "on", "off"])]
+    pub wal_kv_log: String,
+
     // ── MoonStore v2: Vector Warm Tier ──────────────────────────────
     /// mlock vector codes pages into RAM
     #[arg(long = "vec-codes-mlock", default_value = "enable")]
@@ -648,6 +662,19 @@ fn default_data_dir() -> Option<std::path::PathBuf> {
     )
 }
 
+/// Resolved `--wal-kv-log` mode (see the flag docs on `ServerConfig::wal_kv_log`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum WalKvLogMode {
+    /// Skip WAL KV records while the AOF is the recovery authority and no
+    /// CDC subscriber is attached; log otherwise. The per-drain decision is
+    /// made by the shard event loop (it owns the CDC registry).
+    Auto,
+    /// Always log KV records to the WAL (pre-0.6 behavior).
+    On,
+    /// Never log KV records to the WAL.
+    Off,
+}
+
 /// Outcome of persistence-directory auto-resolution (pure decision —
 /// IO happens in [`ServerConfig::resolve_dir`]).
 #[derive(Debug, PartialEq, Eq)]
@@ -733,6 +760,19 @@ impl ServerConfig {
         self.disk_offload == "enable"
     }
 
+    /// Resolve `--wal-kv-log` into its mode. The CLI rejects unknown values
+    /// at parse time (`value_parser`, fail-fast on typos — a silent `Auto`
+    /// fallback could unexpectedly disable WAL KV history needed for
+    /// PITR/CDC); the `_ => Auto` arm below only covers programmatic
+    /// construction in tests.
+    pub fn wal_kv_log_mode(&self) -> WalKvLogMode {
+        match self.wal_kv_log.as_str() {
+            "on" => WalKvLogMode::On,
+            "off" => WalKvLogMode::Off,
+            _ => WalKvLogMode::Auto,
+        }
+    }
+
     /// Returns true when WAL Full Page Images are enabled.
     pub fn wal_fpi_enabled(&self) -> bool {
         self.wal_fpi == "enable"
diff --git a/src/persistence/aof/mod.rs b/src/persistence/aof/mod.rs
index a7580914..9fd1c8e3 100644
--- a/src/persistence/aof/mod.rs
+++ b/src/persistence/aof/mod.rs
@@ -92,6 +92,14 @@ pub enum AofAck {
 pub static AOF_BACKPRESSURE_DROPPED: std::sync::atomic::AtomicU64 =
     std::sync::atomic::AtomicU64::new(0);
 
+/// 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
+/// write, no fsync under everysec — low single-digit ms on a healthy disk)
+/// while keeping the worst-case shard event-loop stall small. Reached only
+/// when the writer channel (10k slots) is completely full.
+pub const AOF_SPSC_BACKPRESSURE_BOUND: std::time::Duration = std::time::Duration::from_millis(5);
+
 /// Result of awaiting an `AppendSync` ack under a bounded timeout (F2).
 ///
 /// Distinguishes the three terminal states the `Always` durability path
diff --git a/src/persistence/aof/pool.rs b/src/persistence/aof/pool.rs
index f549d08b..b75196be 100644
--- a/src/persistence/aof/pool.rs
+++ b/src/persistence/aof/pool.rs
@@ -234,10 +234,12 @@ impl AofWriterPool {
     /// vector identified in the investigation report. For `EverySec` and
     /// `No`, it stays on the fire-and-forget path (zero new latency).
     ///
-    /// Returns `Err(AofAck)` only on the Always path when the write or
-    /// fsync failed (or the writer task is gone). Callers MUST treat
-    /// `Err(_)` as a hard failure — return an error frame to the client,
-    /// do NOT respond `+OK`.
+    /// Returns `Err(AofAck)` on the Always path when the write or fsync
+    /// failed (or the writer task is gone), and on the EverySec/No path
+    /// when the append could not be enqueued within the backpressure bound
+    /// (`AofAck::ChannelFull`) or the writer is gone (`WriteFailed`).
+    /// Callers MUST treat `Err(_)` as a hard failure — return an error
+    /// frame to the client, do NOT respond `+OK`.
     ///
     /// Async because the Always branch awaits a oneshot receiver. The
     /// non-Always branch resolves immediately (no actual suspension) so
@@ -269,8 +271,13 @@ impl AofWriterPool {
                 }
             }
             FsyncPolicy::EverySec | FsyncPolicy::No => {
-                self.try_send_append(shard_id, lsn, bytes);
-                Ok(())
+                // Bounded backpressure instead of the pre-0.6 silent drop:
+                // a full writer channel used to `warn!` and lose the record
+                // while the caller still acked `+OK`. Now the caller awaits
+                // enqueue (NOT fsync) under `fsync_timeout`, and a failed
+                // enqueue surfaces as Err so the client never gets a false
+                // success for a write the durability machinery never saw.
+                self.send_append_backpressure(shard_id, lsn, bytes).await
             }
         }
     }
@@ -400,22 +407,187 @@ impl AofWriterPool {
     /// `ReplicationState::issue_lsn(shard_id, bytes.len() as u64)` for writes
     /// that participate in replication ordering; sites without a
     /// replication-state handle pass 0.
+    ///
+    /// Returns `false` when the entry was NOT enqueued (channel full or
+    /// writer gone) — the record is lost. A `Full` loss is counted in
+    /// [`AOF_BACKPRESSURE_DROPPED`]. Callers that can apply backpressure
+    /// instead of losing the record should prefer
+    /// [`Self::send_append_bounded_blocking`] (sync contexts) or
+    /// [`Self::try_send_append_durable`] (async contexts).
     #[inline]
-    pub fn try_send_append(&self, shard_id: usize, lsn: u64, bytes: Bytes) {
-        if let Err(e) = self
+    pub fn try_send_append(&self, shard_id: usize, lsn: u64, bytes: Bytes) -> bool {
+        match self
             .sender(shard_id)
             .try_send(AofMessage::Append { lsn, bytes })
         {
-            warn!(
-                "AOF append dropped for shard {} (lsn {}): channel {}",
+            Ok(()) => true,
+            Err(e) => {
+                if matches!(e, flume::TrySendError::Full(_)) {
+                    AOF_BACKPRESSURE_DROPPED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                }
+                warn!(
+                    "AOF append dropped for shard {} (lsn {}): channel {}",
+                    shard_id,
+                    lsn,
+                    match e {
+                        flume::TrySendError::Full(_) => "full",
+                        flume::TrySendError::Disconnected(_) => "disconnected",
+                    }
+                );
+                false
+            }
+        }
+    }
+
+    /// Append with bounded *blocking* backpressure — for synchronous callers
+    /// (the shard event loop's SPSC drain) that cannot await.
+    ///
+    /// Fast path: `try_send` (zero cost while the writer keeps up; `budget`
+    /// untouched). When the channel is full, blocks the calling thread up to
+    /// the remaining `budget` waiting for the writer to drain a group-commit
+    /// batch, converting what used to be a **silent, client-acked write
+    /// loss** into a short stall. Only after the budget is exhausted is the
+    /// record dropped — counted in [`AOF_BACKPRESSURE_DROPPED`] and logged
+    /// at `error!`.
+    ///
+    /// `budget` is decremented by the time actually spent blocking so that
+    /// batched callers (MultiExecute / PipelineBatch drain arms) share ONE
+    /// bound across the whole batch — without this, an N-command batch under
+    /// sustained backpressure could hold the shard thread for N×bound.
+    /// Single-command callers pass a fresh
+    /// `&mut AOF_SPSC_BACKPRESSURE_BOUND.clone()`-style local.
+    ///
+    /// Deliberate trade-off: blocking the shard event loop is normally
+    /// forbidden, but the block is (a) bounded, (b) only reachable when the
+    /// AOF writer is >CHANNEL_CAP appends behind (disk badly stalled), and
+    /// (c) strictly better than diverging the AOF from memory state — a
+    /// dropped record makes every later replay of a dependent command wrong
+    /// (e.g. INCR replayed without its SET).
+    pub fn send_append_bounded_blocking(
+        &self,
+        shard_id: usize,
+        lsn: u64,
+        bytes: Bytes,
+        budget: &mut Duration,
+    ) -> bool {
+        let mut msg = AofMessage::Append { lsn, bytes };
+        match self.sender(shard_id).try_send(msg) {
+            Ok(()) => return true,
+            Err(flume::TrySendError::Disconnected(_)) => {
+                warn!(
+                    "AOF append dropped for shard {} (lsn {}): channel disconnected",
+                    shard_id, lsn
+                );
+                return false;
+            }
+            Err(flume::TrySendError::Full(returned)) => msg = returned,
+        }
+        if budget.is_zero() {
+            AOF_BACKPRESSURE_DROPPED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            tracing::error!(
+                "AOF append LOST for shard {} (lsn {}): batch backpressure budget exhausted; \
+                 backpressure_dropped={}",
                 shard_id,
                 lsn,
-                match e {
-                    flume::TrySendError::Full(_) => "full",
-                    flume::TrySendError::Disconnected(_) => "disconnected",
+                AOF_BACKPRESSURE_DROPPED.load(std::sync::atomic::Ordering::Relaxed),
+            );
+            return false;
+        }
+        // Slow path only (channel full, about to block for up to `budget` —
+        // milliseconds): two `Instant::now()` calls are noise here, and the
+        // shard-cached timestamp is too coarse to meter the budget.
+        let blocked_at = std::time::Instant::now();
+        let result = self.sender(shard_id).send_timeout(msg, *budget);
+        let spent = *budget;
+        *budget = budget.saturating_sub(blocked_at.elapsed());
+        match result {
+            Ok(()) => true,
+            Err(e) => {
+                if matches!(e, flume::SendTimeoutError::Timeout(_)) {
+                    AOF_BACKPRESSURE_DROPPED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                 }
+                tracing::error!(
+                    "AOF append LOST for shard {} (lsn {}): writer still {} after {:?} \
+                     backpressure bound; backpressure_dropped={}",
+                    shard_id,
+                    lsn,
+                    match e {
+                        flume::SendTimeoutError::Timeout(_) => "full",
+                        flume::SendTimeoutError::Disconnected(_) => "disconnected",
+                    },
+                    spent,
+                    AOF_BACKPRESSURE_DROPPED.load(std::sync::atomic::Ordering::Relaxed),
+                );
+                false
+            }
+        }
+    }
+
+    /// Async bounded-backpressure append for `everysec`/`no` durable callers.
+    ///
+    /// Fast path: `try_send` (identical cost to the old fire-and-forget
+    /// path). When the writer channel is full, awaits `send_async` under the
+    /// pool's `fsync_timeout` bound (`Duration::ZERO` = unbounded await,
+    /// consistent with [`Self::await_ack`]) instead of silently dropping the
+    /// record and letting the caller ack `+OK` for a write that never
+    /// reached the durability machinery.
+    async fn send_append_backpressure(
+        &self,
+        shard_id: usize,
+        lsn: u64,
+        bytes: Bytes,
+    ) -> Result<(), AofAck> {
+        let mut msg = AofMessage::Append { lsn, bytes };
+        match self.sender(shard_id).try_send(msg) {
+            Ok(()) => return Ok(()),
+            Err(flume::TrySendError::Disconnected(_)) => return Err(AofAck::WriteFailed),
+            Err(flume::TrySendError::Full(returned)) => msg = returned,
+        }
+        // Cancel-safety note: flume's `send_async` future is NOT cancel-safe.
+        // If the timeout wins the race after the future has already enqueued
+        // the message, the record IS delivered while the caller sees
+        // `ChannelFull` (and the drop counter over-counts by one). That is
+        // the safe failure direction: the client receives an error and can
+        // retry an idempotent write; we never ack a write that was NOT
+        // enqueued. Standard ack-timeout semantics — not silent loss.
+        let send_fut = self.sender(shard_id).send_async(msg);
+        let timeout = self.fsync_timeout;
+        if timeout.is_zero() {
+            return match send_fut.await {
+                Ok(()) => Ok(()),
+                Err(_) => Err(AofAck::WriteFailed),
+            };
+        }
+        let outcome: Result<(), AofAck>;
+        #[cfg(feature = "runtime-monoio")]
+        {
+            outcome = monoio::select! {
+                res = send_fut => match res {
+                    Ok(()) => Ok(()),
+                    Err(_) => Err(AofAck::WriteFailed),
+                },
+                _ = monoio::time::sleep(timeout) => Err(AofAck::ChannelFull),
+            };
+        }
+        #[cfg(all(feature = "runtime-tokio", not(feature = "runtime-monoio")))]
+        {
+            outcome = match tokio::time::timeout(timeout, send_fut).await {
+                Ok(Ok(())) => Ok(()),
+                Ok(Err(_)) => Err(AofAck::WriteFailed),
+                Err(_) => Err(AofAck::ChannelFull),
+            };
+        }
+        if outcome == Err(AofAck::ChannelFull) {
+            AOF_BACKPRESSURE_DROPPED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            warn!(
+                "AOF writer channel full (shard {}): everysec append dropped after {:?} \
+                 backpressure bound; backpressure_dropped={}",
+                shard_id,
+                timeout,
+                AOF_BACKPRESSURE_DROPPED.load(std::sync::atomic::Ordering::Relaxed),
             );
         }
+        outcome
     }
 
     /// Synchronous (fsync-before-ack) append for `appendfsync=always`
@@ -1767,4 +1939,188 @@ mod pool_tests {
             "dead writer must resolve to WriteFailed"
         );
     }
+
+    // -----------------------------------------------------------------------
+    // Everysec fail-loud (2026-07 write-path durability): a full writer
+    // channel must surface as backpressure/Err — never a silent drop while
+    // the caller acks `+OK` to the client.
+    // -----------------------------------------------------------------------
+
+    /// RED before the fix: `try_send_append_durable` under EverySec returned
+    /// `Ok(())` while the append was silently dropped. Now the caller awaits
+    /// enqueue under `fsync_timeout` and gets `Err(AofAck::ChannelFull)` when
+    /// the writer stays full past the bound — so it returns an error frame
+    /// instead of a false `+OK`.
+    #[cfg(feature = "runtime-tokio")]
+    #[tokio::test]
+    async fn everysec_full_channel_errs_instead_of_silent_drop() {
+        use std::sync::atomic::Ordering;
+        let (tx0, _rx0) = channel::mpsc_bounded::<AofMessage>(1);
+        // Fill the only slot; keep _rx0 alive so the channel is Full, not Disconnected.
+        tx0.try_send(AofMessage::Append {
+            lsn: 0,
+            bytes: Bytes::from_static(b"x"),
+        })
+        .unwrap();
+        let pool = AofWriterPool::top_level_with_policy(
+            tx0,
+            FsyncPolicy::EverySec,
+            Duration::from_millis(50),
+        );
+
+        let before = AOF_BACKPRESSURE_DROPPED.load(Ordering::Relaxed);
+        let res = pool
+            .try_send_append_durable(0, 1, Bytes::from_static(b"SET k v"))
+            .await;
+        assert_eq!(
+            res,
+            Err(AofAck::ChannelFull),
+            "full writer channel must fail loud under everysec, not ack a lost write"
+        );
+        assert!(
+            AOF_BACKPRESSURE_DROPPED.load(Ordering::Relaxed) > before,
+            "dropped everysec append must be counted"
+        );
+    }
+
+    /// The backpressure path must WAIT, not fail, when the writer frees a
+    /// slot within the bound — converting the old silent loss into a short
+    /// stall with zero data loss.
+    #[cfg(feature = "runtime-tokio")]
+    #[tokio::test]
+    async fn everysec_backpressure_succeeds_when_writer_drains_in_time() {
+        let (tx0, rx0) = channel::mpsc_bounded::<AofMessage>(1);
+        tx0.try_send(AofMessage::Append {
+            lsn: 0,
+            bytes: Bytes::from_static(b"x"),
+        })
+        .unwrap();
+        let pool = AofWriterPool::top_level_with_policy(
+            tx0,
+            FsyncPolicy::EverySec,
+            Duration::from_millis(500),
+        );
+
+        // Mock writer: drain one message shortly after, freeing the slot.
+        let drainer = std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(20));
+            let first = rx0.recv().expect("first queued append");
+            (first, rx0)
+        });
+
+        let res = pool
+            .try_send_append_durable(0, 1, Bytes::from_static(b"SET k v"))
+            .await;
+        assert_eq!(
+            res,
+            Ok(()),
+            "append must succeed once the writer drains within the bound"
+        );
+        let (_first, rx0) = drainer.join().unwrap();
+        assert!(
+            matches!(rx0.try_recv(), Ok(AofMessage::Append { lsn: 1, .. })),
+            "the awaited append must actually be enqueued, not lost"
+        );
+    }
+
+    /// Sync SPSC-path variant: `send_append_bounded_blocking` blocks up to
+    /// the bound, then reports the loss (false + counter) instead of
+    /// pretending success.
+    #[test]
+    fn spsc_bounded_blocking_reports_loss_after_bound() {
+        use std::sync::atomic::Ordering;
+        let (tx0, _rx0) = channel::mpsc_bounded::<AofMessage>(1);
+        tx0.try_send(AofMessage::Append {
+            lsn: 0,
+            bytes: Bytes::from_static(b"x"),
+        })
+        .unwrap();
+        let pool = AofWriterPool::top_level_with_policy(
+            tx0,
+            FsyncPolicy::EverySec,
+            DEFAULT_AOF_FSYNC_TIMEOUT,
+        );
+
+        let before = AOF_BACKPRESSURE_DROPPED.load(Ordering::Relaxed);
+        let start = std::time::Instant::now();
+        let mut budget = Duration::from_millis(30);
+        let ok =
+            pool.send_append_bounded_blocking(0, 7, Bytes::from_static(b"SET k v"), &mut budget);
+        assert!(!ok, "still-full channel after the bound must report loss");
+        assert!(
+            start.elapsed() >= Duration::from_millis(30),
+            "the call must have blocked for the backpressure bound before giving up"
+        );
+        assert!(
+            budget.is_zero(),
+            "the budget must be fully consumed by the blocked send"
+        );
+        assert!(
+            AOF_BACKPRESSURE_DROPPED.load(Ordering::Relaxed) > before,
+            "the dropped append must be counted"
+        );
+    }
+
+    /// Batch-budget semantics (PR #211 review): once the shared budget is
+    /// exhausted, subsequent appends in the same batch fail IMMEDIATELY —
+    /// the shard thread must stall at most one bound per batch, not
+    /// bound × batch-len.
+    #[test]
+    fn spsc_bounded_blocking_exhausted_budget_fails_immediately() {
+        use std::sync::atomic::Ordering;
+        let (tx0, _rx0) = channel::mpsc_bounded::<AofMessage>(1);
+        tx0.try_send(AofMessage::Append {
+            lsn: 0,
+            bytes: Bytes::from_static(b"x"),
+        })
+        .unwrap();
+        let pool = AofWriterPool::top_level_with_policy(
+            tx0,
+            FsyncPolicy::EverySec,
+            DEFAULT_AOF_FSYNC_TIMEOUT,
+        );
+
+        let before = AOF_BACKPRESSURE_DROPPED.load(Ordering::Relaxed);
+        let mut budget = Duration::ZERO; // batch already spent its bound
+        let start = std::time::Instant::now();
+        let ok =
+            pool.send_append_bounded_blocking(0, 8, Bytes::from_static(b"SET k v"), &mut budget);
+        assert!(!ok, "exhausted budget must report loss");
+        assert!(
+            start.elapsed() < Duration::from_millis(10),
+            "exhausted budget must not block again (was: full bound per append)"
+        );
+        assert!(
+            AOF_BACKPRESSURE_DROPPED.load(Ordering::Relaxed) > before,
+            "the budget-exhausted drop must be counted"
+        );
+    }
+
+    /// Fast path: capacity available → enqueued with no stall and no budget
+    /// consumption (the try_send never touches the clock).
+    #[test]
+    fn spsc_bounded_blocking_fast_path_enqueues() {
+        let (tx0, rx0) = channel::mpsc_bounded::<AofMessage>(4);
+        let pool = AofWriterPool::top_level_with_policy(
+            tx0,
+            FsyncPolicy::EverySec,
+            DEFAULT_AOF_FSYNC_TIMEOUT,
+        );
+        let mut budget = Duration::from_millis(30);
+        assert!(pool.send_append_bounded_blocking(
+            0,
+            9,
+            Bytes::from_static(b"SET a 1"),
+            &mut budget,
+        ));
+        assert_eq!(
+            budget,
+            Duration::from_millis(30),
+            "fast path must not consume budget"
+        );
+        assert!(matches!(
+            rx0.try_recv(),
+            Ok(AofMessage::Append { lsn: 9, .. })
+        ));
+    }
 }
diff --git a/src/server/conn/blocking.rs b/src/server/conn/blocking.rs
index 38cee632..d0b4c02f 100644
--- a/src/server/conn/blocking.rs
+++ b/src/server/conn/blocking.rs
@@ -1422,7 +1422,20 @@ pub(crate) fn try_inline_dispatch(
             shard_id,
             frozen.len(),
         );
-        pool.try_send_append(shard_id, lsn, frozen);
+        // Bounded backpressure (not fire-and-forget): this path writes `+OK`
+        // below, so a silent drop here is a client-acked write that never
+        // reached the durability machinery. Fast path is the same try_send;
+        // the block is reachable only when the writer channel is completely
+        // full. On loss, fail-loud: the SET is applied in memory, but the
+        // client must see an error, not `+OK` (review finding, PR #211).
+        let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
+        if !pool.send_append_bounded_blocking(shard_id, lsn, frozen, &mut aof_budget) {
+            write_buf.extend_from_slice(
+                b"-MOONERR AOF backpressure: write applied in memory but not queued \
+                  for persistence\r\n",
+            );
+            return 1;
+        }
     }
 
     write_buf.extend_from_slice(b"+OK\r\n");
diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs
index b04a747d..ab7c817b 100644
--- a/src/shard/event_loop.rs
+++ b/src/shard/event_loop.rs
@@ -393,6 +393,16 @@ impl super::Shard {
         // (stronger guarantees: per-record LSN + CRC32C), so v2 is skipped entirely
         // to avoid double-write overhead.
         let appendonly_enabled = runtime_config.read().appendonly != "no";
+        // `--wal-kv-log`: whether SPSC-executed KV writes are also logged to
+        // the per-shard WAL. Resolved per drain cycle (Auto is dynamic on the
+        // CDC registry); see wal_append_and_fanout for the rationale.
+        let wal_kv_log_mode = server_config.wal_kv_log_mode();
+        if wal_kv_log_mode == crate::config::WalKvLogMode::Off && !appendonly_enabled {
+            tracing::warn!(
+                shard_id,
+                "--wal-kv-log off with --appendonly no: KV writes have NO durability log"
+            );
+        }
         let mut wal_writer: Option<WalWriter> = match (&persistence_dir, appendonly_enabled) {
             (Some(dir), true) if !server_config.disk_offload_enabled() => {
                 match WalWriter::new(shard_id, std::path::Path::new(dir)) {
@@ -1167,6 +1177,13 @@ impl super::Shard {
                         server_config.graph_dead_edge_trigger,
                         &mut autovacuum_daemon,
                         aof_pool.as_ref(),  // FIX-W1-2
+                        match wal_kv_log_mode {
+                            crate::config::WalKvLogMode::On => true,
+                            crate::config::WalKvLogMode::Off => false,
+                            crate::config::WalKvLogMode::Auto => {
+                                !appendonly_enabled || !cdc_registry.is_empty()
+                            }
+                        },
                     );
                     if hit_cap {
                         // M3: capped drain may have left a tail — re-arm immediately
@@ -1262,6 +1279,13 @@ impl super::Shard {
                         server_config.graph_dead_edge_trigger,
                         &mut autovacuum_daemon,
                         aof_pool.as_ref(),  // FIX-W1-2
+                        match wal_kv_log_mode {
+                            crate::config::WalKvLogMode::On => true,
+                            crate::config::WalKvLogMode::Off => false,
+                            crate::config::WalKvLogMode::Auto => {
+                                !appendonly_enabled || !cdc_registry.is_empty()
+                            }
+                        },
                     );
                     if hit_cap {
                         // M3: capped drain may have left a tail — re-arm immediately
@@ -1833,6 +1857,13 @@ impl super::Shard {
                     server_config.graph_dead_edge_trigger,
                     &mut autovacuum_daemon,
                     aof_pool.as_ref(), // FIX-W1-2
+                    match wal_kv_log_mode {
+                        crate::config::WalKvLogMode::On => true,
+                        crate::config::WalKvLogMode::Off => false,
+                        crate::config::WalKvLogMode::Auto => {
+                            !appendonly_enabled || !cdc_registry.is_empty()
+                        }
+                    },
                 );
                 if hit_cap {
                     // M3: the drain stopped at its per-cycle cap (or a snapshot
diff --git a/src/shard/mod.rs b/src/shard/mod.rs
index 07f4f6b7..902a332e 100644
--- a/src/shard/mod.rs
+++ b/src/shard/mod.rs
@@ -412,6 +412,7 @@ mod tests {
                 crate::shard::autovacuum::AutovacuumConfig::default(),
             ),
             None, // aof_pool — None in tests
+            true, // wal_kv_log — legacy behavior in tests
         );
 
         // Subscriber now receives pre-serialized RESP bytes
@@ -474,6 +475,7 @@ mod tests {
                 crate::shard::autovacuum::AutovacuumConfig::default(),
             ),
             None, // aof_pool — None in tests
+            true, // wal_kv_log — legacy behavior in tests
         );
     }
 
diff --git a/src/shard/spsc_handler.rs b/src/shard/spsc_handler.rs
index 250cba19..b63c635b 100644
--- a/src/shard/spsc_handler.rs
+++ b/src/shard/spsc_handler.rs
@@ -77,6 +77,10 @@ pub(crate) fn drain_spsc_shared(
     // FIX-W1-2: per-shard AOF writer pool. Passed through to handle_shard_message_shared
     // so cross-shard writes (MSET/MultiExecute) also land in the per-shard AOF files.
     aof_pool: Option<&std::sync::Arc<crate::persistence::aof::AofWriterPool>>,
+    // Whether KV command records should be logged to the per-shard WAL this
+    // drain cycle (`--wal-kv-log`; auto = false when the AOF is the recovery
+    // authority and no CDC subscriber is attached — see wal_append_and_fanout).
+    wal_kv_log: bool,
 ) -> bool {
     const MAX_DRAIN_PER_CYCLE: usize = 256;
     let mut drained = 0;
@@ -197,6 +201,7 @@ pub(crate) fn drain_spsc_shared(
                 graph_dead_edge_trigger,
                 autovacuum_daemon,
                 aof_pool, // FIX-W1-2: thread AOF pool through SPSC drain
+                wal_kv_log,
             );
         }
     }
@@ -228,6 +233,7 @@ pub(crate) fn drain_spsc_shared(
             graph_dead_edge_trigger,
             autovacuum_daemon,
             aof_pool, // FIX-W1-2: thread AOF pool through SPSC drain
+            wal_kv_log,
         );
     }
 
@@ -279,6 +285,9 @@ pub(crate) fn handle_shard_message_shared(
     // FIX-W1-2: per-shard AOF writer pool. When Some, each successful write command
     // is also routed to the owning shard's AOF file via fire-and-forget try_send_append.
     aof_pool: Option<&std::sync::Arc<crate::persistence::aof::AofWriterPool>>,
+    // Whether KV command records should be logged to the per-shard WAL
+    // (`--wal-kv-log`; see wal_append_and_fanout).
+    wal_kv_log: bool,
 ) {
     match msg {
         ShardMessage::Execute {
@@ -467,7 +476,7 @@ pub(crate) fn handle_shard_message_shared(
                 // Both require two databases simultaneously; intercept before cmd_dispatch.
                 if cmd.eq_ignore_ascii_case(b"MOVE") {
                     use crate::command::keyspace::move_cmd as ksmv;
-                    let response = match ksmv::parse_move_args(args, db_count) {
+                    let mut response = match ksmv::parse_move_args(args, db_count) {
                         Err(e) => e,
                         Ok((_key, dst_db)) if dst_db == db_idx => {
                             crate::protocol::Frame::Integer(0)
@@ -495,7 +504,8 @@ pub(crate) fn handle_shard_message_shared(
                     };
                     if matches!(response, crate::protocol::Frame::Integer(1)) {
                         let serialized = aof::serialize_command(&command);
-                        wal_append_and_fanout(
+                        let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
+                        if !wal_append_and_fanout(
                             &serialized,
                             wal_writer,
                             wal_v3_writer,
@@ -504,7 +514,13 @@ pub(crate) fn handle_shard_message_shared(
                             repl_state,
                             shard_id,
                             aof_pool, // FIX-W1-2
-                        );
+                            wal_kv_log,
+                            &mut aof_budget,
+                        ) {
+                            response = crate::protocol::Frame::Error(bytes::Bytes::from_static(
+                                AOF_APPEND_LOST_ERR,
+                            ));
+                        }
                     }
                     let _ = reply_tx.send(response);
                     return;
@@ -513,7 +529,7 @@ pub(crate) fn handle_shard_message_shared(
                 if cmd.eq_ignore_ascii_case(b"COPY") {
                     use crate::command::keyspace::move_cmd as ksmv;
                     if let Some(copy_result) = ksmv::parse_copy_db_args(args, db_idx, db_count) {
-                        let response = match copy_result {
+                        let mut response = match copy_result {
                             Err(e) => e,
                             Ok(ca) => {
                                 // Refresh expiry clock on BOTH dbs to mirror the
@@ -541,7 +557,9 @@ pub(crate) fn handle_shard_message_shared(
                         };
                         if matches!(response, crate::protocol::Frame::Integer(1)) {
                             let serialized = aof::serialize_command(&command);
-                            wal_append_and_fanout(
+                            let mut aof_budget =
+                                crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
+                            if !wal_append_and_fanout(
                                 &serialized,
                                 wal_writer,
                                 wal_v3_writer,
@@ -550,7 +568,13 @@ pub(crate) fn handle_shard_message_shared(
                                 repl_state,
                                 shard_id,
                                 aof_pool, // FIX-W1-2
-                            );
+                                wal_kv_log,
+                                &mut aof_budget,
+                            ) {
+                                response = crate::protocol::Frame::Error(
+                                    bytes::Bytes::from_static(AOF_APPEND_LOST_ERR),
+                                );
+                            }
                         }
                         let _ = reply_tx.send(response);
                         return;
@@ -579,9 +603,12 @@ pub(crate) fn handle_shard_message_shared(
                         };
 
                         // WAL append + replication fan-out for successful write commands
+                        let mut aof_ok = true;
                         if is_write && !matches!(frame, crate::protocol::Frame::Error(_)) {
                             let serialized = aof::serialize_command(&command);
-                            wal_append_and_fanout(
+                            let mut aof_budget =
+                                crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
+                            aof_ok = wal_append_and_fanout(
                                 &serialized,
                                 wal_writer,
                                 wal_v3_writer,
@@ -590,6 +617,8 @@ pub(crate) fn handle_shard_message_shared(
                                 repl_state,
                                 shard_id,
                                 aof_pool, // FIX-W1-2
+                                wal_kv_log,
+                                &mut aof_budget,
                             );
                         }
 
@@ -653,7 +682,16 @@ pub(crate) fn handle_shard_message_shared(
                             }
                         }
 
-                        frame
+                        // Fail-loud: the mutation is applied (wake/auto-index above
+                        // ran on real state), but the client must not see success
+                        // for a write whose AOF record was dropped.
+                        if aof_ok {
+                            frame
+                        } else {
+                            crate::protocol::Frame::Error(bytes::Bytes::from_static(
+                                AOF_APPEND_LOST_ERR,
+                            ))
+                        }
                     })
                 };
 
@@ -685,6 +723,10 @@ pub(crate) fn handle_shard_message_shared(
             let db_idx = db_index.min(db_count.saturating_sub(1));
             crate::shard::slice::with_shard_db(db_idx, |guard| {
                 guard.refresh_now_from_cache(cached_clock);
+                // ONE backpressure budget for the whole batch: under sustained
+                // AOF backpressure the shard thread stalls at most BOUND total,
+                // not BOUND × batch-len (review finding, PR #211).
+                let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
                 for (_key, cmd_frame) in &commands {
                     let (cmd, args) = match extract_command_static(cmd_frame) {
                         Some(pair) => pair,
@@ -708,13 +750,20 @@ pub(crate) fn handle_shard_message_shared(
                         DispatchResult::Quit(f) => f,
                     };
 
+                    let mut aof_ok = true;
                     if is_write && !matches!(frame, crate::protocol::Frame::Error(_)) {
                         // Skip the serialization alloc when the fanout would
                         // no-op (persistence + replication all off) — it was
                         // pure waste on every cross-shard write.
-                        if wal_fanout_has_work(wal_writer, wal_v3_writer, replica_txs, aof_pool) {
+                        if wal_fanout_has_work(
+                            wal_writer,
+                            wal_v3_writer,
+                            replica_txs,
+                            aof_pool,
+                            wal_kv_log,
+                        ) {
                             let serialized = aof::serialize_command(cmd_frame);
-                            wal_append_and_fanout(
+                            aof_ok = wal_append_and_fanout(
                                 &serialized,
                                 wal_writer,
                                 wal_v3_writer,
@@ -723,6 +772,8 @@ pub(crate) fn handle_shard_message_shared(
                                 repl_state,
                                 shard_id,
                                 aof_pool, // FIX-W1-2
+                                wal_kv_log,
+                                &mut aof_budget,
                             );
                         }
 
@@ -761,7 +812,13 @@ pub(crate) fn handle_shard_message_shared(
                         }
                     }
 
-                    results.push(frame);
+                    results.push(if aof_ok {
+                        frame
+                    } else {
+                        crate::protocol::Frame::Error(bytes::Bytes::from_static(
+                            AOF_APPEND_LOST_ERR,
+                        ))
+                    });
                 }
             });
             let _ = reply_tx.send(results);
@@ -779,6 +836,10 @@ pub(crate) fn handle_shard_message_shared(
             crate::shard::slice::with_shard(|s| {
                 let guard = &mut s.databases[db_idx];
                 guard.refresh_now_from_cache(cached_clock);
+                // ONE backpressure budget for the whole batch: under sustained
+                // AOF backpressure the shard thread stalls at most BOUND total,
+                // not BOUND × batch-len (review finding, PR #211).
+                let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
                 for cmd_frame in &commands {
                     let (cmd, args) = match extract_command_static(cmd_frame) {
                         Some(pair) => pair,
@@ -802,12 +863,19 @@ pub(crate) fn handle_shard_message_shared(
                         DispatchResult::Quit(f) => f,
                     };
 
+                    let mut aof_ok = true;
                     if is_write && !matches!(frame, crate::protocol::Frame::Error(_)) {
                         // See `wal_fanout_has_work` — skip the serialization alloc
                         // entirely when the fanout would no-op (persistence off).
-                        if wal_fanout_has_work(wal_writer, wal_v3_writer, replica_txs, aof_pool) {
+                        if wal_fanout_has_work(
+                            wal_writer,
+                            wal_v3_writer,
+                            replica_txs,
+                            aof_pool,
+                            wal_kv_log,
+                        ) {
                             let serialized = aof::serialize_command(cmd_frame);
-                            wal_append_and_fanout(
+                            aof_ok = wal_append_and_fanout(
                                 &serialized,
                                 wal_writer,
                                 wal_v3_writer,
@@ -826,6 +894,8 @@ pub(crate) fn handle_shard_message_shared(
                                 // The handler_monoio cross-shard AOF write is removed to avoid
                                 // the double-write that was the original reason for None.
                                 aof_pool, // FIX-C4-FOLD
+                                wal_kv_log,
+                                &mut aof_budget,
                             );
                         }
                     }
@@ -885,7 +955,13 @@ pub(crate) fn handle_shard_message_shared(
                         }
                     }
 
-                    results.push(frame);
+                    results.push(if aof_ok {
+                        frame
+                    } else {
+                        crate::protocol::Frame::Error(bytes::Bytes::from_static(
+                            AOF_APPEND_LOST_ERR,
+                        ))
+                    });
                 }
             });
             let _ = reply_tx.send(results);
@@ -927,9 +1003,12 @@ pub(crate) fn handle_shard_message_shared(
                             DispatchResult::Quit(f) => f,
                         };
 
+                        let mut aof_ok = true;
                         if is_write && !matches!(frame, crate::protocol::Frame::Error(_)) {
                             let serialized = aof::serialize_command(&command);
-                            wal_append_and_fanout(
+                            let mut aof_budget =
+                                crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
+                            aof_ok = wal_append_and_fanout(
                                 &serialized,
                                 wal_writer,
                                 wal_v3_writer,
@@ -938,6 +1017,8 @@ pub(crate) fn handle_shard_message_shared(
                                 repl_state,
                                 shard_id,
                                 aof_pool, // FIX-W1-2
+                                wal_kv_log,
+                                &mut aof_budget,
                             );
                         }
 
@@ -977,7 +1058,15 @@ pub(crate) fn handle_shard_message_shared(
                             }
                         }
 
-                        frame
+                        // Fail-loud: mutation applied, but the client must not
+                        // see success for a write whose AOF record was dropped.
+                        if aof_ok {
+                            frame
+                        } else {
+                            crate::protocol::Frame::Error(bytes::Bytes::from_static(
+                                AOF_APPEND_LOST_ERR,
+                            ))
+                        }
                     })
                 };
                 // Arc-owned slot: deref is safe, refcount keeps it alive.
@@ -995,6 +1084,10 @@ pub(crate) fn handle_shard_message_shared(
             let db_idx = db_index.min(db_count.saturating_sub(1));
             crate::shard::slice::with_shard_db(db_idx, |guard| {
                 guard.refresh_now_from_cache(cached_clock);
+                // ONE backpressure budget for the whole batch: under sustained
+                // AOF backpressure the shard thread stalls at most BOUND total,
+                // not BOUND × batch-len (review finding, PR #211).
+                let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
                 for (_key, cmd_frame) in &commands {
                     let (cmd, args) = match extract_command_static(cmd_frame) {
                         Some(pair) => pair,
@@ -1018,13 +1111,20 @@ pub(crate) fn handle_shard_message_shared(
                         DispatchResult::Quit(f) => f,
                     };
 
+                    let mut aof_ok = true;
                     if is_write && !matches!(frame, crate::protocol::Frame::Error(_)) {
                         // Skip the serialization alloc when the fanout would
                         // no-op (persistence + replication all off) — it was
                         // pure waste on every cross-shard write.
-                        if wal_fanout_has_work(wal_writer, wal_v3_writer, replica_txs, aof_pool) {
+                        if wal_fanout_has_work(
+                            wal_writer,
+                            wal_v3_writer,
+                            replica_txs,
+                            aof_pool,
+                            wal_kv_log,
+                        ) {
                             let serialized = aof::serialize_command(cmd_frame);
-                            wal_append_and_fanout(
+                            aof_ok = wal_append_and_fanout(
                                 &serialized,
                                 wal_writer,
                                 wal_v3_writer,
@@ -1033,6 +1133,8 @@ pub(crate) fn handle_shard_message_shared(
                                 repl_state,
                                 shard_id,
                                 aof_pool, // FIX-W1-2
+                                wal_kv_log,
+                                &mut aof_budget,
                             );
                         }
 
@@ -1071,7 +1173,13 @@ pub(crate) fn handle_shard_message_shared(
                         }
                     }
 
-                    results.push(frame);
+                    results.push(if aof_ok {
+                        frame
+                    } else {
+                        crate::protocol::Frame::Error(bytes::Bytes::from_static(
+                            AOF_APPEND_LOST_ERR,
+                        ))
+                    });
                 }
             });
             // Arc-owned slot: deref is safe, refcount keeps it alive.
@@ -1090,6 +1198,10 @@ pub(crate) fn handle_shard_message_shared(
             crate::shard::slice::with_shard(|s| {
                 let guard = &mut s.databases[db_idx];
                 guard.refresh_now_from_cache(cached_clock);
+                // ONE backpressure budget for the whole batch: under sustained
+                // AOF backpressure the shard thread stalls at most BOUND total,
+                // not BOUND × batch-len (review finding, PR #211).
+                let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
                 for cmd_frame in &commands {
                     let (cmd, args) = match extract_command_static(cmd_frame) {
                         Some(pair) => pair,
@@ -1113,12 +1225,19 @@ pub(crate) fn handle_shard_message_shared(
                         DispatchResult::Quit(f) => f,
                     };
 
+                    let mut aof_ok = true;
                     if is_write && !matches!(frame, crate::protocol::Frame::Error(_)) {
                         // See `wal_fanout_has_work` — skip the serialization alloc
                         // entirely when the fanout would no-op (persistence off).
-                        if wal_fanout_has_work(wal_writer, wal_v3_writer, replica_txs, aof_pool) {
+                        if wal_fanout_has_work(
+                            wal_writer,
+                            wal_v3_writer,
+                            replica_txs,
+                            aof_pool,
+                            wal_kv_log,
+                        ) {
                             let serialized = aof::serialize_command(cmd_frame);
-                            wal_append_and_fanout(
+                            aof_ok = wal_append_and_fanout(
                                 &serialized,
                                 wal_writer,
                                 wal_v3_writer,
@@ -1138,6 +1257,8 @@ pub(crate) fn handle_shard_message_shared(
                                 // cross-shard AOF write (handler_sharded/mod.rs) is removed
                                 // to avoid the double-write this None guard was preventing.
                                 aof_pool, // FIX-C4-FOLD
+                                wal_kv_log,
+                                &mut aof_budget,
                             );
                         }
                     }
@@ -1196,7 +1317,13 @@ pub(crate) fn handle_shard_message_shared(
                         }
                     }
 
-                    results.push(frame);
+                    results.push(if aof_ok {
+                        frame
+                    } else {
+                        crate::protocol::Frame::Error(bytes::Bytes::from_static(
+                            AOF_APPEND_LOST_ERR,
+                        ))
+                    });
                 }
             });
             // Arc-owned slot: deref is safe, refcount keeps it alive.
@@ -1655,7 +1782,11 @@ pub(crate) fn handle_shard_message_shared(
                 crate::protocol::Frame::BulkString(bytes::Bytes::copy_from_slice(b_str.as_bytes())),
             ]);
             let serialized = aof::serialize_command(&wal_frame);
-            wal_append_and_fanout(
+            // No per-client response frame exists here (coordinator broadcast,
+            // reply is `()`): an AOF-append loss is already counted +
+            // error!-logged inside the pool, so the result is discarded.
+            let mut aof_budget = crate::persistence::aof::AOF_SPSC_BACKPRESSURE_BOUND;
+            let _ = wal_append_and_fanout(
                 &serialized,
                 wal_writer,
                 wal_v3_writer,
@@ -1664,6 +1795,8 @@ pub(crate) fn handle_shard_message_shared(
                 repl_state,
                 shard_id,
                 aof_pool, // FIX-W1-2
+                wal_kv_log,
+                &mut aof_budget,
             );
 
             // Perform the in-place swap via ShardSlice (thread-local, no locks needed).
@@ -2431,10 +2564,27 @@ pub(crate) fn wal_fanout_has_work(
     wal_v3_writer: &Option<WalWriterV3>,
     replica_txs: &[(u64, channel::MpscSender<bytes::Bytes>)],
     aof_pool: Option<&std::sync::Arc<crate::persistence::aof::AofWriterPool>>,
+    wal_kv_log: bool,
 ) -> bool {
-    wal_writer.is_some() || wal_v3_writer.is_some() || !replica_txs.is_empty() || aof_pool.is_some()
+    (wal_kv_log && (wal_writer.is_some() || wal_v3_writer.is_some()))
+        || !replica_txs.is_empty()
+        || aof_pool.is_some()
 }
 
+/// Error frame substituted for a write's success frame when the command
+/// mutated memory but its AOF record could not be enqueued within the
+/// backpressure budget — fail-loud so the client knows durability was not
+/// achieved (review finding, PR #211).
+pub(crate) const AOF_APPEND_LOST_ERR: &[u8] =
+    b"MOONERR AOF backpressure: write applied in memory but not queued for persistence";
+
+/// Returns `false` iff the AOF append was NOT enqueued (bounded backpressure
+/// exhausted / writer gone) — callers with a per-command response frame MUST
+/// replace it with [`AOF_APPEND_LOST_ERR`]. WAL/replica fan-out remains
+/// fire-and-forget by design (replication has its own resync path). Batched
+/// callers pass ONE `aof_budget` across the whole batch so sustained
+/// backpressure stalls the shard thread at most `AOF_SPSC_BACKPRESSURE_BOUND`
+/// per drain arm, not per command.
 pub(crate) fn wal_append_and_fanout(
     data: &[u8],
     wal_writer: &mut Option<WalWriter>,
@@ -2444,22 +2594,31 @@ pub(crate) fn wal_append_and_fanout(
     repl_state: &Option<crate::replication::state::OffsetHandle>,
     shard_id: usize,
     aof_pool: Option<&std::sync::Arc<crate::persistence::aof::AofWriterPool>>,
-) {
+    wal_kv_log: bool,
+    aof_budget: &mut std::time::Duration,
+) -> bool {
     // S3.5b (2026-04-27): hot-path bypass when nothing actually has work.
     // See `wal_fanout_has_work` — callers use the same predicate to skip the
     // `aof::serialize_command` alloc entirely when the fanout would no-op.
-    if !wal_fanout_has_work(wal_writer, wal_v3_writer, replica_txs, aof_pool) {
-        return;
+    if !wal_fanout_has_work(wal_writer, wal_v3_writer, replica_txs, aof_pool, wal_kv_log) {
+        return true;
     }
-    // WAL v3 supersedes v2 — skip v2 append when v3 is active to avoid
-    // double-write overhead (2 write syscalls per SPSC drain batch).
-    if let Some(w3) = wal_v3_writer {
-        w3.append(
-            crate::persistence::wal_v3::record::WalRecordType::Command,
-            data,
-        );
-    } else if let Some(w) = wal_writer {
-        w.append(data);
+    // `wal_kv_log == false` (--wal-kv-log auto/off): the AOF is the recovery
+    // authority and no CDC subscriber is attached, so the WAL copy of this
+    // KV command would be written and then discarded by Phase-B recovery —
+    // pure write amplification (measured 2.7× file bytes at shards=4).
+    // FPI/checkpoint/feature records are unaffected (different entry points).
+    if wal_kv_log {
+        // WAL v3 supersedes v2 — skip v2 append when v3 is active to avoid
+        // double-write overhead (2 write syscalls per SPSC drain batch).
+        if let Some(w3) = wal_v3_writer {
+            w3.append(
+                crate::persistence::wal_v3::record::WalRecordType::Command,
+                data,
+            );
+        } else if let Some(w) = wal_writer {
+            w.append(data);
+        }
     }
     // 2. Replication backlog (in-memory circular buffer for partial resync).
     //
@@ -2488,14 +2647,24 @@ pub(crate) fn wal_append_and_fanout(
         }
     }
     // 5. Per-shard AOF pool (FIX-W1-2): route to the owning shard's writer.
-    // Uses fire-and-forget (`try_send_append`) because this function is sync
-    // and cannot await the fsync rendezvous. The `appendfsync=always` ack is
-    // handled by the async connection handler (handler_sharded / handler_single).
-    // LSN=0 is safe here: per-shard order is preserved by write order; the LSN
-    // is only meaningful for cross-shard TXN merge (RFC step 5, not yet wired).
+    // Bounded-blocking (`send_append_bounded_blocking`) because this function
+    // is sync and cannot await the fsync rendezvous: the fast path is the same
+    // try_send as before; only when the writer channel is FULL (writer >10k
+    // appends behind) does the shard thread block up to the bound instead of
+    // silently losing a record the client already got `+OK` for. The
+    // `appendfsync=always` ack is handled by the async connection handler
+    // (handler_sharded / handler_single). LSN=0 is safe here: per-shard order
+    // is preserved by write order; the LSN is only meaningful for cross-shard
+    // TXN merge (RFC step 5, not yet wired).
     if let Some(pool) = aof_pool {
-        pool.try_send_append(shard_id, 0, bytes::Bytes::copy_from_slice(data));
+        return pool.send_append_bounded_blocking(
+            shard_id,
+            0,
+            bytes::Bytes::copy_from_slice(data),
+            aof_budget,
+        );
     }
+    true
 }
 
 /// Extract command name and args from a Frame (static helper for SPSC dispatch).
@@ -2540,6 +2709,8 @@ mod wal_append_tests {
             &None, // no repl_state
             0,
             None, // no aof_pool
+            true, // wal_kv_log
+            &mut std::time::Duration::from_millis(5),
         );
 
         let final_end = backlog.lock().as_ref().unwrap().end_offset();
@@ -2568,6 +2739,8 @@ mod wal_append_tests {
             &None,
             0,
             None, // no aof_pool
+            true, // wal_kv_log
+            &mut std::time::Duration::from_millis(5),
         );
 
         let end = backlog.lock().as_ref().unwrap().end_offset();
@@ -2605,6 +2778,8 @@ mod wal_append_tests {
             &None,       // no repl_state
             0,           // shard_id
             Some(&pool), // aof_pool provided — bypass must NOT fire
+            true,        // wal_kv_log
+            &mut std::time::Duration::from_millis(5),
         );
 
         // The pool should have received exactly one message.
@@ -2668,6 +2843,8 @@ mod wal_append_tests {
             &None, // no repl_state
             0,     // shard_id
             None,  // PipelineBatch fix: None prevents double-write
+            true,  // wal_kv_log
+            &mut std::time::Duration::from_millis(5),
         );
         assert!(
             rx0.try_recv().is_err(),
@@ -2692,6 +2869,8 @@ mod wal_append_tests {
             &None,
             0,
             Some(&pool), // MultiExecute: pool must receive this entry
+            true,        // wal_kv_log
+            &mut std::time::Duration::from_millis(5),
         );
         let msg = rx0
             .try_recv()
@@ -2701,6 +2880,77 @@ mod wal_append_tests {
             "expected AofMessage::Append from MultiExecute arm, got unexpected variant",
         );
     }
+
+    /// `--wal-kv-log` gate (2026-07 write-path durability): with
+    /// `wal_kv_log == false` (AOF is the recovery authority, no CDC
+    /// subscriber) the KV Command record must NOT reach the WAL — startup
+    /// recovery wipes WAL-replayed state and replays the AOF, so the WAL
+    /// copy is pure 2× write amplification. The AOF pool must STILL receive
+    /// the entry (it is the surviving durability log).
+    #[test]
+    fn test_kv_append_skipped_when_wal_kv_log_false() {
+        use crate::persistence::aof::{AofMessage, AofWriterPool, FsyncPolicy};
+        use crate::runtime::channel::mpsc_bounded;
+
+        let tmp = tempfile::tempdir().unwrap();
+        let wal_dir = tmp.path().join("wal");
+        let mut w3 = Some(WalWriterV3::new(0, &wal_dir, 16 * 1024 * 1024).unwrap());
+        w3.as_mut().unwrap().flush_sync().unwrap();
+        let seg = wal_dir.join("000000000001.wal");
+        let base_len = std::fs::metadata(&seg).unwrap().len();
+
+        let (tx, rx) = mpsc_bounded::<AofMessage>(16);
+        let pool = AofWriterPool::top_level_with_policy(
+            tx,
+            FsyncPolicy::EverySec,
+            std::time::Duration::ZERO,
+        );
+        let backlog: SharedBacklog = std::sync::Arc::new(parking_lot::Mutex::new(None));
+        let cmd = b"*3\r\n$3\r\nSET\r\n$1\r\nk\r\n$1\r\nv\r\n";
+
+        wal_append_and_fanout(
+            cmd,
+            &mut None,
+            &mut w3,
+            &backlog,
+            &[],
+            &None,
+            0,
+            Some(&pool),
+            false, // wal_kv_log: AOF authoritative, no CDC consumer
+            &mut std::time::Duration::from_millis(5),
+        );
+        w3.as_mut().unwrap().flush_sync().unwrap();
+        assert_eq!(
+            std::fs::metadata(&seg).unwrap().len(),
+            base_len,
+            "KV Command record must be skipped when wal_kv_log is false (double-write gate)"
+        );
+        assert!(
+            matches!(rx.try_recv(), Ok(AofMessage::Append { .. })),
+            "the AOF pool must still receive the entry — it is the surviving KV log"
+        );
+
+        // Control: with wal_kv_log == true (CDC attached / --wal-kv-log on)
+        // the record must land in the WAL as before.
+        wal_append_and_fanout(
+            cmd,
+            &mut None,
+            &mut w3,
+            &backlog,
+            &[],
+            &None,
+            0,
+            Some(&pool),
+            true, // wal_kv_log
+            &mut std::time::Duration::from_millis(5),
+        );
+        w3.as_mut().unwrap().flush_sync().unwrap();
+        assert!(
+            std::fs::metadata(&seg).unwrap().len() > base_len,
+            "with wal_kv_log true the KV record must be logged to the WAL"
+        );
+    }
 }
 
 #[cfg(test)]
@@ -2774,6 +3024,7 @@ mod drain_cap_tests {
             0.2,
             &mut autovacuum,
             None,
+            true, // wal_kv_log
         );
         assert!(
             hit_cap,
@@ -2804,6 +3055,7 @@ mod drain_cap_tests {
             0.2,
             &mut autovacuum,
             None,
+            true, // wal_kv_log
         );
         assert!(
             !hit_cap2,
diff --git a/tests/ft_search_multi_shard_as_of.rs b/tests/ft_search_multi_shard_as_of.rs
index 71cc301c..41b446cc 100644
--- a/tests/ft_search_multi_shard_as_of.rs
+++ b/tests/ft_search_multi_shard_as_of.rs
@@ -98,6 +98,7 @@ fn build_config(port: u16, num_shards: usize) -> ServerConfig {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
diff --git a/tests/ft_search_temporal_parity.rs b/tests/ft_search_temporal_parity.rs
index d327adaa..50a5a5c7 100644
--- a/tests/ft_search_temporal_parity.rs
+++ b/tests/ft_search_temporal_parity.rs
@@ -110,6 +110,7 @@ fn build_config(port: u16, num_shards: usize) -> ServerConfig {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
diff --git a/tests/integration.rs b/tests/integration.rs
index 25cbf001..16135180 100644
--- a/tests/integration.rs
+++ b/tests/integration.rs
@@ -84,6 +84,7 @@ async fn start_server() -> (u16, CancellationToken) {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
@@ -187,6 +188,7 @@ async fn start_server_with_pass(password: &str) -> (u16, CancellationToken) {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
@@ -1380,6 +1382,7 @@ async fn start_server_with_persistence(
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
@@ -2267,6 +2270,7 @@ async fn start_server_with_maxmemory(maxmemory: usize, policy: &str) -> (u16, Ca
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
@@ -2681,6 +2685,7 @@ async fn start_sharded_server(num_shards: usize) -> (u16, CancellationToken) {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
@@ -3896,6 +3901,7 @@ async fn start_cluster_server() -> (u16, CancellationToken) {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
@@ -4564,6 +4570,7 @@ async fn start_server_with_aclfile(acl_path: &str) -> (u16, CancellationToken) {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
diff --git a/tests/kill_snapshot.rs b/tests/kill_snapshot.rs
index ac50bc92..590cdbcc 100644
--- a/tests/kill_snapshot.rs
+++ b/tests/kill_snapshot.rs
@@ -80,6 +80,7 @@ fn base_config(port: u16, num_shards: usize) -> ServerConfig {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
diff --git a/tests/mq_integration.rs b/tests/mq_integration.rs
index 24034449..f289f70c 100644
--- a/tests/mq_integration.rs
+++ b/tests/mq_integration.rs
@@ -100,6 +100,7 @@ async fn start_mq_server(num_shards: usize) -> (u16, CancellationToken) {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
diff --git a/tests/replication_test.rs b/tests/replication_test.rs
index 093b2d5a..f9377f30 100644
--- a/tests/replication_test.rs
+++ b/tests/replication_test.rs
@@ -82,6 +82,7 @@ async fn start_server() -> (u16, CancellationToken) {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
diff --git a/tests/txn_ft_search_snapshot.rs b/tests/txn_ft_search_snapshot.rs
index c81b9c3e..bec00fc6 100644
--- a/tests/txn_ft_search_snapshot.rs
+++ b/tests/txn_ft_search_snapshot.rs
@@ -98,6 +98,7 @@ fn build_config(port: u16, num_shards: usize) -> ServerConfig {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
diff --git a/tests/txn_kv_wiring.rs b/tests/txn_kv_wiring.rs
index 6569bf8b..64ebb943 100644
--- a/tests/txn_kv_wiring.rs
+++ b/tests/txn_kv_wiring.rs
@@ -105,6 +105,7 @@ async fn start_txn_server(num_shards: usize, persistence_dir: &str) -> (u16, Can
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
diff --git a/tests/vacuum_commands.rs b/tests/vacuum_commands.rs
index e31c1f19..cc766fd7 100644
--- a/tests/vacuum_commands.rs
+++ b/tests/vacuum_commands.rs
@@ -87,6 +87,7 @@ fn base_config(port: u16) -> ServerConfig {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
diff --git a/tests/workspace_integration.rs b/tests/workspace_integration.rs
index dda2a649..39c4e38b 100644
--- a/tests/workspace_integration.rs
+++ b/tests/workspace_integration.rs
@@ -93,6 +93,7 @@ async fn start_workspace_server(num_shards: usize) -> (u16, CancellationToken) {
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,
@@ -320,6 +321,7 @@ async fn start_workspace_server_with_auth(
         console_rate_limit: 1000.0,
         console_rate_burst: 2000.0,
         wal_max_checkpoint_lag_ms: 10_000,
+        wal_kv_log: "auto".to_string(),
         recovery_target_lsn: None,
         recovery_target_time: None,
         manifest_tombstone_retain_epochs: 2,