diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6b7a140d4..a46c95306 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,7 +6,46 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Performance
+- **Disk-offload: shard event loop no longer pays manifest-commit fsyncs
+  (task #59 levers 1+2).** Under spill load, `apply_completion_vec` ran one
+  DURABLE manifest commit (up to 2 fsyncs) per spilled file **on the shard
+  event-loop thread** — strace-measured at 1.0–2.1s cumulative block time per
+  8s flood window per shard (single fsyncs up to 1.0s), stalling every
+  connection on the shard. `ShardManifest` now splits into loop-owned RAM
+  state + a per-shard `manifest-sync-{id}` thread owning the file handle:
+  spill-completion commits are deferred, coalescing snapshot sends (correct
+  because that path only runs under `--appendonly yes`, where AOF replay +
+  the orphan sweep reconstruct anything a lost manifest commit recorded);
+  `commit()` keeps its exact blocking durability for the checkpoint protocol
+  and the no-AOF durable batch spill path. The crash-safety write order
+  (overflow pages sync'd before the root that references them; dual-slot
+  atomic root flip) is preserved verbatim. Additionally the spill writer now
+  yields a 4ms quantum after each durable batch flush while a cold read is
+  in flight (`COLD_READS_INFLIGHT` RAII signal from both the async pool and
+  the synchronous MGET/MULTI/Lua read paths), never pacing when the request
+  backlog is deep (pacing must not push eviction into `-OOM`) or at
+  shutdown.
+
 ### Fixed
+- **BGSAVE issued during shard startup could hang forever (pre-existing,
+  reachable in v0.8.0 and earlier).** The listener answers clients as soon as
+  the fastest shard's event loop is up, but each shard seeded its snapshot
+  epoch cursor from the trigger watch channel's *current* value at loop
+  start — a BGSAVE (or auto-save) broadcast while a slower shard was still
+  initializing was silently swallowed by that shard: its snapshot never ran,
+  `BGSAVE_SHARDS_REMAINING` never reached zero, and `rdb_bgsave_in_progress`
+  stuck at `1` with every later BGSAVE refused as already-in-progress until
+  restart. Found as a ~15%/run crash-matrix flake under full-suite CPU
+  contention; reproduced deterministically with a delayed shard start (new
+  test-only `MOON_TEST_SLOW_SHARD_START_MS` fault injection) and pinned by
+  `tests/bgsave_startup_race.rs`. The cursor now starts at 0 (epochs are
+  per-process), so a trigger that arrives during startup is honored on the
+  shard's first tick. The same test exposed a second pre-existing gap: the
+  **tokio** shard loop never called `bgsave_shard_done` after finalizing its
+  snapshot (only the monoio arm did), so under `runtime-tokio` every sharded
+  BGSAVE left `rdb_bgsave_in_progress` stuck at 1 — now decremented in both
+  arms.
 - **Storage: recovery panic `double NeedsSplit after split_segment` in
   `DashTable::insert_or_update`.** The insert-or-update path split an overflowing
   segment exactly once and declared a second `NeedsSplit` unreachable — false under
@@ -17,6 +56,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   the checkpoint was quarantined (code identical back to v0.6.0). Now split-retries
   in a loop, mirroring `insert`'s recursion; regression test brute-forces 56 keys
   sharing the top 12 xxh64 bits to force the double split.
+- **Test harness: five latently broken integration suites unmasked by the
+  task #59 gate battery.** `cmd_flush_dbsize_debug_memory` still passed the
+  `--persistence-dir` flag removed in June (server exited 2 before accepting;
+  CI stayed green only because the suite self-skips without a release
+  binary); it and `memory_doctor_response` ignored `MOON_BIN` via hardcoded
+  `target/release/moon` finders (stale host Mach-O via OrbStack proxy → 30s
+  spawn timeouts). `mem_watchdog`, `oom_bypass_closure`, and `spsc_two_db`
+  now pass `--disk-free-min-pct 0` so a near-full dev volume's diskfull
+  write pause can't shadow the memory-guard/routing behavior they assert.
 
 ## [0.8.0] — 2026-07-16
 
diff --git a/src/persistence/manifest.rs b/src/persistence/manifest.rs
index 1f19cb504..ec7f9ddbd 100644
--- a/src/persistence/manifest.rs
+++ b/src/persistence/manifest.rs
@@ -281,27 +281,46 @@ pub struct ManifestRoot {
 /// process restart implies no in-flight readers holding old snapshot views.
 /// After the configured retention period both tombstone entries are
 /// physically removed from `active_root.entries` by `gc_tombstones`.
+/// Test-only injected delay inside [`ManifestIo::persist`], mirroring the
+/// `cold_read::TEST_INJECT_DELAY_MS` pattern: simulates a slow/contended
+/// device so tests can prove which thread pays for the sync.
+#[cfg(test)]
+pub(crate) static TEST_INJECT_SYNC_DELAY_MS: std::sync::atomic::AtomicU64 =
+    std::sync::atomic::AtomicU64::new(0);
+
+/// Test-only count of [`ManifestIo::persist`] executions (for coalescing
+/// assertions).
+#[cfg(test)]
+pub(crate) static TEST_PERSIST_COUNT: std::sync::atomic::AtomicU64 =
+    std::sync::atomic::AtomicU64::new(0);
+
+/// Serializes tests that use the injected sync delay / persist counter —
+/// `cargo test`'s default parallelism otherwise lets one test's injected
+/// delay leak into an unrelated concurrently-running test (same pattern as
+/// `cold_read::TEST_DELAY_LOCK`).
+#[cfg(test)]
+pub(crate) static TEST_SYNC_KNOB_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+/// The file-I/O half of the manifest: the handle, the dual-slot alternation
+/// state, and the reopen bookkeeping that `persist`/`compact` maintain.
+///
+/// Owned inline by [`ShardManifest`] (synchronous commits on the caller
+/// thread — the default everywhere outside the shard event loop), or moved
+/// onto the per-shard manifest-sync thread by
+/// [`ShardManifest::enable_deferred_sync`] so commit fsyncs never run on the
+/// shard event loop (task #59).
 #[derive(Debug)]
-pub struct ShardManifest {
+pub(crate) struct ManifestIo {
     /// File handle opened for read/write.
     file: std::fs::File,
     /// Path to the manifest file on disk.
     path: PathBuf,
-    /// Currently active root (the last successfully committed state).
-    active_root: ManifestRoot,
     /// Which slot is currently active: 0 = Root A (offset 0), 1 = Root B (offset 4096).
     active_slot: u8,
-    /// In-memory registry of tombstoned files: file_id → (tombstone_epoch, tombstoned_at).
-    ///
-    /// `tombstone_epoch` is the `active_root.epoch` at the moment `remove_file` was called
-    /// (i.e. the epoch of the root that will be committed next, which equals current + 1
-    /// after the commit flip — we record the pre-commit value so age = current - tombstone_epoch).
-    /// `tombstoned_at` is a monotonic `Instant` for wall-clock retention.
-    tombstone_registry: HashMap<u64, (u64, Instant)>,
     /// Set when `compact()` rewrote+renamed the manifest durably but then failed
     /// to reopen `self.file` against the new inode. The old handle now refers to
     /// the pre-rename (orphaned) inode, so writing through it would silently
-    /// discard every subsequent commit. While set, `commit()` reattaches to
+    /// discard every subsequent commit. While set, `persist()` reattaches to
     /// `self.path` BEFORE writing (or fails loudly if that reopen also fails).
     needs_reopen: bool,
     /// Test-only fault injection: when true, `compact()` simulates the
@@ -312,6 +331,37 @@ pub struct ShardManifest {
     fail_compact_reopen: bool,
 }
 
+/// Where a [`ShardManifest`]'s file I/O runs.
+#[derive(Debug)]
+enum IoBackend {
+    /// Commits persist synchronously on the calling thread (the historical
+    /// behavior; used by recovery, tests, and every non-shard-loop caller).
+    Inline(ManifestIo),
+    /// Commits are shipped to the per-shard manifest-sync thread; deferred
+    /// commits return immediately, durable commits block on an ack.
+    Deferred(crate::persistence::manifest_sync::ManifestSyncAgent),
+    /// Transient placeholder during backend swaps, and the terminal state if
+    /// the sync thread died holding the file handle. Commits fail loudly.
+    Poisoned,
+}
+
+#[derive(Debug)]
+pub struct ShardManifest {
+    /// File-I/O backend (see [`IoBackend`]).
+    io: IoBackend,
+    /// Path to the manifest file on disk.
+    path: PathBuf,
+    /// Currently active root (the last successfully committed state).
+    active_root: ManifestRoot,
+    /// In-memory registry of tombstoned files: file_id → (tombstone_epoch, tombstoned_at).
+    ///
+    /// `tombstone_epoch` is the `active_root.epoch` at the moment `remove_file` was called
+    /// (i.e. the epoch of the root that will be committed next, which equals current + 1
+    /// after the commit flip — we record the pre-commit value so age = current - tombstone_epoch).
+    /// `tombstoned_at` is a monotonic `Instant` for wall-clock retention.
+    tombstone_registry: HashMap<u64, (u64, Instant)>,
+}
+
 impl ShardManifest {
     /// Create a new manifest file with an empty Root A at epoch 1.
     ///
@@ -349,14 +399,17 @@ impl ShardManifest {
         }
 
         Ok(Self {
-            file,
+            io: IoBackend::Inline(ManifestIo {
+                file,
+                path: path.to_path_buf(),
+                active_slot: 0,
+                needs_reopen: false,
+                #[cfg(test)]
+                fail_compact_reopen: false,
+            }),
             path: path.to_path_buf(),
             active_root: root,
-            active_slot: 0,
             tombstone_registry: HashMap::new(),
-            needs_reopen: false,
-            #[cfg(test)]
-            fail_compact_reopen: false,
         })
     }
 
@@ -420,14 +473,17 @@ impl ShardManifest {
         }
 
         Ok(Self {
-            file,
+            io: IoBackend::Inline(ManifestIo {
+                file,
+                path: path.to_path_buf(),
+                active_slot,
+                needs_reopen: false,
+                #[cfg(test)]
+                fail_compact_reopen: false,
+            }),
             path: path.to_path_buf(),
             active_root,
-            active_slot,
             tombstone_registry,
-            needs_reopen: false,
-            #[cfg(test)]
-            fail_compact_reopen: false,
         })
     }
 
@@ -438,87 +494,91 @@ impl ShardManifest {
     /// 3. `sync_data()` — this is the atomic commit point
     /// 4. Flip active_slot
     pub fn commit(&mut self) -> std::io::Result<()> {
-        // A prior compaction renamed a fresh manifest into place durably but then
-        // failed to reopen our handle, so `self.file` still points at the
-        // pre-rename (orphaned) inode. Reattach to the live file BEFORE any write;
-        // writing through the stale handle would silently discard this and every
-        // later commit. If the reopen still fails, surface the error (this commit
-        // genuinely cannot be persisted) rather than losing data silently.
-        if self.needs_reopen {
-            self.file = std::fs::OpenOptions::new()
-                .read(true)
-                .write(true)
-                .open(&self.path)?;
-            self.active_slot = 0;
-            self.needs_reopen = false;
+        let snapshot = self.snapshot_for_commit();
+        match &mut self.io {
+            IoBackend::Inline(io) => io.persist(snapshot),
+            IoBackend::Deferred(agent) => agent.commit_durable(snapshot),
+            IoBackend::Poisoned => Err(std::io::Error::other(
+                "manifest io backend lost (sync thread died holding the handle)",
+            )),
         }
-        self.active_root.epoch += 1;
-        let total = self.active_root.entries.len();
-        self.active_root.file_count = total as u32;
-
-        // Entries beyond the inline cap go into append-only overflow pages.
-        // ORDER IS THE CRASH-SAFETY INVARIANT: overflow pages are appended at
-        // EOF and `sync_data`'d BEFORE the root that references them. Because
-        // the run is appended (never overwriting the currently-active slot's
-        // older overflow), a crash before the root's atomic commit leaves the
-        // previously-committed state — its root AND its overflow run — fully
-        // intact, so `open()` falls back to it. Partial loss, never corruption.
-        let inline_count = total.min(MAX_INLINE_ENTRIES);
-        let overflow = &self.active_root.entries[inline_count..];
-        let npages = overflow.len().div_ceil(ENTRIES_PER_OVERFLOW_PAGE);
-
-        let overflow_start_page: u32 = if npages > 0 {
-            let eof = self.file.seek(SeekFrom::End(0))?;
-            // The manifest is always a whole number of 4 KB pages.
-            debug_assert_eq!(eof % PAGE_4K as u64, 0);
-            let start_page = (eof / PAGE_4K as u64) as u32;
-            let mut buf = vec![0u8; npages * PAGE_4K];
-            for (pi, chunk) in overflow.chunks(ENTRIES_PER_OVERFLOW_PAGE).enumerate() {
-                Self::serialize_overflow_page(chunk, &mut buf[pi * PAGE_4K..(pi + 1) * PAGE_4K]);
-            }
-            self.file.seek(SeekFrom::Start(eof))?;
-            self.file.write_all(&buf)?;
-            self.file.sync_data()?; // overflow durable BEFORE the root points at it
-            start_page
-        } else {
-            0
-        };
-
-        self.active_root.entry_page_count = npages as u32;
-
-        let mut page = [0u8; PAGE_4K];
-        Self::serialize_root(&self.active_root, overflow_start_page, &mut page);
-
-        // Write to the inactive slot
-        let write_offset = if self.active_slot == 0 {
-            ROOT_B_OFFSET
-        } else {
-            ROOT_A_OFFSET
-        };
+    }
 
-        self.file.seek(SeekFrom::Start(write_offset))?;
-        self.file.write_all(&page)?;
-        self.file.sync_data()?; // ATOMIC COMMIT POINT
+    /// Commit without waiting for durability: the snapshot is handed to the
+    /// manifest-sync thread and this returns immediately (task #59). Falls
+    /// back to a synchronous inline persist when deferred sync is not
+    /// enabled, so semantics degrade to `commit()` — never to a silent no-op.
+    ///
+    /// ONLY correct for callers whose durability backstop is elsewhere (the
+    /// async-spill completion path runs exclusively under `--appendonly yes`,
+    /// where AOF replay + the orphan sweep reconstruct anything a lost
+    /// manifest commit would have recorded). Paths where the manifest IS the
+    /// durability record must call `commit()`.
+    pub fn commit_deferred(&mut self) -> std::io::Result<()> {
+        let snapshot = self.snapshot_for_commit();
+        match &mut self.io {
+            IoBackend::Inline(io) => io.persist(snapshot),
+            IoBackend::Deferred(agent) => agent.commit_deferred(snapshot),
+            IoBackend::Poisoned => Err(std::io::Error::other(
+                "manifest io backend lost (sync thread died holding the handle)",
+            )),
+        }
+    }
 
-        // Flip active slot
-        self.active_slot = if self.active_slot == 0 { 1 } else { 0 };
+    /// Advance the epoch and clone the root for a commit. Each snapshot is a
+    /// COMPLETE manifest state: the sync agent may coalesce a run of queued
+    /// snapshots down to the newest one with no loss.
+    fn snapshot_for_commit(&mut self) -> ManifestRoot {
+        self.active_root.epoch += 1;
+        self.active_root.file_count = self.active_root.entries.len() as u32;
+        self.active_root.clone()
+    }
 
-        // Bound append-only growth: when the file is mostly dead overflow from
-        // superseded commits, rewrite it compactly. Best-effort — the commit is
-        // already durable, so a compaction failure must not fail the commit.
-        if npages > 0 {
-            if let Ok(file_len) = self.file.seek(SeekFrom::End(0)) {
-                let live_pages = 2 + npages as u64;
-                let live_bytes = live_pages * PAGE_4K as u64;
-                if file_len > live_bytes.saturating_mul(4) && file_len > 16 * PAGE_4K as u64 {
-                    if let Err(e) = self.compact() {
-                        tracing::warn!(error = %e, "manifest compaction failed (commit already durable)");
+    /// Move the file-I/O half onto a dedicated `manifest-sync-{shard_id}`
+    /// thread. From here on, `commit()` blocks on an ack from that thread
+    /// (unchanged durability) while `commit_deferred()` returns immediately.
+    /// Idempotent.
+    pub fn enable_deferred_sync(&mut self, shard_id: usize) {
+        let cur = std::mem::replace(&mut self.io, IoBackend::Poisoned);
+        self.io = match cur {
+            IoBackend::Inline(io) => {
+                match crate::persistence::manifest_sync::ManifestSyncAgent::spawn(io, shard_id) {
+                    Ok(agent) => IoBackend::Deferred(agent),
+                    // Degrade, don't destroy: a failed thread spawn keeps the
+                    // working inline backend (commits fsync on the shard
+                    // thread again — slow, but durable and loud about it).
+                    Err((io, e)) => {
+                        tracing::error!(
+                            shard_id, error = %e,
+                            "manifest-sync: thread spawn failed — staying inline \
+                             (manifest commits fsync on the shard thread)"
+                        );
+                        IoBackend::Inline(io)
                     }
                 }
             }
-        }
+            other => other,
+        };
+    }
 
-        Ok(())
+    /// Flush every pending deferred commit, stop the sync thread, and take
+    /// the file handle back inline so later commits (e.g. during shutdown
+    /// finalization) persist synchronously again. Idempotent.
+    pub fn shutdown_deferred(&mut self) {
+        let cur = std::mem::replace(&mut self.io, IoBackend::Poisoned);
+        self.io = match cur {
+            IoBackend::Deferred(agent) => match agent.shutdown() {
+                Some(io) => IoBackend::Inline(io),
+                None => {
+                    tracing::error!(
+                        "manifest-sync thread lost the manifest file handle; \
+                         further commits on this shard will fail"
+                    );
+                    IoBackend::Poisoned
+                }
+            },
+            other => other,
+        };
     }
 
     /// Add a file entry to the manifest (in-memory only until commit).
@@ -663,8 +723,38 @@ impl ShardManifest {
     }
 
     /// Return the currently active slot (0 = Root A, 1 = Root B).
+    ///
+    /// Only meaningful while the io backend is inline (tests, recovery); the
+    /// slot lives on the sync thread once deferred sync is enabled.
     pub fn active_slot(&self) -> u8 {
-        self.active_slot
+        match &self.io {
+            IoBackend::Inline(io) => io.active_slot,
+            IoBackend::Deferred(_) | IoBackend::Poisoned => 0,
+        }
+    }
+
+    /// Test-only: arm/disarm the compact-reopen fault injection on the inline io.
+    #[cfg(test)]
+    pub(crate) fn set_fail_compact_reopen(&mut self, armed: bool) {
+        if let IoBackend::Inline(io) = &mut self.io {
+            io.fail_compact_reopen = armed;
+        }
+    }
+
+    /// Test-only: whether the inline io is flagged for reattach.
+    #[cfg(test)]
+    pub(crate) fn needs_reopen(&self) -> bool {
+        matches!(&self.io, IoBackend::Inline(io) if io.needs_reopen)
+    }
+
+    /// Test-only: run a compaction of the current state through the inline io.
+    #[cfg(test)]
+    pub(crate) fn compact_for_test(&mut self) -> std::io::Result<()> {
+        let mut root = self.active_root.clone();
+        match &mut self.io {
+            IoBackend::Inline(io) => io.compact(&mut root),
+            _ => Err(std::io::Error::other("compact_for_test requires inline io")),
+        }
     }
 
     /// Return the path to the manifest file.
@@ -816,33 +906,129 @@ impl ShardManifest {
         }
         Some(root)
     }
+}
+
+impl ManifestIo {
+    /// Persist a complete root snapshot: append+sync its overflow run, write
+    /// the root to the inactive slot, sync (the atomic commit point), flip
+    /// slots, then opportunistically compact. This is the historical body of
+    /// `ShardManifest::commit()`, made independent of the in-RAM manifest
+    /// state so it can run on the manifest-sync thread (task #59).
+    pub(crate) fn persist(&mut self, mut root: ManifestRoot) -> std::io::Result<()> {
+        #[cfg(test)]
+        {
+            TEST_PERSIST_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            let ms = TEST_INJECT_SYNC_DELAY_MS.load(std::sync::atomic::Ordering::SeqCst);
+            if ms > 0 {
+                std::thread::sleep(std::time::Duration::from_millis(ms));
+            }
+        }
+        // A prior compaction renamed a fresh manifest into place durably but then
+        // failed to reopen our handle, so `self.file` still points at the
+        // pre-rename (orphaned) inode. Reattach to the live file BEFORE any write;
+        // writing through the stale handle would silently discard this and every
+        // later commit. If the reopen still fails, surface the error (this commit
+        // genuinely cannot be persisted) rather than losing data silently.
+        if self.needs_reopen {
+            self.file = std::fs::OpenOptions::new()
+                .read(true)
+                .write(true)
+                .open(&self.path)?;
+            self.active_slot = 0;
+            self.needs_reopen = false;
+        }
+        let total = root.entries.len();
+
+        // Entries beyond the inline cap go into append-only overflow pages.
+        // ORDER IS THE CRASH-SAFETY INVARIANT: overflow pages are appended at
+        // EOF and `sync_data`'d BEFORE the root that references them. Because
+        // the run is appended (never overwriting the currently-active slot's
+        // older overflow), a crash before the root's atomic commit leaves the
+        // previously-committed state — its root AND its overflow run — fully
+        // intact, so `open()` falls back to it. Partial loss, never corruption.
+        let inline_count = total.min(MAX_INLINE_ENTRIES);
+        let overflow = &root.entries[inline_count..];
+        let npages = overflow.len().div_ceil(ENTRIES_PER_OVERFLOW_PAGE);
+
+        let overflow_start_page: u32 = if npages > 0 {
+            let eof = self.file.seek(SeekFrom::End(0))?;
+            // The manifest is always a whole number of 4 KB pages.
+            debug_assert_eq!(eof % PAGE_4K as u64, 0);
+            let start_page = (eof / PAGE_4K as u64) as u32;
+            let mut buf = vec![0u8; npages * PAGE_4K];
+            for (pi, chunk) in overflow.chunks(ENTRIES_PER_OVERFLOW_PAGE).enumerate() {
+                ShardManifest::serialize_overflow_page(
+                    chunk,
+                    &mut buf[pi * PAGE_4K..(pi + 1) * PAGE_4K],
+                );
+            }
+            self.file.seek(SeekFrom::Start(eof))?;
+            self.file.write_all(&buf)?;
+            self.file.sync_data()?; // overflow durable BEFORE the root points at it
+            start_page
+        } else {
+            0
+        };
+
+        root.entry_page_count = npages as u32;
+
+        let mut page = [0u8; PAGE_4K];
+        ShardManifest::serialize_root(&root, overflow_start_page, &mut page);
+
+        // Write to the inactive slot
+        let write_offset = if self.active_slot == 0 {
+            ROOT_B_OFFSET
+        } else {
+            ROOT_A_OFFSET
+        };
+
+        self.file.seek(SeekFrom::Start(write_offset))?;
+        self.file.write_all(&page)?;
+        self.file.sync_data()?; // ATOMIC COMMIT POINT
+
+        // Flip active slot
+        self.active_slot = if self.active_slot == 0 { 1 } else { 0 };
+
+        // Bound append-only growth: when the file is mostly dead overflow from
+        // superseded commits, rewrite it compactly. Best-effort — the commit is
+        // already durable, so a compaction failure must not fail the commit.
+        if npages > 0 {
+            if let Ok(file_len) = self.file.seek(SeekFrom::End(0)) {
+                let live_pages = 2 + npages as u64;
+                let live_bytes = live_pages * PAGE_4K as u64;
+                if file_len > live_bytes.saturating_mul(4) && file_len > 16 * PAGE_4K as u64 {
+                    if let Err(e) = self.compact(&mut root) {
+                        tracing::warn!(error = %e, "manifest compaction failed (commit already durable)");
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
 
     /// Rewrite the manifest compactly, reclaiming dead overflow pages from
     /// superseded commits. Layout becomes `[Root A][Root B][fresh overflow]`
     /// with the active root written to BOTH slots (same epoch) so either is a
     /// valid recovery target. Crash-safe via temp-file + atomic rename: the
     /// live manifest stays valid until the rename completes.
-    fn compact(&mut self) -> std::io::Result<()> {
-        let total = self.active_root.entries.len();
+    fn compact(&mut self, root: &mut ManifestRoot) -> std::io::Result<()> {
+        let total = root.entries.len();
         let inline_count = total.min(MAX_INLINE_ENTRIES);
-        let overflow = &self.active_root.entries[inline_count..];
+        let overflow = &root.entries[inline_count..];
         let npages = overflow.len().div_ceil(ENTRIES_PER_OVERFLOW_PAGE);
 
         // Overflow starts immediately after the two root pages.
         let overflow_start_page: u32 = if npages > 0 { 2 } else { 0 };
-        self.active_root.entry_page_count = npages as u32;
+        root.entry_page_count = npages as u32;
 
         let mut buf = vec![0u8; (2 + npages) * PAGE_4K];
         for (pi, chunk) in overflow.chunks(ENTRIES_PER_OVERFLOW_PAGE).enumerate() {
             let o = (2 + pi) * PAGE_4K;
-            Self::serialize_overflow_page(chunk, &mut buf[o..o + PAGE_4K]);
+            ShardManifest::serialize_overflow_page(chunk, &mut buf[o..o + PAGE_4K]);
         }
-        Self::serialize_root(&self.active_root, overflow_start_page, &mut buf[0..PAGE_4K]);
-        Self::serialize_root(
-            &self.active_root,
-            overflow_start_page,
-            &mut buf[PAGE_4K..2 * PAGE_4K],
-        );
+        ShardManifest::serialize_root(root, overflow_start_page, &mut buf[0..PAGE_4K]);
+        ShardManifest::serialize_root(root, overflow_start_page, &mut buf[PAGE_4K..2 * PAGE_4K]);
 
         let tmp = self.path.with_extension("manifest.compact.tmp");
         std::fs::write(&tmp, &buf)?;
@@ -897,7 +1083,9 @@ impl ShardManifest {
             }
         }
     }
+}
 
+impl ShardManifest {
     /// Try to parse a root page from a 4KB buffer.
     ///
     /// Returns `None` if magic/type mismatch or CRC32C fails. Recognizes both
@@ -1243,21 +1431,21 @@ mod tests {
 
         // Simulate the race: compaction rewrote + renamed durably (data safe on
         // disk) but then failed to reopen the handle.
-        m.fail_compact_reopen = true;
-        let err = m.compact().unwrap_err();
+        m.set_fail_compact_reopen(true);
+        let err = m.compact_for_test().unwrap_err();
         assert_eq!(err.kind(), std::io::ErrorKind::Other);
         assert!(
-            m.needs_reopen,
+            m.needs_reopen(),
             "compact() must flag needs_reopen when it loses the file handle"
         );
-        m.fail_compact_reopen = false;
+        m.set_fail_compact_reopen(false);
 
         // The next commit MUST reattach to the real manifest first. Without the
         // guard this write lands in the orphaned inode and is lost on recovery.
         m.add_file(make_entry(999));
         m.commit().unwrap();
         assert!(
-            !m.needs_reopen,
+            !m.needs_reopen(),
             "commit() must clear needs_reopen after reattaching to self.path"
         );
 
diff --git a/src/persistence/manifest_sync.rs b/src/persistence/manifest_sync.rs
new file mode 100644
index 000000000..1d64c783e
--- /dev/null
+++ b/src/persistence/manifest_sync.rs
@@ -0,0 +1,404 @@
+//! Off-loop manifest commit agent (task #59).
+//!
+//! Under spill load, `ShardManifest::commit()` fsyncs used to run on the
+//! shard event-loop thread — measured at 163–167 fsync calls per 8s window,
+//! 1.0–2.1s cumulative block time, single calls up to 1.0s on a contended
+//! device (2026-07-16 strace decomposition, `tmp/t59-decompose.sh`). Every
+//! connection on the shard stalls for each one.
+//!
+//! Under `--appendonly yes` (the only config where the async-spill completion
+//! path runs), the AOF is the durability backstop for spill placement: a
+//! crash that loses a manifest commit costs an orphan-file sweep plus AOF
+//! replay of the affected keys — never data loss. The commit therefore does
+//! not need to block the event loop. This agent owns the manifest's file
+//! handle ([`ManifestIo`]) on a dedicated per-shard thread and performs the
+//! exact same ordered overflow→root→fsync sequence off-loop.
+//!
+//! Two commit flavors, both handed over via a single latest-snapshot slot
+//! (never a queue — a bounded queue behind a stalled fsync would block the
+//! shard thread, re-creating the very stall this agent removes):
+//! - **Deferred** (`commit_deferred`): fire-and-forget slot replace + wake;
+//!   the spill-completion path uses this. Consecutive snapshots coalesce by
+//!   construction — each is a complete root, so only the newest needs to
+//!   reach disk.
+//! - **Durable** (`commit_durable`): slot replace + wake, then block on ack.
+//!   Paths where the manifest IS the durability record (checkpoint protocol,
+//!   no-AOF durable batch spill) keep exactly their old blocking semantics
+//!   through this.
+//!
+//! Coalescing + ack correctness: every ack is fired after the NEWEST snapshot
+//! persists. A newer root strictly supersedes an older one (each snapshot is
+//! the full entry list at a higher epoch), so "your commit is durable"
+//! remains true for the older requester.
+
+use std::io;
+use std::sync::Arc;
+
+use parking_lot::Mutex;
+
+use super::manifest::{ManifestIo, ManifestRoot};
+
+/// State shared between callers and the sync thread. Commits REPLACE the
+/// `latest` slot instead of queueing — each snapshot is a complete root at a
+/// monotonically higher epoch, so only the newest must ever reach disk. The
+/// slot never fills up, so a deferred commit can never block the shard
+/// thread behind a slow persist (a saturated-disk fsync stall would have
+/// filled any bounded queue and re-created exactly the stall this agent
+/// exists to remove).
+struct SyncShared {
+    latest: Option<ManifestRoot>,
+    /// Durable waiters. Always pushed together with a `latest` write, and
+    /// taken together with it; firing after a NEWER snapshot persists is
+    /// correct because a newer root strictly supersedes the waiter's own.
+    acks: Vec<flume::Sender<io::Result<()>>>,
+    /// Set once by `shutdown()`; the thread gives back its [`ManifestIo`]
+    /// on this and exits after a final flush of the slot.
+    shutdown: Option<flume::Sender<ManifestIo>>,
+}
+
+/// Handle to the per-shard manifest-sync thread. Owned by `ShardManifest`
+/// while deferred sync is enabled; `shutdown()` reclaims the [`ManifestIo`]
+/// so post-shutdown commits can run inline again.
+pub(crate) struct ManifestSyncAgent {
+    shared: Arc<Mutex<SyncShared>>,
+    /// Wake token, capacity 1. `Full` on send is fine: a token is already
+    /// pending and state was written before the send, so the thread's next
+    /// round observes it. `Disconnected` means the thread is gone — fail
+    /// loudly, never a silent no-op.
+    wake: flume::Sender<()>,
+    join: Option<std::thread::JoinHandle<()>>,
+}
+
+impl std::fmt::Debug for ManifestSyncAgent {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ManifestSyncAgent")
+            .field("pending", &self.shared.lock().latest.is_some())
+            .finish()
+    }
+}
+
+impl ManifestSyncAgent {
+    /// Spawn the sync thread. On failure (thread limit, EAGAIN) the caller
+    /// gets its [`ManifestIo`] BACK so it can keep committing inline — a
+    /// failed spawn must degrade to on-loop fsyncs, never to a manifest that
+    /// can no longer commit at all. The io is handed to the thread over a
+    /// channel (not moved into the closure) precisely so it survives the
+    /// failure path.
+    pub(crate) fn spawn(io: ManifestIo, shard_id: usize) -> Result<Self, (ManifestIo, io::Error)> {
+        let shared = Arc::new(Mutex::new(SyncShared {
+            latest: None,
+            acks: Vec::new(),
+            shutdown: None,
+        }));
+        let (wake, rx) = flume::bounded(1);
+        let (io_tx, io_rx) = flume::bounded::<ManifestIo>(1);
+        let thread_shared = Arc::clone(&shared);
+        match std::thread::Builder::new()
+            .name(format!("manifest-sync-{shard_id}"))
+            .spawn(move || {
+                // The spawner sends the io immediately after a successful
+                // spawn; Err here means the agent was dropped first — exit.
+                let Ok(io) = io_rx.recv() else { return };
+                run(io, thread_shared, rx)
+            }) {
+            Ok(join) => match io_tx.send(io) {
+                Ok(()) => Ok(Self {
+                    shared,
+                    wake,
+                    join: Some(join),
+                }),
+                // Thread died before its recv (can only be a panic in thread
+                // start-up glue) — flume returns the unsent io in the error.
+                Err(flume::SendError(io)) => {
+                    let _ = join.join();
+                    Err((io, io::Error::other("manifest-sync thread died at startup")))
+                }
+            },
+            Err(e) => Err((io, e)),
+        }
+    }
+
+    fn notify(&self) -> io::Result<()> {
+        match self.wake.try_send(()) {
+            Ok(()) | Err(flume::TrySendError::Full(())) => Ok(()),
+            Err(flume::TrySendError::Disconnected(())) => {
+                Err(io::Error::other("manifest-sync thread unavailable"))
+            }
+        }
+    }
+
+    /// Fire-and-forget commit of a complete root snapshot. Never blocks:
+    /// replaces the shared slot and posts a wake token.
+    pub(crate) fn commit_deferred(&self, root: ManifestRoot) -> io::Result<()> {
+        self.shared.lock().latest = Some(root);
+        self.notify()
+    }
+
+    /// Commit and block until the snapshot (or a newer one) is durable.
+    pub(crate) fn commit_durable(&self, root: ManifestRoot) -> io::Result<()> {
+        let (ack_tx, ack_rx) = flume::bounded::<io::Result<()>>(1);
+        {
+            let mut s = self.shared.lock();
+            s.latest = Some(root);
+            s.acks.push(ack_tx);
+        }
+        self.notify()?;
+        ack_rx
+            .recv()
+            .map_err(|_| io::Error::other("manifest-sync thread died before ack"))?
+    }
+
+    /// Flush any pending commit and reclaim the file-I/O state. Returns
+    /// `None` only if the thread died abnormally (its panic already logged).
+    pub(crate) fn shutdown(mut self) -> Option<ManifestIo> {
+        let (gb_tx, gb_rx) = flume::bounded::<ManifestIo>(1);
+        self.shared.lock().shutdown = Some(gb_tx);
+        let _ = self.notify();
+        let io = gb_rx.recv().ok();
+        if let Some(j) = self.join.take() {
+            let _ = j.join();
+        }
+        io
+    }
+}
+
+fn run(mut io: ManifestIo, shared: Arc<Mutex<SyncShared>>, rx: flume::Receiver<()>) {
+    while rx.recv().is_ok() {
+        // Take the whole state atomically. Anything written after this take
+        // comes with its own wake token (writers post the token AFTER the
+        // state write; `Full` implies an unconsumed token), so it is handled
+        // in a later round — no lost updates.
+        let (latest, acks, shutdown) = {
+            let mut s = shared.lock();
+            (
+                s.latest.take(),
+                std::mem::take(&mut s.acks),
+                s.shutdown.take(),
+            )
+        };
+
+        let res = match latest {
+            Some(root) => {
+                let r = io.persist(root);
+                if let Err(ref e) = r {
+                    tracing::warn!(error = %e, "manifest-sync: commit failed (placement metadata only; AOF replay + orphan sweep recover on restart)");
+                }
+                r
+            }
+            // Empty slot: a previous round already persisted a snapshot at
+            // least as new as anything these (necessarily-raced) wakes saw.
+            None => Ok(()),
+        };
+        for ack in acks {
+            let mirrored = match &res {
+                Ok(()) => Ok(()),
+                // io::Error is not Clone; mirror kind + message per waiter.
+                Err(e) => Err(io::Error::new(e.kind(), e.to_string())),
+            };
+            let _ = ack.send(mirrored);
+        }
+
+        if let Some(give_back) = shutdown {
+            let _ = give_back.send(io);
+            return;
+        }
+    }
+    // Wake channel disconnected without an explicit Shutdown: `ShardManifest`
+    // was dropped. Every accepted commit's token is delivered before the
+    // disconnect is reported (flume drains queued messages first), so the
+    // slot was flushed; `io` (and its fd) drop here.
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::manifest::{FileEntry, FileStatus, ShardManifest, StorageTier};
+    use std::time::{Duration, Instant};
+
+    fn make_entry(file_id: u64) -> FileEntry {
+        FileEntry {
+            file_id,
+            file_type: crate::persistence::page::PageType::KvLeaf as u8,
+            status: FileStatus::Active,
+            tier: StorageTier::Hot,
+            page_size_log2: 12,
+            page_count: 1,
+            byte_size: 4096,
+            created_lsn: file_id,
+            min_key_hash: 0,
+            max_key_hash: u64::MAX,
+            last_modified_lsn: file_id,
+        }
+    }
+
+    #[test]
+    fn deferred_commit_does_not_block_caller_and_reaches_disk() {
+        #[allow(clippy::unwrap_used)] // test-only; poisoning would already be a failed test
+        let _knob = super::super::manifest::TEST_SYNC_KNOB_LOCK.lock().unwrap();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let path = tmp.path().join("shard-0.manifest");
+        let mut m = ShardManifest::create(&path).expect("create");
+        m.enable_deferred_sync(0);
+
+        super::super::manifest::TEST_INJECT_SYNC_DELAY_MS
+            .store(200, std::sync::atomic::Ordering::SeqCst);
+        let t0 = Instant::now();
+        m.add_file(make_entry(1));
+        m.commit_deferred().expect("deferred commit send");
+        let elapsed = t0.elapsed();
+        super::super::manifest::TEST_INJECT_SYNC_DELAY_MS
+            .store(0, std::sync::atomic::Ordering::SeqCst);
+        assert!(
+            elapsed < Duration::from_millis(100),
+            "deferred commit must not block the caller for the injected sync \
+             delay (took {elapsed:?})"
+        );
+
+        // Shutdown flushes pending commits and reclaims the io handle.
+        m.shutdown_deferred();
+        drop(m);
+        let reopened = ShardManifest::open(&path).expect("reopen");
+        assert_eq!(reopened.files().len(), 1, "deferred commit must reach disk");
+    }
+
+    #[test]
+    fn durable_commit_blocks_until_persisted() {
+        #[allow(clippy::unwrap_used)] // test-only; poisoning would already be a failed test
+        let _knob = super::super::manifest::TEST_SYNC_KNOB_LOCK.lock().unwrap();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let path = tmp.path().join("shard-1.manifest");
+        let mut m = ShardManifest::create(&path).expect("create");
+        m.enable_deferred_sync(1);
+
+        super::super::manifest::TEST_INJECT_SYNC_DELAY_MS
+            .store(150, std::sync::atomic::Ordering::SeqCst);
+        let t0 = Instant::now();
+        m.add_file(make_entry(7));
+        m.commit().expect("durable commit");
+        let elapsed = t0.elapsed();
+        super::super::manifest::TEST_INJECT_SYNC_DELAY_MS
+            .store(0, std::sync::atomic::Ordering::SeqCst);
+        assert!(
+            elapsed >= Duration::from_millis(150),
+            "durable commit must wait for the persist (took {elapsed:?})"
+        );
+
+        // Durable means durable NOW: reopen from a second handle without any
+        // flush and the entry must already be there.
+        let reopened = ShardManifest::open(&path).expect("reopen");
+        assert_eq!(reopened.files().len(), 1);
+        m.shutdown_deferred();
+    }
+
+    #[test]
+    fn queued_deferred_commits_coalesce_to_fewer_persists() {
+        #[allow(clippy::unwrap_used)] // test-only; poisoning would already be a failed test
+        let _knob = super::super::manifest::TEST_SYNC_KNOB_LOCK.lock().unwrap();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let path = tmp.path().join("shard-2.manifest");
+        let mut m = ShardManifest::create(&path).expect("create");
+        m.enable_deferred_sync(2);
+
+        let before =
+            super::super::manifest::TEST_PERSIST_COUNT.load(std::sync::atomic::Ordering::SeqCst);
+        // Get the worker INSIDE a stalled persist before the burst, and keep
+        // every persist stalled until after the shutdown flush — otherwise
+        // scheduling can let the worker race the burst commit-by-commit and
+        // weaken the coalescing bound into flakiness.
+        super::super::manifest::TEST_INJECT_SYNC_DELAY_MS
+            .store(100, std::sync::atomic::Ordering::SeqCst);
+        m.add_file(make_entry(99));
+        m.commit_deferred().expect("priming deferred send");
+        std::thread::sleep(Duration::from_millis(50));
+        for i in 0..20u64 {
+            m.add_file(make_entry(100 + i));
+            m.commit_deferred().expect("deferred send");
+        }
+        m.shutdown_deferred();
+        super::super::manifest::TEST_INJECT_SYNC_DELAY_MS
+            .store(0, std::sync::atomic::Ordering::SeqCst);
+        let persists = super::super::manifest::TEST_PERSIST_COUNT
+            .load(std::sync::atomic::Ordering::SeqCst)
+            - before;
+        assert!(
+            persists < 20,
+            "20 queued deferred commits must coalesce (saw {persists} persists)"
+        );
+
+        drop(m);
+        let reopened = ShardManifest::open(&path).expect("reopen");
+        assert_eq!(
+            reopened.files().len(),
+            21,
+            "coalescing must still land the final (superset) snapshot"
+        );
+    }
+
+    /// The old bounded(64) request queue blocked the CALLER once a stalled
+    /// persist let 64 snapshots pile up — a saturated-disk fsync stall would
+    /// re-create on the shard thread exactly the stall the agent exists to
+    /// remove. The latest-slot handoff must absorb an arbitrarily large burst
+    /// without the caller ever waiting on the persist.
+    #[test]
+    fn burst_of_deferred_commits_never_blocks_the_caller() {
+        #[allow(clippy::unwrap_used)] // test-only; poisoning would already be a failed test
+        let _knob = super::super::manifest::TEST_SYNC_KNOB_LOCK.lock().unwrap();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let path = tmp.path().join("shard-4.manifest");
+        let mut m = ShardManifest::create(&path).expect("create");
+        m.enable_deferred_sync(4);
+
+        // Stall every persist and get the worker INSIDE one before bursting:
+        // a burst fired while the worker is still draining its queue gets
+        // consumed as fast as it is sent, which is why a plain burst never
+        // reproduced the block. Blocking needs commits to arrive while the
+        // worker is stuck in persist with no one draining.
+        super::super::manifest::TEST_INJECT_SYNC_DELAY_MS
+            .store(400, std::sync::atomic::Ordering::SeqCst);
+        m.add_file(make_entry(999));
+        m.commit_deferred().expect("priming deferred send");
+        std::thread::sleep(Duration::from_millis(50));
+
+        let t0 = Instant::now();
+        for i in 0..100u64 {
+            m.add_file(make_entry(1000 + i));
+            m.commit_deferred().expect("deferred send");
+        }
+        let elapsed = t0.elapsed();
+        super::super::manifest::TEST_INJECT_SYNC_DELAY_MS
+            .store(0, std::sync::atomic::Ordering::SeqCst);
+        assert!(
+            elapsed < Duration::from_millis(200),
+            "a 100-commit burst against a stalled persist must not block the \
+             caller (took {elapsed:?})"
+        );
+
+        m.shutdown_deferred();
+        drop(m);
+        let reopened = ShardManifest::open(&path).expect("reopen");
+        assert_eq!(
+            reopened.files().len(),
+            101,
+            "the final (superset) snapshot must land despite coalescing"
+        );
+    }
+
+    #[test]
+    fn shutdown_reclaims_io_and_inline_commits_still_work() {
+        #[allow(clippy::unwrap_used)] // test-only; poisoning would already be a failed test
+        let _knob = super::super::manifest::TEST_SYNC_KNOB_LOCK.lock().unwrap();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let path = tmp.path().join("shard-3.manifest");
+        let mut m = ShardManifest::create(&path).expect("create");
+        m.enable_deferred_sync(3);
+        m.add_file(make_entry(1));
+        m.commit_deferred().expect("deferred send");
+        m.shutdown_deferred();
+
+        // Post-shutdown the manifest is inline again: durable commit works.
+        m.add_file(make_entry(2));
+        m.commit().expect("inline commit after shutdown");
+        drop(m);
+        let reopened = ShardManifest::open(&path).expect("reopen");
+        assert_eq!(reopened.files().len(), 2);
+    }
+}
diff --git a/src/persistence/mod.rs b/src/persistence/mod.rs
index 4b8340da4..d6cc72492 100644
--- a/src/persistence/mod.rs
+++ b/src/persistence/mod.rs
@@ -9,6 +9,7 @@ pub mod control;
 pub mod fsync;
 pub mod kv_page;
 pub mod manifest;
+pub(crate) mod manifest_sync;
 pub mod migrate_aof;
 pub mod page;
 pub mod page_cache;
diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs
index 8b8cb7b06..1f7cb0a6d 100644
--- a/src/shard/event_loop.rs
+++ b/src/shard/event_loop.rs
@@ -629,6 +629,14 @@ impl super::Shard {
             } else {
                 None
             };
+        // Task #59: manifest commit fsyncs measured up to 1.0s each on this
+        // event-loop thread under spill flood (every connection on the shard
+        // stalls behind them). Move the file-I/O half onto the per-shard
+        // manifest-sync thread: durable commits keep their blocking ack,
+        // spill-completion commits become deferred sends.
+        if let Some(ref mut m) = shard_manifest {
+            m.enable_deferred_sync(shard_id);
+        }
         // Task #55: background reclaim of crash-orphaned heap files that
         // recovery only CLASSIFIED (see `Shard::restore_from_persistence` /
         // `persistence::recovery::recover_shard_v3_pitr`). Classification is
@@ -756,7 +764,28 @@ impl super::Shard {
             .map(|rs| rs.read().is_replica_mirror.clone());
 
         // Track last seen snapshot epoch to detect watch channel triggers
-        let mut last_snapshot_epoch = snapshot_trigger_rx.borrow();
+        // Test-only fault injection: delay every non-zero shard's loop start so
+        // integration tests can deterministically exercise the startup window
+        // where the listener already answers (fastest shard up) while other
+        // shards are still here. Used by tests/bgsave_startup_race.rs; never
+        // set in production.
+        if shard_id != 0
+            && let Ok(ms) = std::env::var("MOON_TEST_SLOW_SHARD_START_MS")
+            && let Ok(ms) = ms.parse::<u64>()
+        {
+            std::thread::sleep(std::time::Duration::from_millis(ms));
+        }
+        // Start the cursor at 0, NOT at the watch channel's current value: the
+        // listener accepts clients as soon as the fastest shard is up, so a
+        // BGSAVE (or auto-save) can broadcast epoch N while a slower shard is
+        // still in this setup code. Seeding the cursor with borrow() would
+        // swallow that pending trigger — this shard never snapshots, the
+        // BGSAVE_SHARDS_REMAINING counter never reaches zero, and every later
+        // BGSAVE reports "already in progress" forever (observed as the
+        // 20s-stuck `rdb_bgsave_in_progress:1` crash-matrix flake; the race
+        // reproduces deterministically with a delayed shard start). Epochs are
+        // per-process and start at 0, so 0 is always a safe "nothing seen yet".
+        let mut last_snapshot_epoch = 0u64;
 
         // Sub-timer intervals: tokio uses separate select! branches for each.
         // monoio uses counter-based dispatch from a single periodic tick to avoid
@@ -1558,10 +1587,17 @@ impl super::Shard {
                                     &mut snapshot_state, &mut snapshot_reply_tx, shard_id,
                                     &e.to_string(),
                                 );
+                                // Decrement the BGSAVE fan-in counter (same as
+                                // the monoio arm below — this was missing here,
+                                // so tokio BGSAVE left rdb_bgsave_in_progress
+                                // stuck at 1 forever). Safe for auto-save
+                                // snapshots too: the counter ignores calls at 0.
+                                crate::command::persistence::bgsave_shard_done(false);
                             } else {
                                 persistence_tick::finalize_snapshot_success(
                                     &mut snapshot_state, &mut snapshot_reply_tx, shard_id,
                                 );
+                                crate::command::persistence::bgsave_shard_done(true);
                                 bgsave_checkpoint_requested = true;
                             }
                         }
@@ -1854,6 +1890,11 @@ impl super::Shard {
                     if let Some(ref mut wal) = wal_writer {
                         let _ = wal.flush_sync();
                     }
+                    // Task #59: flush pending deferred manifest commits and
+                    // stop the manifest-sync thread before the loop exits.
+                    if let Some(ref mut m) = shard_manifest {
+                        m.shutdown_deferred();
+                    }
                     break;
                 }
             }
@@ -1994,6 +2035,11 @@ impl super::Shard {
                     if let Some(ref mut wal) = wal_writer {
                         let _ = wal.flush_sync();
                     }
+                    // Task #59: flush pending deferred manifest commits and
+                    // stop the manifest-sync thread before the loop exits.
+                    if let Some(ref mut m) = shard_manifest {
+                        m.shutdown_deferred();
+                    }
                     break;
                 }
 
diff --git a/src/shard/persistence_tick.rs b/src/shard/persistence_tick.rs
index 9f22238e7..3d7b22500 100644
--- a/src/shard/persistence_tick.rs
+++ b/src/shard/persistence_tick.rs
@@ -587,6 +587,15 @@ fn apply_completion_vec(
         return;
     }
 
+    // Task #59: this runs on the shard event-loop thread, so it must not pay
+    // for manifest fsyncs — one DEFERRED commit for the whole batch (below),
+    // shipped to the manifest-sync thread. Correct because this path only
+    // runs under `--appendonly yes` (evict_one_async_spill bails otherwise):
+    // AOF replay + the orphan sweep reconstruct anything a lost manifest
+    // commit would have recorded. Previously: one durable commit (up to 2
+    // fsyncs) per flushed file, measured blocking the loop 1.0-2.1s per 8s
+    // window under spill flood, single calls up to 1.0s.
+    let mut manifest_dirty = false;
     for c in completions {
         if !c.success {
             tracing::warn!(
@@ -598,16 +607,10 @@ fn apply_completion_vec(
 
         let file_id = c.file_entry.file_id;
 
-        // ONE manifest add_file + commit per flushed file.
+        // RAM-only manifest update; durability handled once per batch below.
         if let Some(ref mut manifest) = *shard_manifest {
             manifest.add_file(c.file_entry);
-            if let Err(e) = manifest.commit() {
-                tracing::warn!(
-                    file_id,
-                    error = %e,
-                    "Manifest commit failed for spill completion"
-                );
-            }
+            manifest_dirty = true;
         }
 
         // Insert one ColdIndex entry per KV within this file. `ttl_ms` rides
@@ -628,6 +631,20 @@ fn apply_completion_vec(
             });
         }
     }
+
+    // ONE deferred commit for the whole drained batch (see the task #59 note
+    // at the top of this function). Non-blocking when the manifest-sync
+    // thread is attached; degrades to a synchronous persist otherwise.
+    if manifest_dirty {
+        if let Some(ref mut manifest) = *shard_manifest {
+            if let Err(e) = manifest.commit_deferred() {
+                tracing::warn!(
+                    error = %e,
+                    "Deferred manifest commit failed for spill completion batch"
+                );
+            }
+        }
+    }
 }
 
 // ---------------------------------------------------------------------------
diff --git a/src/storage/tiered/cold_read.rs b/src/storage/tiered/cold_read.rs
index b9e5a3e8a..c9ebc9a75 100644
--- a/src/storage/tiered/cold_read.rs
+++ b/src/storage/tiered/cold_read.rs
@@ -130,6 +130,11 @@ pub(crate) fn read_cold_entry(
     now_ms: u64,
     page_cache: Option<&PageCache>,
 ) -> ColdReadOutcome {
+    // Task #59 lever 2: any cold read — including the synchronous MGET /
+    // MULTI / Lua paths that never go through the async pool — signals the
+    // spill writer to briefly yield the device. Double-counting with the
+    // async path's own guard is harmless (the signal is "readers > 0").
+    let _inflight = super::cold_read_pool::ColdReadInflightGuard::new();
     #[cfg(test)]
     {
         let delay_ms = TEST_INJECT_DELAY_MS.load(std::sync::atomic::Ordering::Relaxed);
diff --git a/src/storage/tiered/cold_read_pool.rs b/src/storage/tiered/cold_read_pool.rs
index a1a1a8d61..640a2a837 100644
--- a/src/storage/tiered/cold_read_pool.rs
+++ b/src/storage/tiered/cold_read_pool.rs
@@ -41,6 +41,38 @@ use super::cold_read::{ColdReadOutcome, read_cold_entry};
 /// backlogged disk is bottlenecked on the disk itself, not on worker count.
 const DEFAULT_POOL_THREADS: usize = 2;
 
+/// Number of cold reads currently queued or executing (task #59 lever 2).
+/// The spill writer consults this after each durable batch flush and briefly
+/// yields the device to readers (`spill_thread::spill_pace_after_flush`).
+/// Relaxed ordering: a heuristic pacing signal, not a synchronization edge.
+pub(crate) static COLD_READS_INFLIGHT: std::sync::atomic::AtomicUsize =
+    std::sync::atomic::AtomicUsize::new(0);
+
+/// RAII increment of [`COLD_READS_INFLIGHT`] — drop-based so the count stays
+/// correct when an awaiting connection task is cancelled mid-read.
+pub(crate) struct ColdReadInflightGuard(&'static std::sync::atomic::AtomicUsize);
+
+impl ColdReadInflightGuard {
+    pub(crate) fn new() -> Self {
+        Self::on(&COLD_READS_INFLIGHT)
+    }
+
+    /// Guard an explicit counter. Exists so tests can verify the RAII
+    /// mechanics on a private counter — the global one is shared with every
+    /// concurrently running test that touches the cold-read path, so exact
+    /// assertions on it are inherently racy.
+    pub(crate) fn on(counter: &'static std::sync::atomic::AtomicUsize) -> Self {
+        counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        Self(counter)
+    }
+}
+
+impl Drop for ColdReadInflightGuard {
+    fn drop(&mut self) {
+        self.0.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
+    }
+}
+
 struct ColdReadJob {
     shard_dir: PathBuf,
     location: ColdLocation,
@@ -98,6 +130,9 @@ pub async fn read_cold_entry_async(
     location: ColdLocation,
     now_ms: u64,
 ) -> ColdReadOutcome {
+    // Signal the spill writer that a reader is waiting on the device (task
+    // #59 lever 2). Held across the await: queue-wait counts as waiting.
+    let _inflight = ColdReadInflightGuard::new();
     let (reply_tx, reply_rx) = flume::bounded::<ColdReadOutcome>(1);
     let job = ColdReadJob {
         shard_dir: shard_dir.to_path_buf(),
diff --git a/src/storage/tiered/spill_thread.rs b/src/storage/tiered/spill_thread.rs
index 1d64d3354..aa2cd5beb 100644
--- a/src/storage/tiered/spill_thread.rs
+++ b/src/storage/tiered/spill_thread.rs
@@ -79,6 +79,42 @@ const FLUSH_ENTRY_CAP: usize = 256;
 /// path (one value's worth of pages) rather than a multiplicative one.
 const BATCH_BYTES_CAP: usize = 4 * 1024 * 1024;
 
+/// Request-channel depth (see `SpillThread::new`). Also referenced by the
+/// pacing cutoff below so "deep backlog" stays defined relative to capacity.
+const REQUEST_QUEUE_CAP: usize = 4096;
+
+/// Task #59 lever 2 — read/write device fairness, the pacing decision.
+///
+/// After each durable batch flush (a multi-MB pwrite + fsync), the spill
+/// thread asks whether to briefly YIELD the device before building the next
+/// batch. Under a sustained spill flood the device queue is otherwise
+/// permanently deep with writer traffic, and a cold reader's pread waits
+/// behind all of it (measured p99 208ms for a raw 64KB O_DIRECT read during
+/// the G2-shape flood, vs 6ms idle).
+///
+/// Rules, in priority order:
+/// - Nobody is reading → never pause (zero throughput cost when idle).
+/// - Request backlog ≥ half capacity → never pause: a paced spill thread
+///   must not push eviction into `try_send` failure, which surfaces as -OOM
+///   to write clients. Full-speed drain wins over read latency there.
+/// - Otherwise → one small fixed quantum. 4ms per flush caps the throughput
+///   tax at ~a few percent of a typical flush's own write+fsync time while
+///   giving a queued pread a window in which the device queue is empty.
+///
+/// Pure function of its inputs for unit-testability.
+pub(crate) fn spill_pace_after_flush(
+    reads_inflight: usize,
+    backlog_len: usize,
+) -> std::time::Duration {
+    const PACE_QUANTUM_MS: u64 = 4;
+    const BACKLOG_CUTOFF: usize = REQUEST_QUEUE_CAP / 2;
+    if reads_inflight == 0 || backlog_len >= BACKLOG_CUTOFF {
+        std::time::Duration::ZERO
+    } else {
+        std::time::Duration::from_millis(PACE_QUANTUM_MS)
+    }
+}
+
 /// Cumulative count of `SpillCompletion`s dropped. Dropping is **data loss**:
 /// the `.mpf` file is on disk but its manifest `add_file` + `cold_index` insert
 /// happen only when the event loop consumes the completion, and the keys are
@@ -417,7 +453,7 @@ impl SpillThread {
     /// rebuilds `cold_index` from the manifest, so dropping is safe (though
     /// we count it for observability).
     pub fn new(shard_id: usize) -> Self {
-        let (request_tx, request_rx) = flume::bounded::<SpillRequest>(4096);
+        let (request_tx, request_rx) = flume::bounded::<SpillRequest>(REQUEST_QUEUE_CAP);
         let (completion_tx, completion_rx) = flume::bounded::<SpillCompletion>(8192);
         let stop_flag = Arc::new(AtomicBool::new(false));
         let stop_flag_bg = stop_flag.clone();
@@ -452,6 +488,19 @@ impl SpillThread {
     ) {
         let mut buffer: Vec<SpillRequest> = Vec::with_capacity(FLUSH_ENTRY_CAP);
 
+        // Task #59 lever 2: after a durable batch flush, briefly yield the
+        // device to any waiting cold reader. Never paces at shutdown or when
+        // the request backlog is deep (see `spill_pace_after_flush`).
+        let pace = |request_rx: &flume::Receiver<SpillRequest>| {
+            let pause = spill_pace_after_flush(
+                super::cold_read_pool::COLD_READS_INFLIGHT.load(Ordering::Relaxed),
+                request_rx.len(),
+            );
+            if !pause.is_zero() {
+                std::thread::sleep(pause);
+            }
+        };
+
         loop {
             // Liveness heartbeat (R5): stamped every loop tick so INFO can
             // surface a silently-dead spill thread. Relaxed max-store is
@@ -481,6 +530,7 @@ impl SpillThread {
                             flush_buffer(&mut buffer),
                             &stop_flag,
                         );
+                        pace(&request_rx);
                     }
                 }
                 Err(flume::RecvTimeoutError::Timeout) => {
@@ -491,6 +541,7 @@ impl SpillThread {
                             flush_buffer(&mut buffer),
                             &stop_flag,
                         );
+                        pace(&request_rx);
                     }
                 }
                 Err(flume::RecvTimeoutError::Disconnected) => {
@@ -639,6 +690,60 @@ mod tests {
         completions
     }
 
+    #[test]
+    fn pace_is_zero_when_no_readers_waiting() {
+        assert_eq!(
+            spill_pace_after_flush(0, 0),
+            std::time::Duration::ZERO,
+            "pacing must cost nothing when nobody is reading"
+        );
+        assert_eq!(
+            spill_pace_after_flush(0, REQUEST_QUEUE_CAP),
+            std::time::Duration::ZERO
+        );
+    }
+
+    #[test]
+    fn pace_is_zero_under_deep_backlog_even_with_readers() {
+        // A paced spill thread must never push eviction into try_send
+        // failure (-OOM to write clients): deep backlog always wins.
+        assert_eq!(
+            spill_pace_after_flush(3, REQUEST_QUEUE_CAP / 2),
+            std::time::Duration::ZERO
+        );
+        assert_eq!(
+            spill_pace_after_flush(100, REQUEST_QUEUE_CAP),
+            std::time::Duration::ZERO
+        );
+    }
+
+    #[test]
+    fn pace_yields_a_quantum_when_reader_waits_and_backlog_shallow() {
+        let d = spill_pace_after_flush(1, 0);
+        assert!(d > std::time::Duration::ZERO);
+        assert!(
+            d <= std::time::Duration::from_millis(10),
+            "quantum must stay small — this is a yield, not a stall ({d:?})"
+        );
+        assert_eq!(d, spill_pace_after_flush(64, REQUEST_QUEUE_CAP / 2 - 1));
+    }
+
+    #[test]
+    fn reader_inflight_guard_counts_and_releases() {
+        use super::super::cold_read_pool::ColdReadInflightGuard;
+        // Own counter, NOT the global COLD_READS_INFLIGHT: sibling tests in
+        // this binary exercise the cold-read path concurrently, so exact
+        // assertions on the shared global are racy. The RAII mechanics are
+        // identical either way.
+        static LOCAL: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
+        {
+            let _g1 = ColdReadInflightGuard::on(&LOCAL);
+            let _g2 = ColdReadInflightGuard::on(&LOCAL);
+            assert_eq!(LOCAL.load(Ordering::Relaxed), 2);
+        }
+        assert_eq!(LOCAL.load(Ordering::Relaxed), 0);
+    }
+
     #[test]
     fn test_spill_thread_new_returns_valid_handles() {
         let st = SpillThread::new(0);
diff --git a/tests/bgsave_startup_race.rs b/tests/bgsave_startup_race.rs
new file mode 100644
index 000000000..6cff6a47c
--- /dev/null
+++ b/tests/bgsave_startup_race.rs
@@ -0,0 +1,122 @@
+//! BGSAVE-vs-shard-startup race (found via crash_matrix_cross_plane flake,
+//! 2026-07-17): the listener answers clients as soon as the FASTEST shard is
+//! up, so a BGSAVE can broadcast its snapshot epoch while a slower shard is
+//! still initializing its event loop. The per-shard epoch cursor used to be
+//! seeded from the watch channel's CURRENT value at loop start, silently
+//! swallowing such a pending trigger — that shard never snapshotted,
+//! `BGSAVE_SHARDS_REMAINING` never reached zero, and `rdb_bgsave_in_progress`
+//! stuck at 1 forever (every later BGSAVE refused as already-in-progress).
+//!
+//! `MOON_TEST_SLOW_SHARD_START_MS` (test-only fault injection in
+//! `event_loop.rs`) makes the race deterministic: shards 1..N sleep before
+//! entering their loops while shard 0 answers the client. Pre-fix this test
+//! times out at the 10s deadline on every run; post-fix (cursor starts at 0,
+//! pending triggers are honored on the first tick) it completes in <2s.
+//!
+//! Run with:
+//!   cargo test --release --test bgsave_startup_race
+
+#![allow(clippy::unwrap_used)]
+
+mod common;
+
+use std::io::{Read, Write};
+use std::time::{Duration, Instant};
+
+fn find_moon_binary() -> std::path::PathBuf {
+    if let Ok(bin) = std::env::var("MOON_BIN") {
+        let p = std::path::PathBuf::from(bin);
+        if p.exists() {
+            return p;
+        }
+    }
+    std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon"))
+}
+
+struct Conn(std::net::TcpStream);
+
+impl Conn {
+    fn open(port: u16) -> Self {
+        let deadline = Instant::now() + Duration::from_secs(30);
+        loop {
+            if let Ok(s) = std::net::TcpStream::connect(("127.0.0.1", port)) {
+                s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
+                return Self(s);
+            }
+            assert!(Instant::now() < deadline, "server never accepted");
+            std::thread::sleep(Duration::from_millis(50));
+        }
+    }
+
+    /// Send an inline command, return the raw reply up to one read() chunk.
+    fn cmd(&mut self, line: &str) -> String {
+        self.0
+            .write_all(format!("{line}\r\n").as_bytes())
+            .expect("write");
+        let mut buf = [0u8; 64 * 1024];
+        let n = self.0.read(&mut buf).expect("read");
+        String::from_utf8_lossy(&buf[..n]).into_owned()
+    }
+}
+
+#[test]
+fn bgsave_issued_during_shard_startup_still_completes() {
+    let base = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("target/bgsave-race-tmp");
+    std::fs::create_dir_all(&base).expect("create tmp base");
+    let dir = tempfile::Builder::new()
+        .prefix("bgrace-")
+        .tempdir_in(&base)
+        .expect("tempdir");
+
+    let port = common::reserve_port();
+    let mut child = std::process::Command::new(find_moon_binary());
+    child
+        .args([
+            "--port",
+            &port.to_string(),
+            "--dir",
+            &dir.path().to_string_lossy(),
+            "--shards",
+            "4",
+            "--appendonly",
+            "yes",
+            "--disk-free-min-pct",
+            "0",
+        ])
+        // Shards 1-3 enter their event loops ~500ms late; shard 0 (and the
+        // listener) come up immediately, so the BGSAVE below lands inside
+        // the startup window.
+        .env("MOON_TEST_SLOW_SHARD_START_MS", "500")
+        .stdout(std::process::Stdio::null())
+        .stderr(std::process::Stdio::null());
+    let mut child = child.spawn().expect("spawn moon");
+
+    let mut c = Conn::open(port);
+    // PING to confirm command plane is up, then fire BGSAVE immediately —
+    // while shards 1-3 are still asleep in setup.
+    assert!(c.cmd("PING").contains("PONG"), "server must answer PING");
+    let reply = c.cmd("BGSAVE");
+    assert!(
+        reply.contains("Background saving started"),
+        "BGSAVE must be accepted, got: {reply}"
+    );
+
+    // The snapshot must COMPLETE even though 3 of 4 shards received the
+    // trigger before their loops started. Pre-fix: in_progress sticks at 1.
+    let deadline = Instant::now() + Duration::from_secs(10);
+    loop {
+        let info = c.cmd("INFO persistence");
+        if info.contains("rdb_bgsave_in_progress:0") {
+            break;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "BGSAVE never completed — a shard swallowed the snapshot trigger \
+             broadcast during its startup (INFO: {info})"
+        );
+        std::thread::sleep(Duration::from_millis(100));
+    }
+
+    common::sigkill(&mut child);
+    let _ = child.wait();
+}
diff --git a/tests/cmd_flush_dbsize_debug_memory.rs b/tests/cmd_flush_dbsize_debug_memory.rs
index 874fbc9fe..f6363f269 100644
--- a/tests/cmd_flush_dbsize_debug_memory.rs
+++ b/tests/cmd_flush_dbsize_debug_memory.rs
@@ -25,7 +25,10 @@ fn redis_cli_available() -> bool {
 }
 
 fn release_binary() -> std::path::PathBuf {
-    std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("target/release/moon")
+    // MOON_BIN-aware: the hardcoded target/release/moon fallback inside
+    // find_moon_binary is a stale-binary trap on shared checkouts (VM runs
+    // exec a host Mach-O via OrbStack's proxy and never accept in-VM).
+    common::find_moon_binary()
 }
 
 /// Running moon instance. Auto-killed on drop.
@@ -71,7 +74,7 @@ fn spawn_moon() -> Option<Moon> {
                 "0",
                 "--appendonly",
                 "no",
-                "--persistence-dir",
+                "--dir",
                 tmp_dir.to_str().unwrap(),
             ])
             .stdout(Stdio::null())
diff --git a/tests/crash_matrix_cross_plane/scenarios.rs b/tests/crash_matrix_cross_plane/scenarios.rs
index 726e237e8..488cf010e 100644
--- a/tests/crash_matrix_cross_plane/scenarios.rs
+++ b/tests/crash_matrix_cross_plane/scenarios.rs
@@ -475,6 +475,14 @@ pub fn mq_isolated(cfg: &Config) {
     let marker = "XPLANE-MQ-MARKER".to_string();
     mq_push(&mut c, ordersq, "final", &marker);
     harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, marker.as_bytes());
+    // The marker proves ORDERSQ's shard flushed its WAL stream — it says
+    // nothing about the shard that owns the DLQ queue: per-shard WAL streams
+    // flush independently on each shard's own 1ms tick, and under CPU
+    // starvation (full-suite parallelism) that tick can slip past this kill.
+    // Sync on the DLQ queue's own bytes before crashing, or this cell
+    // measures cross-shard flush-timing luck instead of DLQ durability
+    // (observed: dlqtestq absent from EVERY WAL at kill time, 2/12 runs).
+    harness::wait_for_wal_v3_bytes_any_shard(&dir, cfg.shards, dlqtestq.as_bytes());
     harness::crash(guard, port);
 
     let (guard2, port2) = harness::spawn_moon_on(&dir, cfg, &[]);
diff --git a/tests/mem_watchdog.rs b/tests/mem_watchdog.rs
index 2a298f6d0..66f163751 100644
--- a/tests/mem_watchdog.rs
+++ b/tests/mem_watchdog.rs
@@ -96,6 +96,12 @@ fn spawn_moon_mem(
             "no",
             "--mem-full-pct",
             &mem_full_pct.to_string(),
+            // This suite tests the RSS watchdog, not the disk guard; on
+            // near-full dev volumes the 5%-free write pause would shadow
+            // every assertion with MOONERR diskfull (see
+            // gotcha_diskfull_guard_gutted_crash_tests).
+            "--disk-free-min-pct",
+            "0",
         ]);
         if let Some(limit) = limit_bytes_override {
             cmd.env("MOON_MEM_LIMIT_BYTES", limit.to_string());
diff --git a/tests/memory_doctor_response.rs b/tests/memory_doctor_response.rs
index fb3b4d1f8..9097149f6 100644
--- a/tests/memory_doctor_response.rs
+++ b/tests/memory_doctor_response.rs
@@ -23,7 +23,10 @@ fn redis_cli_available() -> bool {
 }
 
 fn release_binary() -> std::path::PathBuf {
-    std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("target/release/moon")
+    // MOON_BIN-aware: the bare target/release/moon fallback is a stale-binary
+    // trap on shared checkouts (an OrbStack VM exec's a host Mach-O via proxy
+    // and the server never accepts in-VM — 30s spawn_listening timeout).
+    common::find_moon_binary()
 }
 
 /// Running moon instance. Auto-killed on drop.
diff --git a/tests/oom_bypass_closure.rs b/tests/oom_bypass_closure.rs
index 7f0f8a08f..4ee52e64f 100644
--- a/tests/oom_bypass_closure.rs
+++ b/tests/oom_bypass_closure.rs
@@ -99,6 +99,11 @@ fn spawn_moon_oom(dir: &std::path::Path, shards: u32, maxmemory_bytes: u64) -> (
                 &maxmemory_bytes.to_string(),
                 "--maxmemory-policy",
                 "noeviction",
+                // Under test here is the -OOM eviction gate, not the disk
+                // guard; a near-full dev volume would otherwise shadow every
+                // write with MOONERR diskfull.
+                "--disk-free-min-pct",
+                "0",
             ])
             .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log"))
             .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log"))
@@ -688,6 +693,10 @@ fn spawn_moon_no_maxmemory(dir: &std::path::Path, shards: u32) -> (ServerGuard,
                 // starts the atomic definitively unset.
                 "--maxmemory",
                 "0",
+                // Same rationale as spawn_moon_oom: keep the disk guard from
+                // shadowing the maxmemory-publish behavior under test.
+                "--disk-free-min-pct",
+                "0",
             ])
             .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log"))
             .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log"))
diff --git a/tests/spsc_two_db.rs b/tests/spsc_two_db.rs
index 039d8edce..69747e674 100644
--- a/tests/spsc_two_db.rs
+++ b/tests/spsc_two_db.rs
@@ -94,6 +94,11 @@ fn spawn_moon(dir: &std::path::Path, shards: u32) -> (ServerGuard, u16) {
                 &shards.to_string(),
                 "--appendonly",
                 "no",
+                // COPY/MOVE routing is under test, not the disk guard; a
+                // near-full dev volume would otherwise fail every setup SET
+                // with MOONERR diskfull.
+                "--disk-free-min-pct",
+                "0",
             ])
             .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("stdout log"))
             .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("stderr log"))