perf #422: fix WAL fsync serialization regression — concurrent fsync pipeline

.dockerignore

@@ -1,2 +1,3 @@
 **/target/
 .git/
+examples/

CHANGELOG.md

@@ -26,6 +26,12 @@ All notable changes to this project will be documented in this file.
 
 - **`ClusterConfig::default()` now consistent with serde default** (#415): `Default::default()` previously produced an empty `initial_cluster`, but `#[serde(default)]` returned `[{id:1, ...}]`. `RaftNodeConfig::new()` uses the Rust `Default` impl, so programmatic configs always got an empty cluster. Fixed.
 
+- **WAL fsync serialization regression (#422)**: #407 enabled per-batch `flush_wal(true)` (Level 3 durability),
+  but the IO thread blocked inline on fsync — every entry paid the full fsync penalty individually under
+  moderate load. Introduced `FsyncCoordinator`: fsync is dispatched to `spawn_blocking`, the IO thread
+  returns immediately, and entries arriving during an in-flight fsync are coalesced into the same physical
+  disk flush. Storage-level group commit is restored without artificial batching windows.
+
 ### Changed
 
 - `[raft.read_actor]` replaces the previous flat `read_actor_channel_capacity` / `read_actor_max_drain` fields in `[raft]`. Update existing config files accordingly.
@@ -34,6 +40,14 @@ All notable changes to this project will be documented in this file.
 
 - **KV encoding moved out of `d-engine-core`** (#415): `ClientWriteRequest.command` is now `Option<Bytes>` (pre-serialized). Serialization happens in the server transport layer (embedded/standalone/gRPC handler). Core is encoding-agnostic.
 
+- **Metrics renamed with `core.` namespace prefix**: all `d-engine-core` metrics now use the `core.*`
+  prefix (e.g. `core.raft.fsync.duration_ms`, `core.raft.write.propose_to_apply_ms`,
+  `core.state_machine.apply_chunk.duration_ms`). The `[raft.metrics]` config section (`MetricsConfig`)
+  has been removed — metrics are always-on, disabled only by not installing a recorder.
+
+- **New `shutdown_timeout_ms` in `[raft.persistence]`** (default `5000`): bounds `close()` wait
+  against a stuck fsync task. The task itself continues running in the background.
+
 ---
 
 ## [v0.2.4] - 2026-05-23

Makefile

@@ -27,6 +27,31 @@
 # Cargo executable
 CARGO ?= cargo
 
+# On macOS with Homebrew: auto-detect compression lib paths to skip bundled C++
+# compilation of RocksDB dependencies, which fails under macOS 26 + Xcode 26
+# (Clang 16 lacks __builtin_ctzg/__builtin_clzg from LLVM 18+ SDK headers).
+# brew --prefix resolves correctly on both Apple Silicon (/opt/homebrew) and
+# Intel Mac (/usr/local). Silently no-ops when brew or a lib is absent.
+SNAPPY_PREFIX    := $(shell brew --prefix snappy 2>/dev/null)
+LZ4_PREFIX       := $(shell brew --prefix lz4    2>/dev/null)
+ZSTD_PREFIX      := $(shell brew --prefix zstd   2>/dev/null)
+BREW_ROCKSDB_ENV :=
+
+ifneq ($(SNAPPY_PREFIX),)
+ifneq ($(wildcard $(SNAPPY_PREFIX)/lib),)
+  BREW_ROCKSDB_ENV += SNAPPY_LIB_DIR=$(SNAPPY_PREFIX)/lib
+endif
+endif
+ifneq ($(LZ4_PREFIX),)
+  BREW_ROCKSDB_ENV += LZ4_LIB_DIR=$(LZ4_PREFIX)/lib
+endif
+ifneq ($(ZSTD_PREFIX),)
+  BREW_ROCKSDB_ENV += ZSTD_LIB_DIR=$(ZSTD_PREFIX)/lib
+endif
+ifneq ($(BREW_ROCKSDB_ENV),)
+  CARGO := $(BREW_ROCKSDB_ENV) $(CARGO)
+endif
+
 # Rust logging level for tests
 RUST_LOG_LEVEL ?= d_engine_server=debug,d_engine_core=debug,d_engine_client=debug,d_engine=debug
 

benches/embedded-bench/Cargo.toml

@@ -17,3 +17,4 @@ rand = "0.9"
 futures = "0.3"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tracing = { version = "0.1" }
+metrics-exporter-prometheus = "0.17.2"

benches/embedded-bench/Makefile

@@ -11,6 +11,10 @@ NODE ?= n1
 CONFIG_PATH := $(CONFIG_DIR)/$(NODE).toml
 SINGLE_NODE_CONFIG_PATH := $(CONFIG_DIR)/single_node.toml
 
+# Metrics port per node — same scheme as examples/three-nodes-standalone
+# (8081/8082/8083), reused because the two setups never run concurrently.
+METRICS_PORT := $(if $(filter n2,$(NODE)),8082,$(if $(filter n3,$(NODE)),8083,8081))
+
 # ===============================
 # Global Variables
 # ===============================
@@ -21,10 +25,38 @@ KEY_SIZE := 8
 VALUE_SIZE := 256
 SEQUENTIAL := --sequential-keys
 
+# Concurrent client tasks (embedded-bench is in-process — no --conns knob exists
+# in the binary, only --clients). Override: make test-high-conc-write CLIENTS=64
+CLIENTS ?= 1000
+
 # Optional write verification (VERIFY=true to enable)
 VERIFY ?= false
 VERIFY_FLAG := $(if $(filter true,$(VERIFY)),--verify-write,)
 
+
+# On macOS with Homebrew: auto-detect compression lib paths to skip bundled C++
+# compilation of RocksDB dependencies, which fails under macOS 26 + Xcode 26
+# (Clang 16 lacks __builtin_ctzg/__builtin_clzg from LLVM 18+ SDK headers).
+# brew --prefix resolves correctly on both Apple Silicon (/opt/homebrew) and
+# Intel Mac (/usr/local). Silently no-ops when brew or a lib is absent.
+SNAPPY_PREFIX    := $(shell brew --prefix snappy 2>/dev/null)
+LZ4_PREFIX       := $(shell brew --prefix lz4    2>/dev/null)
+ZSTD_PREFIX      := $(shell brew --prefix zstd   2>/dev/null)
+BREW_ROCKSDB_ENV :=
+
+ifneq ($(SNAPPY_PREFIX),)
+ifneq ($(wildcard $(SNAPPY_PREFIX)/lib),)
+  BREW_ROCKSDB_ENV += SNAPPY_LIB_DIR=$(SNAPPY_PREFIX)/lib
+endif
+endif
+ifneq ($(LZ4_PREFIX),)
+  BREW_ROCKSDB_ENV += LZ4_LIB_DIR=$(LZ4_PREFIX)/lib
+endif
+ifneq ($(ZSTD_PREFIX),)
+  BREW_ROCKSDB_ENV += ZSTD_LIB_DIR=$(ZSTD_PREFIX)/lib
+endif
+
+
 help:
 	@echo "Embedded-bench Makefile - Performance Testing"
 	@echo ""
@@ -59,7 +91,7 @@ help:
 
 build:
 	@echo "Building embedded-bench..."
-	cargo build --release --jobs 8
+	$(BREW_ROCKSDB_ENV) cargo build --release --jobs 8
 
 clean:
 	@echo "Cleaning build artifacts and data directories..."
@@ -80,13 +112,14 @@ clean-log-db:
 # ============================================
 
 # Single client write (10K requests)
-# Comparable to: embedded-bench --conns 1 --clients 1 --total 10000
+# Comparable to: embedded-bench --clients 1 --total 10000
 test-single-write: build
 	@echo "========================================"
 	@echo "Single Client Write (10K requests)"
 	@echo "Node: $(NODE)"
 	@echo "========================================"
 	CONFIG_PATH=$(SINGLE_NODE_CONFIG_PATH) \
+	METRICS_PORT=$(METRICS_PORT) \
 	RUST_LOG=embedded-bench=$(LOG_LEVEL),d_engine=$(LOG_LEVEL),timing=$(LOG_LEVEL) \
 	$(BENCH_BIN) \
 		--mode local \
@@ -99,18 +132,19 @@ test-single-write: build
 		put
 
 # High concurrency write (100K requests)
-# Comparable to: embedded-bench --conns 200 --clients 1000 --total 100000
+# Comparable to: embedded-bench --clients $(CLIENTS) --total 100000
 test-high-conc-write: build
 	@echo "========================================"
 	@echo "High Concurrency Write (100K requests)"
 	@echo "Node: $(NODE)"
 	@echo "========================================"
 	CONFIG_PATH=$(CONFIG_PATH) \
+	METRICS_PORT=$(METRICS_PORT) \
 	RUST_LOG=embedded-bench=$(LOG_LEVEL),d_engine=$(LOG_LEVEL),timing=$(LOG_LEVEL) \
 	$(BENCH_BIN) \
 		--mode local \
 		--total 100000 \
-		--clients 1000 \
+		--clients $(CLIENTS) \
 		--key-size $(KEY_SIZE) \
 		--value-size $(VALUE_SIZE) \
 		$(SEQUENTIAL) \
@@ -122,72 +156,76 @@ test-high-conc-write: build
 # ============================================
 
 # Linearizable read (100K requests)
-# Comparable to: embedded-bench --conns 200 --clients 1000 --total 100000 range --consistency l
+# Comparable to: embedded-bench --clients $(CLIENTS) --total 100000 range --consistency l
 test-linearizable-read: build
 	@echo "========================================"
 	@echo "Linearizable Read (100K requests)"
 	@echo "Node: $(NODE)"
 	@echo "========================================"
 	CONFIG_PATH=$(CONFIG_PATH) \
+	METRICS_PORT=$(METRICS_PORT) \
 	RUST_LOG=embedded-bench=$(LOG_LEVEL),d_engine=$(LOG_LEVEL),timing=$(LOG_LEVEL) \
 	$(BENCH_BIN) \
 		--mode local \
 		--total 100000 \
-		--clients 1000 \
+		--clients $(CLIENTS) \
 		--key-size $(KEY_SIZE) \
 		--value-size $(VALUE_SIZE) \
 		$(SEQUENTIAL) \
 		get --consistency l
 
 # Lease-based read (100K requests)
-# Comparable to: embedded-bench --conns 200 --clients 1000 --total 100000 range --consistency s
+# Comparable to: embedded-bench --clients $(CLIENTS) --total 100000 range --consistency s
 test-lease-read: build
 	@echo "========================================"
 	@echo "Lease-Based Read (100K requests)"
 	@echo "Node: $(NODE)"
 	@echo "========================================"
 	CONFIG_PATH=$(CONFIG_PATH) \
+	METRICS_PORT=$(METRICS_PORT) \
 	RUST_LOG=embedded-bench=$(LOG_LEVEL),d_engine=$(LOG_LEVEL),timing=$(LOG_LEVEL) \
 	$(BENCH_BIN) \
 		--mode local \
 		--total 100000 \
-		--clients 1000 \
+		--clients $(CLIENTS) \
 		--key-size $(KEY_SIZE) \
 		--value-size $(VALUE_SIZE) \
 		$(SEQUENTIAL) \
 		get --consistency s
 
 # Eventual consistency read (100K requests)
-# Comparable to: embedded-bench --conns 200 --clients 1000 --total 100000 range --consistency e
+# Comparable to: embedded-bench --clients $(CLIENTS) --total 100000 range --consistency e
 test-eventual-read: build
 	@echo "========================================"
 	@echo "Eventual Consistency Read (100K requests)"
 	@echo "Node: $(NODE)"
 	@echo "========================================"
 	CONFIG_PATH=$(CONFIG_PATH) \
+	METRICS_PORT=$(METRICS_PORT) \
 	RUST_LOG=embedded-bench=$(LOG_LEVEL),d_engine=$(LOG_LEVEL),timing=$(LOG_LEVEL) \
 	$(BENCH_BIN) \
 		--mode local \
 		--total 100000 \
-		--clients 1000 \
+		--clients $(CLIENTS) \
 		--key-size $(KEY_SIZE) \
 		--value-size $(VALUE_SIZE) \
 		$(SEQUENTIAL) \
 		get --consistency e
 
 # Hot-key test (100K requests, 10 keys)
-# Comparable to: embedded-bench --conns 200 --clients 1000 --total 100000 --key-space 10 range --consistency l
+# Comparable to: embedded-bench --clients $(CLIENTS) --total 100000 --key-space 10 range --consistency l
 test-hot-key: build
 	@echo "========================================"
 	@echo "Hot-Key Test (100K requests, 10 keys)"
 	@echo "Node: $(NODE)"
 	@echo "========================================"
 	CONFIG_PATH=$(CONFIG_PATH) \
+	METRICS_PORT=$(METRICS_PORT) \
 	RUST_LOG=embedded-bench=$(LOG_LEVEL),d_engine=$(LOG_LEVEL),timing=$(LOG_LEVEL) \
 	$(BENCH_BIN) \
 		--mode local \
 		--total 100000 \
-		--clients 1000 \
+		--clients $(CLIENTS) \
 		--key-size $(KEY_SIZE) \
 		--key-space 10 \
 		$(SEQUENTIAL) \
@@ -203,11 +241,12 @@ all-tests: build
 	@echo "Node: $(NODE)"
 	@echo "========================================"
 	CONFIG_PATH=$(CONFIG_PATH) \
+	METRICS_PORT=$(METRICS_PORT) \
 	RUST_LOG=embedded-bench=$(LOG_LEVEL),d_engine=$(LOG_LEVEL),timing=$(LOG_LEVEL) \
 	$(BENCH_BIN) \
 		--mode local \
 		--batch \
-		--clients 1000 \
+		--clients $(CLIENTS) \
 		--key-size $(KEY_SIZE) \
 		--value-size $(VALUE_SIZE) \
 		$(SEQUENTIAL) \

benches/embedded-bench/src/main.rs

@@ -18,7 +18,6 @@ use serde::{Deserialize, Serialize};
 
 // Batch test configuration constants
 const BATCH_TEST_TOTAL_REQUESTS: u64 = 100000;
-const BATCH_TEST_CONCURRENT_CLIENTS: usize = 100;
 
 #[derive(Parser, Debug, Clone)]
 #[command(author, version, about, long_about = None)]
@@ -185,6 +184,14 @@ async fn main() {
         cli.config_path = path;
     }
 
+    // Same port scheme as examples/three-nodes-standalone (8081/8082/8083 per node) —
+    // the two never run at the same time, so reusing the ports keeps the existing
+    // Prometheus "dengine-host" scrape job working for both.
+    let metrics_port: u16 = std::env::var("METRICS_PORT")
+        .map(|v| v.parse::<u16>().expect("METRICS_PORT must be a valid port"))
+        .unwrap_or(8081);
+    tokio::spawn(start_metrics_server(metrics_port));
+
     match cli.mode.as_str() {
         "local" => run_local_benchmark(cli).await,
         "server" => run_http_server(cli).await,
@@ -199,6 +206,28 @@ async fn main() {
     }
 }
 
+/// Starts the Prometheus metrics exporter with exponential _ms buckets matching
+/// examples/three-nodes-standalone/src/main.rs, so bench data stays comparable
+/// with that setup's Grafana dashboards. Default Prometheus buckets start at
+/// 5ms, collapsing all sub-5ms detail; these exponential buckets cover 0.1ms–1.6s.
+async fn start_metrics_server(port: u16) {
+    println!("Metrics server will start at http://0.0.0.0:{port}/metrics");
+
+    const MS_BUCKETS: &[f64] = &[
+        0.1, 0.2, 0.4, 0.8, 1.6, 3.2, 6.4, 12.8, 25.6, 51.2, 102.4, 204.8, 409.6, 819.2, 1638.4,
+    ];
+
+    metrics_exporter_prometheus::PrometheusBuilder::new()
+        .with_http_listener(([0, 0, 0, 0], port))
+        .set_buckets_for_metric(
+            metrics_exporter_prometheus::Matcher::Suffix("_ms".to_string()),
+            MS_BUCKETS,
+        )
+        .expect("failed to configure _ms buckets")
+        .install()
+        .expect("failed to start Prometheus metrics exporter");
+}
+
 /// Run a single benchmark test with specified parameters
 #[allow(clippy::too_many_arguments)]
 async fn run_benchmark_task(
@@ -325,7 +354,7 @@ async fn run_batch_tests(
         "High Concurrency Write (100K requests)",
         Commands::Put,
         BATCH_TEST_TOTAL_REQUESTS,
-        BATCH_TEST_CONCURRENT_CLIENTS,
+        cli.clients,
         cli,
     )
     .await
@@ -345,7 +374,7 @@ async fn run_batch_tests(
             consistency: "l".to_string(),
         },
         BATCH_TEST_TOTAL_REQUESTS,
-        BATCH_TEST_CONCURRENT_CLIENTS,
+        cli.clients,
         cli,
     )
     .await
@@ -364,7 +393,7 @@ async fn run_batch_tests(
             consistency: "s".to_string(),
         },
         BATCH_TEST_TOTAL_REQUESTS,
-        BATCH_TEST_CONCURRENT_CLIENTS,
+        cli.clients,
         cli,
     )
     .await
@@ -383,7 +412,7 @@ async fn run_batch_tests(
             consistency: "e".to_string(),
         },
         BATCH_TEST_TOTAL_REQUESTS,
-        BATCH_TEST_CONCURRENT_CLIENTS,
+        cli.clients,
         cli,
     )
     .await
@@ -404,7 +433,7 @@ async fn run_batch_tests(
             consistency: "l".to_string(),
         },
         BATCH_TEST_TOTAL_REQUESTS,
-        BATCH_TEST_CONCURRENT_CLIENTS,
+        cli.clients,
         &cli_hotkey,
     )
     .await