fix #423: PreVote + leader lease guard, vote protocol fixes, write admission

CHANGELOG.md

@@ -18,6 +18,10 @@ All notable changes to this project will be documented in this file.
 
 - **`start_with()` accepts `impl AsRef<Path>`** (#415): Callers can now pass `&str`, `String`, `&Path`, or `PathBuf` — backward compatible, no migration needed.
 
+- **PreVote and leader lease guard** (#423): An isolated node no longer inflates its term and forces a healthy leader to step down. Elections run a PreVote round first (reusing `VoteRequest`/`VoteResponse` over a new `PreVote` RPC; no term bump, nothing persisted), and followers and leaders ignore vote/PreVote requests while a leader is alive (Raft dissertation §4.2.3).
+
+- **Write admission and CheckQuorum** (#423): A leader without a recent quorum ACK rejects new client writes with `NotLeader` after `election_timeout_min`, before anything is appended, so clients can retry safely. After `election_timeout_max x write_admission_election_timeout_multiple` (new key in `[raft.backpressure]`, default `2`) without a quorum ACK it steps down. A new leader gets one window for its first ACK; single-voter clusters are exempt.
+
 ### Fixed
 
 - **fix(ttl) #398**: Removed `lease.enabled` flag — TTL expiration is always active. Fixes fatal crash when calling `put_with_ttl` without setting the (now-removed) `lease.enabled = true`.
@@ -44,6 +48,10 @@ All notable changes to this project will be documented in this file.
   replicas — see [Throughput Optimization Guide](./d-engine/src/docs/performance/throughput-optimization-guide.md)
   for tuning `idle_flush_interval_ms`.
 
+- **Vote protocol hardening** (#423): Hard state (term + vote) is now fsynced before replying to a vote request and a flush failure is fatal (a lost vote could allow two leaders in one term). A vote from an older term no longer blocks a request in a newer term, a same-term step-down keeps the term's vote, a candidate adopts a higher term even when it cannot grant the vote (§5.1), only active voters campaign (none while a committed membership change is unapplied), the first election round starts immediately instead of after two election timeouts, and the read lease is published only after the leader's noop is applied (§8). After a long snapshot install the follower re-arms its election timer and leader-contact guard.
+
+- **Quorum confirmation no longer derives from replicated history** (#423): In a 5+ voter cluster, a leader cut off from the majority but still reached by one follower kept renewing its read lease and never stepped down, because the commit median over every peer's stored `match_index` stayed true. A leader is now confirmed only when a quorum of voters acknowledged a newer heartbeat round; unreachable peers stop counting. Applies to lease reads, queued linearizable reads, CheckQuorum, write admission and the vote guard.
+
 ### Changed
 
 - **MSRV raised to Rust 1.89**: The `data_dir` startup lock (prevents two node processes from
@@ -69,6 +77,12 @@ All notable changes to this project will be documented in this file.
 - **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.
 
+- **⚠️ Two new startup config checks** (#423). A config that violated them started before and now fails with a config error:
+  - `election.election_timeout_min` must be at least 3 x `replication.rpc_append_entries_clock_in_ms` (defaults 500 / 100). Raise `election_timeout_min` or lower the heartbeat interval.
+  - `backpressure.max_pending_writes` / `max_pending_reads` must be greater than `batching.max_batch_size`, or `0` (unlimited).
+
+- **Writes are at-least-once across a leader failover** (#423): a write already proposed when the leader steps down can still be committed by the new leader although the client received an error (`NotLeader`, `ProposeFailed` or `TermOutdated`). There is no request-id deduplication yet, so retrying a non-idempotent command can apply it twice.
+
 - **`data_dir` removed from `ClusterConfig`** — it is now a required explicit argument to all engine constructors (`EmbeddedEngine::start(data_dir)`, `StandaloneEngine::run(data_dir, shutdown_rx)`, etc.). Remove `[cluster] data_dir` / `[cluster] db_root_dir` from all config files; the constructor argument always wins and the config field is silently ignored.
 
 - **`cluster.log_dir` removed** — was validated but never consumed.

README.md

@@ -5,7 +5,7 @@
 [![codecov](https://codecov.io/gh/deventlab/d-engine/graph/badge.svg?token=K3BEDM45V8)](https://codecov.io/gh/deventlab/d-engine)
 ![Static Badge](https://img.shields.io/badge/license-MIT%20%7C%20Apache--2.0-blue)
 [![CI](https://github.com/deventlab/d-engine/actions/workflows/ci.yml/badge.svg)](https://github.com/deventlab/d-engine/actions/workflows/ci.yml)
-[![Ask DeepWiki](https://deepwiki.com/badge.svg)](https://deepwiki.com/deventlab/d-engine)
+[![DeepWiki](https://img.shields.io/badge/DeepWiki-d--engine-blue.svg?logo=data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAACwAAAAyCAYAAAAnWDnqAAAAAXNSR0IArs4c6QAAA05JREFUaEPtmUtyEzEQhtWTQyQLHNak2AB7ZnyXZMEjXMGeK/AIi+QuHrMnbChYY7MIh8g01fJoopFb0uhhEqqcbWTp06/uv1saEDv4O3n3dV60RfP947Mm9/SQc0ICFQgzfc4CYZoTPAswgSJCCUJUnAAoRHOAUOcATwbmVLWdGoH//PB8mnKqScAhsD0kYP3j/Yt5LPQe2KvcXmGvRHcDnpxfL2zOYJ1mFwrryWTz0advv1Ut4CJgf5uhDuDj5eUcAUoahrdY/56ebRWeraTjMt/00Sh3UDtjgHtQNHwcRGOC98BJEAEymycmYcWwOprTgcB6VZ5JK5TAJ+fXGLBm3FDAmn6oPPjR4rKCAoJCal2eAiQp2x0vxTPB3ALO2CRkwmDy5WohzBDwSEFKRwPbknEggCPB/imwrycgxX2NzoMCHhPkDwqYMr9tRcP5qNrMZHkVnOjRMWwLCcr8ohBVb1OMjxLwGCvjTikrsBOiA6fNyCrm8V1rP93iVPpwaE+gO0SsWmPiXB+jikdf6SizrT5qKasx5j8ABbHpFTx+vFXp9EnYQmLx02h1QTTrl6eDqxLnGjporxl3NL3agEvXdT0WmEost648sQOYAeJS9Q7bfUVoMGnjo4AZdUMQku50McDcMWcBPvr0SzbTAFDfvJqwLzgxwATnCgnp4wDl6Aa+Ax283gghmj+vj7feE2KBBRMW3FzOpLOADl0Isb5587h/U4gGvkt5v60Z1VLG8BhYjbzRwyQZemwAd6cCR5/XFWLYZRIMpX39AR0tjaGGiGzLVyhse5C9RKC6ai42ppWPKiBagOvaYk8lO7DajerabOZP46Lby5wKjw1HCRx7p9sVMOWGzb/vA1hwiWc6jm3MvQDTogQkiqIhJV0nBQBTU+3okKCFDy9WwferkHjtxib7t3xIUQtHxnIwtx4mpg26/HfwVNVDb4oI9RHmx5WGelRVlrtiw43zboCLaxv46AZeB3IlTkwouebTr1y2NjSpHz68WNFjHvupy3q8TFn3Hos2IAk4Ju5dCo8B3wP7VPr/FGaKiG+T+v+TQqIrOqMTL1VdWV1DdmcbO8KXBz6esmYWYKPwDL5b5FA1a0hwapHiom0r/cKaoqr+27/XcrS5UwSMbQAAAABJRU5ErkJggg==)](https://deepwiki.com/deventlab/d-engine)
 
 **d-engine** is a lightweight, embeddable distributed coordination engine for Rust — strong-consistency KV, watch streams, and leader election, running inside your process.
 

d-engine-client/src/lib.rs

@@ -179,8 +179,10 @@ impl Client {
     /// **What this does NOT do:**
     /// - Does not guarantee that the caller's in-flight requests succeeded;
     ///   requests sent before `refresh()` may have failed and need to be retried
-    /// - Does not implement application-level retry — the caller is responsible
-    ///   for re-issuing any operations that failed during the failover window
+    /// - Writes are at-least-once across a failover: a write that was already proposed when
+    ///   the leader stepped down may still be committed by the new leader even though the
+    ///   caller received an error. Re-issuing a non-idempotent command can apply it twice
+    ///   (the engine has no request-id dedup yet).
     /// - Does not update endpoints permanently; pass `new_endpoints` to change
     ///   the bootstrap list for this and future refreshes
     ///

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

@@ -180,11 +180,21 @@ impl RaftConfig {
         self.replication.validate()?;
         self.batching.validate()?;
         self.election.validate()?;
+        // Write admission rejects after `election_timeout_min` without a quorum ACK, so a few
+        // lost heartbeats must fit inside it or normal jitter rejects client writes.
+        let heartbeat_ms = self.replication.rpc_append_entries_clock_in_ms;
+        if self.election.election_timeout_min < heartbeat_ms.saturating_mul(3) {
+            return Err(Error::Config(ConfigError::Message(format!(
+                "election_timeout_min {}ms must be at least 3 x rpc_append_entries_clock_in_ms {}ms",
+                self.election.election_timeout_min, heartbeat_ms
+            ))));
+        }
         self.membership.validate()?;
         self.state_machine.validate()?;
         self.snapshot.validate()?;
         self.read_consistency.validate(self.election.election_timeout_min)?;
         self.read_actor.validate()?;
+        self.backpressure.validate(self.batching.max_batch_size)?;
         self.watch.validate()?;
         self.persistence.validate()?;
 
@@ -936,13 +946,25 @@ pub struct BackpressureConfig {
     /// **Default**: 50000 (0 = unlimited)
     #[serde(default = "default_max_pending_reads")]
     pub max_pending_reads: usize,
+
+    /// CheckQuorum step-down limit, as a multiple of `election.election_timeout_max`.
+    ///
+    /// A leader that no quorum has answered within this limit steps down (same term). New
+    /// client writes are rejected earlier, after `election_timeout_min` without an ACK, and
+    /// that rejection never changes the role.
+    ///
+    /// **Default**: 2 (2 s with the default 1 s `election_timeout_max`). Must be at least 1.
+    #[serde(default = "default_write_admission_election_timeout_multiple")]
+    pub write_admission_election_timeout_multiple: u64,
 }
 
 impl Default for BackpressureConfig {
     fn default() -> Self {
         Self {
             max_pending_writes: default_max_pending_writes(),
             max_pending_reads: default_max_pending_reads(),
+            write_admission_election_timeout_multiple:
+                default_write_admission_election_timeout_multiple(),
         }
     }
 }
@@ -955,7 +977,43 @@ fn default_max_pending_reads() -> usize {
     50_000
 }
 
+fn default_write_admission_election_timeout_multiple() -> u64 {
+    2
+}
+
 impl BackpressureConfig {
+    /// Validates the write admission window and the pending limits.
+    ///
+    /// `max_batch_size` is `batching.max_batch_size`: one Raft loop round pushes up to
+    /// `max_batch_size + 1` commands before the buffer is flushed, so a non-zero limit at or
+    /// below it rejects requests of a perfectly healthy node.
+    pub(crate) fn validate(
+        &self,
+        max_batch_size: usize,
+    ) -> Result<()> {
+        if self.write_admission_election_timeout_multiple == 0 {
+            return Err(Error::Config(ConfigError::Message(
+                "write_admission_election_timeout_multiple must be at least 1 \
+                     (0 leaves no window: the leader would reject every write)"
+                    .into(),
+            )));
+        }
+
+        for (name, limit) in [
+            ("max_pending_writes", self.max_pending_writes),
+            ("max_pending_reads", self.max_pending_reads),
+        ] {
+            if limit != 0 && limit <= max_batch_size {
+                return Err(Error::Config(ConfigError::Message(format!(
+                    "{name} {limit} must be greater than batching.max_batch_size \
+                         {max_batch_size} (0 = unlimited): a drained batch alone would trip the limit"
+                ))));
+            }
+        }
+
+        Ok(())
+    }
+
     /// Check if write request should be rejected due to backpressure
     ///
     /// Returns true if the current pending count exceeds the limit.

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

@@ -99,6 +99,7 @@ fn test_backpressure_unlimited_when_zero() {
     let config = BackpressureConfig {
         max_pending_writes: 0,
         max_pending_reads: 0,
+        ..Default::default()
     };
 
     // 0 = unlimited, should never reject
@@ -111,6 +112,7 @@ fn test_backpressure_write_limit_enforcement() {
     let config = BackpressureConfig {
         max_pending_writes: 100,
         max_pending_reads: 200,
+        ..Default::default()
     };
 
     // Below limit - allow
@@ -130,6 +132,7 @@ fn test_backpressure_read_limit_enforcement() {
     let config = BackpressureConfig {
         max_pending_writes: 100,
         max_pending_reads: 200,
+        ..Default::default()
     };
 
     // Below limit - allow
@@ -149,6 +152,7 @@ fn test_backpressure_write_and_read_independent() {
     let config = BackpressureConfig {
         max_pending_writes: 100,
         max_pending_reads: 200,
+        ..Default::default()
     };
 
     // Write limit doesn't affect read checks
@@ -167,11 +171,13 @@ fn test_backpressure_zero_vs_nonzero() {
     let config_unlimited_writes = BackpressureConfig {
         max_pending_writes: 0,
         max_pending_reads: 100,
+        ..Default::default()
     };
 
     let config_limited_writes = BackpressureConfig {
         max_pending_writes: 100,
         max_pending_reads: 0,
+        ..Default::default()
     };
 
     // Unlimited writes, limited reads
@@ -528,3 +534,133 @@ fn load_toml(text: &str) -> RaftConfig {
         .try_deserialize()
         .expect("config must deserialize")
 }
+
+// ============================================================================
+// Write admission / step-down config and the cross-field checks around it
+// ============================================================================
+
+#[test]
+fn test_write_admission_multiple_default_is_two() {
+    assert_eq!(
+        BackpressureConfig::default().write_admission_election_timeout_multiple,
+        2
+    );
+}
+
+/// A missing key parses to the default, so existing config files keep working.
+#[test]
+fn test_write_admission_multiple_missing_key_parses_to_default() {
+    let parsed: BackpressureConfig = config::Config::builder()
+        .add_source(config::File::from_str(
+            "max_pending_writes = 500",
+            config::FileFormat::Toml,
+        ))
+        .build()
+        .unwrap()
+        .try_deserialize()
+        .unwrap();
+
+    assert_eq!(parsed.max_pending_writes, 500);
+    assert_eq!(parsed.write_admission_election_timeout_multiple, 2);
+    assert_eq!(parsed.max_pending_reads, 50_000);
+}
+
+#[test]
+fn test_write_admission_multiple_zero_is_rejected() {
+    let mut config = RaftConfig::default();
+    config.backpressure.write_admission_election_timeout_multiple = 0;
+
+    assert!(config.validate().is_err());
+}
+
+#[test]
+fn test_write_admission_multiple_one_and_large_values_are_accepted() {
+    for multiple in [1, 2, 10, 1_000] {
+        let mut config = RaftConfig::default();
+        config.backpressure.write_admission_election_timeout_multiple = multiple;
+
+        assert!(
+            config.validate().is_ok(),
+            "multiple {multiple} must be accepted"
+        );
+    }
+}
+
+/// `max_pending_writes` / `max_pending_reads` must exceed `batching.max_batch_size` (one loop
+/// round pushes up to max_batch_size + 1 commands), unless 0 = unlimited.
+#[test]
+fn test_pending_limits_must_exceed_max_batch_size() {
+    let batch = 100;
+    let with = |writes: usize, reads: usize| {
+        let mut config = RaftConfig::default();
+        config.batching.max_batch_size = batch;
+        config.backpressure.max_pending_writes = writes;
+        config.backpressure.max_pending_reads = reads;
+        config.validate().is_ok()
+    };
+
+    assert!(
+        !with(batch, 50_000),
+        "writes limit == max_batch_size must be rejected"
+    );
+    assert!(
+        !with(batch - 1, 50_000),
+        "writes limit below max_batch_size must be rejected"
+    );
+    assert!(
+        with(batch + 1, 50_000),
+        "writes limit == max_batch_size + 1 must be accepted"
+    );
+    assert!(
+        !with(10_000, batch),
+        "reads limit == max_batch_size must be rejected"
+    );
+    assert!(
+        with(10_000, batch + 1),
+        "reads limit == max_batch_size + 1 must be accepted"
+    );
+    assert!(with(0, 0), "0 = unlimited must be accepted for both");
+    assert!(
+        with(0, 50_000) && with(10_000, 0),
+        "each limit is independent"
+    );
+}
+
+/// Write admission rejects after `election_timeout_min` without a quorum ACK, so that window must
+/// hold at least 3 heartbeats (two can be lost).
+#[test]
+fn test_election_timeout_min_must_hold_three_heartbeats() {
+    let validate_with = |election_min: u64, heartbeat: u64| {
+        let mut config = RaftConfig::default();
+        // Keep the read-lease rule out of the way: only the heartbeat ratio is under test.
+        config.read_consistency.lease_duration_ms = 10;
+        config.election.election_timeout_min = election_min;
+        config.election.election_timeout_max = 3_000;
+        config.replication.rpc_append_entries_clock_in_ms = heartbeat;
+        config.validate()
+    };
+
+    let too_small = validate_with(299, 100).expect_err("299 ms holds fewer than 3 heartbeats");
+    assert!(
+        too_small.to_string().contains("rpc_append_entries_clock_in_ms"),
+        "the error must name the heartbeat setting, got: {too_small}"
+    );
+    assert!(
+        validate_with(300, 100).is_ok(),
+        "exactly 3 heartbeats is accepted"
+    );
+    assert!(
+        validate_with(500, 100).is_ok(),
+        "the defaults (500 / 100) are accepted"
+    );
+    assert!(
+        validate_with(30, 10).is_ok(),
+        "the rule is a ratio, not a fixed size"
+    );
+    assert!(validate_with(29, 10).is_err());
+}
+
+#[test]
+fn test_default_config_is_valid() {
+    assert!(RaftConfig::default().validate().is_ok());
+}