Skip to content

fix(persistence): TopLevel-monoio AOF writer honors EverySec bound when idle #233

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pilotspacex-byte merged 2 commits into from
Jul 7, 2026
Merged

fix(persistence): TopLevel-monoio AOF writer honors EverySec bound when idle #233

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e9936ed
fix(persistence): TopLevel-monoio AOF writer honors EverySec bound wh…
TinDang97 Jul 7, 2026
fa04884
test(persistence): crash-matrix harness — pin SET reply, serialize se…
TinDang97 Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

## [Unreleased]

### Fixed — TopLevel-monoio AOF writer: EverySec fsync deferred indefinitely when idle (PR #TBD)

@coderabbitai coderabbitai Bot Jul 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Placeholder PR number left in changelog heading.

(PR #TBD) should be replaced with the actual PR number before merge.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@CHANGELOG.md` at line 9, The changelog heading still contains the placeholder
PR reference for the TopLevel-monoio AOF writer fix; update the entry in
CHANGELOG.md to replace “(PR `#TBD`)” with the actual merged PR number, keeping
the rest of the heading unchanged.


- **`src/persistence/aof/writer_task.rs`**: the TopLevel monoio AOF writer
blocked on an **untimed** `rx.recv()`, with its EverySec deadline check
living only inside the batch-commit path. A batch written under
`appendfsync everysec` gets no per-batch fsync, so if the client stopped
writing right after a burst, the buffered bytes only became durable when
the NEXT message happened to arrive — the 1s fsync bound was deferred
indefinitely while idle. Exposure is host-crash-only (the per-batch
`flush()` already reaches the kernel page cache, so a plain process kill
loses nothing), which is exactly the window EverySec exists to bound.
The loop now mirrors the audited PerShard writers: bounded
`recv_timeout` on the wave-5 `IdleWait` ladder (50ms → 250ms → 1s,
pinned at the floor while a batch awaits its fsync) plus an end-of-loop
proactive fsync that runs on message AND timeout iterations. This
supersedes wave 5's "TopLevel monoio needs none of this" note — it was
the one writer loop left without the ≤ ~1s idle durability bound.
- **`tests/crash_matrix_per_shard_aof.rs` harness hardening** (found while
validating the above): (1) `redis_set` asserted only redis-cli's exit
status, which is 0 even for server ERROR replies — a tripped diskfull
guard (host <5% free) turned every SET into a silent no-op and surfaced
as a bogus "200 keys missing after recovery"; the helper now pins the
reply to `+OK`. (2) The three server-spawning tests split-brain when run
in parallel: `unique_port()` hands out OS-sequential ephemeral ports, the
other tests offset +1/+2, and SO_REUSEPORT lets two tests bind the SAME
port without an error — one test's redis-cli traffic lands on another
test's server (rotating total-loss false alarms). A shared mutex now
serializes them.

### Fixed — RSS/CPU remediation wave 5 (PR #TBD)

- **Item A — mmap the exact-rerank f16 sidecar on segment reload**
Expand Down
253 changes: 143 additions & 110 deletions src/persistence/aof/writer_task.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,9 +25,8 @@ use super::group_commit::{GroupCommitSink, commit_group_commit_batch};
/// Idle-adaptive wake cadence for a background AOF writer's channel poll
/// (RSS/CPU wave 5, item B).
///
/// The steady-state writer loops (PerShard monoio/tokio, TopLevel tokio —
/// TopLevel monoio blocks on an untimed `rx.recv()` and needs none of this)
/// poll their channel with a bounded timeout so the EverySec proactive-fsync
/// All four steady-state writer loops (PerShard monoio/tokio, TopLevel
/// monoio/tokio) poll their channel with a bounded timeout so the EverySec proactive-fsync
/// deadline check that follows every wake still fires when no new Appends
/// ever arrive. A FIXED cadence forever (previously 50ms monoio / 200ms
/// tokio) means an idle server's AOF writer thread wakes 5-20 times a
Expand Down Expand Up @@ -356,13 +355,34 @@ pub async fn aof_writer_task(
// Read once at task startup; zero cost in production (var absent).
let fail_fsync_for_test = std::env::var("MOON_TEST_AOF_FSYNC_FAIL").as_deref() == Ok("1");

// Idle-adaptive channel-poll wake cadence (RSS/CPU wave 5, item B) —
// see `IdleWait` docs near the top of this file. This loop used to
// block on an UNTIMED `rx.recv()`, with the EverySec deadline check
// only inside the batch path: a batch buffered without a per-batch
// fsync only became durable when the NEXT message happened to
// arrive, so idle-after-a-burst deferred the fsync indefinitely
// (host-crash exposure only — the per-batch flush already reaches
// the kernel page cache, so a plain process kill loses nothing).
// The bounded recv + end-of-loop proactive fsync below restore the
// ~1s EverySec bound exactly like the PerShard writers.
let mut idle_wait = IdleWait::new();

loop {
// Group commit: block for one message, then opportunistically drain
// whatever else is already queued into a bounded batch so a single
// fsync makes the whole batch durable (TopLevel = plain RESP bytes).
let first = match rx.recv() {
Ok(m) => m,
Err(_) => {
// Group commit: wait (bounded) for one message, then
// opportunistically drain whatever else is already queued into a
// bounded batch so a single fsync makes the whole batch durable
// (TopLevel = plain RESP bytes). On timeout, fall through (None)
// to the EverySec proactive fsync at the end of the loop.
let first = match rx.recv_timeout(idle_wait.current()) {
Ok(m) => {
idle_wait.on_message();
Some(m)
}
Err(flume::RecvTimeoutError::Timeout) => {
idle_wait.on_timeout();
None
}
Err(flume::RecvTimeoutError::Disconnected) => {
// Channel disconnected — final sync + shut down.
if !write_error {
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
Expand All @@ -373,123 +393,136 @@ pub async fn aof_writer_task(
break;
}
};
let mut batch = collect_group_commit_batch(
first,
|| rx.try_recv().ok(),
AOF_GROUP_COMMIT_MAX_BATCH,
AOF_GROUP_COMMIT_MAX_BYTES,
);

// -- commit the data batch (one fsync under Always; deadline under everysec) --
if !batch.data.is_empty() {
if write_error {
// Persistent I/O failure latched: drop appends and fail every
// AppendSync waiter — never a false durability claim.
let _ = group_commit::ack_batch(&mut batch, BatchAck::WriteFailed);
} else {
let do_fsync = matches!(fsync, FsyncPolicy::Always);
let mut sink = FileGroupSink {
file: &mut file,
fail_sync: fail_fsync_for_test,
};
let outcome = commit_group_commit_batch(&mut sink, &mut batch, do_fsync);
if outcome.write_failed {
// A torn write may leave a partial record — latch so no
// further bytes are appended after the tear.
error!(
"AOF batch write failed (seq {}). Persistence degraded.",
manifest.seq
);
write_error = true;
}
// EverySec: the batch was written but not per-batch-fsynced
// (do_fsync=false; there are no AppendSync waiters under
// everysec). Honor the 1s deadline exactly as the old
// per-Append path did.
if fsync == FsyncPolicy::EverySec
&& !write_error
&& last_fsync.elapsed() >= std::time::Duration::from_secs(1)
{
let t = Instant::now();
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF sync failed (seq {}, everysec): {}", manifest.seq, e);
// Non-fatal for everysec: retry next interval
} else {
crate::admin::metrics_setup::record_aof_fsync(
t.elapsed().as_micros() as u64
if let Some(first) = first {
let mut batch = collect_group_commit_batch(
first,
|| rx.try_recv().ok(),
AOF_GROUP_COMMIT_MAX_BATCH,
AOF_GROUP_COMMIT_MAX_BYTES,
);

// -- commit the data batch (one fsync under Always; deadline under everysec) --
if !batch.data.is_empty() {
if write_error {
// Persistent I/O failure latched: drop appends and fail every
// AppendSync waiter — never a false durability claim.
let _ = group_commit::ack_batch(&mut batch, BatchAck::WriteFailed);
} else {
let do_fsync = matches!(fsync, FsyncPolicy::Always);
let mut sink = FileGroupSink {
file: &mut file,
fail_sync: fail_fsync_for_test,
};
let outcome = commit_group_commit_batch(&mut sink, &mut batch, do_fsync);
if outcome.write_failed {
// A torn write may leave a partial record — latch so no
// further bytes are appended after the tear.
error!(
"AOF batch write failed (seq {}). Persistence degraded.",
manifest.seq
);
last_fsync = Instant::now();
write_error = true;
}
// EverySec: the batch was written but not per-batch-fsynced
// (do_fsync=false; there are no AppendSync waiters under
// everysec). The end-of-loop proactive fsync makes it
// durable within the 1s bound — pin the idle wait at its
// fast floor until that fsync clears it.
if fsync == FsyncPolicy::EverySec && !write_error {
idle_wait.mark_pending();
}
}
}
}

// -- handle the control message that ended the drain (if any) --
// A control message is NEVER absorbed into the batch: the batch above
// is already committed before the control message is handled
// (batch_straddles_control is structurally impossible).
match batch.deferred_control {
None => {}
Some(AofMessage::Shutdown) => {
if !write_error {
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF final sync failed (seq {}): {}", manifest.seq, e);
// -- handle the control message that ended the drain (if any) --
// A control message is NEVER absorbed into the batch: the batch above
// is already committed before the control message is handled
// (batch_straddles_control is structurally impossible).
match batch.deferred_control {
None => {}
Some(AofMessage::Shutdown) => {
if !write_error {
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF final sync failed (seq {}): {}", manifest.seq, e);
}
}
info!("AOF writer shutting down (monoio, seq {})", manifest.seq);
break;
}
info!("AOF writer shutting down (monoio, seq {})", manifest.seq);
break;
}
Some(AofMessage::Rewrite(db)) => {
if !write_error {
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF pre-rewrite sync failed (seq {}): {}", manifest.seq, e);
Some(AofMessage::Rewrite(db)) => {
if !write_error {
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF pre-rewrite sync failed (seq {}): {}", manifest.seq, e);
}
}
}
match do_rewrite_single(&db, &mut manifest, &mut file, &rx) {
Ok(()) => {
write_error = false; // Reset on successful rewrite
match do_rewrite_single(&db, &mut manifest, &mut file, &rx) {
Ok(()) => {
write_error = false; // Reset on successful rewrite
}
Err(e) => error!("AOF rewrite failed (seq {}): {}", manifest.seq, e),
}
Err(e) => error!("AOF rewrite failed (seq {}): {}", manifest.seq, e),
crate::command::persistence::AOF_REWRITE_IN_PROGRESS
.store(false, std::sync::atomic::Ordering::SeqCst);
}
crate::command::persistence::AOF_REWRITE_IN_PROGRESS
.store(false, std::sync::atomic::Ordering::SeqCst);
}
Some(AofMessage::RewriteSharded(shard_dbs)) => {
if !write_error {
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF pre-rewrite sync failed (seq {}): {}", manifest.seq, e);
Some(AofMessage::RewriteSharded(shard_dbs)) => {
if !write_error {
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF pre-rewrite sync failed (seq {}): {}", manifest.seq, e);
}
}
}
// C4 TopLevel cooperative fold: pass the wired fold channels
// (producer + notifier for shard 0) so do_rewrite_sharded can
// use the AofFold SPSC protocol instead of the deleted RwLock
// path. `fold_channels` is `None` only if main.rs failed to
// wire them at startup (Arc::get_mut race — logged at boot).
match do_rewrite_sharded(
&shard_dbs,
&mut manifest,
&mut file,
&rx,
fold_channels.as_ref(),
) {
Ok(()) => {
write_error = false;
// C4 TopLevel cooperative fold: pass the wired fold channels
// (producer + notifier for shard 0) so do_rewrite_sharded can
// use the AofFold SPSC protocol instead of the deleted RwLock
// path. `fold_channels` is `None` only if main.rs failed to
// wire them at startup (Arc::get_mut race — logged at boot).
match do_rewrite_sharded(
&shard_dbs,
&mut manifest,
&mut file,
&rx,
fold_channels.as_ref(),
) {
Ok(()) => {
write_error = false;
}
Err(e) => error!("AOF rewrite failed (seq {}): {}", manifest.seq, e),
}
Err(e) => error!("AOF rewrite failed (seq {}): {}", manifest.seq, e),
crate::command::persistence::AOF_REWRITE_IN_PROGRESS
.store(false, std::sync::atomic::Ordering::SeqCst);
}
crate::command::persistence::AOF_REWRITE_IN_PROGRESS
.store(false, std::sync::atomic::Ordering::SeqCst);
// [F6] A TopLevel writer never owns per-shard files; receiving
// RewritePerShard means a routing bug. Self-abort so the
// coordinator's countdown completes and the flag clears.
Some(AofMessage::RewritePerShard { coord, .. }) => {
warn!(
"AOF TopLevel writer received RewritePerShard — routing bug; aborting"
);
coord.mark_failed();
coord.shard_done();
}
// collect_group_commit_batch only ever defers a control message.
Some(_) => {}
}
Comment on lines +438 to 506

@coderabbitai coderabbitai Bot Jul 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Check whether the PerShard writer's rewrite handling back-dates
# last_fsync/marks idle_wait pending, to confirm the intended parity.
rg -n "last_fsync|idle_wait" src/persistence/aof/*.rs | rg -n "rewrite" -A5 -B5 -i
fd -e rs . src/persistence/aof

Repository: pilotspace/moon

Length of output: 310

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Map the relevant files first.
ast-grep outline src/persistence/aof/writer_task.rs --view expanded
printf '\n---\n'
ast-grep outline src/persistence/aof/rewrite.rs --view expanded
printf '\n---\n'
ast-grep outline src/persistence/aof/mod.rs --view expanded

Repository: pilotspace/moon

Length of output: 5156

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Read the monoio writer loop around the rewrite handling.
sed -n '430,520p' src/persistence/aof/writer_task.rs

printf '\n--- TOKIO / TOPLEVEL REWRITE HANDLING ---\n'
sed -n '680,770p' src/persistence/aof/writer_task.rs

printf '\n--- REWRITE HELPERS ---\n'
sed -n '780,980p' src/persistence/aof/rewrite.rs

Repository: pilotspace/moon

Length of output: 19393

🏁 Script executed:

#!/bin/bash
set -euo pipefail

# Find every place the writer back-dates fsync or marks pending,
# to compare monoio vs tokio handling around rewrites.
rg -n "last_fsync\s*=|mark_pending\(\)" src/persistence/aof/writer_task.rs

Repository: pilotspace/moon

Length of output: 1313

Back-date fsync state after monoio rewrites src/persistence/aof/writer_task.rs:453-493 — after a successful Rewrite / RewriteSharded, the monoio path resets write_error but leaves last_fsync and idle_wait untouched. That can leave the post-rewrite backlog on an escalated timeout and delay the next proactive fsync by up to the full idle cadence. Mirror the tokio path here with the same back-date + idle_wait.mark_pending().

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/persistence/aof/writer_task.rs` around lines 438 - 506, The monoio
rewrite handling in writer_task’s deferred control message match resets
write_error after successful Rewrite and RewriteSharded, but it does not refresh
the fsync scheduling state. Update the successful paths in
do_rewrite_single/do_rewrite_sharded handling to mirror the tokio behavior by
back-dating last_fsync and calling idle_wait.mark_pending(), so the next
proactive fsync is not delayed after a rewrite.

// [F6] A TopLevel writer never owns per-shard files; receiving
// RewritePerShard means a routing bug. Self-abort so the
// coordinator's countdown completes and the flag clears.
Some(AofMessage::RewritePerShard { coord, .. }) => {
warn!("AOF TopLevel writer received RewritePerShard — routing bug; aborting");
coord.mark_failed();
coord.shard_done();
}

// EverySec proactive fsync — runs after every loop iteration
// (message processed OR timeout); the only path that guarantees
// the ~1s durability bound when no further messages arrive after
// a buffered batch (idle-after-a-burst).
if fsync == FsyncPolicy::EverySec
&& !write_error
&& last_fsync.elapsed() >= std::time::Duration::from_secs(1)
{
let t = Instant::now();
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF sync failed (seq {}, everysec): {}", manifest.seq, e);
// Non-fatal for everysec: retry next interval
} else {
crate::admin::metrics_setup::record_aof_fsync(t.elapsed().as_micros() as u64);
last_fsync = Instant::now();
idle_wait.clear_pending();
}
// collect_group_commit_batch only ever defers a control message.
Some(_) => {}
}
}
return;
Expand Down
Loading
Loading