fix #415, #418: embedded API gaps and snapshot last_included mismatch

CHANGELOG.md

@@ -12,14 +12,28 @@ All notable changes to this project will be documented in this file.
 
 - **ReadActor fast path for Eventual/LeaseRead** (#392): Dedicated `ReadActor` serves `EventualConsistency` and `LeaseRead` off the Raft loop, eliminating channel contention under high read concurrency. Lease Read +10.9%, Eventual Read +13.1% vs v0.2.4 (100 concurrent clients, local embedded).
 
+- **`Command::Batch` — atomic multi-key writes** (#415): New `client.batch(ops)` API commits multiple Insert/Delete operations as a single Raft entry. All ops in a batch succeed or fail together. Library wrappers (e.g. service registration, distributed locks) no longer need workarounds like reserved-key encoding.
+
+- **`start_node()` public API** (#415): Both `EmbeddedEngine` and `StandaloneEngine` now expose `start_node(config, storage, sm)` — start an engine with a programmatic `RaftNodeConfig`, no config file required. For library wrappers that build their own config layer in Rust.
+
+- **`start_with()` accepts `impl AsRef<Path>`** (#415): Callers can now pass `&str`, `String`, `&Path`, or `PathBuf` — backward compatible, no migration needed.
+
 ### 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`.
 
+- **🛑 Snapshot data corruption — label/data mismatch** (#418): `last_included` in snapshot metadata was computed by subtracting `retained_log_entries` from `last_applied`, but the snapshot data already reflected the full `last_applied` state. Followers installing such snapshots would re-apply entries 91–100 after restoring state through index 100, causing double-application and permanent cluster state divergence. Also fixes a fabricated `LogId` where `last_included.term` was copied from `last_applied.term` instead of queried from the log. Snapshot label is now always truthful: `last_included == last_applied`.
+
+- **`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.
+
 ### Changed
 
 - `[raft.read_actor]` replaces the previous flat `read_actor_channel_capacity` / `read_actor_max_drain` fields in `[raft]`. Update existing config files accordingly.
 
+- **⚠️ `StateMachine::entry_term()` removed from trait** (#418): Term lookup belongs to the log layer, not the state machine. Custom `StateMachine` implementations must delete this method — it is no longer part of the trait. See [Migration Guide](./MIGRATION_GUIDE.md) for details.
+
+- **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.
+
 ---
 
 ## [v0.2.4] - 2026-05-23

MIGRATION_GUIDE.md

@@ -128,13 +128,13 @@ grep "WAL" /var/log/d-engine.log
 
 ## Timeline
 
-| Version       | WAL Format          | Wire Protocol        | Migration Required                                      |
-| ------------- | ------------------- | -------------------- | ------------------------------------------------------- |
-| v0.1.x        | Relative TTL        | Compatible           | -                                                       |
-| 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) |
+| Version       | WAL Format          | Wire Protocol         | Migration Required                                                                                                              |
+| ------------- | ------------------- | --------------------- | ------------------------------------------------------------------------------------------------------------------------------- |
+| v0.1.x        | Relative TTL        | Compatible            | -                                                                                                                               |
+| 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`; remove `entry_term()` from custom SM; use `start_node` instead of `start_custom_with_config`) |
 
 ---
 
@@ -463,6 +463,7 @@ Existing v0.2.3 snapshots cannot be loaded by v0.2.4 nodes.
 ### What Changed
 
 Watch buffer overflow is no longer silent. When a per-watcher channel fills up, the server now:
+
 1. Sends a `CANCELED` sentinel (`event_type = WATCH_EVENT_TYPE_CANCELED`, `error = WATCH_BUFFER_OVERFLOW`)
 2. Unregisters the watcher — no further events are delivered on that stream
 
@@ -494,6 +495,65 @@ See [Watch Feature Guide](https://docs.rs/d-engine/latest/d_engine/docs/server_g
 
 ---
 
+## For v0.2.4 Users: `StateMachine::entry_term()` Removed in v0.2.5 (#418)
+
+### What Changed
+
+The `entry_term()` method has been removed from the `StateMachine` trait. Term lookup belongs
+to the Raft log layer, not the state machine — this was always an architectural layering issue.
+
+### Impact
+
+⚠️ **Breaking for custom `StateMachine` implementations.**
+
+If you implemented a custom `StateMachine`, delete the `entry_term` method from your impl block.
+Built-in implementations (`FileStateMachine`, `RocksDBStateMachine`, `SledStateMachine`) are
+already updated and require no changes.
+
+```rust
+// Old (v0.2.4) — delete this block
+fn entry_term(&self, entry_id: u64) -> Option<u64> {
+    Some(1)
+}
+```
+
+Compilation will fail with "method not found in trait" if you forget to remove it.
+
+---
+
+## For v0.2.4 Users: `start_custom_with_config` → `start_node()` (#415)
+
+### What Changed
+
+The public API now exposes `start_node()` directly instead of routing through a thin wrapper.
+
+```rust
+// Old (v0.2.4) — removed
+EmbeddedEngine::start_custom_with_config(storage, sm, config).await?;
+
+// New (v0.2.5)
+EmbeddedEngine::start_node(config, storage, sm).await?;           // embedded
+StandaloneEngine::start_node(config, storage, sm, shutdown_rx).await?;  // standalone
+```
+
+`start_node` calls `config.validate()` internally — callers may pass validated or unvalidated configs.
+
+---
+
+## For v0.2.4 Users: Snapshot Label Fix (#418)
+
+### What Changed
+
+Snapshot `last_included` was previously computed by subtracting `retained_log_entries`
+from `last_applied`, introducing a label/data mismatch. Fixed: `last_included == last_applied`
+always.
+
+### Impact
+
+Transparent — no action required. Nodes that installed snapshots with the old label may have
+double-applied entries; upgrading to v0.2.5 prevents this from recurring. Existing snapshots
+are compatible.
+
 ---
 
 ## For v0.2.4 Users: TTL Config Change in v0.2.5 (#398)

d-engine-client/src/grpc_client.rs

@@ -1,10 +1,14 @@
-use std::sync::Arc;
-
+use super::ClientInner;
+use crate::ClientApiError;
+use crate::ClientResponseExt;
+use crate::scoped_timer::ScopedTimer;
 use arc_swap::ArcSwap;
 use bytes::Bytes;
+use d_engine_core::BatchOp;
 use d_engine_core::ScanResult;
 use d_engine_core::client::ErrorCode;
 use d_engine_core::client::KvEntry;
+use d_engine_core::client::{ClientApi, ClientApiResult};
 use d_engine_core::config::ReadConsistencyPolicy;
 use d_engine_proto::client::ClientReadRequest;
 use d_engine_proto::client::ClientWriteRequest;
@@ -15,21 +19,20 @@ use d_engine_proto::client::WatchRequest;
 use d_engine_proto::client::WatchResponse;
 use d_engine_proto::client::WriteCommand;
 use d_engine_proto::client::raft_client_service_client::RaftClientServiceClient;
+use d_engine_proto::client::write_command::BatchOp as ProtoBatchOp;
+use d_engine_proto::client::write_command::Delete;
+use d_engine_proto::client::write_command::Insert;
+use d_engine_proto::client::write_command::batch_op;
 use rand::Rng;
 use rand::SeedableRng;
 use rand::rngs::StdRng;
+use std::sync::Arc;
 use tonic::codec::CompressionEncoding;
 use tonic::transport::Channel;
 use tracing::debug;
 use tracing::error;
 use tracing::warn;
 
-use super::ClientInner;
-use crate::ClientApiError;
-use crate::ClientResponseExt;
-use crate::scoped_timer::ScopedTimer;
-use d_engine_core::client::{ClientApi, ClientApiResult};
-
 /// gRPC-based KV client for standalone mode. Obtained via [`crate::ClientBuilder`].
 ///
 /// For embedded mode use `EmbeddedClient` instead. Both implement `ClientApi`.
@@ -354,6 +357,59 @@ impl ClientApi for GrpcClient {
         }
     }
 
+    async fn batch(
+        &self,
+        ops: Vec<BatchOp>,
+    ) -> ClientApiResult<()> {
+        if ops.is_empty() {
+            return Err(ClientApiError::Business {
+                code: ErrorCode::InvalidRequest,
+                message: "batch ops must not be empty".into(),
+                required_action: None,
+            });
+        }
+
+        // Performance tracking for batch operation
+        let _timer = ScopedTimer::new("client::batch");
+
+        let client_inner = self.client_inner.load();
+        let proto_ops: Vec<ProtoBatchOp> = ops
+            .into_iter()
+            .map(|op| match op {
+                BatchOp::Insert { key, value } => ProtoBatchOp {
+                    op: Some(batch_op::Op::Insert(Insert {
+                        key,
+                        value,
+                        ttl_secs: 0,
+                    })),
+                },
+                BatchOp::Delete { key } => ProtoBatchOp {
+                    op: Some(batch_op::Op::Delete(Delete { key })),
+                },
+            })
+            .collect();
+        let command = WriteCommand::batch(proto_ops);
+
+        let request = ClientWriteRequest {
+            client_id: client_inner.client_id,
+            command: Some(command),
+        };
+
+        // Send write request to leader node (strong consistency required)
+        let mut client = self.make_leader_client().await?;
+        match client.handle_client_write(request).await {
+            Ok(response) => {
+                debug!("[:GrpcClient:batch] response: {:?}", response);
+                let client_response = response.get_ref();
+                client_response.validate_error()
+            }
+            Err(status) => {
+                error!("[:GrpcClient:batch] status: {:?}", status);
+                Err(Into::<ClientApiError>::into(ClientApiError::from(status)))
+            }
+        }
+    }
+
     async fn get(
         &self,
         key: impl AsRef<[u8]> + Send,

d-engine-core/benches/leader_state_bench.rs

@@ -175,7 +175,6 @@ impl BenchFixture {
         state_machine.expect_stop().returning(|| Ok(()));
         state_machine.expect_is_running().returning(|| true);
         state_machine.expect_get().returning(|_| Ok(None));
-        state_machine.expect_entry_term().returning(|_| None);
         state_machine.expect_apply_chunk().returning(|_| Ok(vec![]));
         state_machine.expect_len().returning(|| 0);
         state_machine.expect_update_last_applied().returning(|_| ());

d-engine-core/src/client/client_api.rs

@@ -25,6 +25,7 @@
 //! }
 //! ```
 
+use crate::BatchOp;
 use crate::ScanResult;
 use crate::client::client_api_error::ClientApiResult;
 use crate::config::ReadConsistencyPolicy;
@@ -97,6 +98,30 @@ pub trait ClientApi: Send + Sync {
         ttl_secs: u64,
     ) -> ClientApiResult<()>;
 
+    /// Atomically commits multiple write operations as a single Raft log entry.
+    ///
+    /// All operations succeed or none apply — all-or-nothing. Accepts only writes
+    /// (insert / delete); conditional writes belong to transactions (#298).
+    ///
+    /// # Errors
+    ///
+    /// - `InvalidArgument` if `ops` is empty
+    /// - `Network` if the node is shutting down or the request times out
+    /// - `Business` for server-side rejections (e.g., not the current leader)
+    ///
+    /// # Example
+    ///
+    /// ```rust,ignore
+    /// client.batch(vec![
+    ///     BatchOp::Insert { key: Bytes::from("a"), value: Bytes::from("1") },
+    ///     BatchOp::Delete { key: Bytes::from("b") },
+    /// ]).await?;
+    /// ```
+    async fn batch(
+        &self,
+        ops: Vec<BatchOp>,
+    ) -> ClientApiResult<()>;
+
     /// Retrieves the value associated with a key.
     ///
     /// Uses linearizable reads by default, ensuring the returned value

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

@@ -11,7 +11,7 @@ pub use client_api::ClientApi;
 pub use client_api_error::{ClientApiError, ClientApiResult};
 pub use types::{
     ClientReadRequest, ClientResponse, ClientResponsePayload, ClientWriteRequest, ErrorCode,
-    KvEntry, LeaderHint, ReadResults, WriteOperation, WriteResult,
+    KvEntry, LeaderHint, ReadResults, WriteResult,
 };
 // ReadConsistencyPolicy lives in config but is part of the public client API surface.
 pub use crate::config::ReadConsistencyPolicy;

d-engine-core/src/client/types.rs

@@ -29,31 +29,7 @@ pub use crate::config::ReadConsistencyPolicy;
 #[derive(Debug, Clone, PartialEq)]
 pub struct ClientWriteRequest {
     pub client_id: u32,
-    pub command: Option<WriteOperation>,
-}
-
-/// Decoded write operation — the unit submitted by a client.
-///
-/// Mirrors proto `WriteCommand` in shape but carries no prost annotations.
-/// Core owns the serialization to Raft log bytes (`WriteOperation → proto::WriteCommand → bytes`);
-/// transport adapters work with this native type only.
-#[derive(Debug, Clone, PartialEq)]
-pub enum WriteOperation {
-    Insert {
-        key: Bytes,
-        value: Bytes,
-        /// `None` = no expiration. Proto encodes this as `ttl_secs = 0`.
-        ttl_secs: Option<u64>,
-    },
-    Delete {
-        key: Bytes,
-    },
-    CompareAndSwap {
-        key: Bytes,
-        /// `None` means the key must not exist for the swap to succeed.
-        expected: Option<Bytes>,
-        new_value: Bytes,
-    },
+    pub command: Option<Bytes>,
 }
 
 // ─── Read request ─────────────────────────────────────────────────────────────