fix(persistence): close 8 CodeRabbit PR #136 durability findings + aof_manifest split

[pilotspacex-byte]

Summary

Closes all 8 findings from the CodeRabbit review of #136 (review 4420584384) that the merged PR did not address — AOF / disk-offload durability bugs plus the aof_manifest.rs line-cap refactor. Each fix was validated with red/green TDD in the OrbStack Linux VM and is green on both runtimes.

Verification (all green): cargo test --lib monoio 3457 passed, tokio,jemalloc 2861 passed; cargo clippy -- -D warnings (default + tokio,jemalloc); cargo fmt --check. Per-shard AOF SIGKILL recovery + per-shard BGREWRITEAOF crash-recovery integration suites green.

Fixes (issue -> commit)

#	Issue	Fix	Commit
5	spill completion drop	dropped SpillCompletion = permanent cold-tier data loss -> blocking send_timeout backpressure + drain-on-shutdown	6cac58f
4	inline-batch spill	wholesale success:false on inline-batch failure -> per-entry fallback salvage	deae60d
6 (#141)	migrated tokio conn	spawn_migrated_tokio_connection dropped spill context -> thread it through, mirroring the 3 sibling spawn paths	cf90800
7 (#142)	embedded parity	run_embedded skipped cold recovery under appendonly=no + re-enabled the per-shard-accept race -> should_run_recovery gate + central accept	4828b3d
2 (#138)	rewrite fan-out	partial fan-out send failure wedged AOF_REWRITE_IN_PROGRESS forever -> mark_failed + shard_done the unsent tail	fa9afe9
1	rewrite abort	folded writers kept appending into the pruned new generation after an abort (silent loss) -> barrier-before-resume (PerShardRewriteCoord publishes the committed seq; writers await + reopen onto it) + ShardDoneGuard so a fold panic can't hang every other writer at the barrier (closes the #138 panic-wedge too)	c919ff5
3 (#140)	AppendSync ack	drained AppendSync acked Synced before the boundary fsync (breaks appendfsync=always) -> park acks, fulfil Synced/FsyncFailed strictly after sync_data	73d9514
8 (#143)	1500-line cap	split aof_manifest.rs (3058 lines) -> aof_manifest/{mod.rs, shard_replay.rs, shard_rewrite.rs}, all under cap, public paths preserved via pub use	14c572b

Notes / design

• Deadlock-free barrier: each per-shard AOF writer runs on its own dedicated OS thread (aof-writer-{sid} -> block_on_local) on both runtimes, so a synchronous barrier wait can never starve the thread that must publish.
• Panic safety is load-bearing: [profile.release] is panic = "unwind" (no panic = "abort"), so the ShardDoneGuard is what prevents a save_snapshot_to_bytes OOM-unwind from hanging the whole writer pool.
• Pre-existing flake (not introduced here): tests/aof_fsync_err_subscribe_ordering::aof_fsync_err_propagates_before_subscribe_single_shard (an #[ignore]d normal-append-path test; the multi-shard twin passes) fails identically on the base commit.
• aof.rs itself is now 4379 lines (>1500) after these durability fixes — tracked separately from Refactor: split src/persistence/aof_manifest.rs (>1500-line cap) into per-shard rewrite/replay submodules #143, not split here.

Test plan

• cargo test --lib (monoio) — 3457 pass
• cargo test --lib --no-default-features --features runtime-tokio,jemalloc — 2861 pass
• cargo clippy -- -D warnings (default + tokio,jemalloc)
• cargo fmt --check
• crash_matrix_per_shard_aof + crash_matrix_per_shard_bgrewriteaof (--ignored) — 5 pass

Summary by CodeRabbit

• Bug Fixes

  • Fixed rewrite deadlocks and stalled rewrite state; ensured rewrite aborts never leave writers blocked and addressed ack ordering/durability across rewrite boundaries. Prevented silent data loss on shutdown by surfacing final spill completions and blocking when completion channel is full.

• New Features

  • Added per-shard AOF layout with multi-part replay and coordinated per-shard rewrite/prune lifecycle. Improved drain/ack boundary semantics and configurable fsync durability.

• Tests

  • Extensive unit tests for rewrite fan-out, panic-safety, replay correctness, spill shutdown, and metrics concurrency.

• Chores

  • Preserve disk-offload context across connection migration and refined recovery/accept behavior.

[qodo-code-review]

Qodo reviews are paused for this user.

Troubleshooting steps vary by plan Learn more →

On a Teams plan?
Reviews resume once this user has a paid seat and their Git account is linked in Qodo.
Link Git account →

Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center?
These require an Enterprise plan - Contact us
Contact us →

[coderabbitai]

Warning

Review limit reached

@TinDang97, we couldn't start this review because you've reached your PR review rate limit.

More reviews will be available in 13 minutes and 18 seconds. Learn how PR review limits work.

Your organization has run out of usage credits. Purchase more in the billing tab.

⌛ How to resolve this issue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available.

Please see our Fair Usage Limits Policy for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 2e4c56ab-c7cb-4e7a-bedc-964f07eece19

📥 Commits

Reviewing files that changed from the base of the PR and between edffce3 and e41f065.

📒 Files selected for processing (2)

• CHANGELOG.md
• src/command/vector_search/tests.rs

📝 Walkthrough

Walkthrough

This PR adds a per-shard AOF manifest (v1/v2) and per-shard replay, introduces a committed-generation barrier and panic-safe shard-done guard for BGREWRITEAOF coordination with deferred ack fulfillment across fsync boundaries, prevents spill-thread completion loss by blocking sends and returning final completions on shutdown, and preserves disk-offload state across migrated connections and startup recovery.

Changes

Per-Shard Persistence and Durability

Layer / File(s)	Summary
AOF manifest types and foundation src/persistence/aof_manifest/mod.rs	Defines AofLayout (TopLevel vs PerShard), ShardManifest and AofManifest with path helpers, init/load/write/advance/migrate, orphan cleanup, and detailed v2 tests.
Per-shard rewrite initialization and advancement src/persistence/aof_manifest/shard_rewrite.rs	Adds initialize_multi, try_initialize_multi, advance_shard, and prune_shard_files with atomic tmp+fsync+rename semantics, rollback on partial failure, and tests for idempotency and crash-safety ordering.
Per-shard AOF replay with ordered merge src/persistence/aof_manifest/shard_replay.rs	Implements replay_multi_part, replay_incr_resp, framed per-shard replay buffering ordered entries, replay_per_shard (parallel per-shard replay), and replay_ordered_merge with torn-commit heuristics and tests.
Per-shard BGREWRITEAOF coordination with ack-ordering barrier src/persistence/aof.rs	Adds PerShardRewriteCoord outcome slot + Condvar, ShardDoneGuard for panic-safe publication, fan-out partial-failure clearing, parked AppendSync ack tracking (pending_acks), and sync_and_fulfill_drain helpers to fsync then fulfill acks in correct durability order; updates all rewrite paths and writer handlers.
Rewrite implementations and writer integration src/persistence/aof/rewrite.rs, src/persistence/aof/writer_task.rs	Adds drain helpers (framed/non-framed), per-shard fold/single/sharded rewrite implementations, synchronous rewrite helpers, and runtime-gated writer-task integration with framed writes, ack semantics, and test hooks.
AOF writer pool and tests src/persistence/aof/pool.rs, src/persistence/aof/mod.rs	Introduces AofWriterPool, durable append APIs, per-shard rewrite fan-out with partial-send accounting, fsync policies, ordered-append LSN tagging, and many unit tests for routing, ack/backpressure, and rewrite scenarios.
Spill thread completion blocking and shutdown safety src/storage/tiered/spill_thread.rs	Replaces drop-on-full try_send with blocking send_timeout loop sensitive to shutdown, adds spill_single_entry, makes shutdown() return Vec<SpillCompletion> of unapplied final completions, and extends tests for blocking and returned completions.
Server startup recovery and connection migration disk-offload wiring src/server/embedded.rs, src/shard/conn_accept.rs, src/shard/event_loop.rs, src/shard/persistence_tick.rs	Adds should_run_recovery to decide pre-loop restore when persistence or disk-offload is present, preserves spill context across migrated connections by extending spawn_migrated_tokio_connection and wiring call sites, and applies final spill completions on shutdown via apply_completion_vec.

Sequence Diagram(s)

sequenceDiagram
  participant Client
  participant AofWriterPool
  participant PerShardWriter
  participant PerShardRewriteCoord
  participant Filesystem
  Client->>AofWriterPool: send RewritePerShard(coord)
  AofWriterPool->>PerShardWriter: RewritePerShard(coord)
  PerShardWriter->>PerShardRewriteCoord: await_outcome()
  PerShardRewriteCoord->>Filesystem: publish outcome (new_seq / old_seq)
  PerShardRewriteCoord-->>PerShardWriter: wake with committed seq
  PerShardWriter->>Filesystem: reopen append file on committed generation

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~75 minutes

Possibly related issues

• per-shard BGREWRITEAOF: panic in do_rewrite_per_shard wedges rewrite flag + kills shard AOF writer #138: Changes implement panic-safety and shard-done guard behavior matching that issue.
• Embedded: run_embedded() missing main.rs startup parity (disk-offload recovery gate + central-accept) #142: Startup recovery gating and accept-path adjustments overlap with this PR's embedded/server changes.
• Refactor: split src/persistence/aof_manifest.rs (>1500-line cap) into per-shard rewrite/replay submodules #143: The shard_replay/shard_rewrite extraction and manifest work correspond to the manifest/replay additions here.

Possibly related PRs

• pilotspace/moon#136: Earlier per-shard rewrite coordination and fan-out work closely related to this PR.
• feat(tier-2): Lane A — SWAPDB, MOVE, COPY DB n, CLUSTER REPLICAS/SLAVES, COUNT-FAILURE-REPORTS #100: Prior BGREWRITEAOF coordination refinements that interact with rewrite-in-progress gating.
• feat: multi-part AOF persistence + fast RDB loader #63: Initial sharded BGREWRITEAOF command wiring that this PR builds upon.

Suggested labels

enhancement

"From burrows deep I hop and write,
Manifests in twos take flight tonight.
Acks wait patient at the fsync gate,
Spill threads block to guard our crate.
Guards stand firm when rewrites roar — hooray! 🥕"

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly summarizes the main focus: fixing 8 CodeRabbit PR #136 durability findings and splitting aof_manifest.
Description check	✅ Passed	The description is comprehensive with a clear summary, detailed fix table, notes on design decisions, and explicit test verification across both runtimes.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch fix/coderabbit-pr136-durability

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

src/shard/conn_accept.rs (1)

277-320: ⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Gate spawn_migrated_tokio_connection behind #[cfg(unix)] (or add a non-Unix stub)

In src/shard/conn_accept.rs, spawn_migrated_tokio_connection is only #[cfg(feature = "runtime-tokio")], but its signature/body use Unix-only std APIs (std::os::unix::io::RawFd, FromRawFd, and std::net::TcpStream::from_raw_fd). src/shard/event_loop.rs calls it under the same runtime-tokio-only guard, with no unix gating—so enabling runtime-tokio on non-Unix targets will not compile. Add #[cfg(unix)] around the function (and its call sites) or provide a non-Unix stub.

🤖 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/shard/conn_accept.rs` around lines 277 - 320, The function
spawn_migrated_tokio_connection uses Unix-only APIs but is only feature-gated by
runtime-tokio; gate it for Unix targets (e.g. add #[cfg(all(feature =
"runtime-tokio", unix))] to the function) or add a non-Unix stub with the same
signature (behind #[cfg(all(feature = "runtime-tokio", not(unix)))] ) that
returns early/panics with a clear message; also apply the same cfg change to
every call site (e.g. in src/shard/event_loop.rs) so calls are only compiled on
the matching cfg or call the stub, ensuring consistent signatures across
targets.

🧹 Nitpick comments (4)

src/persistence/aof_manifest/shard_rewrite.rs (1)

73-106: 💤 Low value

Consider cleaning up incr files during rollback.

The rollback logic removes base RDB files for successfully-created shards, but doesn't remove the corresponding incr files created at line 87. While this is not a correctness issue (orphan cleanup will remove them on next load), it would be more complete to also remove incr files during rollback.

♻️ Optional: also remove incr files during rollback

         if let Err(e) = loop_result {
             // Rollback: remove base RDB files for all successfully-created shards.
             for sid in created_shards {
                 let base = manifest.shard_base_path(sid);
                 if let Err(re) = std::fs::remove_file(&base) {
                     warn!(
                         "initialize_multi rollback: failed to remove {}: {}",
                         base.display(),
                         re
                     );
                 }
+                let incr = manifest.shard_incr_path(sid);
+                if let Err(re) = std::fs::remove_file(&incr) {
+                    warn!(
+                        "initialize_multi rollback: failed to remove {}: {}",
+                        incr.display(),
+                        re
+                    );
+                }
             }
             return Err(e);
         }

🤖 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_manifest/shard_rewrite.rs` around lines 73 - 106, The
rollback after loop_result currently only removes base RDB files for entries in
created_shards; extend that rollback to also remove the corresponding incr files
(use manifest.shard_incr_path(sid)) inside the same for sid in created_shards
loop or a follow-up loop, attempting std::fs::remove_file and emitting the same
warn! message on failure (include incr path.display() and the error) so orphan
incr files are cleaned up on rollback.

src/persistence/aof_manifest/shard_replay.rs (3)

236-243: ⚡ Quick win

Replace expect() with infallible conversion pattern.

The expect() calls at lines 240 and 243 are technically unreachable (the bounds check at line 228 guarantees the slices are valid), but the coding guidelines discourage expect() in library code outside tests. The conversion from a length-matched slice to a fixed-size array is actually infallible.

♻️ Use infallible array conversion

-        #[allow(clippy::unwrap_used)] // bounds-checked above; try_into is statically length-matched
-        let raw_lsn = u64::from_le_bytes(data[offset..offset + 8].try_into().expect("8 bytes"));
-        #[allow(clippy::unwrap_used)] // same bounds-check guarantee
-        let len =
-            u32::from_le_bytes(data[offset + 8..offset + 12].try_into().expect("4 bytes")) as usize;
+        // SAFETY: bounds-checked at line 228; slice lengths match array sizes exactly.
+        let raw_lsn = u64::from_le_bytes(
+            <[u8; 8]>::try_from(&data[offset..offset + 8])
+                .unwrap_or_else(|_| unreachable!()),
+        );
+        let len = u32::from_le_bytes(
+            <[u8; 4]>::try_from(&data[offset + 8..offset + 12])
+                .unwrap_or_else(|_| unreachable!()),
+        ) as usize;

Alternatively, since the conversion is statically guaranteed, you could use the more idiomatic pattern with direct slice indexing and array construction:

-        #[allow(clippy::unwrap_used)] // bounds-checked above; try_into is statically length-matched
-        let raw_lsn = u64::from_le_bytes(data[offset..offset + 8].try_into().expect("8 bytes"));
-        #[allow(clippy::unwrap_used)] // same bounds-check guarantee
-        let len =
-            u32::from_le_bytes(data[offset + 8..offset + 12].try_into().expect("4 bytes")) as usize;
+        // SAFETY: bounds-checked at line 228; slice lengths match array sizes exactly.
+        let lsn_bytes: [u8; 8] = data[offset..offset + 8]
+            .try_into()
+            .unwrap_or([0; 8]); // unreachable fallback
+        let len_bytes: [u8; 4] = data[offset + 8..offset + 12]
+            .try_into()
+            .unwrap_or([0; 4]); // unreachable fallback
+        let raw_lsn = u64::from_le_bytes(lsn_bytes);
+        let len = u32::from_le_bytes(len_bytes) as usize;

🤖 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_manifest/shard_replay.rs` around lines 236 - 243, Replace
the infallible slice-to-array conversions that currently call expect() for
raw_lsn and len with explicit fixed-size buffers and copy_from_slice to avoid
panicking; e.g. create a [0u8;8] buffer, call
copy_from_slice(&data[offset..offset+8]) and then u64::from_le_bytes(buf) for
raw_lsn, and similarly a [0u8;4] buffer for len and u32::from_le_bytes(buf) as
usize — apply these changes where raw_lsn and len are computed in
shard_replay.rs.

---

577-607: 💤 Low value

Torn commit detection heuristic may have false positives for heterogeneous batches.

The comment at lines 586-591 notes that this is heuristic and assumes uniform cardinality per LSN. If a future emitter produces legitimately heterogeneous batches (e.g., single-shard ordered ops mixed with multi-shard ones), this would incorrectly drop valid entries.

The current implementation is correct for the documented use case (production emitters guarantee uniform cardinality), and the note at lines 588-591 explicitly acknowledges this limitation. No change needed, but consider adding a debug_assert or invariant check when future cross-shard TXN work lands.

🤖 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_manifest/shard_replay.rs` around lines 577 - 607, Add a
debug-only invariant/assertion around the per-LSN cardinality check to catch
heterogeneous batches during development: after computing counts (the BTreeMap
named counts built from entries) and max_count, add a debug_assert (or a
conditional check gated by cfg(debug_assertions) or a feature flag) that
validates all LSNs either have count == max_count or clearly belong to a
documented single-shard op type; keep the existing torn_lsns logic for runtime
safety but emit a debug-level summary (or panic in debug mode) when the
invariant is violated to surface false-positive scenarios early for the
OrderedAcrossShards/torn commit detection code path.

---

510-523: 💤 Low value

Thread join panic handling uses non-informative error message.

When a shard thread panics, the error message is just "replay_per_shard worker thread panicked" without any indication of which shard panicked. For debugging, it would be helpful to include the shard index.

♻️ Include shard index in panic error message

         // Collect results in shard order.
         handles
             .into_iter()
-            .map(|h| {
+            .enumerate()
+            .map(|(idx, h)| {
                 h.join().unwrap_or_else(|_| {
                     Err(crate::error::MoonError::from(
                         crate::error::AofError::RewriteFailed {
-                            detail: "replay_per_shard worker thread panicked".to_owned(),
+                            detail: format!("replay_per_shard worker thread for shard {} panicked", idx),
                         },
                     ))
                 })
             })
             .collect()

🤖 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_manifest/shard_replay.rs` around lines 510 - 523, The
panic handler for shard worker threads is not reporting which shard panicked;
update the code that builds handles so you store the shard index with each
handle (e.g., Vec<(usize, JoinHandle<...>)>) and then change this collection
step to iterate with that index: replace handles.into_iter().map(|h| { ... })
with handles.into_iter().map(|(shard_idx, h)| h.join().unwrap_or_else(|_|
Err(crate::error::MoonError::from(crate::error::AofError::RewriteFailed {
detail: format!("replay_per_shard worker thread panicked for shard {}",
shard_idx) })))) so the panic message includes the shard index; refer to the
replay_per_shard worker creation site and the handles variable to add the index
pairing.

🤖 Prompt for all review comments with 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.

Inline comments:
In `@src/storage/tiered/spill_thread.rs`:
- Around line 429-476: The stall in send_one_completion can block the spill
thread and let eviction.rs drop SpillRequest (via try_send) after the key is
removed from RAM, causing permanent data loss; fix by making the eviction path
provide end-to-end backpressure: in the code that enqueues SpillRequest (the
eviction path in eviction.rs that currently uses try_send), switch to a
blocking/awaiting send or loop until the request queue accepts the SpillRequest
(respecting stop_flag) so eviction does not remove the in-memory key until the
request is enqueued, or alternatively check the completion/request queue
capacity before evicting and postpone/remap the eviction if the queue is full;
keep references to send_one_completion, SpillRequest, and request_rx when
locating the changes.

---

Outside diff comments:
In `@src/shard/conn_accept.rs`:
- Around line 277-320: The function spawn_migrated_tokio_connection uses
Unix-only APIs but is only feature-gated by runtime-tokio; gate it for Unix
targets (e.g. add #[cfg(all(feature = "runtime-tokio", unix))] to the function)
or add a non-Unix stub with the same signature (behind #[cfg(all(feature =
"runtime-tokio", not(unix)))] ) that returns early/panics with a clear message;
also apply the same cfg change to every call site (e.g. in
src/shard/event_loop.rs) so calls are only compiled on the matching cfg or call
the stub, ensuring consistent signatures across targets.

---

Nitpick comments:
In `@src/persistence/aof_manifest/shard_replay.rs`:
- Around line 236-243: Replace the infallible slice-to-array conversions that
currently call expect() for raw_lsn and len with explicit fixed-size buffers and
copy_from_slice to avoid panicking; e.g. create a [0u8;8] buffer, call
copy_from_slice(&data[offset..offset+8]) and then u64::from_le_bytes(buf) for
raw_lsn, and similarly a [0u8;4] buffer for len and u32::from_le_bytes(buf) as
usize — apply these changes where raw_lsn and len are computed in
shard_replay.rs.
- Around line 577-607: Add a debug-only invariant/assertion around the per-LSN
cardinality check to catch heterogeneous batches during development: after
computing counts (the BTreeMap named counts built from entries) and max_count,
add a debug_assert (or a conditional check gated by cfg(debug_assertions) or a
feature flag) that validates all LSNs either have count == max_count or clearly
belong to a documented single-shard op type; keep the existing torn_lsns logic
for runtime safety but emit a debug-level summary (or panic in debug mode) when
the invariant is violated to surface false-positive scenarios early for the
OrderedAcrossShards/torn commit detection code path.
- Around line 510-523: The panic handler for shard worker threads is not
reporting which shard panicked; update the code that builds handles so you store
the shard index with each handle (e.g., Vec<(usize, JoinHandle<...>)>) and then
change this collection step to iterate with that index: replace
handles.into_iter().map(|h| { ... }) with handles.into_iter().map(|(shard_idx,
h)| h.join().unwrap_or_else(|_|
Err(crate::error::MoonError::from(crate::error::AofError::RewriteFailed {
detail: format!("replay_per_shard worker thread panicked for shard {}",
shard_idx) })))) so the panic message includes the shard index; refer to the
replay_per_shard worker creation site and the handles variable to add the index
pairing.

In `@src/persistence/aof_manifest/shard_rewrite.rs`:
- Around line 73-106: The rollback after loop_result currently only removes base
RDB files for entries in created_shards; extend that rollback to also remove the
corresponding incr files (use manifest.shard_incr_path(sid)) inside the same for
sid in created_shards loop or a follow-up loop, attempting std::fs::remove_file
and emitting the same warn! message on failure (include incr path.display() and
the error) so orphan incr files are cleaned up on rollback.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 2ecc15a2-494e-452d-b4a2-5ba73017efe4

📥 Commits

Reviewing files that changed from the base of the PR and between ea25177 and 14c572b.

📒 Files selected for processing (10)

• src/persistence/aof.rs
• src/persistence/aof_manifest.rs
• src/persistence/aof_manifest/mod.rs
• src/persistence/aof_manifest/shard_replay.rs
• src/persistence/aof_manifest/shard_rewrite.rs
• src/server/embedded.rs
• src/shard/conn_accept.rs
• src/shard/event_loop.rs
• src/shard/persistence_tick.rs
• src/storage/tiered/spill_thread.rs

[pilotspacex-byte]

Re: the outside-diff 🟠 Major on src/shard/conn_accept.rs:277-320 — gate spawn_migrated_tokio_connection behind cfg(unix).

Skipping with reason. The observation is factually correct (the fn uses std::os::unix::io::RawFd / FromRawFd / from_raw_fd), but it is not a defect for moon's supported targets. Moon targets Linux + macOS only — both Unix (per CLAUDE.md; connection migration is itself a Linux feature). On every supported target cfg(unix) is always true, so the function compiles and cargo clippy --no-default-features --features runtime-tokio,jemalloc -- -D warnings passes (verified on macOS + the ubuntu CI matrix).

The gap only manifests on unsupported Windows builds, and adding the gate would require matching #[cfg(unix)] on the event_loop.rs call site too — risking cfg drift on the one build that matters (supported runtime-tokio) for zero supported-platform benefit. Keeping changes minimal per the review prompt.

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

src/storage/eviction.rs (1)

272-285: ⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

Handle read_dir iteration errors as hard failures for spill seeding.

Using entries.flatten() silently drops ReadDir errors. A partial scan can miss the highest heap-*.mpf id and return a seed that reuses an existing filename, risking cold-tier overwrite after restart. Please make incomplete scans fail the spill-seed path (e.g., return Result<u64, io::Error> and propagate), instead of continuing silently.

🤖 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/storage/eviction.rs` around lines 272 - 285, The directory scan currently
uses entries.flatten() which silently ignores ReadDir errors and can produce an
incorrect spill seed; change the scan in the spill-seeding code (the loop that
iterates over entries, the max_id calculation and the match that returns m + 1
or 1) to propagate any IO errors instead of flattening: stop using flatten,
iterate over entries and handle each Result<DirEntry, io::Error> by returning
Err(e) on error, change the function signature to return Result<u64, io::Error>
(or propagate the error to the caller) so failed ReadDir iterations become hard
failures, and adjust call sites to handle the Result accordingly. Ensure the
final seed logic still returns Ok(m + 1) or Ok(1) on success.

🤖 Prompt for all review comments with 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.

Outside diff comments:
In `@src/storage/eviction.rs`:
- Around line 272-285: The directory scan currently uses entries.flatten() which
silently ignores ReadDir errors and can produce an incorrect spill seed; change
the scan in the spill-seeding code (the loop that iterates over entries, the
max_id calculation and the match that returns m + 1 or 1) to propagate any IO
errors instead of flattening: stop using flatten, iterate over entries and
handle each Result<DirEntry, io::Error> by returning Err(e) on error, change the
function signature to return Result<u64, io::Error> (or propagate the error to
the caller) so failed ReadDir iterations become hard failures, and adjust call
sites to handle the Result accordingly. Ensure the final seed logic still
returns Ok(m + 1) or Ok(1) on success.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 65794868-1d7f-4efc-a9fc-891367185399

📥 Commits

Reviewing files that changed from the base of the PR and between 14c572b and 566869d.

📒 Files selected for processing (1)

• src/storage/eviction.rs

[coderabbitai]

Actionable comments posted: 6

🤖 Prompt for all review comments with 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.

Inline comments:
In `@src/persistence/aof/mod.rs`:
- Around line 543-557: The bug is that malformed top-level frames increment
count even though engine.replay_command is never called; remove the count += 1
from the early continue branches and instead increment count only after a
command has been successfully processed (e.g., after engine.replay_command
returns success). Update the match arms around Frame::Array handling so the
branches that skip processing do not mutate count, and ensure count is
incremented immediately after the successful replay path (reference symbols:
count, frame, Frame::Array, engine.replay_command).
- Around line 290-295: In await_outcome(), remove the final expect() and instead
extract the value using pattern matching: convert the current
while-wait-plus-expect into a loop that waits while self.outcome is None (using
self.outcome_cv.wait) and then uses if let Some(v) = self.outcome.lock().take()
{ return v } (or match) to return the u64; reference the symbols await_outcome,
self.outcome and self.outcome_cv and ensure you call take() (or match on the
Option) to consume and return the value without using expect/unwrap.

In `@src/persistence/aof/pool.rs`:
- Around line 238-242: The fire-and-forget helper try_send_append currently
swallows TrySendError and must record dropped appends like
try_send_append_ordered does: detect the TrySendError returned from
self.sender(shard_id).try_send(AofMessage::Append { lsn, bytes }) and on failure
increment the AOF_BACKPRESSURE_DROPPED metric (and/or use the same counter used
by try_send_append_ordered) and emit a concise log mentioning the shard_id and
lsn; mirror the exact error-handling/metrics/logging logic used in
try_send_append_ordered so both unordered and ordered paths count and report
dropped Append messages consistently.

In `@src/persistence/aof/rewrite.rs`:
- Around line 225-235: The current rewrite emits a relative PEXPIRE using
remaining_ms (created in the block where entry.has_expiry() and exp_ms =
entry.expires_at_ms(base_ts)), which extends TTLs after downtime; instead emit
an absolute-time PEXPIREAT using the absolute exp_ms value. Change the Frame
creation that currently builds Frame::BulkString(b"PEXPIRE") and
remaining_ms.to_string() to build Frame::BulkString(b"PEXPIREAT") and use
exp_ms.to_string() (keep the existing check exp_ms > now_ms and skip expired
keys), updating the serialize call accordingly so plain keys use absolute expiry
like the HashWithTtl branch.
- Around line 838-875: The AOF rewrite helpers (e.g., rewrite_aof_sync in
src/persistence/aof/rewrite.rs) snapshot only under read locks, so messages
still queued on the writer task channel (rx) can get serialized into the new
base and then appended again — causing double-apply; fix by ensuring the writer
task drains or stops accepting/flushes rx before invoking the rewrite helpers
(or alternately acquire the writer-side exclusive lock that prevents enqueueing
during snapshot), i.e., in writer_task.rs before calling rewrite_aof_sync (and
the counterpart used at 883-923), drain the rx channel (or atomically take and
hold pending messages) so no pending writes remain queued when the snapshot is
taken.

In `@src/persistence/aof/writer_task.rs`:
- Around line 39-43: The EverySec flush loop in aof_writer_task currently uses
interval.tick().await which can be starved by a ready receive arm; replace the
interval/tick usage with an explicit per-iteration sleep tied to last_fsync
(e.g. compute next_deadline = last_fsync + Duration::from_secs(1) and use
tokio::time::sleep_until(next_deadline.into()).await or tokio::time::sleep for
the remaining duration) and await that sleep inside the select so the timeout
cannot be permanently starved by other ready branches; update the code paths
that reference interval.tick() (including the initial tick consume) to the new
deadline-based sleep logic so the EverySec appendfsync window is honored.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 3354151b-da97-4e7a-90b5-daf9fc0ce570

📥 Commits

Reviewing files that changed from the base of the PR and between 566869d and 5f5ecb3.

📒 Files selected for processing (8)

• src/persistence/aof.rs
• src/persistence/aof/mod.rs
• src/persistence/aof/pool.rs
• src/persistence/aof/rewrite.rs
• src/persistence/aof/writer_task.rs
• src/shard/conn_accept.rs
• src/shard/event_loop.rs
• src/storage/eviction.rs

🚧 Files skipped from review as they are similar to previous changes (2)

• src/storage/eviction.rs
• src/shard/conn_accept.rs

[coderabbitai]

🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win

Remove the expect() from await_outcome.

The wait loop already establishes the Some invariant, so this panic is avoidable and violates the library-code expect() ban.

Proposed fix

     fn await_outcome(&self) -> u64 {
         let mut slot = self.outcome.lock();
-        while slot.is_none() {
-            self.outcome_cv.wait(&mut slot);
-        }
-        slot.expect("outcome is Some after the wait loop")
+        loop {
+            if let Some(committed_seq) = *slot {
+                return committed_seq;
+            }
+            self.outcome_cv.wait(&mut slot);
+        }
     }

As per coding guidelines, "No unwrap() or expect() in library code outside tests. Use pattern matching, if let, or let-else."

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	fn await_outcome(&self) -> u64 {
	let mut slot = self.outcome.lock();
	while slot.is_none() {
	self.outcome_cv.wait(&mut slot);
	}
	slot.expect("outcome is Some after the wait loop")
	fn await_outcome(&self) -> u64 {
	let mut slot = self.outcome.lock();
	loop {
	if let Some(committed_seq) = *slot {
	return committed_seq;
	}
	self.outcome_cv.wait(&mut slot);
	}
	}

🤖 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/mod.rs` around lines 290 - 295, In await_outcome(),
remove the final expect() and instead extract the value using pattern matching:
convert the current while-wait-plus-expect into a loop that waits while
self.outcome is None (using self.outcome_cv.wait) and then uses if let Some(v) =
self.outcome.lock().take() { return v } (or match) to return the u64; reference
the symbols await_outcome, self.outcome and self.outcome_cv and ensure you call
take() (or match on the Option) to consume and return the value without using
expect/unwrap.

[coderabbitai]

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Only count frames that were actually replayed.

These continue branches increment count even though engine.replay_command never runs, so malformed top-level frames are reported as successfully recovered.

Proposed fix

                     Frame::Array(arr) if !arr.is_empty() => {
                         let name = match &arr[0] {
                             Frame::BulkString(s) => s.as_ref(),
                             Frame::SimpleString(s) => s.as_ref(),
-                            _ => {
-                                count += 1;
-                                continue;
-                            }
+                            _ => continue,
                         };
                         (name as &[u8], &arr[1..])
                     }
-                    _ => {
-                        count += 1;
-                        continue;
-                    }
+                    _ => continue,
                 };
                 engine.replay_command(databases, cmd, cmd_args, &mut selected_db);
                 count += 1;

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	let (cmd, cmd_args) = match &frame {
	Frame::Array(arr) if !arr.is_empty() => {
	let name = match &arr[0] {
	Frame::BulkString(s) => s.as_ref(),
	Frame::SimpleString(s) => s.as_ref(),
	_ => {
	count += 1;
	continue;
	}
	};
	(name as &[u8], &arr[1..])
	}
	_ => {
	count += 1;
	continue;
	let (cmd, cmd_args) = match &frame {
	Frame::Array(arr) if !arr.is_empty() => {
	let name = match &arr[0] {
	Frame::BulkString(s) => s.as_ref(),
	Frame::SimpleString(s) => s.as_ref(),
	_ => continue,
	};
	(name as &[u8], &arr[1..])
	}
	_ => continue,
	};
	engine.replay_command(databases, cmd, cmd_args, &mut selected_db);
	count += 1;

🤖 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/mod.rs` around lines 543 - 557, The bug is that malformed
top-level frames increment count even though engine.replay_command is never
called; remove the count += 1 from the early continue branches and instead
increment count only after a command has been successfully processed (e.g.,
after engine.replay_command returns success). Update the match arms around
Frame::Array handling so the branches that skip processing do not mutate count,
and ensure count is incremented immediately after the successful replay path
(reference symbols: count, frame, Frame::Array, engine.replay_command).

[coderabbitai]

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Count dropped fire-and-forget appends too.

Both append helpers discard TrySendError completely. Under appendfsync=everysec / no, a full writer channel now loses AOF entries without incrementing AOF_BACKPRESSURE_DROPPED or logging anything, even though this module documents dropped Append traffic as observable.

Proposed fix

     pub fn try_send_append(&self, shard_id: usize, lsn: u64, bytes: Bytes) {
-        let _ = self
-            .sender(shard_id)
-            .try_send(AofMessage::Append { lsn, bytes });
+        match self.sender(shard_id).try_send(AofMessage::Append { lsn, bytes }) {
+            Ok(()) => {}
+            Err(flume::TrySendError::Full(_)) => {
+                AOF_BACKPRESSURE_DROPPED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                warn!("AOF writer channel full (shard {}): Append dropped", shard_id);
+            }
+            Err(flume::TrySendError::Disconnected(_)) => {
+                warn!("AOF writer channel disconnected (shard {}): Append dropped", shard_id);
+            }
+        }
     }

Apply the same handling to try_send_append_ordered.

Also applies to: 332-343

🤖 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/pool.rs` around lines 238 - 242, The fire-and-forget
helper try_send_append currently swallows TrySendError and must record dropped
appends like try_send_append_ordered does: detect the TrySendError returned from
self.sender(shard_id).try_send(AofMessage::Append { lsn, bytes }) and on failure
increment the AOF_BACKPRESSURE_DROPPED metric (and/or use the same counter used
by try_send_append_ordered) and emit a concise log mentioning the shard_id and
lsn; mirror the exact error-handling/metrics/logging logic used in
try_send_append_ordered so both unordered and ordered paths count and report
dropped Append messages consistently.

[coderabbitai]

⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

Use absolute expiry when rewriting TTLs.

PEXPIRE stores a relative TTL from rewrite time, so a restart after downtime extends every rewritten key by the downtime interval. The HashWithTtl branch already uses absolute-time semantics via HPEXPIREAT; plain keys need the same treatment.

🤖 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/rewrite.rs` around lines 225 - 235, The current rewrite
emits a relative PEXPIRE using remaining_ms (created in the block where
entry.has_expiry() and exp_ms = entry.expires_at_ms(base_ts)), which extends
TTLs after downtime; instead emit an absolute-time PEXPIREAT using the absolute
exp_ms value. Change the Frame creation that currently builds
Frame::BulkString(b"PEXPIRE") and remaining_ms.to_string() to build
Frame::BulkString(b"PEXPIREAT") and use exp_ms.to_string() (keep the existing
check exp_ms > now_ms and skip expired keys), updating the serialize call
accordingly so plain keys use absolute expiry like the HashWithTtl branch.

[coderabbitai]

⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

The tokio rewrite helpers can double-apply queued writes.

These helpers snapshot under read locks only. In src/persistence/aof/writer_task.rs Lines 315-345, the tokio writer flushes the old file and calls them without draining rx, so a write that updates the DB before the snapshot but is still queued in the channel is serialized into the new base and then appended again after reopen. That breaks recovery for non-idempotent commands.

Also applies to: 883-923

🤖 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/rewrite.rs` around lines 838 - 875, The AOF rewrite
helpers (e.g., rewrite_aof_sync in src/persistence/aof/rewrite.rs) snapshot only
under read locks, so messages still queued on the writer task channel (rx) can
get serialized into the new base and then appended again — causing double-apply;
fix by ensuring the writer task drains or stops accepting/flushes rx before
invoking the rewrite helpers (or alternately acquire the writer-side exclusive
lock that prevents enqueueing during snapshot), i.e., in writer_task.rs before
calling rewrite_aof_sync (and the counterpart used at 883-923), drain the rx
channel (or atomically take and hold pending messages) so no pending writes
remain queued when the snapshot is taken.

[coderabbitai]

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Top-level tokio writer still has the starvable EverySec flush loop.

The per-shard tokio branch below already moved away from interval.tick() because an always-ready receive arm can starve it under sustained writes. aof_writer_task still uses the old pattern, so single-file tokio deployments can buffer writes well past one second and lose more than the configured appendfsync=everysec window on crash.

Also applies to: 372-378

🤖 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 39 - 43, The EverySec flush
loop in aof_writer_task currently uses interval.tick().await which can be
starved by a ready receive arm; replace the interval/tick usage with an explicit
per-iteration sleep tied to last_fsync (e.g. compute next_deadline = last_fsync
+ Duration::from_secs(1) and use
tokio::time::sleep_until(next_deadline.into()).await or tokio::time::sleep for
the remaining duration) and await that sleep inside the select so the timeout
cannot be permanently starved by other ready branches; update the code paths
that reference interval.tick() (including the initial tick consume) to the new
deadline-based sleep logic so the EverySec appendfsync window is honored.

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with 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.

Inline comments:
In `@src/command/vector_search/tests.rs`:
- Around line 3-6: Replace the standard library lock with parking_lot: change
the import from std::sync::RwLock to parking_lot::RwLock and update the static
METRICS_LOCK declaration to use that type; then remove all .unwrap() calls when
acquiring guards (e.g., change .read().unwrap()/.write().unwrap() to .read()/
.write()) for METRICS_LOCK and any other RwLock usages in this test file so the
guards use parking_lot’s infallible API.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 5b288f65-464e-4589-ad9f-16eac7623b6b

📥 Commits

Reviewing files that changed from the base of the PR and between 5f5ecb3 and edffce3.

📒 Files selected for processing (1)

• src/command/vector_search/tests.rs