Skip to content

perf(server): appendfsync-always local writes join per-batch group commit #239

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 1 commit into from
Jul 7, 2026
Merged

perf(server): appendfsync-always local writes join per-batch group commit #239

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
25 changes: 25 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,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`
Expand Down
70 changes: 44 additions & 26 deletions src/server/conn/handler_monoio/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
}
}
}
Expand Down Expand Up @@ -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;
}
}
}
}
Expand Down Expand Up @@ -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);
Expand All @@ -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;
}
}
}
}
Expand Down Expand Up @@ -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.
Expand Down
74 changes: 50 additions & 24 deletions src/server/conn/handler_sharded/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
}
}
}
Expand Down Expand Up @@ -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;
}
}
}
}
Expand Down Expand Up @@ -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;
}
}
}
}
Expand All @@ -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.
Expand Down
Loading
Loading