perf #392: ReadActor fast path for Lease/Eventual Read — fix v0.2.4 throughput regression

.gitignore

@@ -53,3 +53,7 @@ CLAUDE.md
 .claude/
 .rules
 .github/copilot-instructions.md
+
+# MemPalace per-project files (issue #185)
+mempalace.yaml
+entities.json

CHANGELOG.md

@@ -4,6 +4,27 @@ All notable changes to this project will be documented in this file.
 
 ---
 
+## [v0.2.5] - 2026-XX-XX
+
+### Added
+
+- **ReadActor fast path for Eventual/LeaseRead** (#392): Dedicated `ReadActor` task serves `EventualConsistency` and `LeaseRead` without entering the Raft loop, eliminating channel contention under high read concurrency.
+  - `Arc<SM>` is owned exclusively by ReadActor — guarantees RocksDB LOCK release before Raft shutdown on `stop()`, fixing `test_snapshot_recovery_embedded`
+  - `ReadLease.revoke()`: atomic lease invalidation on leader demotion (replaces `invalidate()`)
+  - `EmbeddedClient` routes Eventual/Lease → ReadActor; falls back to `cmd_tx` on `LeaseInvalid`/`SmStopped`
+  - New `[raft.read_actor]` config section: `channel_capacity` (default 512), `max_drain` (default 100)
+  - **Benchmark (local embedded, 100 concurrent clients)**: Lease Read +10.9%, Eventual Read +13.1% vs v0.2.4; write throughput unchanged (see `benches/reports/v0.2.5/`)
+
+### 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`.
+
+### Changed
+
+- `[raft.read_actor]` replaces the previous flat `read_actor_channel_capacity` / `read_actor_max_drain` fields in `[raft]`. Update existing config files accordingly.
+
+---
+
 ## [v0.2.4] - 2026-05-23
 
 ### Added

MIGRATION_GUIDE.md

@@ -134,6 +134,7 @@ grep "WAL" /var/log/d-engine.log
 | v0.2.0–v0.2.2 | Absolute expiration | Compatible           | ✅ Yes (clear WAL from v0.1.x)                          |
 | v0.2.3        | Same as v0.2.0+     | **Incompatible**     | ✅ Yes (protobuf enum changes + API changes)             |
 | v0.2.4        | Same as v0.2.0+     | Compatible (additive)| ✅ Yes (delete `snapshot/` — format changed to CF export)|
+| v0.2.5        | Same as v0.2.0+     | Compatible (additive)| ⚠️ Minor (remove `lease.enabled` from config if present) |
 
 ---
 
@@ -493,4 +494,50 @@ See [Watch Feature Guide](https://docs.rs/d-engine/latest/d_engine/docs/server_g
 
 ---
 
+---
+
+## For v0.2.4 Users: TTL Config Change in v0.2.5 (#398)
+
+### What Changed
+
+The `enabled` flag under `[raft.state_machine.lease]` has been removed. TTL expiration is now **always active** — no opt-in required.
+
+In v0.2.4, omitting `enabled = true` caused a fatal crash on `put_with_ttl`. This bug is fixed in v0.2.5.
+
+### Migration
+
+If your config contains `enabled = true` or `enabled = false`, remove the line:
+
+```toml
+# Old (v0.2.4) — remove this line
+[raft.state_machine.lease]
+enabled = true   # ← delete
+
+# New (v0.2.5) — TTL always active, no flag needed
+[raft.state_machine.lease]
+cleanup_interval_ms = 1000
+```
+
+**Impact**: None if you do not touch the config — unrecognised fields are ignored. The only behavioral change is that TTL expiration is now unconditionally enabled.
+
+---
+
+## For v0.2.4 Users: New `[raft.read_actor]` Config Section in v0.2.5 (#392)
+
+This section is **optional** — both fields have defaults and existing configs work without changes.
+
+```toml
+[raft.read_actor]
+# mpsc channel buffer for Eventual/LeaseRead fast path.
+# Rule of thumb: ≥ 2× peak concurrent readers. Default: 512.
+channel_capacity = 512
+
+# Max reads drained per wakeup. Default: 100.
+max_drain = 100
+```
+
+No migration action required unless you want to tune read concurrency.
+
+---
+
 **Last Updated:** May 2026

README.md

@@ -54,7 +54,7 @@ use std::time::Duration;
 
 #[tokio::main]
 async fn main() {
-    let engine = EmbeddedEngine::start("./data").await.unwrap();
+    let engine = DefaultEmbeddedEngine::start("./data").await.unwrap();
     engine.wait_ready(Duration::from_secs(5)).await.unwrap();
 
     let client = engine.client();
@@ -123,13 +123,13 @@ Implement the `StorageEngine` and `StateMachine` traits for custom backends:
 
 ## Performance
 
-### d-engine v0.2.4 vs etcd
+### d-engine v0.2.5 vs etcd
 
-![d-engine vs etcd comparison](https://raw.githubusercontent.com/deventlab/d-engine/main/benches/reports/v0.2.4/d-engine_comparison_v0.2.4.png)
+![d-engine vs etcd comparison](https://raw.githubusercontent.com/deventlab/d-engine/main/benches/reports/v0.2.5/d-engine_comparison_v0.2.5.png)
 
-### d-engine v0.2.4 vs v0.2.3
+### d-engine v0.2.5 vs v0.2.4
 
-![d-engine v0.2.4 vs v0.2.3 comparison](https://raw.githubusercontent.com/deventlab/d-engine/main/benches/reports/v0.2.4/d-engine_v0.2.3_vs_v0.2.4_embedded_mode.png)
+![d-engine v0.2.5 vs v0.2.4 comparison](https://raw.githubusercontent.com/deventlab/d-engine/main/benches/reports/v0.2.5/d-engine_v0.2.5_vs_v0.2.4_embedded_mode.png)
 
 ```bash
 open benches/reports/

benches/embedded-bench/config/n1.toml

@@ -9,6 +9,10 @@ initial_cluster = [
 db_root_dir = "./data/n1"
 log_dir = "./logs"
 
+[raft.read_actor]
+channel_capacity = 2048
+max_drain = 1024
+
 [raft.batching]
 # Maximum number of commands to accumulate in a single batch during drain operations
 max_batch_size = 200

benches/embedded-bench/config/n2.toml

@@ -9,6 +9,10 @@ initial_cluster = [
 db_root_dir = "./data/n2"
 log_dir = "./logs"
 
+[raft.read_actor]
+channel_capacity = 2048
+max_drain = 1024
+
 [raft.batching]
 # Maximum number of commands to accumulate in a single batch during drain operations
 max_batch_size = 200

benches/embedded-bench/config/n3.toml

@@ -9,6 +9,10 @@ initial_cluster = [
 db_root_dir = "./data/n3"
 log_dir = "./logs"
 
+[raft.read_actor]
+channel_capacity = 2048
+max_drain = 1024
+
 [raft.batching]
 # Maximum number of commands to accumulate in a single batch during drain operations
 max_batch_size = 200

benches/embedded-bench/src/main.rs

@@ -1,4 +1,4 @@
-use d_engine::EmbeddedEngine;
+use d_engine::DefaultEmbeddedEngine;
 use d_engine::protocol::ReadConsistencyPolicy;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
@@ -202,7 +202,7 @@ async fn main() {
 /// Run a single benchmark test with specified parameters
 #[allow(clippy::too_many_arguments)]
 async fn run_benchmark_task(
-    engine: &Arc<EmbeddedEngine>,
+    engine: &Arc<DefaultEmbeddedEngine>,
     test_name: &str,
     command: Commands,
     total: u64,
@@ -294,7 +294,7 @@ async fn run_benchmark_task(
 
 /// Run all benchmark tests in batch mode
 async fn run_batch_tests(
-    engine: &Arc<EmbeddedEngine>,
+    engine: &Arc<DefaultEmbeddedEngine>,
     cli: &Cli,
 ) {
     println!("\n╔════════════════════════════════════════╗");
@@ -420,7 +420,7 @@ async fn run_batch_tests(
 
 /// Helper function to run a single test in batch mode
 async fn run_single_batch_test(
-    engine: &Arc<EmbeddedEngine>,
+    engine: &Arc<DefaultEmbeddedEngine>,
     test_name: &str,
     command: Commands,
     total: u64,
@@ -447,7 +447,7 @@ async fn run_local_benchmark(cli: Cli) {
     println!("Starting local benchmark mode...");
 
     let engine = Arc::new(
-        EmbeddedEngine::start_with(&cli.config_path)
+        DefaultEmbeddedEngine::start_with(&cli.config_path)
             .await
             .expect("Failed to start engine"),
     );
@@ -673,7 +673,7 @@ async fn run_http_server(cli: Cli) {
     println!("Health check port: {}", cli.health_port);
 
     let engine = Arc::new(
-        EmbeddedEngine::start_with(&cli.config_path)
+        DefaultEmbeddedEngine::start_with(&cli.config_path)
             .await
             .expect("Failed to start engine"),
     );
@@ -701,7 +701,7 @@ async fn run_http_server(cli: Cli) {
 }
 
 async fn start_health_check_server(
-    engine: Arc<EmbeddedEngine>,
+    engine: Arc<DefaultEmbeddedEngine>,
     port: u16,
 ) {
     let app = Router::new()
@@ -718,15 +718,15 @@ async fn start_health_check_server(
     axum::serve(listener, app).await.expect("Health check server failed");
 }
 
-async fn health_primary(State(engine): State<Arc<EmbeddedEngine>>) -> StatusCode {
+async fn health_primary(State(engine): State<Arc<DefaultEmbeddedEngine>>) -> StatusCode {
     if engine.is_leader() {
         StatusCode::OK
     } else {
         StatusCode::SERVICE_UNAVAILABLE
     }
 }
 
-async fn health_replica(State(engine): State<Arc<EmbeddedEngine>>) -> StatusCode {
+async fn health_replica(State(engine): State<Arc<DefaultEmbeddedEngine>>) -> StatusCode {
     if !engine.is_leader() {
         StatusCode::OK
     } else {
@@ -735,7 +735,7 @@ async fn health_replica(State(engine): State<Arc<EmbeddedEngine>>) -> StatusCode
 }
 
 async fn start_business_server(
-    engine: Arc<EmbeddedEngine>,
+    engine: Arc<DefaultEmbeddedEngine>,
     port: u16,
 ) {
     let app = Router::new()
@@ -753,7 +753,7 @@ async fn start_business_server(
 }
 
 async fn handle_put(
-    State(engine): State<Arc<EmbeddedEngine>>,
+    State(engine): State<Arc<DefaultEmbeddedEngine>>,
     Json(req): Json<PutRequest>,
 ) -> StatusCode {
     match engine.client().put(req.key.into_bytes(), req.value.into_bytes()).await {
@@ -763,7 +763,7 @@ async fn handle_put(
 }
 
 async fn handle_get(
-    State(engine): State<Arc<EmbeddedEngine>>,
+    State(engine): State<Arc<DefaultEmbeddedEngine>>,
     Path(key): Path<String>,
 ) -> Result<Json<GetResponse>, StatusCode> {
     match engine.client().get_eventual(key.into_bytes()).await {