perf(persistence): local-leg group commit + BITOP/COPY/DEL/UNLINK coordinator leg durability#213
Merged
Merged
Conversation
…commit under appendfsync=always The cross-shard coordinator's LOCAL-leg persist (co-located MSET/MSETNX and scattered-MSET local slices, shipped in v3-4 Finding 1) called try_send_append_durable, which under appendfsync=always awaits one fsync ack PER COMMAND bounded by --aof-fsync-timeout-ms (default 2000ms). A pipeline of coordinated writes stacked these serially — the measured 2000-3000ms always far-tail (tmp/V3-4-GCLOUD-BENCH.md), carried as the [HIGH] follow-up into v3-5. Fix: local legs now use the same fire-and-forget-then-barrier contract the remote SPSC legs have used since the H1-BARRIER fix: - New AofWriterPool::send_append_group enqueues under bounded backpressure and returns immediately; Ok(true) means Always policy — the caller owes a barrier. EverySec/No are unchanged (writer loop owns the fsync cadence). - persist_local_leg switches to it and reports needs-barrier up through coordinate_mset/coordinate_msetnx/coordinate_multi_key. - Both connection handlers (monoio + sharded) collect the response indexes of barrier-pending local-leg writes and issue ONE fsync_barrier(local shard) per pipeline batch, BEFORE response serialization — +OK still implies confirmed durability, but a batch of N coordinated writes costs 1 awaited fsync instead of N. - On barrier failure every affected response is overwritten with AOF_FSYNC_ERR — never a false +OK (design-for-failure preserved). Tests: 3 new red-proven pool unit tests (send_append_group must not await the per-write ack; everysec needs no barrier; dead writer errors); v3-4 coordinator_local_leg_durability crash-recovery suite green on the new path (local legs still persist and replay); full lib suite 3656 pass; clippy clean on default + tokio,jemalloc. Absolute tail magnitude needs a GCE run (OrbStack fsync is near-free); tracked in v3-5. author: Tin Dang
The cross-shard coordinator's in-process legs for BITOP (dest write),
COPY (dst write + PEXPIRE TTL restore), and multi-key DEL/UNLINK (both
the co-located fast path and the scattered local slice) executed in
memory but never appended to the owning shard's AOF — the carried v3-4
follow-up. Failure modes on kill-9 + restart:
- DEL/UNLINK: deleted keys RESURRECTED from their seed writes (the seed
MSET is in the AOF, the local-leg delete never was).
- BITOP/COPY: results whose dest/dst owner == the connection's own shard
silently vanished. Remote legs were always durable (MultiExecute ->
wal_append_and_fanout).
Fix: all four now persist through the same persist_local_leg
group-commit path as MSET/MSETNX (previous commit):
- New run_on_owner_persist mirrors run_on_owner but appends the command
to my_shard's AOF when the owner is local and execution succeeded —
used by BITOP's whole-command forward + synthesized DEL/SET dest legs
and COPY's whole-command forward + SET/PEXPIRE dst legs. Replay-safe:
each persisted command covers only keys owned by my_shard.
- coordinate_multi_del_or_exists persists DEL/UNLINK on the co-located
fast path (whole command) and the scattered local slice (synthesized
over only local keys), skipped when n=0 (nothing removed replays
identically without a record). EXISTS/TOUCH stay read-only.
- All legs ride the batch-end fsync barrier under appendfsync=always
via local_barrier_pending (previous commit's plumbing).
Tests: 4 new red-proven crash-recovery tests in
coordinator_local_leg_durability.rs (DEL scatter resurrection, UNLINK
co-located fast-path resurrection, BITOP dest legs both shapes, COPY dst
legs both shapes) — deterministic via per-shard {hash-tags} regardless
of which shard the connection lands on; suite 7/7 green post-fix. Full
lib suite 3656 pass; clippy clean both feature sets.
author: Tin Dang
…s sleep The cluster_* tests spawned an in-process cluster server, slept a fixed 100ms, then connected with a no-retry unwrap (integration.rs:242). Under full-suite parallelism the listener thread can take longer to bind -> "Connection refused" flakes (observed repeatedly in CI-parity runs; pass in isolation). Replace the sleep with a bounded connect-poll (10s), the same pattern the binary-spawning suites use. author: Tin Dang
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
✅ Files skipped from review due to trivial changes (1)
📝 WalkthroughWalkthroughThis PR adds group-commit durability for coordinator-local write legs, threads barrier tracking through multi-key coordination and connection handling, and adds restart regression coverage plus changelog and startup readiness updates. ChangesCoordinator local-leg durability
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant Coordinator
participant AofWriterPool
participant Shared
Client->>Coordinator: multi-key write
Coordinator->>AofWriterPool: enqueue local leg append
Coordinator->>Shared: resolve_local_leg_barrier
Shared->>AofWriterPool: fsync_barrier(shard_id)
Shared-->>Coordinator: update responses or clear pending indexes
Coordinator-->>Client: send batch responses
Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
PR Summary by QodoCoordinator local-leg group commit + BITOP/COPY/DEL/UNLINK AOF durability
AI Description
Diagram
High-Level Assessment
Files changed (8)
|
Code Review by Qodo
Context used✅ Compliance rules (platform):
40 rules 1.
|
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/shard/coordinator.rs (1)
1192-1266: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftLocal-leg persist failure aborts before remaining shard groups are dispatched.
In the scatter branch,
persist_local_leg(...).await(and its earlyreturnonErr(())) runs inside thefor (shard_id, key_args) in &groupsloop, before all groups have been dispatched and beforepending_shardsis awaited. Ifmy_shard's slice happens before a higher-shard-id group in the BTreeMap ordering and the local AOF enqueue fails, the function returns immediately — the remaining (not-yet-iterated) shard groups never get theirspsc_sendDEL/UNLINK dispatched at all, whilemy_shard's own keys were already deleted in memory and any earlier (lower-shard-id) groups were already fire-and-forget dispatched. The client gets a singleAOF_FSYNC_ERR, but the actual keyspace ends up partially deleted across shards with no way to tell which keys were affected.
coordinate_msetavoids this by deferring its local-leg persist call until after the full group loop and thepending_shardsawait loop.coordinate_multi_del_or_existsshould follow the same pattern: capture the synthesized local DEL/UNLINK parts during the loop, dispatch every group unconditionally, await allpending_shards, and only then attemptpersist_local_leg(returning the error afterward if it fails).🔧 Proposed restructuring
let mut total_count: i64 = 0; let mut pending_shards: Vec<channel::OneshotReceiver<Vec<Frame>>> = Vec::new(); + let mut local_delete_parts: Option<Vec<Frame>> = None; for (shard_id, key_args) in &groups { if *shard_id == my_shard { let mut selected = db_index; let result = crate::shard::slice::with_shard_db(db_index, |db| { db.refresh_now_from_cache(cached_clock); cmd_dispatch(db, cmd, key_args, &mut selected, db_count) }); if let DispatchResult::Response(Frame::Integer(n)) = result { total_count += n; if is_delete && n > 0 { let mut parts: Vec<Frame> = Vec::with_capacity(key_args.len() + 1); parts.push(Frame::BulkString(Bytes::from(cmd_upper.clone()))); parts.extend_from_slice(key_args); - let serialized = - crate::persistence::aof::serialize_command(&Frame::Array(parts.into())); - match persist_local_leg(aof_pool, repl_state, my_shard, serialized).await { - Ok(needs_barrier) => *local_barrier_pending |= needs_barrier, - Err(()) => { - return Frame::Error(Bytes::from_static( - crate::persistence::aof::AOF_FSYNC_ERR, - )); - } - } + local_delete_parts = Some(parts); } } } else { ... spsc_send(dispatch_tx, my_shard, *shard_id, msg, spsc_notifiers).await; pending_shards.push(reply_rx); } } for reply_rx in pending_shards { match reply_rx.recv().await { ... } } + if let Some(parts) = local_delete_parts { + let serialized = crate::persistence::aof::serialize_command(&Frame::Array(parts.into())); + match persist_local_leg(aof_pool, repl_state, my_shard, serialized).await { + Ok(needs_barrier) => *local_barrier_pending |= needs_barrier, + Err(()) => { + return Frame::Error(Bytes::from_static(crate::persistence::aof::AOF_FSYNC_ERR)); + } + } + } + Frame::Integer(total_count)🤖 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/coordinator.rs` around lines 1192 - 1266, The scatter path in coordinate_multi_del_or_exists is returning from persist_local_leg too early inside the shard loop, which can prevent later shard groups from being dispatched. Refactor the loop to always finish collecting/disptaching all groups and awaiting pending_shards first, while capturing the local DEL/UNLINK payload for my_shard during the loop. Then perform the persist_local_leg call after the loop (as coordinate_mset does), and only return AOF_FSYNC_ERR afterward if that persist fails.
🤖 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/server/conn/handler_monoio/mod.rs`:
- Around line 1816-1831: Move the local-leg fsync barrier in
handler_monoio::mod::handle so it runs before any early-return PSYNC handling
and before the blocking-command flush/break path, not only at end-of-batch.
Ensure the same barrier logic that currently drains local_leg_write_idxs and
uses ctx.aof_pool.fsync_barrier(ctx.shard_id) is executed for MSET/MSETNX local
legs before those exits can skip it, so acknowledgements are only serialized
after durability is confirmed.
In `@src/server/conn/handler_sharded/mod.rs`:
- Around line 1719-1735: Batch-end durability barrier is using stale
local_leg_write_idxs after the blocking-command early flush path replaces
responses. In handler_sharded::mod.rs, make sure any pending AOF fsync barrier
for local write legs is resolved before the blocking-commands branch flushes
responses and resets the vector, then clear local_leg_write_idxs so later code
cannot index into the new buffer. Use the existing fsync_barrier logic around
the responses/local_leg_write_idxs handling in the per-frame loop, especially
the blocking-command branch and the final barrier block near the end of batch
processing.
---
Outside diff comments:
In `@src/shard/coordinator.rs`:
- Around line 1192-1266: The scatter path in coordinate_multi_del_or_exists is
returning from persist_local_leg too early inside the shard loop, which can
prevent later shard groups from being dispatched. Refactor the loop to always
finish collecting/disptaching all groups and awaiting pending_shards first,
while capturing the local DEL/UNLINK payload for my_shard during the loop. Then
perform the persist_local_leg call after the loop (as coordinate_mset does), and
only return AOF_FSYNC_ERR afterward if that persist fails.
🪄 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: 4af9ed81-0668-4713-a812-f3d9cb558f92
📒 Files selected for processing (8)
CHANGELOG.mdsrc/persistence/aof/pool.rssrc/server/conn/handler_monoio/dispatch.rssrc/server/conn/handler_monoio/mod.rssrc/server/conn/handler_sharded/mod.rssrc/shard/coordinator.rstests/coordinator_local_leg_durability.rstests/integration.rs
…shes (PR #213 review) Review findings from PR #213 (CodeRabbit + Qodo), all addressed: 1. [CRITICAL, CodeRabbit] The batch-end local-leg barrier could be skipped or corrupted by mid-batch early flushes: - blocking commands (BLPOP...) flush accumulated responses then REPLACE the vec — pending barrier indexes became stale (panic or misattributed AOF_FSYNC_ERR onto the blocking response) and the flushed +OK escaped without confirmed durability; - PSYNC hijack (monoio) and SUBSCRIBE-entry (both runtimes) flush early with the same durability leak. Fix: new shared::resolve_local_leg_barrier(pool, shard, idxs, responses) — always drains, patches AOF_FSYNC_ERR on barrier failure — called at the batch end AND before every early flush (blocking, SUBSCRIBE entry, PSYNC) in both handlers. 2. [Qodo bug] run_on_owner_persist logged no-op writes (COPY SET..NX refusal, DEL of absent dest, PEXPIRE on vanished key) — costing a needless barrier fsync and risking AOF_FSYNC_ERR on a command that wrote nothing. Fix: persist_if predicate per call site (BITOP forward always mutates; DEL only n>0; COPY only :1; SET dst only +OK; PEXPIRE only :1). 3. [Qodo rule] #[allow(clippy::too_many_arguments)] now carries its justification; CHANGELOG latency numbers now state their Linux/GCE measurement context. Tests: coordinator_local_leg_durability 7/7, full lib suite 3656 pass, clippy clean both feature sets. The barrier-failure arm of the early flush paths is not black-box testable (requires an AOF writer dying mid-pipeline); covered by the shared helper's single code path + the existing fsync_barrier unit tests. author: Tin Dang
…commit # Conflicts: # CHANGELOG.md
4 tasks
pilotspacex-byte
added a commit
that referenced
this pull request
Jul 7, 2026
…mmit (#239) Under `appendfsync always`, all three connection handlers awaited one fsync ack PER COMMAND for plain local writes (try_send_append_durable). A 16-deep pipeline therefore paid 16 serialized fsync round-trips per connection, while Redis fsyncs once per event-loop iteration — measured as an 8x SET deficit at P16 (Moon 5.2k vs Redis 44k ops/s, GCE c3-standard-8; tmp/MOON-VS-REDIS-DURABILITY.md). The writer-side group commit (one fsync per drained batch) already existed, and cross-shard writes + coordinator local legs already used the fire-and-forget + barrier contract from PR #213. This change extends that contract to plain local writes: - handler_monoio / handler_sharded: local writes, MOVE, and COPY enqueue via send_append_group(); successful responses join local_leg_write_idxs and the existing end-of-batch resolve_local_leg_barrier() (one fsync_barrier per pipelined batch) converts them to AOF_FSYNC_ERR on failure — never a silent +OK. - handler_single: flush_with_aof_ack collects barrier indexes and issues ONE fsync_barrier(0) for the whole flush; the inline pre-SUBSCRIBE path and single-shard GRAPH.* WAL-record loop use the same pattern. - The AOF writer processes its channel in order, so an acked zero-length AppendSync barrier proves every prior Append durable: the H1 fsync-before-ack guarantee is unchanged. - everysec/no policies unaffected (send_append_group returns Ok(false), no barrier joined). Validation: - crash_matrix_per_shard_aof --ignored: 3/3 green (SIGKILL under appendfsync always still recovers 100% of acked writes). - flush_with_aof_ack H1 ordering test updated to the batch protocol (Append then AppendSync barrier; ack still gates the response). - fmt, clippy x2 (default + tokio,jemalloc), cargo test --lib x2 (monoio 3864 / tokio 3143), cargo test --no-run x2: all green. author: Tin Dang Co-authored-by: Tin Dang <tindang.ht97@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Two v3-5 write-path-durability items, both red/green TDD-proven, plus a test-harness de-flake.
1. Coordinator local legs ride group commit under
appendfsync=always(666d32c)persist_local_leg(co-located MSET/MSETNX, scattered-MSET local slices) awaited one fsync ack per command, each bounded by--aof-fsync-timeout-ms(default 2000ms) — a pipeline of coordinated writes stacked these serially into the measured 2000–3000msalwaysfar-tail (tmp/V3-4-GCLOUD-BENCH.md; the carried [HIGH] follow-up).Local legs now use the same fire-and-forget-then-barrier contract as the remote SPSC legs:
AofWriterPool::send_append_group— bounded-backpressure enqueue, never awaits the per-write ack;Ok(true)= caller owes a barrier (Alwaysonly).fsync_barrier(local shard)per pipeline batch, before response serialization —+OKstill implies confirmed durability; batch of N coordinated writes = 1 awaited fsync instead of N.MOONERR AOF fsync— never a false+OK.everysec/nounchanged.2. BITOP/COPY/DEL/UNLINK local legs now persist (d513e8d)
The coordinator's in-process legs for BITOP (dest write), COPY (dst write + PEXPIRE), and multi-key DEL/UNLINK (co-located fast path AND scattered local slice) executed in memory but never reached the AOF (carried v3-4 follow-up). Failure modes on kill-9 + restart: deleted keys resurrected from their seed writes; BITOP/COPY results on the connection's own shard vanished.
run_on_owner_persistmirrorsrun_on_ownerbut appends the executed command to the local shard's AOF on success — replay-safe (each persisted command covers only locally-owned keys).3. Cluster test harness de-flake (1fd4ac7, test-only)
start_cluster_serverslept a fixed 100ms then connected with a no-retry unwrap — "Connection refused" flakes under full-suite parallelism (observed repeatedly in CI-parity runs, pass in isolation). Replaced with a bounded connect-poll; integration suite now 3/3 green at full parallelism.Testing
send_append_groupmust not await the per-write ack / everysec no-barrier / dead-writer errors)coordinator_local_leg_durability.rs(DEL scatter resurrection, UNLINK fast-path resurrection, BITOP dest both shapes, COPY dst both shapes) — deterministic via per-shard{hash-tags}; suite 7/7 green post-fix-D warningsboth feature sets, monoio--releasefull suite, tokio full suite (zero failures, honest exit code)Follow-up (not in this PR)
always-tail magnitude requires a GCE bench (OrbStack fsync is near-free) — happy to run the A/B on requestSummary by CodeRabbit
Performance
Bug Fixes
Tests