fix #228: Critical: Inconsistent reads in single-node mode

README.md

@@ -29,120 +29,89 @@ with pluggable storage backends, built-in observability, and tokio runtime suppo
 - **Tunable Persistence**: DiskFirst for durability or MemFirst for lower latency
 - **Flexible Read Consistency**: Three-tier model (Linearizable/Lease-Based/Eventual)
 - **Pluggable Storage**: Custom backends supported (RocksDB, Sled, Raw File)
-- **Observability**: Built-in metrics, structured logging, distributed tracing
-- **Extensible Design**: Business logic decoupled from protocol layer
 
 ---
 
-## Quick Start
-
-### Installation
+## Quick Start (Embedded Mode)
 
 Add d-engine to your `Cargo.toml`:
 
 ```toml
-# Default: Embedded mode with RocksDB (recommended for most use cases)
 d-engine = "0.2"
+```
 
-# Standalone mode: Connect to existing d-engine cluster
-d-engine = { version = "0.2", features = ["client"], default-features = false }
+Example code:
 
-# Custom storage backend (disable RocksDB)
-d-engine = { version = "0.2", features = ["server"], default-features = false }
-```
+```rust
+use d_engine::prelude::*;
+use std::time::Duration;
 
-### When to Use
+#[tokio::main]
+async fn main() {
+    let engine = EmbeddedEngine::start().await.unwrap();
+    engine.wait_ready(Duration::from_secs(5)).await.unwrap();
 
-#### Embedded Mode (Default) - Rust Applications
+    let client = engine.client();
+    client.put(b"hello".to_vec(), b"world".to_vec()).await.unwrap();
+    let value = client.get_linearizable(b"hello".to_vec()).await.unwrap();
 
-```toml
-d-engine = "0.2"
+    println!("Retrieved: {}", String::from_utf8_lossy(&value.unwrap()));
+    engine.stop().await.unwrap();
+}
 ```
 
-**Use when**: Building Rust services needing distributed coordination  
-**Examples**: Database HA coordinators, distributed schedulers, control plane services  
-**Why**: Zero-overhead (<0.1ms), single binary, type-safe
-
-❌ **Don't use**: For non-Rust applications (use Standalone mode instead)
+**→ Full example:** [examples/quick-start-embedded](examples/quick-start-embedded/README.md)
 
 ---
 
-#### Standalone Mode - Non-Rust Applications
+## Integration Modes
+
+### Embedded Mode - In-Process Deployment
 
 ```toml
-d-engine = { version = "0.2", features = ["client"], default-features = false }
+d-engine = "0.2"
 ```
 
-**Use when**: Go/Python/Java apps need distributed KV  
-**Examples**: Microservices requiring distributed KV, multi-language deployments  
-**Why**: Language-agnostic gRPC API, optimized for write-heavy workloads
+**Use when**: Building Rust applications that need distributed coordination  
+**Why**: Zero-overhead (<0.1ms), single binary, zero network cost
 
-\*Benchmarks show 51% higher write throughput vs etcd in test environments. Production results vary by hardware. See benches/ for methodology.
+**→ Examples:**
 
-❌ **Don't use**: For Rust apps (embedded mode has zero overhead)
+- [Quick Start Embedded](examples/quick-start-embedded/README.md) - Minimal setup
+- [Service Discovery Embedded](examples/service-discovery-embedded/README.md) - Watch API with sub-millisecond latency
 
 ---
 
-#### Custom Storage - Advanced Use Cases
+### Standalone Mode - Separate Service Deployment
 
 ```toml
-d-engine = { version = "0.2", features = ["server"], default-features = false }
+d-engine = { version = "0.2", features = ["client"], default-features = false }
 ```
 
-**Use when**: Specific storage requirements (Sled, memory-only, cloud storage)  
-**Examples**: Embedded devices, testing environments  
-**See**: [Custom Storage Guide](https://docs.rs/d-engine/latest/d_engine/docs/server_guide/index.html#implementing-custom-storage-engines) | [Sled Example](examples/sled-cluster)
+**Use when**: Application and d-engine run as separate processes  
+**Why**: Language-agnostic (Go/Python/Java/Rust), independent scaling, easier operations
 
----
-
-## Basic Usage (Single-Node Mode)
-
-```rust
-use d_engine::prelude::*;
-use std::time::Duration;
+> **Performance note**: Benchmark shows 45% higher write throughput vs etcd 3.5 in high-concurrency tests (M2 Mac single machine vs etcd on 3 GCE instances). See [benches/](benches/d-engine-bench/reports/v0.2.0/) for methodology and hardware details.
 
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    // Start embedded engine with RocksDB (auto-creates directories)
-    let engine = EmbeddedEngine::with_rocksdb("./data", None).await?;
+**Note**: Rust apps can use both modes - embedded for performance, standalone for operational flexibility
 
-    // Wait for leader election (single-node: instant)
-    engine.wait_ready(Duration::from_secs(5)).await?;
+**→ Example:** [Quick Start Standalone (Go client)](examples/quick-start-standalone/README.md)
 
-    // Get KV client (zero-overhead, in-process)
-    let client = engine.client();
+---
 
-    // Store and retrieve data
-    client.put(b"hello".to_vec(), b"world".to_vec()).await?;
+## Advanced: Custom Storage Backends
 
-    if let Some(value) = client.get(b"hello".to_vec()).await? {
-        println!("Retrieved: hello = {}", String::from_utf8_lossy(&value));
-    }
+For specific storage requirements (Sled, memory-only, cloud storage):
 
-    // Graceful shutdown
-    engine.stop().await?;
-    Ok(())
-}
+```toml
+d-engine = { version = "0.2", features = ["server"], default-features = false }
 ```
 
-## **Custom Storage Implementations**
-
-d-engine provides flexible storage abstraction layers. Implement your own storage engines and state machines by implementing the respective traits:
-
-- **Custom Storage Engines**: See [Implementing Custom Storage Engines](https://docs.rs/d-engine/latest/d_engine/docs/server_guide/index.html#implementing-custom-storage-engines)
-- **Custom State Machines**: See [Implementing Custom State Machines](https://docs.rs/d-engine/latest/d_engine/docs/server_guide/index.html#implementing-custom-state-machines)
-
-Note: Single-node deployment is supported for development and low-traffic production. For high availability, you can dynamically expand to a 3-node cluster with zero downtime.
-
----
-
-## Architecture
-
-**Hybrid threading model**: Single-threaded Raft core + async I/O layer
+Implement custom storage engines and state machines by implementing the respective traits:
 
-- **Consensus logic**: Dedicated event loop (eliminates lock contention)
-- **Network/Storage**: Tokio's async runtime for efficient I/O multiplexing
-- **Design goal**: Maximize single CPU core performance while minimizing resource overhead
+- **Custom Storage Engines**: [Implementation Guide](https://docs.rs/d-engine/latest/d_engine/docs/server_guide/index.html#implementing-custom-storage-engines)
+- **Custom State Machines**: [Implementation Guide](https://docs.rs/d-engine/latest/d_engine/docs/server_guide/index.html#implementing-custom-state-machines)
+- **Example**: [Sled Storage Backend](examples/sled-cluster)
 
 ---
 

config/base/raft.toml

@@ -83,6 +83,10 @@ max_buffered_entries = 10000
 default_policy = "LinearizableRead"
 lease_duration_ms = 250
 allow_client_override = true
+# Timeout to wait for state machine to catch up with commit index (milliseconds)
+# Used by LinearizableRead to ensure reads reflect all committed writes
+# Typical apply latency: <1ms on local SSD. Default: 10ms (safe buffer)
+state_machine_sync_timeout_ms = 10
 
 
 # Granular RPC compression configuration

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

@@ -949,6 +949,14 @@ pub struct ReadConsistencyConfig {
     /// When false, all reads use the cluster's default_policy setting.
     #[serde(default = "default_allow_client_override")]
     pub allow_client_override: bool,
+
+    /// Timeout in milliseconds to wait for state machine to catch up with commit index
+    ///
+    /// Used by LinearizableRead to ensure the state machine has applied all committed
+    /// entries before serving reads. Typical apply latency is <1ms on local SSD.
+    /// Default: 10ms (safe buffer for single-node local deployments)
+    #[serde(default = "default_state_machine_sync_timeout_ms")]
+    pub state_machine_sync_timeout_ms: u64,
 }
 
 impl Default for ReadConsistencyConfig {
@@ -957,6 +965,7 @@ impl Default for ReadConsistencyConfig {
             default_policy: ReadConsistencyPolicy::default(),
             lease_duration_ms: default_lease_duration_ms(),
             allow_client_override: default_allow_client_override(),
+            state_machine_sync_timeout_ms: default_state_machine_sync_timeout_ms(),
         }
     }
 }
@@ -971,6 +980,10 @@ fn default_allow_client_override() -> bool {
     true
 }
 
+fn default_state_machine_sync_timeout_ms() -> u64 {
+    10 // 10ms is safe for typical <1ms apply latency on local SSD
+}
+
 impl ReadConsistencyConfig {
     fn validate(&self) -> Result<()> {
         // Validate read consistency configuration

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

@@ -911,10 +911,13 @@ impl<T: TypeConfig> RaftRoleState for LeaderState<T> {
                                 Err(tonic::Status::failed_precondition(
                                     "enforce_quorum_consensus failed".to_string(),
                                 ))
-                            } else if let Err(e) = self.ensure_state_machine_upto_commit_index(
-                                &ctx.handlers.state_machine_handler,
-                                last_applied_index,
-                            ) {
+                            } else if let Err(e) = self
+                                .ensure_state_machine_upto_commit_index(
+                                    &ctx.handlers.state_machine_handler,
+                                    last_applied_index,
+                                )
+                                .await
+                            {
                                 warn!(
                                     "ensure_state_machine_upto_commit_index failed for linear read request"
                                 );
@@ -1842,7 +1845,7 @@ impl<T: TypeConfig> LeaderState<T> {
         (false, current_commit_index)
     }
 
-    pub fn ensure_state_machine_upto_commit_index(
+    pub async fn ensure_state_machine_upto_commit_index(
         &self,
         state_machine_handler: &Arc<SMHOF<T>>,
         last_applied: u64,
@@ -1856,6 +1859,13 @@ impl<T: TypeConfig> LeaderState<T> {
         if last_applied < commit_index {
             state_machine_handler.update_pending(commit_index);
 
+            // Wait for state machine to catch up
+            // This ensures linearizable reads see all committed writes
+            let timeout_ms = self.node_config.raft.read_consistency.state_machine_sync_timeout_ms;
+            state_machine_handler
+                .wait_applied(commit_index, std::time::Duration::from_millis(timeout_ms))
+                .await?;
+
             debug!("ensure_state_machine_upto_commit_index success");
         }
         Ok(())

d-engine-core/src/state_machine_handler/default_state_machine_handler.rs

@@ -102,6 +102,11 @@ where
     /// Temporary lock when snapshot is generated (to prevent concurrent snapshot generation)
     snapshot_lock: RwLock<()>,
 
+    /// Watch channel to notify when last_applied advances
+    /// Used for linearizable reads to wait until state machine catches up with commit_index
+    applied_notify_tx: tokio::sync::watch::Sender<u64>,
+    applied_notify_rx: tokio::sync::watch::Receiver<u64>,
+
     /// Broadcast channel for watch event notifications (fire-and-forget)
     /// StateMachine sends events here without blocking on watchers
     #[cfg(feature = "watch")]
@@ -155,6 +160,34 @@ where
         }
     }
 
+    async fn wait_applied(
+        &self,
+        target_index: u64,
+        timeout: Duration,
+    ) -> Result<()> {
+        let mut rx = self.applied_notify_rx.clone();
+
+        tokio::time::timeout(timeout, async {
+            loop {
+                let current = *rx.borrow_and_update();
+                if current >= target_index {
+                    return Ok(());
+                }
+                rx.changed()
+                    .await
+                    .map_err(|_| crate::Error::Fatal("apply notify channel closed".into()))?;
+            }
+        })
+        .await
+        .map_err(|_| {
+            let current_applied = *rx.borrow();
+            crate::Error::Fatal(format!(
+                "Timeout waiting for state machine to apply index {target_index} \
+                 (timeout: {timeout:?}, current_applied: {current_applied})"
+            ))
+        })?
+    }
+
     async fn apply_chunk(
         &self,
         chunk: Vec<Entry>,
@@ -210,6 +243,8 @@ where
                 // Efficiently obtain the maximum index: directly get the index of the last entry
                 if let Some(idx) = last_index {
                     self.last_applied.store(idx, Ordering::Release);
+                    // Notify waiters that last_applied has advanced
+                    let _ = self.applied_notify_tx.send(idx);
                 }
 
                 metrics::counter!(
@@ -764,6 +799,9 @@ where
             tokio::sync::broadcast::Sender<d_engine_proto::client::WatchResponse>,
         >,
     ) -> Self {
+        let (applied_notify_tx, applied_notify_rx) =
+            tokio::sync::watch::channel(last_applied_index);
+
         Self {
             node_id,
             last_applied: AtomicU64::new(last_applied_index),
@@ -779,6 +817,8 @@ where
 
             snapshot_lock: RwLock::new(()),
             snapshot_in_progress: AtomicBool::new(false),
+            applied_notify_tx,
+            applied_notify_rx,
             #[cfg(feature = "watch")]
             watch_event_tx,
         }

d-engine-core/src/state_machine_handler/mod.rs

@@ -78,6 +78,18 @@ where
         new_commit: u64,
     );
 
+    /// Waits until the state machine has applied entries up to the target index.
+    /// Returns error if timeout is reached before the target is applied.
+    ///
+    /// This is used to ensure linearizable reads: after leader confirms a log entry
+    /// is committed (via quorum), we must wait for the state machine to apply it
+    /// before reading to guarantee the read reflects all committed writes.
+    async fn wait_applied(
+        &self,
+        target_index: u64,
+        timeout: std::time::Duration,
+    ) -> Result<()>;
+
     /// Applies a batch of committed log entries to the state machine
     async fn apply_chunk(
         &self,

d-engine-server/src/network/connection_cache_test.rs

@@ -108,15 +108,16 @@ async fn test_remove_node() {
     let cache = ConnectionCache::new(test_config());
     let node1 = 1;
     let node2 = 2;
-    let (address, _tx) = mock_address().await;
 
-    // Create connections for two nodes
-    cache.get_channel(node1, ConnectionType::Data, address.clone()).await.unwrap();
-    cache
-        .get_channel(node1, ConnectionType::Control, address.clone())
-        .await
-        .unwrap();
-    cache.get_channel(node2, ConnectionType::Data, address.clone()).await.unwrap();
+    // Create connections for two nodes - use independent mock addresses
+    let (address1, _tx1) = mock_address().await;
+    cache.get_channel(node1, ConnectionType::Data, address1).await.unwrap();
+
+    let (address2, _tx2) = mock_address().await;
+    cache.get_channel(node1, ConnectionType::Control, address2).await.unwrap();
+
+    let (address3, _tx3) = mock_address().await;
+    cache.get_channel(node2, ConnectionType::Data, address3).await.unwrap();
 
     assert_eq!(cache.cache.len(), 3);