fix #446: gate commit quorum and follower ACKs on fsync-durable index (RPO=0)

.dockerignore

@@ -1,3 +1,5 @@
 **/target/
 .git/
-examples/
+examples/*
+!examples/three-nodes-standalone
+!examples/client-usage-standalone

CHANGELOG.md

@@ -32,6 +32,18 @@ All notable changes to this project will be documented in this file.
   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.
 
+- **🛑 Client-acknowledged writes could be lost on correlated power loss (#446)**: Raft commit quorum
+  counted the leader's own log contribution using its in-memory tail (`last_entry_id()`), not its
+  fsync-confirmed position (`durable_index()`) — a write could reach a majority-looking commit index,
+  and be acknowledged to the client, before enough replicas had actually synced it to disk. If those
+  nodes then lost power before their next fsync, the acknowledged write was gone. Fixed: leader quorum
+  calculation, follower `AppendEntries` ACK timing (a follower now withholds its response until its own
+  `durable_index` reaches the acknowledged entry), and single-voter clusters (previously exempted from
+  this class of fix, see #329) all gate on `durable_index`. RPO=0 for acknowledged writes is now a
+  mandatory invariant. Net effect: write acknowledgment latency now includes fsync time on a quorum of
+  replicas — see [Throughput Optimization Guide](./d-engine/src/docs/performance/throughput-optimization-guide.md)
+  for tuning `idle_flush_interval_ms`.
+
 ### Changed
 
 - **MSRV raised to Rust 1.89**: The `data_dir` startup lock (prevents two node processes from
@@ -65,6 +77,17 @@ All notable changes to this project will be documented in this file.
 
 - **`NodeBuilder` is no longer public** — use `EmbeddedEngine::start_custom`/`StandaloneEngine::run_custom` to plug in a custom storage engine or state machine. See [Migration Guide](./MIGRATION_GUIDE.md) for details.
 
+- **⚠️ `[raft] ordered_channel_capacity` renamed to `max_pending_append_responses`** (#446): Follows the
+  gRPC `AppendEntries` forwarder rewrite (`FuturesUnordered`-based, no longer strict-FIFO) that shipped
+  alongside the durability fix above. Old field name is silently ignored, not an error — update existing
+  configs to the new name to keep the setting in effect.
+
+- **⚠️ `[raft.persistence] strategy` removed** (#446): `PersistenceStrategy` was a single-variant enum
+  (`MemFirst`) left over from #268; its only meaning now lives in whether an entry has reached
+  `durable_index`, which is no longer a configurable choice. Existing configs setting `strategy =
+  "MemFirst"` or `"DiskFirst"` are silently ignored, not an error — remove the field, `flush_policy`
+  is the only persistence knob now.
+
 ---
 
 ## [v0.2.4] - 2026-05-23

benches/embedded-bench/config/n1.toml

@@ -16,11 +16,7 @@ max_drain = 1024
 max_batch_size = 200
 
 [raft.persistence]
-strategy = "MemFirst"
 flush_policy = { Batch = { idle_flush_interval_ms = 1000 } }
-# Maximum number of log entries to buffer in memory
-# when using async persistence strategies (MemFirst/Batched)
-max_buffered_entries = 10000
 
 [raft.metrics]
 enable_backpressure = false

benches/embedded-bench/config/n2.toml

@@ -16,11 +16,7 @@ max_drain = 1024
 max_batch_size = 200
 
 [raft.persistence]
-strategy = "MemFirst"
 flush_policy = { Batch = { idle_flush_interval_ms = 1000 } }
-# Maximum number of log entries to buffer in memory
-# when using async persistence strategies (MemFirst/Batched)
-max_buffered_entries = 10000
 
 [raft.metrics]
 enable_backpressure = false

benches/embedded-bench/config/n3.toml

@@ -16,11 +16,7 @@ max_drain = 1024
 max_batch_size = 200
 
 [raft.persistence]
-strategy = "MemFirst"
 flush_policy = { Batch = { idle_flush_interval_ms = 1000 } }
-# Maximum number of log entries to buffer in memory
-# when using async persistence strategies (MemFirst/Batched)
-max_buffered_entries = 10000
 
 [raft.metrics]
 enable_backpressure = false

benches/reports/v0.2.5/bench_report_v0.2.5.md

@@ -59,7 +59,7 @@ _(v0.2.5: 6-round average; v0.2.4: 4-round average; v0.2.3: 4-round average (Lea
 _(v0.2.5: 5-round average; v0.2.4: 5-round average; v0.2.3: 5-round average. All manually collected. 2026-07-12: 4-round average (conns=200, clients=200, Docker monitoring stack stopped).)_
 
 | **Scenario**        | **Metric**  | **v0.2.3**   | **v0.2.4**   | **v0.2.5**   | **Δ (v0.2.4→v0.2.5)** | **0712**      | **Δ (v0.2.5→0712)** |
-| ------------------- | ----------- | ------------ | ------------ | ------------ | --------------------- | ------------ | ------------------- |
+| ------------------- | ----------- | ------------ | ------------ | ------------ | --------------------- | ------------- | ------------------- |
 | Single Client Write | Throughput  | 6,421 ops/s  | 5,245 ops/s  | 5,234 ops/s  | stable                | 9,450 ops/s   | **+80.5%** ✅       |
 |                     | Avg Latency | 0.155 ms     | 0.190 ms     | 0.190 ms     | stable                | 0.105 ms      | **-44.6%** ✅       |
 |                     | p99 Latency | 0.200 ms     | 0.235 ms     | 0.237 ms     | stable                | 0.223 ms      | -6.0% →             |
@@ -214,7 +214,6 @@ read_actor_channel_capacity = 10240
 read_actor_max_drain = 2000
 
 [raft.persistence]
-strategy = "MemFirst"
 flush_policy = { Batch = { idle_flush_interval_ms = 1000 } }
 
 [raft.batching]

d-engine-core/src/config/raft.rs

@@ -79,11 +79,15 @@ pub struct RaftConfig {
     #[serde(default = "default_cmd_channel_capacity")]
     pub cmd_channel_capacity: usize,
 
-    /// Ordered channel capacity for stream_append_entries ordering
-    /// Controls buffering of response receivers in FIFO order
-    /// Default value is set via default_ordered_channel_capacity() function
-    #[serde(default = "default_ordered_channel_capacity")]
-    pub ordered_channel_capacity: usize,
+    /// Max in-flight AppendEntries requests on `stream_append_entries` that can be
+    /// dispatched to the Raft loop and awaiting their response at once. Once this many
+    /// are pending, the stream stops reading new requests until one completes — this
+    /// bounds memory/task growth if this node's own durable_index stalls (RPO=0, #446).
+    /// Also used directly as the output channel's buffer size, since completed
+    /// responses can never outnumber in-flight requests.
+    /// Default value is set via default_max_pending_append_responses() function
+    #[serde(default = "default_max_pending_append_responses")]
+    pub max_pending_append_responses: usize,
 
     /// ReadActor configuration — tuning for the dedicated Eventual/LeaseRead fast path.
     #[serde(default)]
@@ -141,7 +145,7 @@ impl Default for RaftConfig {
             auto_join: AutoJoinConfig::default(),
             snapshot_rpc_timeout_ms: default_snapshot_rpc_timeout_ms(),
             cmd_channel_capacity: default_cmd_channel_capacity(),
-            ordered_channel_capacity: default_ordered_channel_capacity(),
+            max_pending_append_responses: default_max_pending_append_responses(),
             read_actor: ReadActorConfig::default(),
             read_consistency: ReadConsistencyConfig::default(),
             backpressure: BackpressureConfig::default(),
@@ -201,7 +205,7 @@ fn default_cmd_channel_capacity() -> usize {
     1024
 }
 
-fn default_ordered_channel_capacity() -> usize {
+fn default_max_pending_append_responses() -> usize {
     1024
 }
 
@@ -817,27 +821,6 @@ impl Default for PromotionConfig {
 fn default_stale_learner_threshold() -> Duration {
     Duration::from_secs(300)
 }
-/// Defines how Raft log entries are persisted and accessed.
-///
-/// All strategies use a configurable [`FlushPolicy`] to control when memory contents
-/// are flushed to disk, affecting write latency and durability guarantees.
-///
-/// **Note:** Both strategies now fully load all log entries from disk into memory at startup.
-/// The in-memory `SkipMap` serves as the primary data structure for reads in all modes.
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
-pub enum PersistenceStrategy {
-    /// Memory-first persistence strategy.
-    ///
-    /// - **Write path**: On append, the log entry is first written to the in-memory `SkipMap` and
-    ///   acknowledged immediately. Disk persistence happens asynchronously in the background,
-    ///   governed by [`FlushPolicy`].
-    ///
-    /// - **Read path**: Reads are always served from the in-memory `SkipMap`.
-    ///
-    /// - **Startup behavior**: All log entries are loaded from disk into memory at startup.
-    ///
-    MemFirst,
-}
 
 /// Controls when in-memory logs should be flushed to disk.
 ///
@@ -857,40 +840,20 @@ pub enum FlushPolicy {
 /// Configuration parameters for log persistence behavior
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct PersistenceConfig {
-    /// Strategy for persisting Raft logs
-    ///
-    /// This controls the trade-off between durability guarantees and performance
-    /// characteristics. The choice impacts both write throughput and recovery
-    /// behavior after node failures.
-    #[serde(default = "default_persistence_strategy")]
-    pub strategy: PersistenceStrategy,
-
     /// Flush policy for asynchronous strategies
     ///
     /// This controls when log entries are flushed to disk. The choice impacts
     /// write performance and durability guarantees.
     #[serde(default = "default_flush_policy")]
     pub flush_policy: FlushPolicy,
 
-    /// Maximum number of in-memory log entries to buffer when using async strategies
-    ///
-    /// This acts as a safety valve to prevent memory exhaustion during periods of
-    /// high write throughput or when disk persistence is slow.
-    #[serde(default = "default_max_buffered_entries")]
-    pub max_buffered_entries: usize,
-
     /// Maximum time to wait, on shutdown, for an in-flight fsync task to finish
     /// before giving up. Bounds close() against a stuck/slow disk — the task
     /// itself is not cancelled, it keeps running in the background regardless.
     #[serde(default = "default_shutdown_timeout_ms")]
     pub shutdown_timeout_ms: u64,
 }
 
-/// Default persistence strategy (optimized for balanced workloads)
-fn default_persistence_strategy() -> PersistenceStrategy {
-    PersistenceStrategy::MemFirst
-}
-
 /// Default flush policy for asynchronous strategies
 ///
 /// This controls when log entries are flushed to disk. The choice impacts
@@ -901,11 +864,6 @@ fn default_flush_policy() -> FlushPolicy {
     }
 }
 
-/// Default maximum buffered log entries
-fn default_max_buffered_entries() -> usize {
-    10_000
-}
-
 fn default_shutdown_timeout_ms() -> u64 {
     5_000
 }
@@ -933,9 +891,7 @@ impl PersistenceConfig {
 impl Default for PersistenceConfig {
     fn default() -> Self {
         Self {
-            strategy: default_persistence_strategy(),
             flush_policy: default_flush_policy(),
-            max_buffered_entries: default_max_buffered_entries(),
             shutdown_timeout_ms: default_shutdown_timeout_ms(),
         }
     }

d-engine-core/src/event.rs

@@ -71,6 +71,14 @@ pub enum InternalEvent {
         durable_index: u64,
     },
 
+    /// Raw fsync-completion signal — NOT yet validated. Consumer must call
+    /// `raft_log().try_advance_durable_index(index, term)`, which re-checks
+    /// content before actually advancing `durable_index`.
+    FsyncCompleted {
+        index: u64,
+        term: u64,
+    },
+
     /// AppendEntries result from a per-follower ReplicationWorker back to the Raft loop.
     /// Leader processes this in handle_append_result: updates match_index, re-calculates commit,
     /// and drains pending_client_writes when quorum is achieved.

d-engine-core/src/lib.rs

@@ -173,6 +173,11 @@ pub(crate) fn if_higher_term_found(
 /// entries in the  logs. If the logs have last entries with different terms, then the log with the
 /// later term is more up-to-date. If the logs end with the same term, then whichever log is longer
 /// is more up-to-date.
+///
+/// #446: callers must pass the in-memory last-log-id (last_entry_id), never durable_index.
+/// A node with an un-fsynced tail must still be able to reject a candidate whose log is
+/// genuinely less up to date — voting eligibility and commit-durability are separate
+/// concerns and must not share the same index source.
 pub(crate) fn is_target_log_more_recent(
     my_last_log_index: u64,
     my_last_log_term: u64,

d-engine-core/src/raft.rs

@@ -609,6 +609,15 @@ where
                     .handle_log_flushed(durable_index, &self.ctx, &self.internal_event_tx)
                     .await;
             }
+            InternalEvent::FsyncCompleted { index, term } => {
+                if let Some(new_durable) =
+                    self.ctx.raft_log().try_advance_durable_index(index, term)
+                {
+                    self.role
+                        .handle_log_flushed(new_durable, &self.ctx, &self.internal_event_tx)
+                        .await;
+                }
+            }
             InternalEvent::AppendResult {
                 follower_id,
                 result,

d-engine-core/src/raft_role/follower_state.rs

@@ -7,6 +7,7 @@ use d_engine_proto::server::cluster::ClusterConfUpdateResponse;
 use d_engine_proto::server::cluster::LeaderDiscoveryResponse;
 use d_engine_proto::server::election::VoteResponse;
 use d_engine_proto::server::storage::SnapshotMetadata;
+use std::collections::BTreeMap;
 use std::fmt::Debug;
 use std::marker::PhantomData;
 use std::sync::Arc;
@@ -43,6 +44,7 @@ use crate::RaftNodeConfig;
 use crate::Result;
 use crate::StateTransitionError;
 use crate::TypeConfig;
+use crate::role_state::PendingAck;
 use crate::role_state::schedule_and_execute_purge;
 use crate::utils::cluster::error;
 use crate::utils::cluster_printer::print_role_transition_line;
@@ -73,6 +75,10 @@ pub struct FollowerState<T: TypeConfig> {
     /// Last physically purged log index (inclusive)
     pub last_purged_index: Option<LogId>,
 
+    /// AppendEntries responses withheld pending this node's own durable_index.
+    /// See `role_state::PendingAck`.
+    pending_append_acks: BTreeMap<u64, PendingAck>,
+
     // -- Snapshot Management --
     /// Prevents concurrent snapshot creation
     ///
@@ -463,6 +469,12 @@ impl<T: TypeConfig> RaftRoleState for FollowerState<T> {
     fn pending_purge_upto_mut(&mut self) -> Option<&mut Option<LogId>> {
         Some(&mut self.pending_purge_upto)
     }
+
+    fn pending_append_acks_mut(
+        &mut self
+    ) -> Option<&mut std::collections::BTreeMap<u64, super::role_state::PendingAck>> {
+        Some(&mut self.pending_append_acks)
+    }
 }
 
 impl<T: TypeConfig> FollowerState<T> {
@@ -484,6 +496,7 @@ impl<T: TypeConfig> FollowerState<T> {
                 node_config.raft.election.election_timeout_max,
             )),
             node_config,
+            pending_append_acks: BTreeMap::new(),
             snapshot_in_progress: AtomicBool::new(false),
             _marker: PhantomData,
             last_purged_index: None,
@@ -511,6 +524,7 @@ impl<T: TypeConfig> From<&CandidateState<T>> for FollowerState<T> {
             )),
             node_config: candidate_state.node_config.clone(),
             snapshot_in_progress: AtomicBool::new(false),
+            pending_append_acks: BTreeMap::new(),
             last_purged_index: candidate_state.last_purged_index,
             // scheduled_purge_upto: None,
             _marker: PhantomData,
@@ -527,6 +541,7 @@ impl<T: TypeConfig> From<&LeaderState<T>> for FollowerState<T> {
                 leader_state.node_config.raft.election.election_timeout_max,
             )),
             node_config: leader_state.node_config.clone(),
+            pending_append_acks: BTreeMap::new(),
             snapshot_in_progress: AtomicBool::new(
                 leader_state.snapshot_in_progress.load(Ordering::SeqCst),
             ),
@@ -548,6 +563,7 @@ impl<T: TypeConfig> From<&LearnerState<T>> for FollowerState<T> {
             )),
             node_config: learner_state.node_config.clone(),
             snapshot_in_progress: AtomicBool::new(false),
+            pending_append_acks: BTreeMap::new(),
             last_purged_index: learner_state.last_purged_index,
             pending_purge_upto: learner_state.pending_purge_upto,
             _marker: PhantomData,