diff --git a/CHANGELOG.md b/CHANGELOG.md index 536d7352..a2569b01 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed — `appendfsync always`: local writes now group-commit per pipeline batch (PR #TBD) + +- **All three connection handlers** (`handler_monoio`, `handler_sharded`, + `handler_single`): plain local writes (plus MOVE/COPY and single-shard + GRAPH.* WAL records) no longer await one fsync ack **per command** under + `appendfsync always`. Appends are enqueued fire-and-forget + (`send_append_group`) and the whole pipelined batch is confirmed by ONE + `fsync_barrier` before response serialization — the same contract + cross-shard writes and coordinator local legs already used (PR #213). + The writer processes its channel in order, so an acked barrier proves + every prior append durable; the fsync-before-ack H1 guarantee is + unchanged (a failed barrier converts every joined response to an error, + never a silent `+OK`). +- **Why**: a 16-deep pipeline paid 16 serialized fsync round-trips per + connection while Redis fsyncs once per event-loop iteration — measured + 8× SET-throughput deficit at P16 (Moon 5.2k vs Redis 44k ops/s, GCE + c3-standard-8, tmp/MOON-VS-REDIS-DURABILITY.md). Writer-side group + commit existed but could only batch *across* connections; the per-command + await defeated it *within* a connection. +- everysec/no policies are unchanged (`send_append_group` degrades to the + same bounded-backpressure enqueue). +- `flush_with_aof_ack`'s discriminating H1 ordering test updated to the + batch protocol (one `Append` + one `AppendSync` barrier; ack still gates + the first response). + ### Changed — WAL v3 fsync moved off the shard event loop (PR #TBD) - **`src/persistence/wal_v3/sync_agent.rs` (new)**: per-shard `WalSyncAgent` diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index 54f3b980..6ca68fd0 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -1163,15 +1163,19 @@ pub(crate) async fn handle_connection_sharded_monoio< ctx.shard_id, serialized.len(), ); - if pool - .try_send_append_durable(ctx.shard_id, lsn, serialized) - .await - .is_err() - { - responses.push(Frame::Error(bytes::Bytes::from_static( - aof::AOF_FSYNC_ERR, - ))); - continue; + match pool.send_append_group(ctx.shard_id, lsn, serialized).await { + // Always: durability confirmed by ONE + // fsync_barrier per batch (resolve_local_leg_barrier + // before serialization) instead of an awaited + // fsync per pipelined command. + Ok(true) => local_leg_write_idxs.push(responses.len()), + Ok(false) => {} + Err(_) => { + responses.push(Frame::Error(bytes::Bytes::from_static( + aof::AOF_FSYNC_ERR, + ))); + continue; + } } } } @@ -1224,15 +1228,16 @@ pub(crate) async fn handle_connection_sharded_monoio< ctx.shard_id, serialized.len(), ); - if pool - .try_send_append_durable(ctx.shard_id, lsn, serialized) - .await - .is_err() - { - responses.push(Frame::Error(bytes::Bytes::from_static( - aof::AOF_FSYNC_ERR, - ))); - continue; + match pool.send_append_group(ctx.shard_id, lsn, serialized).await { + // Same one-barrier-per-batch contract as MOVE. + Ok(true) => local_leg_write_idxs.push(responses.len()), + Ok(false) => {} + Err(_) => { + responses.push(Frame::Error(bytes::Bytes::from_static( + aof::AOF_FSYNC_ERR, + ))); + continue; + } } } } @@ -1475,6 +1480,14 @@ pub(crate) async fn handle_connection_sharded_monoio< // invalidation, etc.) below — the client must see // the failure, not a silent inconsistency. let mut aof_failed = false; + // Always-mode local writes join the per-batch group commit: + // the append is enqueued fire-and-forget here and confirmed + // by ONE fsync_barrier before response serialization + // (resolve_local_leg_barrier), amortizing the fsync across + // the whole pipelined batch — previously each command + // awaited its own fsync ack (~1 fsync per write, the + // measured 8x deficit vs Redis at P16). + let mut aof_barrier_pending = false; if !matches!(response, Frame::Error(_)) && is_write { if let Some(ref pool) = ctx.aof_pool { let serialized = aof::serialize_command(&frame); @@ -1483,14 +1496,14 @@ pub(crate) async fn handle_connection_sharded_monoio< ctx.shard_id, serialized.len(), ); - if pool - .try_send_append_durable(ctx.shard_id, lsn, serialized) - .await - .is_err() - { - response = - Frame::Error(bytes::Bytes::from_static(aof::AOF_FSYNC_ERR)); - aof_failed = true; + match pool.send_append_group(ctx.shard_id, lsn, serialized).await { + Ok(true) => aof_barrier_pending = true, + Ok(false) => {} + Err(_) => { + response = + Frame::Error(bytes::Bytes::from_static(aof::AOF_FSYNC_ERR)); + aof_failed = true; + } } } } @@ -1530,6 +1543,11 @@ pub(crate) async fn handle_connection_sharded_monoio< if let Some(ws_id) = conn.workspace_id.as_ref() { strip_workspace_prefix_from_response(ws_id, cmd, &mut response); } + // Only successful writes join the barrier set — an error + // response must not be overwritten by a barrier failure. + if aof_barrier_pending && !matches!(response, Frame::Error(_)) { + local_leg_write_idxs.push(responses.len()); + } responses.push(response); } else { // Snapshot visibility filter for active cross-store transactions. diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 61192e04..c0b8e132 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -1221,15 +1221,23 @@ pub(crate) async fn handle_connection_sharded_inner< if let Some(ref bytes) = aof_bytes { if let Some(ref pool) = ctx.aof_pool { let lsn = aof::AofWriterPool::issue_append_lsn(&ctx.repl_state, ctx.shard_id, bytes.len()); - if pool - .try_send_append_durable(ctx.shard_id, lsn, bytes.clone()) + match pool + .send_append_group(ctx.shard_id, lsn, bytes.clone()) .await - .is_err() { - responses.push(Frame::Error(Bytes::from_static( - aof::AOF_FSYNC_ERR, - ))); - continue; + // Always: one fsync_barrier per batch + // (resolve_local_leg_barrier) instead of + // an awaited fsync per command. + Ok(true) => { + local_leg_write_idxs.push(responses.len()) + } + Ok(false) => {} + Err(_) => { + responses.push(Frame::Error(Bytes::from_static( + aof::AOF_FSYNC_ERR, + ))); + continue; + } } } } @@ -1271,15 +1279,23 @@ pub(crate) async fn handle_connection_sharded_inner< if let Some(ref bytes) = aof_bytes { if let Some(ref pool) = ctx.aof_pool { let lsn = aof::AofWriterPool::issue_append_lsn(&ctx.repl_state, ctx.shard_id, bytes.len()); - if pool - .try_send_append_durable(ctx.shard_id, lsn, bytes.clone()) + match pool + .send_append_group(ctx.shard_id, lsn, bytes.clone()) .await - .is_err() { - responses.push(Frame::Error(Bytes::from_static( - aof::AOF_FSYNC_ERR, - ))); - continue; + // Always: one fsync_barrier per batch + // (resolve_local_leg_barrier) instead of + // an awaited fsync per command. + Ok(true) => { + local_leg_write_idxs.push(responses.len()) + } + Ok(false) => {} + Err(_) => { + responses.push(Frame::Error(Bytes::from_static( + aof::AOF_FSYNC_ERR, + ))); + continue; + } } } } @@ -1554,21 +1570,26 @@ pub(crate) async fn handle_connection_sharded_inner< } } } - // H1: durable path under appendfsync=always. + // Always-mode local writes join the per-batch group + // commit: append enqueued fire-and-forget, confirmed by + // ONE fsync_barrier before serialization + // (resolve_local_leg_barrier) — previously each command + // awaited its own fsync (the 8x P16 deficit vs Redis). let mut aof_failed = false; + let mut aof_barrier_pending = false; if let Some(bytes) = aof_bytes { if !matches!(response, Frame::Error(_)) { if let Some(ref pool) = ctx.aof_pool { let lsn = aof::AofWriterPool::issue_append_lsn(&ctx.repl_state, ctx.shard_id, bytes.len()); - if pool - .try_send_append_durable(ctx.shard_id, lsn, bytes) - .await - .is_err() - { - response = Frame::Error(Bytes::from_static( - aof::AOF_FSYNC_ERR, - )); - aof_failed = true; + match pool.send_append_group(ctx.shard_id, lsn, bytes).await { + Ok(true) => aof_barrier_pending = true, + Ok(false) => {} + Err(_) => { + response = Frame::Error(Bytes::from_static( + aof::AOF_FSYNC_ERR, + )); + aof_failed = true; + } } } } @@ -1590,6 +1611,11 @@ pub(crate) async fn handle_connection_sharded_inner< if let Some(ws_id) = conn.workspace_id.as_ref() { strip_workspace_prefix_from_response(ws_id, cmd, &mut response); } + // Only successful writes join the barrier set — an error + // response must not be overwritten by a barrier failure. + if aof_barrier_pending && !matches!(response, Frame::Error(_)) { + local_leg_write_idxs.push(responses.len()); + } responses.push(response); } else { // Snapshot visibility filter for active cross-store transactions. diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index 965e1eb5..4525193c 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -67,18 +67,37 @@ pub(crate) async fn flush_with_aof_ack( where S: futures::Sink + Unpin, { - // Phase 1 — await every fsync ack; patch failed slots. + // Phase 1 — group commit: enqueue every append fire-and-forget, then + // confirm the whole batch with ONE fsync barrier (Always) instead of an + // awaited fsync per entry — same contract as the sharded handlers' + // resolve_local_leg_barrier. On barrier failure every enqueued write in + // the batch is unconfirmed, so every joined slot is patched. + let mut barrier_idxs: Vec = Vec::new(); for (resp_idx, bytes) in aof_entries { let lsn = crate::persistence::aof::AofWriterPool::issue_append_lsn(repl_state, 0, bytes.len()); - if pool.try_send_append_durable(0, lsn, bytes).await.is_err() && resp_idx < responses.len() - { - responses[resp_idx] = Frame::Error(Bytes::from_static(b"WRITEFAIL aof fsync failed")); + match pool.send_append_group(0, lsn, bytes).await { + Ok(true) => barrier_idxs.push(resp_idx), + Ok(false) => {} + Err(_) => { + if resp_idx < responses.len() { + responses[resp_idx] = + Frame::Error(Bytes::from_static(b"WRITEFAIL aof fsync failed")); + } + } } if let Some(counter) = change_counter { counter.fetch_add(1, Ordering::Relaxed); } } + if !barrier_idxs.is_empty() && pool.fsync_barrier(0).await.is_err() { + for resp_idx in barrier_idxs { + if resp_idx < responses.len() { + responses[resp_idx] = + Frame::Error(Bytes::from_static(b"WRITEFAIL aof fsync failed")); + } + } + } // Phase 2 — all acks received; flush responses to client. let mut break_outer = false; for response in responses { @@ -953,17 +972,28 @@ pub async fn handle_connection( // failure mode discards the entire response buffer; future // per-slot patching could use it. let mut aof_write_failed = false; + let mut aof_barrier_needed = false; for (_resp_idx, bytes) in aof_entries.drain(..) { if let Some(ref pool) = aof_pool { let lsn = crate::persistence::aof::AofWriterPool::issue_append_lsn(&repl_state, 0, bytes.len()); - if let Err(_aof_err) = pool.try_send_append_durable(0, lsn, bytes).await { - aof_write_failed = true; + match pool.send_append_group(0, lsn, bytes).await { + Ok(true) => aof_barrier_needed = true, + Ok(false) => {} + Err(_) => aof_write_failed = true, } } if let Some(ref counter) = change_counter { counter.fetch_add(1, Ordering::Relaxed); } } + // ONE fsync confirms the whole batch (group commit). + if aof_barrier_needed { + if let Some(ref pool) = aof_pool { + if pool.fsync_barrier(0).await.is_err() { + aof_write_failed = true; + } + } + } if aof_write_failed { // Discard buffered +OK responses — the writes are not // durable. Log at warn level so operators can correlate @@ -1558,23 +1588,33 @@ pub async fn handle_connection( (resp, records) }; let mut graph_aof_failed = false; + let mut graph_barrier_needed = false; for record in wal_records { if let Some(ref pool) = aof_pool { // Single-shard mode (shard_id = 0). - // `try_send_append_durable` awaits writer - // ack under `appendfsync=always` (H1 - // closure) and is fire-and-forget for + // Group commit: enqueue all records, + // ONE fsync barrier below confirms them + // (Always); fire-and-forget for // everysec/no. let bytes = bytes::Bytes::from(record); let lsn = crate::persistence::aof::AofWriterPool::issue_append_lsn(&repl_state, 0, bytes.len()); - if let Err(_aof_err) = pool.try_send_append_durable(0, lsn, bytes).await { - graph_aof_failed = true; + match pool.send_append_group(0, lsn, bytes).await { + Ok(true) => graph_barrier_needed = true, + Ok(false) => {} + Err(_) => graph_aof_failed = true, } } if let Some(ref counter) = change_counter { counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed); } } + if graph_barrier_needed { + if let Some(ref pool) = aof_pool { + if pool.fsync_barrier(0).await.is_err() { + graph_aof_failed = true; + } + } + } if graph_aof_failed { responses.push(Frame::Error(bytes::Bytes::from_static( crate::persistence::aof::AOF_FSYNC_ERR, @@ -2462,25 +2502,20 @@ mod tests { } } - /// FIX-W1-1 r3: discriminating ordering test for `flush_with_aof_ack`. + /// FIX-W1-1 r3 (updated for group commit): discriminating ordering test + /// for `flush_with_aof_ack`. /// /// This test calls the **real** `flush_with_aof_ack` function that the /// production handler uses — not an inline copy. The H1 contract is: /// - /// All AOF fsync acks MUST be awaited BEFORE any response is sent. - /// - /// Red state (broken ordering — send-before-ack): - /// If the fn were to flush responses BEFORE awaiting acks, the mock - /// 60ms fsync delay would NOT gate the first response, so - /// `elapsed_ms < 55` → test fails. - /// - /// Green state (ack-before-send — current production code): - /// `flush_with_aof_ack` awaits the 60ms ack BEFORE any `start_send`, - /// so `elapsed_ms >= 55` → test passes. + /// AOF durability MUST be confirmed BEFORE any response is sent. /// - /// Red verification was performed by temporarily inverting the fn body - /// (Phase 2 before Phase 1) and confirming `elapsed_ms ≈ 0ms` → FAIL. - /// The fn was then restored and the test passes consistently. + /// Under the group-commit protocol the writer channel carries the + /// entry's fire-and-forget `Append` followed by ONE zero-length + /// `AppendSync` barrier for the whole batch (previously: one awaited + /// `AppendSync` per entry). The 60ms mock fsync delay gates the barrier + /// ack; a broken ordering (flush responses before the barrier ack) + /// would send the first response at ~0ms → `elapsed_ms < 55` → FAIL. #[tokio::test] async fn flush_with_aof_ack_ack_precedes_response() { // Build an Always-policy pool backed by a real bounded channel. @@ -2491,16 +2526,22 @@ mod tests { std::time::Duration::ZERO, ); - // Mock writer: receives one AppendSync, sleeps 60ms to simulate fsync, - // then sends Synced. Runs on a blocking thread because flume's + // Mock writer: receives the entry's fire-and-forget Append, then the + // batch's ONE AppendSync barrier; sleeps 60ms to simulate the fsync, + // then acks Synced. Runs on a blocking thread because flume's // `Receiver::recv()` is synchronous. let mock_writer = tokio::task::spawn_blocking(move || { - let msg = rx.recv().expect("mock writer: message received"); + let first = rx.recv().expect("mock writer: append received"); + assert!( + matches!(first, AofMessage::Append { .. }), + "group commit enqueues the entry fire-and-forget (Append first)" + ); + let msg = rx.recv().expect("mock writer: barrier received"); if let AofMessage::AppendSync { ack, .. } = msg { std::thread::sleep(Duration::from_millis(60)); let _ = ack.send(AofAck::Synced); } else { - panic!("Always policy MUST send AppendSync; got non-AppendSync variant"); + panic!("Always policy MUST send an AppendSync barrier"); } });