From 0eed0a1b7d3a168d77c3095cbc942ac90818998b Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 16 Jul 2026 14:23:06 +0700 Subject: [PATCH 01/12] =?UTF-8?q?perf(persistence):=20off-loop=20manifest?= =?UTF-8?q?=20commits=20=E2=80=94=20per-shard=20manifest-sync=20thread=20(?= =?UTF-8?q?task=20#59=20lever=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of the residual task #59 tail (205ms cold-GET during spill flood, G2 re-run): apply_completion_vec ran ONE DURABLE manifest commit (up to 2 fsyncs) PER SPILLED FILE on the shard event-loop thread. strace decomposition (2026-07-16, tmp/t59-decompose.sh + tmp/t59-strace.sh): each shard-N thread spent 1.0-2.1s of an 8s flood window blocked in fsync (163-167 calls, single calls up to 1.0s) — every connection on the shard stalls behind each one. PING p50 through a fresh connection measured 103ms under flood AND under post-flood drain; raw device preads only p50=4ms — the loop, not the disk, was the dominant stall. Fix: split ShardManifest into loop-owned RAM state + ManifestIo (file handle, dual-slot alternation, reopen bookkeeping). enable_deferred_sync() moves ManifestIo onto a per-shard manifest-sync-{id} thread: - commit() keeps EXACTLY its old blocking durability (send + ack) — the checkpoint protocol and the no-AOF durable batch spill path, where the manifest IS the durability record, are unchanged. - commit_deferred() (new) ships a complete root snapshot and returns immediately; consecutive queued snapshots coalesce to the newest (each is a full root at a higher epoch). Used ONLY by the spill-completion apply path, which runs exclusively under --appendonly yes where AOF replay + the orphan sweep reconstruct anything a lost manifest commit recorded. - apply_completion_vec: one deferred commit per drained batch (was: durable commit per file), manifest add_file + cold-index inserts stay RAM-only. - Shutdown flushes pending commits and reclaims the handle inline (shutdown_deferred) before the event loop exits, both runtime arms. Crash-safety invariant (overflow pages sync'd BEFORE the root that points at them; dual-slot atomic root flip) is preserved verbatim — persist() is the old commit() body running on a different thread. TDD (mirrors cold_read's TEST_INJECT_DELAY_MS pattern): - deferred_commit_does_not_block_caller_and_reaches_disk — 200ms injected sync delay, caller returns <100ms, entry durable after flush (fails by construction against the old synchronous path). - durable_commit_blocks_until_persisted — commit() still waits and is immediately reopenable from a second handle. - queued_deferred_commits_coalesce_to_fewer_persists — 20 queued commits coalesce, final superset snapshot lands. - shutdown_reclaims_io_and_inline_commits_still_work. Full lib suite: 4343 passed / 0 failed. Part of task #59 (read-vs-spill fairness); lever 2 (spill pacing) follows. author: Tin Dang --- src/persistence/manifest.rs | 399 ++++++++++++++++++++++--------- src/persistence/manifest_sync.rs | 307 ++++++++++++++++++++++++ src/persistence/mod.rs | 1 + src/shard/event_loop.rs | 18 ++ src/shard/persistence_tick.rs | 33 ++- 5 files changed, 638 insertions(+), 120 deletions(-) create mode 100644 src/persistence/manifest_sync.rs diff --git a/src/persistence/manifest.rs b/src/persistence/manifest.rs index 1f19cb504..052a57c22 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, /// 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, +} + 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,78 @@ 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; + /// 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)", + )), + } + } - let mut page = [0u8; PAGE_4K]; - Self::serialize_root(&self.active_root, overflow_start_page, &mut page); + /// 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() + } - // Write to the inactive slot - let write_offset = if self.active_slot == 0 { - ROOT_B_OFFSET - } else { - ROOT_A_OFFSET + /// 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) => IoBackend::Deferred( + crate::persistence::manifest_sync::ManifestSyncAgent::spawn(io, shard_id), + ), + other => other, }; + } - 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() { - tracing::warn!(error = %e, "manifest compaction failed (commit already durable)"); - } + /// 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 } - } - } - - Ok(()) + }, + other => other, + }; } /// Add a file entry to the manifest (in-memory only until commit). @@ -663,8 +710,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 +893,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 +1070,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 +1418,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..5752713e6 --- /dev/null +++ b/src/persistence/manifest_sync.rs @@ -0,0 +1,307 @@ +//! 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: +//! - **Deferred** (`commit_deferred`): fire-and-forget snapshot send; the +//! spill-completion path uses this. Consecutive queued snapshots coalesce — +//! each is a complete root, so only the newest needs to reach disk. +//! - **Durable** (`commit_durable`): send + 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: when several commits coalesce, 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 super::manifest::{ManifestIo, ManifestRoot}; + +/// Queue depth for pending commit requests. Deferred sends block only if the +/// agent falls this many complete snapshots behind (each bounded by one +/// `persist`, i.e. ~2 fsyncs) — bounded, and still off the per-file cadence +/// the loop used to pay. +const QUEUE_CAP: usize = 64; + +pub(crate) enum SyncRequest { + Commit { + root: ManifestRoot, + ack: Option>>, + }, + Shutdown { + give_back: flume::Sender, + }, +} + +/// 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 { + tx: flume::Sender, + join: Option>, +} + +impl std::fmt::Debug for ManifestSyncAgent { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ManifestSyncAgent") + .field("queued", &self.tx.len()) + .finish() + } +} + +impl ManifestSyncAgent { + pub(crate) fn spawn(io: ManifestIo, shard_id: usize) -> Self { + let (tx, rx) = flume::bounded(QUEUE_CAP); + let join = match std::thread::Builder::new() + .name(format!("manifest-sync-{shard_id}")) + .spawn(move || run(io, rx)) + { + Ok(j) => Some(j), + Err(e) => { + // Receiver is dropped with the failed spawn, so every send + // fails loudly; durable callers surface the error, deferred + // callers warn. No silent no-op. + tracing::error!(shard_id, error = %e, "manifest-sync: thread spawn failed"); + None + } + }; + Self { tx, join } + } + + /// Fire-and-forget commit of a complete root snapshot. + pub(crate) fn commit_deferred(&self, root: ManifestRoot) -> io::Result<()> { + self.tx + .send(SyncRequest::Commit { root, ack: None }) + .map_err(|_| io::Error::other("manifest-sync thread unavailable")) + } + + /// 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::>(1); + self.tx + .send(SyncRequest::Commit { + root, + ack: Some(ack_tx), + }) + .map_err(|_| io::Error::other("manifest-sync thread unavailable"))?; + ack_rx + .recv() + .map_err(|_| io::Error::other("manifest-sync thread died before ack"))? + } + + /// Flush all pending commits 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 { + let (gb_tx, gb_rx) = flume::bounded::(1); + let _ = self.tx.send(SyncRequest::Shutdown { give_back: gb_tx }); + let io = gb_rx.recv().ok(); + if let Some(j) = self.join.take() { + let _ = j.join(); + } + io + } +} + +fn run(mut io: ManifestIo, rx: flume::Receiver) { + while let Ok(req) = rx.recv() { + let (mut latest, mut acks) = match req { + SyncRequest::Shutdown { give_back } => { + let _ = give_back.send(io); + return; + } + SyncRequest::Commit { root, ack } => (root, ack.into_iter().collect::>()), + }; + + // Coalesce everything already queued: each snapshot is a complete + // root at a monotonically higher epoch, so only the newest must hit + // disk. Acks collected along the way fire after that newest persist. + let mut pending_shutdown: Option> = None; + while let Ok(more) = rx.try_recv() { + match more { + SyncRequest::Commit { root, ack } => { + latest = root; + acks.extend(ack); + } + SyncRequest::Shutdown { give_back } => { + pending_shutdown = Some(give_back); + break; + } + } + } + + let res = io.persist(latest); + if let Err(ref e) = res { + tracing::warn!(error = %e, "manifest-sync: commit failed (placement metadata only; AOF replay + orphan sweep recover on restart)"); + } + 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) = pending_shutdown { + let _ = give_back.send(io); + return; + } + } + // Channel disconnected without an explicit Shutdown: `ShardManifest` was + // dropped. flume delivers everything already queued before reporting the + // disconnect, so no accepted commit is lost; `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); + // Stall the agent on its first persist so the rest of the burst + // queues up behind it and coalesces. + super::super::manifest::TEST_INJECT_SYNC_DELAY_MS + .store(100, std::sync::atomic::Ordering::SeqCst); + for i in 0..20u64 { + m.add_file(make_entry(100 + i)); + m.commit_deferred().expect("deferred send"); + } + super::super::manifest::TEST_INJECT_SYNC_DELAY_MS + .store(0, std::sync::atomic::Ordering::SeqCst); + m.shutdown_deferred(); + 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(), + 20, + "coalescing must still land the final (superset) snapshot" + ); + } + + #[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..698143014 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 @@ -1854,6 +1862,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 +2007,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" + ); + } + } + } } // --------------------------------------------------------------------------- From ec67579eb90a9c87c5e317f2a64dc33c50f0522c Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 16 Jul 2026 14:31:41 +0700 Subject: [PATCH 02/12] =?UTF-8?q?perf(storage):=20spill-write=20pacing=20?= =?UTF-8?q?=E2=80=94=20yield=20the=20device=20to=20waiting=20cold=20reader?= =?UTF-8?q?s=20(task=20#59=20lever=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Even with all fsyncs off the event loop (lever 1), a cold pread still waits behind the spill writer's continuous multi-MB pwrite+fsync stream at the device level: raw 64KB O_DIRECT reads measured p99 208ms during a G2-shape SET flood vs 6ms idle (tmp/t59-decompose.sh, 2026-07-16). Fix: after each durable batch flush, the spill thread consults COLD_READS_INFLIGHT (a Relaxed atomic incremented by RAII guard in BOTH the async cold-read pool path and the synchronous MGET/MULTI/Lua read path, cancel-safe via Drop) and yields one 4ms quantum when a reader is waiting. Guardrails (spill_pace_after_flush, pure decision fn): - reads_inflight == 0 -> never pause (zero idle cost) - request backlog >= REQUEST_QUEUE_CAP/2 -> 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 - shutdown drain path never paces IO-priority (ioprio_set) deliberately NOT included: nix 0.31 has no wrapper, a raw libc syscall would add a new unsafe block (requires explicit approval per UNSAFE_POLICY), and it is only worth proposing if the pacing + lever-1 numbers still miss the target. Measure first. Tests: 3 unit tests on the decision function (idle-free, backlog-cutoff, quantum bounds) + RAII guard count/release test. Full lib suite 4347/0. author: Tin Dang --- src/storage/tiered/cold_read.rs | 5 ++ src/storage/tiered/cold_read_pool.rs | 27 ++++++++ src/storage/tiered/spill_thread.rs | 100 ++++++++++++++++++++++++++- 3 files changed, 131 insertions(+), 1 deletion(-) 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..cc177855b 100644 --- a/src/storage/tiered/cold_read_pool.rs +++ b/src/storage/tiered/cold_read_pool.rs @@ -41,6 +41,30 @@ 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; + +impl ColdReadInflightGuard { + pub(crate) fn new() -> Self { + COLD_READS_INFLIGHT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + Self + } +} + +impl Drop for ColdReadInflightGuard { + fn drop(&mut self) { + COLD_READS_INFLIGHT.fetch_sub(1, std::sync::atomic::Ordering::Relaxed); + } +} + struct ColdReadJob { shard_dir: PathBuf, location: ColdLocation, @@ -98,6 +122,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::(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..d86077a52 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::(4096); + let (request_tx, request_rx) = flume::bounded::(REQUEST_QUEUE_CAP); let (completion_tx, completion_rx) = flume::bounded::(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 = 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| { + 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,53 @@ 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::{COLD_READS_INFLIGHT, ColdReadInflightGuard}; + let base = COLD_READS_INFLIGHT.load(Ordering::Relaxed); + { + let _g1 = ColdReadInflightGuard::new(); + let _g2 = ColdReadInflightGuard::new(); + assert_eq!(COLD_READS_INFLIGHT.load(Ordering::Relaxed), base + 2); + } + assert_eq!(COLD_READS_INFLIGHT.load(Ordering::Relaxed), base); + } + #[test] fn test_spill_thread_new_returns_valid_handles() { let st = SpillThread::new(0); From ab6041e5dd7da96f5139066d4e591dffa49671e6 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 16 Jul 2026 15:00:17 +0700 Subject: [PATCH 03/12] docs(changelog): record task #59 off-loop manifest fsync + spill pacing author: Tin Dang --- CHANGELOG.md | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b7a140d4..d8d696520 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,27 @@ 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 - **Storage: recovery panic `double NeedsSplit after split_segment` in `DashTable::insert_or_update`.** The insert-or-update path split an overflowing From cbdcfdb991e740c295f1199f46d1d552575950fb Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 16 Jul 2026 15:00:45 +0700 Subject: [PATCH 04/12] style: cargo fmt on task #59 files author: Tin Dang --- src/persistence/manifest_sync.rs | 4 ++-- src/storage/tiered/spill_thread.rs | 5 ++++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/persistence/manifest_sync.rs b/src/persistence/manifest_sync.rs index 5752713e6..931df0096 100644 --- a/src/persistence/manifest_sync.rs +++ b/src/persistence/manifest_sync.rs @@ -255,8 +255,8 @@ mod tests { 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); + let before = + super::super::manifest::TEST_PERSIST_COUNT.load(std::sync::atomic::Ordering::SeqCst); // Stall the agent on its first persist so the rest of the burst // queues up behind it and coalesces. super::super::manifest::TEST_INJECT_SYNC_DELAY_MS diff --git a/src/storage/tiered/spill_thread.rs b/src/storage/tiered/spill_thread.rs index d86077a52..9ffe21b98 100644 --- a/src/storage/tiered/spill_thread.rs +++ b/src/storage/tiered/spill_thread.rs @@ -697,7 +697,10 @@ mod tests { 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); + assert_eq!( + spill_pace_after_flush(0, REQUEST_QUEUE_CAP), + std::time::Duration::ZERO + ); } #[test] From 47451c712ea9795a3d47144489e1292574c7e8eb Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 16 Jul 2026 15:44:17 +0700 Subject: [PATCH 05/12] =?UTF-8?q?test(harness):=20fix=20cmd=5Fflush=5Fdbsi?= =?UTF-8?q?ze=20suite=20=E2=80=94=20dead=20--persistence-dir=20flag=20+=20?= =?UTF-8?q?MOON=5FBIN-blind=20spawn?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 4 tests in cmd_flush_dbsize_debug_memory.rs have been silently broken since 1d266183 removed the --persistence-dir CLI flag (June 2026): the spawned server exits with status 2 (clap arg error) and the suite either fails or self-skips when target/release/moon is absent — which is why CI (no release binary) stayed green while any run with a built binary went red. Two fixes: - Pass --dir (the current flag) instead of the removed --persistence-dir. - Route binary discovery through common::find_moon_binary() so MOON_BIN is honored; the previous hardcoded target/release/moon fallback is the documented stale-binary trap on shared checkouts (OrbStack VM exec's a host Mach-O via proxy and the server never accepts in-VM). Verified on moon-dev: 4/4 green in 0.7s with MOON_BIN pinned to the tokio release-fast binary. Refs: unmasked while running the task #59 gate battery author: Tin Dang --- tests/cmd_flush_dbsize_debug_memory.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) 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 { "0", "--appendonly", "no", - "--persistence-dir", + "--dir", tmp_dir.to_str().unwrap(), ]) .stdout(Stdio::null()) From f0692599116579c7f8e27539016cc895d3768883 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 16 Jul 2026 15:52:23 +0700 Subject: [PATCH 06/12] test(harness): disk-guard-proof mem/oom/spsc suites + MOON_BIN-aware memory_doctor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Running the task #59 gate battery on a near-full dev volume (/Volumes/Games at <1% free) unmasked four latently environment-fragile suites: - mem_watchdog, oom_bypass_closure, spsc_two_db: root their server dirs on the repo volume, so the 5%-free diskfull write pause shadowed every assertion with MOONERR diskfull. These suites test the RSS watchdog, the -OOM eviction gate, and COPY/MOVE routing — not the disk guard — so they now pass --disk-free-min-pct 0, per the established harness convention (gotcha_diskfull_guard_gutted_crash_tests). - memory_doctor_response: hardcoded target/release/moon finder ignored MOON_BIN — on a shared macOS/Linux checkout the VM exec'd a stale host Mach-O via OrbStack's proxy and spawn_listening timed out after 30s. Now routes through common::find_moon_binary(). All four suites green on moon-dev with MOON_BIN pinned: 16/16 in <8s (previously 13 failures + one 30s timeout). author: Tin Dang --- tests/mem_watchdog.rs | 6 ++++++ tests/memory_doctor_response.rs | 5 ++++- tests/oom_bypass_closure.rs | 9 +++++++++ tests/spsc_two_db.rs | 5 +++++ 4 files changed, 24 insertions(+), 1 deletion(-) 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")) From 5a5e081d3774d866dd1af665a1279fbffdee1ed6 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 16 Jul 2026 15:53:21 +0700 Subject: [PATCH 07/12] docs(changelog): record the five test-harness fixes from the #59 gate battery author: Tin Dang --- CHANGELOG.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d8d696520..2d65a73b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,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 From b0e26ef08514d7fb73e04f04f0e28b7a140316f5 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Fri, 17 Jul 2026 00:13:14 +0700 Subject: [PATCH 08/12] =?UTF-8?q?fix(persistence):=20latest-slot=20handoff?= =?UTF-8?q?=20in=20ManifestSyncAgent=20=E2=80=94=20deferred=20commits=20ca?= =?UTF-8?q?n=20never=20block?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review finding on PR #361, confirmed real: commit_deferred used a blocking flume send into a bounded(64) request queue. The worker's try_recv drain keeps up with any live burst, but commits arriving while the worker is INSIDE a stalled persist (saturated-disk fsync, seconds are realistic) pile up with no drainer — at 64 deep the shard thread blocks on send, re-creating exactly the stall the agent exists to remove. Replace the queue with a latest-snapshot slot (parking_lot::Mutex) plus a bounded(1) wake-token channel: - commit_deferred: replace slot + try_send token — never blocks. Full is fine (state was written before the send, the pending round observes it); Disconnected still fails loudly. - commit_durable: same handoff, then blocks on its ack as before. Acks are pushed with the slot write and taken with it; firing after a newer snapshot persists remains correct (a newer root strictly supersedes). - Coalescing is now by construction: the slot holds only the newest root. Red/green: new test burst_of_deferred_commits_never_blocks_the_caller primes the worker into a 400ms-stalled persist, then bursts 100 deferred commits — the old agent blocks (FAILED, caller waits out the persist), the new agent absorbs it in <200ms. All 5 manifest_sync unit tests, the 46 crash_matrix_cross_plane kill-9 tests, cold_orphan_sweep, and manifest_gc_tombstones pass. refs #361 author: Tin Dang --- src/persistence/manifest_sync.rs | 224 +++++++++++++++++++++---------- 1 file changed, 150 insertions(+), 74 deletions(-) diff --git a/src/persistence/manifest_sync.rs b/src/persistence/manifest_sync.rs index 931df0096..cb1f5102c 100644 --- a/src/persistence/manifest_sync.rs +++ b/src/persistence/manifest_sync.rs @@ -14,100 +14,130 @@ //! handle ([`ManifestIo`]) on a dedicated per-shard thread and performs the //! exact same ordered overflow→root→fsync sequence off-loop. //! -//! Two commit flavors: -//! - **Deferred** (`commit_deferred`): fire-and-forget snapshot send; the -//! spill-completion path uses this. Consecutive queued snapshots coalesce — -//! each is a complete root, so only the newest needs to reach disk. -//! - **Durable** (`commit_durable`): send + 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. +//! 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: when several commits coalesce, 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. +//! 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 super::manifest::{ManifestIo, ManifestRoot}; +use parking_lot::Mutex; -/// Queue depth for pending commit requests. Deferred sends block only if the -/// agent falls this many complete snapshots behind (each bounded by one -/// `persist`, i.e. ~2 fsyncs) — bounded, and still off the per-file cadence -/// the loop used to pay. -const QUEUE_CAP: usize = 64; +use super::manifest::{ManifestIo, ManifestRoot}; -pub(crate) enum SyncRequest { - Commit { - root: ManifestRoot, - ack: Option>>, - }, - Shutdown { - give_back: flume::Sender, - }, +/// 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, + /// 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>>, + /// Set once by `shutdown()`; the thread gives back its [`ManifestIo`] + /// on this and exits after a final flush of the slot. + shutdown: Option>, } /// 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 { - tx: flume::Sender, + shared: Arc>, + /// 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>, } impl std::fmt::Debug for ManifestSyncAgent { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("ManifestSyncAgent") - .field("queued", &self.tx.len()) + .field("pending", &self.shared.lock().latest.is_some()) .finish() } } impl ManifestSyncAgent { pub(crate) fn spawn(io: ManifestIo, shard_id: usize) -> Self { - let (tx, rx) = flume::bounded(QUEUE_CAP); + let shared = Arc::new(Mutex::new(SyncShared { + latest: None, + acks: Vec::new(), + shutdown: None, + })); + let (wake, rx) = flume::bounded(1); + let thread_shared = Arc::clone(&shared); let join = match std::thread::Builder::new() .name(format!("manifest-sync-{shard_id}")) - .spawn(move || run(io, rx)) + .spawn(move || run(io, thread_shared, rx)) { Ok(j) => Some(j), Err(e) => { - // Receiver is dropped with the failed spawn, so every send + // Receiver is dropped with the failed spawn, so every wake // fails loudly; durable callers surface the error, deferred // callers warn. No silent no-op. tracing::error!(shard_id, error = %e, "manifest-sync: thread spawn failed"); None } }; - Self { tx, join } + Self { shared, wake, join } + } + + 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. + /// 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.tx - .send(SyncRequest::Commit { root, ack: None }) - .map_err(|_| io::Error::other("manifest-sync thread unavailable")) + 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::>(1); - self.tx - .send(SyncRequest::Commit { - root, - ack: Some(ack_tx), - }) - .map_err(|_| io::Error::other("manifest-sync thread unavailable"))?; + { + 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 all pending commits and reclaim the file-I/O state. Returns + /// 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 { let (gb_tx, gb_rx) = flume::bounded::(1); - let _ = self.tx.send(SyncRequest::Shutdown { give_back: gb_tx }); + 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(); @@ -116,37 +146,33 @@ impl ManifestSyncAgent { } } -fn run(mut io: ManifestIo, rx: flume::Receiver) { - while let Ok(req) = rx.recv() { - let (mut latest, mut acks) = match req { - SyncRequest::Shutdown { give_back } => { - let _ = give_back.send(io); - return; - } - SyncRequest::Commit { root, ack } => (root, ack.into_iter().collect::>()), +fn run(mut io: ManifestIo, shared: Arc>, 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(), + ) }; - // Coalesce everything already queued: each snapshot is a complete - // root at a monotonically higher epoch, so only the newest must hit - // disk. Acks collected along the way fire after that newest persist. - let mut pending_shutdown: Option> = None; - while let Ok(more) = rx.try_recv() { - match more { - SyncRequest::Commit { root, ack } => { - latest = root; - acks.extend(ack); - } - SyncRequest::Shutdown { give_back } => { - pending_shutdown = Some(give_back); - break; + 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 } - } - - let res = io.persist(latest); - if let Err(ref e) = res { - tracing::warn!(error = %e, "manifest-sync: commit failed (placement metadata only; AOF replay + orphan sweep recover on restart)"); - } + // 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(()), @@ -156,14 +182,15 @@ fn run(mut io: ManifestIo, rx: flume::Receiver) { let _ = ack.send(mirrored); } - if let Some(give_back) = pending_shutdown { + if let Some(give_back) = shutdown { let _ = give_back.send(io); return; } } - // Channel disconnected without an explicit Shutdown: `ShardManifest` was - // dropped. flume delivers everything already queued before reporting the - // disconnect, so no accepted commit is lost; `io` (and its fd) drop here. + // 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)] @@ -285,6 +312,55 @@ mod tests { ); } + /// 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 From c4cd94c2c60a9e561134684267df6d6b968fe61e Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Fri, 17 Jul 2026 01:23:48 +0700 Subject: [PATCH 09/12] =?UTF-8?q?fix(shard):=20honor=20snapshot=20triggers?= =?UTF-8?q?=20broadcast=20during=20shard=20startup=20=E2=80=94=20BGSAVE=20?= =?UTF-8?q?no=20longer=20hangs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The listener accepts clients as soon as the FASTEST shard's event loop is up, so a BGSAVE (or auto-save) can broadcast its snapshot epoch while a slower shard is still initializing. Each shard seeded its epoch cursor from the watch channel's CURRENT value at loop start (`last_snapshot_epoch = snapshot_trigger_rx.borrow()`), 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 until restart. Pre-existing (not introduced by this branch); surfaced as a ~15%/run crash_matrix_cross_plane flake under full-suite CPU contention, where a stack sample of a stuck instance showed all four shard threads idle in kevent (state-machine hang, not a blocked thread) and INFO showed rdb_last_save_time:0 — the FIRST BGSAVE of the instance's life had hung. Fix: start the cursor at 0. Epochs are per-process and start at 0, so 0 is always "nothing seen yet"; a trigger that raced startup is honored on the shard's first tick. Red/green: new test-only fault injection MOON_TEST_SLOW_SHARD_START_MS delays every non-zero shard's loop entry, making the race deterministic — tests/bgsave_startup_race.rs times out at 10s on the pre-fix cursor every run and completes in ~1s post-fix. 18 full crash-matrix suite runs after the fix: zero BGSAVE hangs (was the dominant failure mode). refs #361 author: Tin Dang --- CHANGELOG.md | 14 ++++ src/shard/event_loop.rs | 23 ++++++- tests/bgsave_startup_race.rs | 122 +++++++++++++++++++++++++++++++++++ 3 files changed, 158 insertions(+), 1 deletion(-) create mode 100644 tests/bgsave_startup_race.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d65a73b3..1c258292c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. - **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 diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 698143014..939a54e52 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -764,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::() + { + 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 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(); +} From 0598d8647a2227e5e0be7f058149c4aea0a59284 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Fri, 17 Jul 2026 01:23:59 +0700 Subject: [PATCH 10/12] test(crash-matrix): sync the DLQ shard's own WAL stream before the mq kill-9 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cross_plane_mq_isolated synced durability on a marker pushed to ORDERSQ and then killed -9 — but the DLQ queue lives on a different shard with an independent WAL stream that flushes on its own 1ms tick. Under full-suite CPU contention that tick can slip past the kill, so the cell measured cross-shard flush-timing luck instead of DLQ durability: autopsy of a failing run's data dir showed the dlqtestq bytes (create + both pushes + DLQ routing) absent from EVERY shard's WAL at kill time (2/12 suite runs; 0/30 isolated runs). Wait for the DLQ queue's own bytes to appear in a WAL before crashing — the product contract under test (DLQ routing recorded in the WAL survives kill-9) is unchanged and now tested exactly. 15 suite runs after the fix: zero mq failures. refs #361 author: Tin Dang --- tests/crash_matrix_cross_plane/scenarios.rs | 8 ++++++++ 1 file changed, 8 insertions(+) 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, &[]); From 25f8aa67fd74e5c478c70b42999046cad31e809b Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Fri, 17 Jul 2026 01:24:12 +0700 Subject: [PATCH 11/12] =?UTF-8?q?fix(persistence):=20review=20round=20?= =?UTF-8?q?=E2=80=94=20spawn-failure=20io=20recovery=20+=20deterministic?= =?UTF-8?q?=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three remaining review findings on PR #361, all verified real: 1. A failed manifest-sync thread spawn (thread limit, EAGAIN) destroyed the working inline backend: the ManifestIo moved into the dropped closure, leaving a manifest that could never commit again until restart. spawn() now hands the io to the thread over a channel and returns Result — on failure enable_deferred_sync keeps IoBackend::Inline and logs, degrading to on-loop fsyncs instead of no durability at all. 2. queued_deferred_commits_coalesce_to_fewer_persists raced its own delay knob: the injected persist stall was reset right after the burst, so scheduling could let the worker miss it and weaken the coalescing bound into flakiness. The test now primes the worker INSIDE a stalled persist before bursting and keeps the delay armed through the shutdown flush. 3. reader_inflight_guard_counts_and_releases asserted exact values on the process-global COLD_READS_INFLIGHT while sibling tests in the same binary exercise the cold-read path concurrently. The guard gains ColdReadInflightGuard::on(&'static AtomicUsize) and the test verifies the RAII mechanics on its own local counter. refs #361 author: Tin Dang --- src/persistence/manifest.rs | 19 ++++++++-- src/persistence/manifest_sync.rs | 57 +++++++++++++++++++--------- src/storage/tiered/cold_read_pool.rs | 16 ++++++-- src/storage/tiered/spill_thread.rs | 16 +++++--- 4 files changed, 77 insertions(+), 31 deletions(-) diff --git a/src/persistence/manifest.rs b/src/persistence/manifest.rs index 052a57c22..ec7f9ddbd 100644 --- a/src/persistence/manifest.rs +++ b/src/persistence/manifest.rs @@ -541,9 +541,22 @@ impl ShardManifest { 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) => IoBackend::Deferred( - crate::persistence::manifest_sync::ManifestSyncAgent::spawn(io, shard_id), - ), + 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, }; } diff --git a/src/persistence/manifest_sync.rs b/src/persistence/manifest_sync.rs index cb1f5102c..1d64c783e 100644 --- a/src/persistence/manifest_sync.rs +++ b/src/persistence/manifest_sync.rs @@ -78,28 +78,44 @@ impl std::fmt::Debug for ManifestSyncAgent { } impl ManifestSyncAgent { - pub(crate) fn spawn(io: ManifestIo, shard_id: usize) -> Self { + /// 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 { 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::(1); let thread_shared = Arc::clone(&shared); - let join = match std::thread::Builder::new() + match std::thread::Builder::new() .name(format!("manifest-sync-{shard_id}")) - .spawn(move || run(io, thread_shared, rx)) - { - Ok(j) => Some(j), - Err(e) => { - // Receiver is dropped with the failed spawn, so every wake - // fails loudly; durable callers surface the error, deferred - // callers warn. No silent no-op. - tracing::error!(shard_id, error = %e, "manifest-sync: thread spawn failed"); - None - } - }; - Self { shared, wake, join } + .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<()> { @@ -284,17 +300,22 @@ mod tests { let before = super::super::manifest::TEST_PERSIST_COUNT.load(std::sync::atomic::Ordering::SeqCst); - // Stall the agent on its first persist so the rest of the burst - // queues up behind it and coalesces. + // 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); - m.shutdown_deferred(); let persists = super::super::manifest::TEST_PERSIST_COUNT .load(std::sync::atomic::Ordering::SeqCst) - before; @@ -307,7 +328,7 @@ mod tests { let reopened = ShardManifest::open(&path).expect("reopen"); assert_eq!( reopened.files().len(), - 20, + 21, "coalescing must still land the final (superset) snapshot" ); } diff --git a/src/storage/tiered/cold_read_pool.rs b/src/storage/tiered/cold_read_pool.rs index cc177855b..640a2a837 100644 --- a/src/storage/tiered/cold_read_pool.rs +++ b/src/storage/tiered/cold_read_pool.rs @@ -50,18 +50,26 @@ pub(crate) static COLD_READS_INFLIGHT: std::sync::atomic::AtomicUsize = /// 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; +pub(crate) struct ColdReadInflightGuard(&'static std::sync::atomic::AtomicUsize); impl ColdReadInflightGuard { pub(crate) fn new() -> Self { - COLD_READS_INFLIGHT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - 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) { - COLD_READS_INFLIGHT.fetch_sub(1, std::sync::atomic::Ordering::Relaxed); + self.0.fetch_sub(1, std::sync::atomic::Ordering::Relaxed); } } diff --git a/src/storage/tiered/spill_thread.rs b/src/storage/tiered/spill_thread.rs index 9ffe21b98..aa2cd5beb 100644 --- a/src/storage/tiered/spill_thread.rs +++ b/src/storage/tiered/spill_thread.rs @@ -730,14 +730,18 @@ mod tests { #[test] fn reader_inflight_guard_counts_and_releases() { - use super::super::cold_read_pool::{COLD_READS_INFLIGHT, ColdReadInflightGuard}; - let base = COLD_READS_INFLIGHT.load(Ordering::Relaxed); + 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::new(); - let _g2 = ColdReadInflightGuard::new(); - assert_eq!(COLD_READS_INFLIGHT.load(Ordering::Relaxed), base + 2); + let _g1 = ColdReadInflightGuard::on(&LOCAL); + let _g2 = ColdReadInflightGuard::on(&LOCAL); + assert_eq!(LOCAL.load(Ordering::Relaxed), 2); } - assert_eq!(COLD_READS_INFLIGHT.load(Ordering::Relaxed), base); + assert_eq!(LOCAL.load(Ordering::Relaxed), 0); } #[test] From 737c08a787b83ca99b70adc4b1c4290dc53bbe27 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Fri, 17 Jul 2026 08:24:18 +0700 Subject: [PATCH 12/12] fix(shard): tokio arm never decremented the BGSAVE fan-in counter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tests/bgsave_startup_race.rs failed in CI (runtime-tokio): the tokio shard loop's snapshot-finalize path called finalize_snapshot_success/error but — unlike the monoio arm — never called bgsave_shard_done, so BGSAVE_SHARDS_REMAINING stayed at num_shards and rdb_bgsave_in_progress stuck at 1 after EVERY sharded BGSAVE under runtime-tokio. Pre-existing and CI-invisible until now: the crash suites that await BGSAVE are `#[ignore]`d (run only in the local VM battery, which uses monoio), so no CI test had ever waited for a tokio BGSAVE to complete. Mirror the monoio arm: bgsave_shard_done(true/false) after finalize. Safe for auto-save-triggered snapshots — the counter ignores calls at 0. Red: CI Check job, 3/3 retries timing out at the 10s deadline (and the same locally with --features runtime-tokio,jemalloc). Green: 0.99s local tokio run; monoio leg unaffected (1.35s). refs #361 author: Tin Dang --- CHANGELOG.md | 6 +++++- src/shard/event_loop.rs | 7 +++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c258292c..a46c95306 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,7 +41,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. + 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 diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 939a54e52..1f7cb0a6d 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -1587,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; } }