From 9a9ff49ed670f28c8864ce56e834c7acd51c1b4e Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Sun, 17 May 2026 20:41:55 +0800 Subject: [PATCH 1/7] feat(scan) #378: add scan_prefix API for zero-race-window watch reconnection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add linearizable prefix scan across all storage backends and client interfaces, enabling clients to reconnect watches without missing events. Protocol: - Add KvEntry, ScanRequest, ScanResponse messages to client_api.proto - Add HandleClientScan RPC to RaftClientService Core: - Add ScanResult{entries, revision} to StateMachine trait (default impl returns error) - Add ClientCmd::Scan variant; leader serves immediately from state machine, followers reject with failed_precondition("Not leader") - Add scan_prefix() to ClientApi trait Storage backends: - RocksDB: set_iterate_upper_bound(prefix_successor) — O(results), no full scan - File: HashMap::iter().filter(starts_with) — O(total keys), unavoidable for HashMap - Sled: sled::Tree::scan_prefix() — O(results), native B-tree support Server/Client: - Add handle_client_scan gRPC handler in grpc_raft_service - Add scan_prefix() to EmbeddedClient and GrpcClient - Re-export ScanResult from d-engine-server and d-engine crates Tests: - 4 RocksDB unit tests covering prefix isolation, upper-bound exclusion, empty prefix, and revision correctness Docs/Examples: - Update watch-feature.md with real Watch->Scan->filter reconnect pattern - Update service-discovery-standalone watcher.rs to use scan_prefix() --- d-engine-client/src/grpc_client.rs | 31 +++++ d-engine-client/src/mock_rpc.rs | 9 ++ d-engine-core/src/client/client_api.rs | 25 ++++ d-engine-core/src/event.rs | 5 + d-engine-core/src/raft_role/leader_state.rs | 4 + d-engine-core/src/raft_role/role_state.rs | 3 + d-engine-core/src/storage/state_machine.rs | 32 +++++ d-engine-core/src/test_utils/mock/mock_rpc.rs | 9 ++ d-engine-proto/proto/client/client_api.proto | 29 ++++ .../src/generated/d_engine.client.rs | 114 ++++++++++++++++ d-engine-server/src/api/embedded_client.rs | 35 +++++ d-engine-server/src/lib.rs | 2 + .../src/network/grpc/grpc_raft_service.rs | 43 ++++++ .../adaptors/file/file_state_machine.rs | 15 ++ .../adaptors/rocksdb/rocksdb_state_machine.rs | 42 ++++++ .../rocksdb/rocksdb_state_machine_test.rs | 128 ++++++++++++++++++ .../src/test_utils/mock/mock_rpc.rs | 9 ++ .../src/docs/server_guide/watch-feature.md | 26 ++-- examples/service-discovery-embedded/README.md | 2 +- examples/service-discovery-embedded/server.rs | 2 +- .../service-discovery-standalone/README.md | 2 +- .../service-discovery-standalone/watcher.rs | 50 ++++--- .../sled-cluster/src/sled_state_machine.rs | 16 ++- 23 files changed, 602 insertions(+), 31 deletions(-) diff --git a/d-engine-client/src/grpc_client.rs b/d-engine-client/src/grpc_client.rs index 60513f0d..c2a2bfcb 100644 --- a/d-engine-client/src/grpc_client.rs +++ b/d-engine-client/src/grpc_client.rs @@ -2,11 +2,13 @@ use std::sync::Arc; use arc_swap::ArcSwap; use bytes::Bytes; +use d_engine_core::ScanResult; use d_engine_proto::client::ClientReadRequest; use d_engine_proto::client::ClientResult; use d_engine_proto::client::ClientWriteRequest; use d_engine_proto::client::MembershipSnapshot; use d_engine_proto::client::ReadConsistencyPolicy; +use d_engine_proto::client::ScanRequest; use d_engine_proto::client::WatchMembershipRequest; use d_engine_proto::client::WatchRequest; use d_engine_proto::client::WatchResponse; @@ -516,4 +518,33 @@ impl ClientApi for GrpcClient { Err(e) => Err(e), } } + + async fn scan_prefix( + &self, + prefix: impl AsRef<[u8]> + Send, + ) -> ClientApiResult { + let client_inner = self.client_inner.load(); + let mut client = self.make_leader_client().await?; + + let request = ScanRequest { + client_id: client_inner.client_id, + prefix: Bytes::copy_from_slice(prefix.as_ref()), + consistency_policy: None, + }; + + let response = client + .handle_client_scan(request) + .await + .map_err(|status| ClientApiError::Business { + code: d_engine_proto::error::ErrorCode::Uncategorized, + message: format!("scan_prefix RPC failed: {}", status.message()), + required_action: None, + })? + .into_inner(); + + Ok(ScanResult { + entries: response.entries.into_iter().map(|e| (e.key, e.value)).collect(), + revision: response.revision, + }) + } } diff --git a/d-engine-client/src/mock_rpc.rs b/d-engine-client/src/mock_rpc.rs index baab07ca..78f24bd4 100644 --- a/d-engine-client/src/mock_rpc.rs +++ b/d-engine-client/src/mock_rpc.rs @@ -5,6 +5,8 @@ use d_engine_proto::client::ClientReadRequest; use d_engine_proto::client::ClientResponse; use d_engine_proto::client::ClientWriteRequest; use d_engine_proto::client::MembershipSnapshot; +use d_engine_proto::client::ScanRequest; +use d_engine_proto::client::ScanResponse; use d_engine_proto::client::WatchMembershipRequest; use d_engine_proto::client::WatchRequest; use d_engine_proto::client::WatchResponse; @@ -173,4 +175,11 @@ impl RaftClientService for MockRpcService { )), } } + + async fn handle_client_scan( + &self, + _request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + Err(tonic::Status::unimplemented("not used in mock")) + } } diff --git a/d-engine-core/src/client/client_api.rs b/d-engine-core/src/client/client_api.rs index 29aa9f2e..26d70115 100644 --- a/d-engine-core/src/client/client_api.rs +++ b/d-engine-core/src/client/client_api.rs @@ -25,6 +25,7 @@ //! } //! ``` +use crate::ScanResult; use crate::client::client_api_error::ClientApiResult; use bytes::Bytes; use d_engine_proto::client::ReadConsistencyPolicy; @@ -363,4 +364,28 @@ pub trait ClientApi: Send + Sync { &self, key: impl AsRef<[u8]> + Send, ) -> ClientApiResult>; + + /// Returns all `(key, value)` pairs whose key starts with `prefix`, plus + /// a `revision` that anchors watch event deduplication. + /// + /// Use as the second step of the reliable watch reconnection pattern: + /// `watch_prefix` first, then `scan_prefix`, then drain watch events + /// filtering by `event.revision <= scan_result.revision`. + /// + /// Linearizable by default — must see the latest committed state to avoid + /// silent gaps when used for watch reconnection. + /// + /// # Example + /// + /// ```rust,ignore + /// loop { + /// let watcher = client.watch_prefix("/services/").await?; + /// let snapshot = client.scan_prefix("/services/").await?; + /// // apply snapshot, then drain watcher filtering by snapshot.revision + /// } + /// ``` + async fn scan_prefix( + &self, + prefix: impl AsRef<[u8]> + Send, + ) -> ClientApiResult; } diff --git a/d-engine-core/src/event.rs b/d-engine-core/src/event.rs index 6e768f03..3e384164 100644 --- a/d-engine-core/src/event.rs +++ b/d-engine-core/src/event.rs @@ -25,6 +25,7 @@ use tonic::Status; use crate::ApplyResult; use crate::MaybeCloneOneshotSender; use crate::Result; +use crate::ScanResult; use crate::StreamResponseSender; #[derive(Debug, Clone, PartialEq)] @@ -46,6 +47,10 @@ pub enum ClientCmd { ClientReadRequest, MaybeCloneOneshotSender>, ), + Scan( + bytes::Bytes, + MaybeCloneOneshotSender>, + ), } #[derive(Debug)] diff --git a/d-engine-core/src/raft_role/leader_state.rs b/d-engine-core/src/raft_role/leader_state.rs index 402a8bad..acc14ecd 100644 --- a/d-engine-core/src/raft_role/leader_state.rs +++ b/d-engine-core/src/raft_role/leader_state.rs @@ -1049,6 +1049,10 @@ impl RaftRoleState for LeaderState { } } } + ClientCmd::Scan(prefix, sender) => { + let result = ctx.state_machine().scan_prefix(&prefix); + let _ = sender.send(result.map_err(|e| Status::internal(e.to_string()))); + } } } diff --git a/d-engine-core/src/raft_role/role_state.rs b/d-engine-core/src/raft_role/role_state.rs index a8e699b1..4decc059 100644 --- a/d-engine-core/src/raft_role/role_state.rs +++ b/d-engine-core/src/raft_role/role_state.rs @@ -291,6 +291,9 @@ pub trait RaftRoleState: Send + Sync + 'static { let status = tonic::Status::failed_precondition("Not leader"); let _ = sender.send(Err(status)); } + ClientCmd::Scan(_, sender) => { + let _ = sender.send(Err(tonic::Status::failed_precondition("Not leader"))); + } ClientCmd::Read(req, sender) => { // Determine effective read policy, mirroring leader_state logic: // 1. If client specified a policy AND server allows override, use client policy. diff --git a/d-engine-core/src/storage/state_machine.rs b/d-engine-core/src/storage/state_machine.rs index fb9a08f3..97ac804c 100644 --- a/d-engine-core/src/storage/state_machine.rs +++ b/d-engine-core/src/storage/state_machine.rs @@ -11,6 +11,18 @@ use d_engine_proto::server::storage::SnapshotMetadata; use mockall::automock; use crate::Error; +use crate::StorageError; + +/// All `(key, value)` pairs returned by a prefix scan, plus the revision anchor. +/// +/// `revision` equals `last_applied_index` at scan time — clients use it as the +/// filter threshold when draining a watch buffer: skip events where +/// `event.revision <= scan_result.revision` to avoid double-applying. +#[derive(Debug, Clone)] +pub struct ScanResult { + pub entries: Vec<(Bytes, Bytes)>, + pub revision: u64, +} /// Result of applying a single log entry to the state machine /// @@ -213,6 +225,26 @@ pub trait StateMachine: Send + Sync + 'static { /// Async operation as it may involve cleaning up files and data. async fn reset(&self) -> Result<(), Error>; + /// Returns all `(key, value)` pairs whose key starts with `prefix`, plus + /// the `last_applied_index` at the time of the scan. + /// + /// Clients use `revision` as the filter anchor when draining watch events + /// after reconnection: skip events where `event.revision <= revision`. + /// + /// # Default implementation + /// Returns an error. Implementations that support prefix scan must override. + fn scan_prefix( + &self, + prefix: &[u8], + ) -> Result { + let _ = prefix; + Err(Error::System(crate::SystemError::Storage( + StorageError::StateMachineError( + "scan_prefix not supported by this state machine".into(), + ), + ))) + } + /// Background lease cleanup hook. /// /// Called periodically by the background cleanup task (if enabled). diff --git a/d-engine-core/src/test_utils/mock/mock_rpc.rs b/d-engine-core/src/test_utils/mock/mock_rpc.rs index 46a1f66a..397f623f 100644 --- a/d-engine-core/src/test_utils/mock/mock_rpc.rs +++ b/d-engine-core/src/test_utils/mock/mock_rpc.rs @@ -4,6 +4,8 @@ use d_engine_proto::client::ClientReadRequest; use d_engine_proto::client::ClientResponse; use d_engine_proto::client::ClientWriteRequest; use d_engine_proto::client::MembershipSnapshot as ProtoMembershipSnapshot; +use d_engine_proto::client::ScanRequest; +use d_engine_proto::client::ScanResponse; use d_engine_proto::client::WatchMembershipRequest; use d_engine_proto::client::WatchRequest; use d_engine_proto::client::WatchResponse; @@ -214,6 +216,13 @@ impl RaftClientService for MockRpcService { ) -> std::result::Result, tonic::Status> { Err(tonic::Status::unimplemented("not used in mock")) } + + async fn handle_client_scan( + &self, + _request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + Err(tonic::Status::unimplemented("not used in mock")) + } } #[tonic::async_trait] diff --git a/d-engine-proto/proto/client/client_api.proto b/d-engine-proto/proto/client/client_api.proto index 9e6981d6..20acbab2 100644 --- a/d-engine-proto/proto/client/client_api.proto +++ b/d-engine-proto/proto/client/client_api.proto @@ -100,6 +100,34 @@ message ReadResults { repeated ClientResult results = 1; } +// A single key-value pair returned by a prefix scan. +message KvEntry { + bytes key = 1; + bytes value = 2; +} + +// Request to scan all keys under a namespace prefix. +message ScanRequest { + uint32 client_id = 1; + + // Prefix to scan. Must start with '/'. e.g. b"/services/". + bytes prefix = 2; + + // Optional consistency policy. + // Omit to use the cluster default (linearizable, recommended for watch reconnection). + optional ReadConsistencyPolicy consistency_policy = 3; +} + +// Response from a prefix scan. +message ScanResponse { + // All (key, value) pairs whose key starts with the requested prefix. + repeated KvEntry entries = 1; + + // Raft applied index at the moment of the scan. + // Use as the watch-event filter anchor: skip events where event.revision <= revision. + uint64 revision = 2; +} + // Request to start a server-side membership watch stream. message WatchMembershipRequest { uint32 client_id = 1; @@ -172,6 +200,7 @@ message WatchResponse { service RaftClientService { rpc HandleClientWrite(ClientWriteRequest) returns (ClientResponse); rpc HandleClientRead(ClientReadRequest) returns (ClientResponse); + rpc HandleClientScan(ScanRequest) returns (ScanResponse); // Watch for changes to a specific key. // diff --git a/d-engine-proto/src/generated/d_engine.client.rs b/d-engine-proto/src/generated/d_engine.client.rs index 124640a1..68bad9b3 100644 --- a/d-engine-proto/src/generated/d_engine.client.rs +++ b/d-engine-proto/src/generated/d_engine.client.rs @@ -118,6 +118,41 @@ pub struct ReadResults { #[prost(message, repeated, tag = "1")] pub results: ::prost::alloc::vec::Vec, } +/// A single key-value pair returned by a prefix scan. +#[derive(serde::Serialize, serde::Deserialize)] +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct KvEntry { + #[prost(bytes = "bytes", tag = "1")] + pub key: ::prost::bytes::Bytes, + #[prost(bytes = "bytes", tag = "2")] + pub value: ::prost::bytes::Bytes, +} +/// Request to scan all keys under a namespace prefix. +#[derive(serde::Serialize, serde::Deserialize)] +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct ScanRequest { + #[prost(uint32, tag = "1")] + pub client_id: u32, + /// Prefix to scan. Must start with '/'. e.g. b"/services/". + #[prost(bytes = "bytes", tag = "2")] + pub prefix: ::prost::bytes::Bytes, + /// Optional consistency policy. + /// Omit to use the cluster default (linearizable, recommended for watch reconnection). + #[prost(enumeration = "ReadConsistencyPolicy", optional, tag = "3")] + pub consistency_policy: ::core::option::Option, +} +/// Response from a prefix scan. +#[derive(serde::Serialize, serde::Deserialize)] +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct ScanResponse { + /// All (key, value) pairs whose key starts with the requested prefix. + #[prost(message, repeated, tag = "1")] + pub entries: ::prost::alloc::vec::Vec, + /// Raft applied index at the moment of the scan. + /// Use as the watch-event filter anchor: skip events where event.revision <= revision. + #[prost(uint64, tag = "2")] + pub revision: u64, +} /// Request to start a server-side membership watch stream. #[derive(serde::Serialize, serde::Deserialize)] #[derive(Clone, Copy, PartialEq, ::prost::Message)] @@ -411,6 +446,32 @@ pub mod raft_client_service_client { ); self.inner.unary(req, path, codec).await } + pub async fn handle_client_scan( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| { + tonic::Status::unknown( + format!("Service was not ready: {}", e.into()), + ) + })?; + let codec = tonic::codec::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static( + "/d_engine.client.RaftClientService/HandleClientScan", + ); + let mut req = request.into_request(); + req.extensions_mut() + .insert( + GrpcMethod::new( + "d_engine.client.RaftClientService", + "HandleClientScan", + ), + ); + self.inner.unary(req, path, codec).await + } /// Watch for changes to a specific key. /// /// Returns a stream of WatchResponse events whenever the watched key changes. @@ -512,6 +573,10 @@ pub mod raft_client_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn handle_client_scan( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; /// Server streaming response type for the Watch method. type WatchStream: tonic::codegen::tokio_stream::Stream< Item = std::result::Result, @@ -735,6 +800,55 @@ pub mod raft_client_service_server { }; Box::pin(fut) } + "/d_engine.client.RaftClientService/HandleClientScan" => { + #[allow(non_camel_case_types)] + struct HandleClientScanSvc(pub Arc); + impl< + T: RaftClientService, + > tonic::server::UnaryService + for HandleClientScanSvc { + type Response = super::ScanResponse; + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { + ::handle_client_scan( + &inner, + request, + ) + .await + }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = HandleClientScanSvc(inner); + let codec = tonic::codec::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/d_engine.client.RaftClientService/Watch" => { #[allow(non_camel_case_types)] struct WatchSvc(pub Arc); diff --git a/d-engine-server/src/api/embedded_client.rs b/d-engine-server/src/api/embedded_client.rs index 97cad936..7a2f84a2 100644 --- a/d-engine-server/src/api/embedded_client.rs +++ b/d-engine-server/src/api/embedded_client.rs @@ -24,6 +24,7 @@ use bytes::Bytes; use d_engine_core::MaybeCloneOneshot; use d_engine_core::RaftEvent; use d_engine_core::RaftOneshot; +use d_engine_core::ScanResult; use d_engine_core::client::{ClientApi, ClientApiError, ClientApiResult}; use d_engine_proto::client::ClientReadRequest; use d_engine_proto::client::ClientWriteRequest; @@ -500,6 +501,33 @@ impl EmbeddedClient { }) } + /// Scan all keys under `prefix` with linearizable consistency. + /// + /// Returns a `ScanResult` containing all matching `(key, value)` pairs and the + /// `revision` (applied index) at scan time. Use `revision` to filter watch events + /// during reconnection: skip events where `event.revision <= revision`. + pub async fn scan_prefix( + &self, + prefix: impl AsRef<[u8]>, + ) -> ClientApiResult { + let (resp_tx, resp_rx) = MaybeCloneOneshot::new(); + + self.cmd_tx + .send(d_engine_core::ClientCmd::Scan( + Bytes::copy_from_slice(prefix.as_ref()), + resp_tx, + )) + .await + .map_err(|_| channel_closed_error())?; + + let result = tokio::time::timeout(self.timeout, resp_rx) + .await + .map_err(|_| timeout_error(self.timeout))? + .map_err(|_| channel_closed_error())?; + + result.map_err(|status| server_error(format!("RPC error: {}", status.message()))) + } + /// Internal helper: Get cluster membership via ClusterConf event async fn get_cluster_membership( &self @@ -693,4 +721,11 @@ impl ClientApi for EmbeddedClient { ) -> ClientApiResult> { self.get_eventual(key).await } + + async fn scan_prefix( + &self, + prefix: impl AsRef<[u8]> + Send, + ) -> ClientApiResult { + self.scan_prefix(prefix).await + } } diff --git a/d-engine-server/src/lib.rs b/d-engine-server/src/lib.rs index 25f0524a..48dcba04 100644 --- a/d-engine-server/src/lib.rs +++ b/d-engine-server/src/lib.rs @@ -124,6 +124,8 @@ pub use d_engine_core::LogStore; pub use d_engine_core::MetaStore; /// Unified result type (equivalent to Result) pub use d_engine_core::Result; +/// Prefix scan result: matching entries + revision anchor for watch deduplication +pub use d_engine_core::ScanResult; /// Storage-specific error type pub use d_engine_core::StorageError; // Internal types required by storage implementations — not part of user API diff --git a/d-engine-server/src/network/grpc/grpc_raft_service.rs b/d-engine-server/src/network/grpc/grpc_raft_service.rs index 660b2617..e1e51830 100644 --- a/d-engine-server/src/network/grpc/grpc_raft_service.rs +++ b/d-engine-server/src/network/grpc/grpc_raft_service.rs @@ -17,7 +17,10 @@ use d_engine_core::WatchError; use d_engine_proto::client::ClientReadRequest; use d_engine_proto::client::ClientResponse; use d_engine_proto::client::ClientWriteRequest; +use d_engine_proto::client::KvEntry; use d_engine_proto::client::MembershipSnapshot as ProtoMembershipSnapshot; +use d_engine_proto::client::ScanRequest; +use d_engine_proto::client::ScanResponse; use d_engine_proto::client::WatchMembershipRequest; use d_engine_proto::client::WatchRequest; use d_engine_proto::client::raft_client_service_server::RaftClientService; @@ -462,6 +465,46 @@ where handle_rpc_timeout(resp_rx, timeout_duration, "handle_client_read").await } + /// Scan all keys under a prefix. + /// + /// Routes through the Raft command channel so the leader serves the scan + /// (linearizable by default). Returns all matching entries plus the applied + /// index at scan time — clients use the revision to filter watch events + /// during reconnection. + async fn handle_client_scan( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + if !self.is_rpc_ready() { + warn!("handle_client_scan: Node-{} is not ready!", self.node_id); + return Err(Status::unavailable("Service is not ready")); + } + + let req = request.into_inner(); + let (resp_tx, resp_rx) = MaybeCloneOneshot::new(); + + self.cmd_tx + .send(d_engine_core::ClientCmd::Scan(req.prefix, resp_tx)) + .await + .map_err(|_| Status::internal("Command channel closed"))?; + + let timeout_duration = + Duration::from_millis(self.node_config.raft.general_raft_timeout_duration_in_ms); + + let scan_result = handle_rpc_timeout(resp_rx, timeout_duration, "handle_client_scan") + .await? + .into_inner(); + + Ok(tonic::Response::new(ScanResponse { + entries: scan_result + .entries + .into_iter() + .map(|(k, v)| KvEntry { key: k, value: v }) + .collect(), + revision: scan_result.revision, + })) + } + /// Watch for changes to a specific key /// /// Returns a stream of events (PUT/DELETE) for the specified key. diff --git a/d-engine-server/src/storage/adaptors/file/file_state_machine.rs b/d-engine-server/src/storage/adaptors/file/file_state_machine.rs index 6695f4fd..7bf4d1f3 100644 --- a/d-engine-server/src/storage/adaptors/file/file_state_machine.rs +++ b/d-engine-server/src/storage/adaptors/file/file_state_machine.rs @@ -96,6 +96,7 @@ use bytes::Bytes; use d_engine_core::ApplyResult; use d_engine_core::Error; use d_engine_core::Lease; +use d_engine_core::ScanResult; use d_engine_core::StateMachine; use d_engine_core::StorageError; use d_engine_proto::client::WriteCommand; @@ -1553,4 +1554,18 @@ impl StateMachine for FileStateMachine { Ok(expired_keys) } + + fn scan_prefix( + &self, + prefix: &[u8], + ) -> Result { + let data = self.data.read(); + let entries: Vec<(Bytes, Bytes)> = data + .iter() + .filter(|(k, _)| k.starts_with(prefix)) + .map(|(k, (v, _))| (k.clone(), v.clone())) + .collect(); + let revision = self.last_applied_index.load(Ordering::SeqCst); + Ok(ScanResult { entries, revision }) + } } diff --git a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs index 53a480a1..b9ab9b23 100644 --- a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs +++ b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs @@ -9,6 +9,7 @@ use bytes::Bytes; use d_engine_core::ApplyResult; use d_engine_core::Error; use d_engine_core::Lease; +use d_engine_core::ScanResult; use d_engine_core::StateMachine; use d_engine_core::StorageError; use d_engine_proto::client::WriteCommand; @@ -28,6 +29,7 @@ use async_trait::async_trait; use rocksdb::Cache; use rocksdb::ColumnFamilyDescriptor; use rocksdb::DB; +use rocksdb::Direction; use rocksdb::ExportImportFilesMetaData; use rocksdb::ImportColumnFamilyOptions; use rocksdb::IteratorMode; @@ -583,6 +585,46 @@ impl RocksDBStateMachine { }; StorageError::DbError(msg) } + + /// Scans all entries whose key starts with `prefix`. + /// + /// Uses `set_iterate_upper_bound(prefix_successor)` so RocksDB stops at the + /// block level — O(results), not O(total keys). No `prefix_extractor` needed. + /// `revision` in the result equals `last_applied_index` — the watch-filter anchor. + pub fn scan_prefix( + &self, + prefix: &[u8], + ) -> Result { + let mut upper = prefix.to_vec(); + match upper.last_mut() { + Some(b) => *b = b.wrapping_add(1), + None => { + let revision = self.last_applied_index.load(Ordering::SeqCst); + return Ok(ScanResult { + entries: vec![], + revision, + }); + } + } + + let mut opts = ReadOptions::default(); + opts.set_iterate_upper_bound(upper); + + let db = self.db.load(); + let cf = db + .cf_handle(STATE_MACHINE_CF) + .ok_or_else(|| StorageError::DbError("STATE_MACHINE_CF not found".into()))?; + let iter = db.iterator_cf_opt(&cf, opts, IteratorMode::From(prefix, Direction::Forward)); + + let mut entries = Vec::new(); + for item in iter { + let (k, v) = item.map_err(|e| StorageError::DbError(e.to_string()))?; + entries.push((Bytes::copy_from_slice(&k), Bytes::copy_from_slice(&v))); + } + + let revision = self.last_applied_index.load(Ordering::SeqCst); + Ok(ScanResult { entries, revision }) + } } #[async_trait] diff --git a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs index 091f440f..a36e5698 100644 --- a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs +++ b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs @@ -156,6 +156,117 @@ async fn test_get_rejected_when_not_serving() { ); } +// ── scan_prefix tests (#378) ────────────────────────────────────────────────── + +/// scan_prefix returns exactly the entries whose key starts with the given prefix, +/// excluding all keys from other namespaces. +/// +/// This is the core correctness guarantee: prefix matching must be exact — +/// no adjacent namespace keys may leak through even if they share a substring. +#[tokio::test] +async fn test_scan_prefix_returns_only_matching_keys() { + let dir = tempfile::TempDir::new().unwrap(); + let (_storage, sm) = RocksDBUnifiedEngine::open(dir.path()).unwrap(); + + sm.apply_chunk(vec![ + encode_insert_at(b"/services/node1", b"10.0.0.1", 1), + encode_insert_at(b"/services/node2", b"10.0.0.2", 2), + encode_insert_at(b"/other/key", b"must_not_appear", 3), + ]) + .await + .unwrap(); + + let result = sm.scan_prefix(b"/services/").unwrap(); + + assert_eq!( + result.entries.len(), + 2, + "only /services/ keys should be returned" + ); + let keys: Vec<&Bytes> = result.entries.iter().map(|(k, _)| k).collect(); + assert!(keys.contains(&&Bytes::from_static(b"/services/node1"))); + assert!(keys.contains(&&Bytes::from_static(b"/services/node2"))); +} + +/// scan_prefix stops at prefix_successor (last byte + 1), not at the first +/// key that does not share the prefix string — these are different. +/// +/// Without set_iterate_upper_bound, a hand-rolled break would stop at the first +/// non-matching key after the prefix range, but RocksDB's block-level skipping +/// would not apply, making the scan O(total_keys) in the worst case. +/// This test verifies that a key in a lexicographically adjacent namespace +/// (/t/ > /s/) is not returned and that the upper-bound mechanism is effective. +#[tokio::test] +async fn test_scan_prefix_upper_bound_excludes_adjacent_namespace() { + let dir = tempfile::TempDir::new().unwrap(); + let (_storage, sm) = RocksDBUnifiedEngine::open(dir.path()).unwrap(); + + sm.apply_chunk(vec![ + encode_insert_at(b"/services/last", b"v", 1), + // '/t' > '/s' in byte order — sits right after '/services/' in the keyspace + encode_insert_at(b"/t/trap", b"must_not_appear", 2), + ]) + .await + .unwrap(); + + let result = sm.scan_prefix(b"/services/").unwrap(); + + assert_eq!( + result.entries.len(), + 1, + "/t/trap must not appear in /services/ scan" + ); + assert_eq!(result.entries[0].0, Bytes::from_static(b"/services/last")); +} + +/// scan_prefix on a prefix with no matching keys returns an empty entries list, +/// not an error. An empty namespace is a valid business state (e.g. no services +/// registered yet), and callers must not need to distinguish it from an error. +#[tokio::test] +async fn test_scan_prefix_empty_namespace_returns_empty_vec() { + let dir = tempfile::TempDir::new().unwrap(); + let (_storage, sm) = RocksDBUnifiedEngine::open(dir.path()).unwrap(); + + sm.apply_chunk(vec![encode_insert_at(b"/other/key", b"v", 1)]).await.unwrap(); + + let result = sm.scan_prefix(b"/missing/").unwrap(); + + assert!( + result.entries.is_empty(), + "missing prefix must return empty entries, not an error" + ); +} + +/// scan_prefix revision is >= the applied index at the time of the call. +/// +/// This is the linearizability anchor for the watch→scan pattern: +/// after scan returns revision=R, callers filter watch events with +/// event.revision <= R (already in the snapshot) vs > R (must be applied). +/// If revision were stale, the filter boundary would be wrong and events +/// would be silently double-applied or missed. +#[tokio::test] +async fn test_scan_prefix_revision_reflects_applied_index() { + let dir = tempfile::TempDir::new().unwrap(); + let (_storage, sm) = RocksDBUnifiedEngine::open(dir.path()).unwrap(); + + // Apply 3 entries as a single chunk; last_applied_index becomes 3 + sm.apply_chunk(vec![ + encode_insert_at(b"/s/a", b"1", 1), + encode_insert_at(b"/s/b", b"2", 2), + encode_insert_at(b"/s/c", b"3", 3), + ]) + .await + .unwrap(); + + let result = sm.scan_prefix(b"/s/").unwrap(); + + assert_eq!(result.entries.len(), 3); + assert_eq!( + result.revision, 3, + "revision must equal the applied index after the writes" + ); +} + // ── Helpers ─────────────────────────────────────────────────────────────────── fn encode_insert( @@ -173,6 +284,23 @@ fn encode_insert( encode_entry(cmd, 1, 1) } +/// Like encode_insert but with an explicit log index, required when placing +/// multiple entries in a single apply_chunk call (indices must be strictly increasing). +fn encode_insert_at( + key: &[u8], + value: &[u8], + index: u64, +) -> Entry { + let cmd = WriteCommand { + operation: Some(Operation::Insert(Insert { + key: key.to_vec().into(), + value: value.to_vec().into(), + ttl_secs: 0, + })), + }; + encode_entry(cmd, index, 1) +} + fn encode_delete( key: &[u8], index: u64, diff --git a/d-engine-server/src/test_utils/mock/mock_rpc.rs b/d-engine-server/src/test_utils/mock/mock_rpc.rs index e80b51c8..20da5707 100644 --- a/d-engine-server/src/test_utils/mock/mock_rpc.rs +++ b/d-engine-server/src/test_utils/mock/mock_rpc.rs @@ -4,6 +4,8 @@ use d_engine_proto::client::ClientReadRequest; use d_engine_proto::client::ClientResponse; use d_engine_proto::client::ClientWriteRequest; use d_engine_proto::client::MembershipSnapshot as ProtoMembershipSnapshot; +use d_engine_proto::client::ScanRequest; +use d_engine_proto::client::ScanResponse; use d_engine_proto::client::WatchMembershipRequest; use d_engine_proto::client::WatchRequest; use d_engine_proto::client::WatchResponse; @@ -215,6 +217,13 @@ impl RaftClientService for MockRpcService { ) -> std::result::Result, tonic::Status> { Err(tonic::Status::unimplemented("not used in mock")) } + + async fn handle_client_scan( + &self, + _request: tonic::Request, + ) -> std::result::Result, tonic::Status> { + Err(tonic::Status::unimplemented("not used in mock")) + } } #[tonic::async_trait] diff --git a/d-engine/src/docs/server_guide/watch-feature.md b/d-engine/src/docs/server_guide/watch-feature.md index 8608db69..d7e7578e 100644 --- a/d-engine/src/docs/server_guide/watch-feature.md +++ b/d-engine/src/docs/server_guide/watch-feature.md @@ -114,31 +114,35 @@ DOES NOT fire for: /services/paymentextra/node ❌ different namespace ``` -### Reconnection pattern +### Reconnection pattern (zero race window) -On disconnect or `CANCELED` (buffer overflow), re-sync current state before re-watching: +On disconnect or `CANCELED` (buffer overflow), use **watch-first, scan-second** to eliminate the race window: ```rust,ignore loop { - // 1. Read current namespace state (scan_prefix coming in ticket #301) - // For now: read known keys individually or accept a brief gap + // Step 1: register watch FIRST — server buffers events from this moment let mut watcher = client.watch_prefix("/services/payment/").await?; + // Step 2: linearizable scan — any write before/during scan is in the snapshot + let snapshot = client.scan_prefix("/services/payment/").await?; + let scan_revision = snapshot.revision; + let mut registry: HashMap = snapshot.entries.into_iter().collect(); + + // Step 3: drain watch buffer, skip events already captured by the scan while let Some(event) = watcher.next().await { match event { - Ok(e) if e.event_type == WatchEventType::Canceled as i32 => { - // Buffer overflow — events were dropped. Re-sync and reconnect. - resync_from_read_api().await; - break; - } - Ok(e) => apply_to_registry(e), - Err(_) => break, // network error — reconnect + Ok(e) if e.revision <= scan_revision => continue, // already in snapshot + Ok(e) if e.event_type == WatchEventType::Canceled as i32 => break, + Ok(e) => apply_to_registry(e, &mut registry), + Err(_) => break, } } tokio::time::sleep(Duration::from_millis(500)).await; } ``` +**Why this is race-free:** any write before the watch registered appears in the scan; any write after watch registered (even during the scan) is buffered. The two sets are non-overlapping when filtered by `revision ≤ scan_revision`. + ### Revision field Every `WatchEvent` carries `revision: u64` — the Raft applied index when the write was committed. diff --git a/examples/service-discovery-embedded/README.md b/examples/service-discovery-embedded/README.md index 408f5ccb..d0c94eac 100644 --- a/examples/service-discovery-embedded/README.md +++ b/examples/service-discovery-embedded/README.md @@ -110,7 +110,7 @@ tokio::spawn(async move { while let Some(e) = rx.recv().await { ... } }); If the per-watcher buffer overflows, d-engine sends a `CANCELED` sentinel. In production, respond by clearing the local registry, re-scanning with `scan_prefix` -(ticket #301), and re-watching. +(#378), and re-watching. ## Key Differences from Standalone diff --git a/examples/service-discovery-embedded/server.rs b/examples/service-discovery-embedded/server.rs index 5b1abbd7..aa7cad69 100644 --- a/examples/service-discovery-embedded/server.rs +++ b/examples/service-discovery-embedded/server.rs @@ -126,7 +126,7 @@ async fn main() -> Result<(), Box> { } e if e == WatchEventType::Canceled as i32 => { // Buffer overflow: registry may be stale. - // Production code: re-scan the prefix then re-watch (ticket #301). + // Production code: re-watch then scan_prefix (see ADR-003 reconnect pattern). println!( "[prefix-watch] CANCELED — buffer overflow on {key_str}; registry may be stale" ); diff --git a/examples/service-discovery-standalone/README.md b/examples/service-discovery-standalone/README.md index 42822329..e9a1131c 100644 --- a/examples/service-discovery-standalone/README.md +++ b/examples/service-discovery-standalone/README.md @@ -110,7 +110,7 @@ Expected watcher output: | Scope | One specific key | All keys under a namespace | | Typical use | Own config entry, distributed lock | API gateway routing table, load balancer | | Watchers registered | N (one per node) | 1 | -| Reconnection | Re-read + re-watch the key | Re-scan prefix + re-watch (ticket #301) | +| Reconnection | Re-read + re-watch the key | Re-scan prefix + re-watch (#378) | ### `revision` field diff --git a/examples/service-discovery-standalone/watcher.rs b/examples/service-discovery-standalone/watcher.rs index c5af664a..26f4753b 100644 --- a/examples/service-discovery-standalone/watcher.rs +++ b/examples/service-discovery-standalone/watcher.rs @@ -25,8 +25,10 @@ //! # Reconnection pattern //! //! When the stream ends (server restart, buffer overflow CANCELED event), this -//! example restarts: re-establish the watch stream. In production, pair this -//! with a Scan of current state first (see ticket #301). +//! example demonstrates the zero-race-window reconnect pattern: +//! 1. Register watch FIRST (server buffers events immediately) +//! 2. Scan current state +//! 3. Drain watch buffer, skipping events at revision ≤ scan revision use std::collections::HashMap; use std::time::Duration; @@ -34,7 +36,7 @@ use std::time::Duration; use anyhow::Result; use clap::Parser; use d_engine::protocol::{WatchEventType, WatchResponse}; -use d_engine::{Client, ClientApi}; +use d_engine::{Client, ClientApi, ScanResult}; use futures::StreamExt; #[derive(Parser)] @@ -134,7 +136,7 @@ async fn run_exact_watch( } // --------------------------------------------------------------------------- -// Prefix watch — live service registry (namespace observer pattern) +// Prefix watch — live service registry (zero race-window reconnect pattern) // --------------------------------------------------------------------------- // // Business scenario: an API Gateway needs a live view of all nodes belonging @@ -143,40 +145,56 @@ async fn run_exact_watch( // // One prefix watcher replaces N per-node exact-key watchers. // -// Reconnection strategy: -// On CANCELED (buffer overflow) or stream error → break inner loop → -// reconnect. In production, precede reconnect with a full namespace Scan -// to recover any events missed during the gap (see ticket #301). +// Reconnection strategy (zero race window): +// 1. Register watch FIRST — server starts buffering events immediately +// 2. scan_prefix() — linearizable snapshot at revision R +// 3. Drain watch buffer, skip events where event.revision <= R +// On CANCELED (buffer overflow) or stream error → restart from step 1 async fn run_prefix_watch( client: &Client, prefix: &str, ) -> Result<()> { - // In-memory registry: key → endpoint value. - // Maintained entirely from watch events in this demo. - // In production, populate it first with scan_prefix() (ticket #301). - let mut registry: HashMap, String> = HashMap::new(); - println!("=== Prefix Watch: Live Service Registry ==="); println!("Namespace: {prefix}"); println!("Events update the in-memory registry in real time.\n"); loop { + // Step 1: register watch FIRST — server buffers events from this moment let mut stream = client .watch_prefix(prefix) .await .map_err(|e| anyhow::anyhow!("Prefix watch failed: {e:?}"))?; - println!("[connected] watching {prefix}"); + // Step 2: linearizable scan — any write before/during scan is in the snapshot + let snapshot: ScanResult = match client.scan_prefix(prefix).await { + Ok(s) => s, + Err(e) => { + eprintln!("[scan error] {e:?} — reconnecting in 1s"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + }; + let scan_revision = snapshot.revision; + let mut registry: HashMap, String> = snapshot + .entries + .into_iter() + .map(|entry| (entry.0.to_vec(), String::from_utf8_lossy(&entry.1).to_string())) + .collect(); + println!("[connected] watching {prefix} (scan_revision={scan_revision})"); + print_registry(®istry); + + // Step 3: drain watch buffer, skip events already captured by the scan while let Some(event_result) = stream.next().await { match event_result { Ok(response) => { + if response.revision <= scan_revision { + continue; // already in scan snapshot + } let canceled = apply_event_to_registry(&response, &mut registry); print_registry(®istry); if canceled { - // CANCELED sentinel received: do not wait for server to close the stream. - // Break immediately so the outer loop reconnects deterministically. break; } } diff --git a/examples/sled-cluster/src/sled_state_machine.rs b/examples/sled-cluster/src/sled_state_machine.rs index d6f2d5fe..273d780c 100644 --- a/examples/sled-cluster/src/sled_state_machine.rs +++ b/examples/sled-cluster/src/sled_state_machine.rs @@ -5,7 +5,7 @@ use async_trait::async_trait; use bincode::config; use bytes::Bytes; use d_engine::{ - ApplyResult, Result, SnapshotError, StateMachine, StorageError, + ApplyResult, Result, ScanResult, SnapshotError, StateMachine, StorageError, client::{ WriteCommand, write_command::{CompareAndSwap, Delete, Insert, Operation}, @@ -552,6 +552,20 @@ impl StateMachine for SledStateMachine { // Return empty vector (no cleanup performed) Ok(vec![]) } + + fn scan_prefix( + &self, + prefix: &[u8], + ) -> Result { + let tree = self.current_tree(); + let mut entries = Vec::new(); + for result in tree.scan_prefix(prefix) { + let (k, v) = result.map_err(|e| StorageError::DbError(e.to_string()))?; + entries.push((Bytes::copy_from_slice(&k), Bytes::copy_from_slice(&v))); + } + let revision = self.last_applied_index.load(Ordering::SeqCst); + Ok(ScanResult { entries, revision }) + } } impl SledStateMachine { From 9ba1947b2f28639ce7136409df0664eb2675b5e5 Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Mon, 18 May 2026 10:00:38 +0800 Subject: [PATCH 2/7] fix(scan) #378: wire RocksDBStateMachine::scan_prefix into StateMachine trait impl MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit impl StateMachine for RocksDBStateMachine was missing scan_prefix delegation. The trait default impl always returns an error; any call through an SM: StateMachine bound (all production paths via SMOF) silently failed. The pub fn scan_prefix at impl RocksDBStateMachine was unreachable through the trait — only direct struct dispatch (e.g. unit tests) could hit it. --- .../src/storage/adaptors/rocksdb/rocksdb_state_machine.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs index b9ab9b23..a6c4056d 100644 --- a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs +++ b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs @@ -1095,6 +1095,13 @@ impl StateMachine for RocksDBStateMachine { Ok(expired_keys) } + + fn scan_prefix( + &self, + prefix: &[u8], + ) -> Result { + self.scan_prefix(prefix) + } } impl Drop for RocksDBStateMachine { fn drop(&mut self) { From 50e11aa856d3a640d094e5daaded470bfafe8e49 Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Mon, 18 May 2026 11:03:55 +0800 Subject: [PATCH 3/7] =?UTF-8?q?fix(scan)=20#378:=20address=20CodeRabbit=20?= =?UTF-8?q?review=20=E2=80=94=20error=20mapping,=20dead=20field,=200xFF=20?= =?UTF-8?q?prefix=20boundary?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit grpc_client.rs: use ClientApiError::from(status) for scan RPC failures instead of hardcoded Uncategorized, consistent with all other RPCs and preserving NotLeader/ network semantics for caller retry logic. proto + generated: remove consistency_policy from ScanRequest — field was dead code (handler extracted only prefix, policy was silently dropped). Documented in #381 where scan consistency guarantees will be addressed together with LinearizableRead. rocksdb_state_machine.rs: fix prefix_successor carry propagation for 0xFF byte suffix. Previous wrapping_add(1) turned 0xFF → 0x00, producing an upper bound lexicographically below the prefix — RocksDB iterator returned 0 rows silently. Now trims trailing 0xFF bytes before incrementing; returns None (no upper bound, starts_with guard used instead) when all bytes are 0xFF. watcher.rs: watch_prefix setup failure in reconnect loop now continues to next iteration instead of propagating error out of the function — consistent with how scan errors are handled and correctly demonstrates the resilient reconnect pattern. New test: test_scan_prefix_0xff_suffix_carry_propagation verifies the fix. client_read_test.rs: add RED regression test for #381 (LinearizableRead served without quorum in minority partition) — documents the bug, expected to fail until #381 is fixed. --- d-engine-client/src/grpc_client.rs | 7 +- .../leader_state_test/client_read_test.rs | 166 ++++++++++++++++++ d-engine-proto/proto/client/client_api.proto | 6 +- .../src/generated/d_engine.client.rs | 6 +- .../adaptors/rocksdb/rocksdb_state_machine.rs | 44 +++-- .../rocksdb/rocksdb_state_machine_test.rs | 30 ++++ .../service-discovery-standalone/watcher.rs | 16 +- 7 files changed, 245 insertions(+), 30 deletions(-) diff --git a/d-engine-client/src/grpc_client.rs b/d-engine-client/src/grpc_client.rs index c2a2bfcb..b0ca40db 100644 --- a/d-engine-client/src/grpc_client.rs +++ b/d-engine-client/src/grpc_client.rs @@ -529,17 +529,12 @@ impl ClientApi for GrpcClient { let request = ScanRequest { client_id: client_inner.client_id, prefix: Bytes::copy_from_slice(prefix.as_ref()), - consistency_policy: None, }; let response = client .handle_client_scan(request) .await - .map_err(|status| ClientApiError::Business { - code: d_engine_proto::error::ErrorCode::Uncategorized, - message: format!("scan_prefix RPC failed: {}", status.message()), - required_action: None, - })? + .map_err(ClientApiError::from)? .into_inner(); Ok(ScanResult { diff --git a/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs b/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs index c7fbed02..62049acf 100644 --- a/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs +++ b/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs @@ -13,9 +13,12 @@ use crate::test_utils::node_config; use crate::{Error, MockRaftLog, MockReplicationCore, RaftNodeConfig}; use bytes::Bytes; use d_engine_proto::client::ClientReadRequest; +use d_engine_proto::common::NodeRole::Follower; +use d_engine_proto::common::NodeStatus; use d_engine_proto::error::ErrorCode; use d_engine_proto::server::cluster::NodeMeta; use std::sync::Arc; +use std::time::Duration; use tokio::sync::mpsc; use tokio::sync::watch; use tokio::time::Instant; @@ -2046,3 +2049,166 @@ async fn test_lease_read_empty_payload_verification_hangs_in_multi_node() { "pending_client_writes must be empty" ); } + +// ============================================================================ +// Regression test: Ticket #381 +// LinearizableRead must NOT be served without quorum in a multi-voter cluster +// ============================================================================ + +/// REGRESSION (#381): LinearizableRead served without quorum in minority partition. +/// +/// ## Why existing tests miss this bug +/// +/// The "quorum success" tests either: +/// - use empty voter lists (`single_voter = true`) — self is the quorum, safe to serve +/// - call `handle_apply_completed` directly — bypasses the missing quorum step entirely +/// +/// None test a multi-voter cluster where `flush_cmd_buffers` is called with a pure +/// read (no concurrent write) and no follower ever responds. +/// +/// ## Bug path in `execute_and_process_raft_rpc` +/// +/// ``` +/// Phase 3: if last_applied >= read_index { +/// execute_pending_reads(read_batch, ctx); // ← IMMEDIATE SERVE, no quorum check +/// } +/// Phase 4: if append_requests.is_empty() { return Ok(()); } // ← no peers → bail +/// Phase 5: // never reached +/// ``` +/// +/// For a pure read (no writes), `prepare_batch_requests` returns an empty +/// `PrepareResult`. Phase 4 bails immediately — no AppendEntries is sent to +/// any follower. Phase 3 has already responded to the client. +/// +/// ## Preconditions that trigger the fast-path +/// +/// - `noop_log_id = Some(0)` — noop guard (`is_none()` check) does not fire +/// - `commit_index = 0`, `last_applied = 0` (defaults) → `read_index = 0` +/// - `last_applied(0) >= read_index(0)` is **true** → fast-path executes +/// +/// ## Expected correct behavior (Raft §8) +/// +/// `flush_cmd_buffers` returns immediately (non-blocking) AND the response +/// channel stays **empty**. The read must be deferred until an AppendEntries +/// heartbeat is sent and a quorum ACK confirms the node is still the leader. +/// +/// ## Current behavior (bug) +/// +/// `resp_rx.try_recv()` returns `Ok(...)` immediately after `flush_cmd_buffers`. +/// No follower was contacted. A leader isolated by a network partition will +/// return data that may have been overwritten by the new leader's commits. +/// +/// **This test FAILS with the current code.** +/// It must PASS after the fix for Ticket #381. +#[tokio::test] +#[traced_test] +async fn test_linearizable_read_served_without_quorum_in_minority_partition() { + let (_graceful_tx, graceful_rx) = watch::channel(()); + + // Pure read, no log entries → prepare_batch_requests returns an empty PrepareResult. + // Empty PrepareResult means no append_requests are generated, so Phase 4 in + // execute_and_process_raft_rpc returns early: no AppendEntries is sent to any peer. + // This simulates a partitioned cluster where followers are silent. + let mut replication_handler = MockReplicationCore::new(); + replication_handler + .expect_prepare_batch_requests() + .times(1) + .returning(|_, _, _, _, _| Ok(crate::PrepareResult::default())); + + let mut node_config = RaftNodeConfig::default(); + node_config.raft.read_consistency.allow_client_override = true; + node_config.raft.batching.max_batch_size = 1; + node_config.raft.general_raft_timeout_duration_in_ms = 2000; + + let ctx = MockBuilder::new(graceful_rx) + .with_db_path("/tmp/test_linearizable_read_without_quorum_partition") + .with_replication_handler(replication_handler) + .with_node_config(node_config) + .build_context(); + + let mut state = LeaderState::::new(1, ctx.node_config.clone()); + + // Precondition 1: noop committed at index 0 (bypasses the LeaderNotReady guard). + // calculate_read_index() = max(commit_index=0, noop_index=0) = 0. + // Default mock SM returns last_applied().index = 0. + // Fast-path condition: last_applied(0) >= read_index(0) → TRUE. + state.noop_log_id = Some(0); + + // Precondition 2: multi-voter cluster — quorum requires peer ACKs. + // A single-voter leader is its own quorum and may safely serve immediately; + // the bug only manifests when leadership confirmation requires a majority of peers. + let mut membership = MockMembership::new(); + membership.expect_voters().returning(|| { + vec![ + NodeMeta { + id: 2, + address: "".into(), + status: NodeStatus::Active as i32, + role: Follower.into(), + }, + NodeMeta { + id: 3, + address: "".into(), + status: NodeStatus::Active as i32, + role: Follower.into(), + }, + ] + }); + membership.expect_replication_peers().returning(|| { + vec![ + NodeMeta { + id: 2, + address: "".into(), + status: NodeStatus::Active as i32, + role: Follower.into(), + }, + NodeMeta { + id: 3, + address: "".into(), + status: NodeStatus::Active as i32, + role: Follower.into(), + }, + ] + }); + state.init_cluster_metadata(&Arc::new(membership)).await.unwrap(); + assert!(!state.cluster_metadata.single_voter, "precondition: multi-voter cluster"); + + let (role_tx, _role_rx) = mpsc::unbounded_channel(); + + // Client sends a pure LinearizableRead (no concurrent write). + let req = ClientReadRequest { + client_id: 1, + keys: vec![safe_kv_bytes(1)], + consistency_policy: Some(ReadConsistencyPolicy::LinearizableRead as i32), + }; + let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new(); + state.push_client_cmd(ClientCmd::Read(req, resp_tx), &ctx); + + // flush_cmd_buffers must return quickly — it must NOT block awaiting quorum. + tokio::time::timeout( + Duration::from_millis(200), + state.flush_cmd_buffers(&ctx, &role_tx), + ) + .await + .expect("flush_cmd_buffers must not block") + .expect("flush must succeed"); + + // ───────────────────────────────────────────────────────────────────────── + // CORE ASSERTION (Raft §8 requirement): + // + // The response channel must be EMPTY immediately after flush. + // In a multi-voter cluster, a LinearizableRead must be deferred until an + // AppendEntries heartbeat is sent and a quorum ACK confirms leadership. + // Only after `handle_append_result` confirms quorum may the read be served. + // + // CURRENT BUG: try_recv() returns Ok(...) — the read was served in Phase 3 + // before any follower was contacted. This test FAILS with the current code. + // ───────────────────────────────────────────────────────────────────────── + assert!( + resp_rx.try_recv().is_err(), + "BUG #381: LinearizableRead was served immediately in Phase 3 without any \ + quorum confirmation. A partitioned leader can return stale data. \ + The read must be deferred until handle_append_result confirms quorum \ + (same as the LeaseRead expired path in process_lease_read)." + ); +} diff --git a/d-engine-proto/proto/client/client_api.proto b/d-engine-proto/proto/client/client_api.proto index 20acbab2..36038986 100644 --- a/d-engine-proto/proto/client/client_api.proto +++ b/d-engine-proto/proto/client/client_api.proto @@ -107,15 +107,13 @@ message KvEntry { } // Request to scan all keys under a namespace prefix. +// Scan is served from the leader's state machine at last_applied. +// consistency_policy is not supported; tracked in #381. message ScanRequest { uint32 client_id = 1; // Prefix to scan. Must start with '/'. e.g. b"/services/". bytes prefix = 2; - - // Optional consistency policy. - // Omit to use the cluster default (linearizable, recommended for watch reconnection). - optional ReadConsistencyPolicy consistency_policy = 3; } // Response from a prefix scan. diff --git a/d-engine-proto/src/generated/d_engine.client.rs b/d-engine-proto/src/generated/d_engine.client.rs index 68bad9b3..1fee1271 100644 --- a/d-engine-proto/src/generated/d_engine.client.rs +++ b/d-engine-proto/src/generated/d_engine.client.rs @@ -128,6 +128,8 @@ pub struct KvEntry { pub value: ::prost::bytes::Bytes, } /// Request to scan all keys under a namespace prefix. +/// Scan is served from the leader's state machine at last_applied. +/// consistency_policy is not supported; tracked in #381. #[derive(serde::Serialize, serde::Deserialize)] #[derive(Clone, PartialEq, ::prost::Message)] pub struct ScanRequest { @@ -136,10 +138,6 @@ pub struct ScanRequest { /// Prefix to scan. Must start with '/'. e.g. b"/services/". #[prost(bytes = "bytes", tag = "2")] pub prefix: ::prost::bytes::Bytes, - /// Optional consistency policy. - /// Omit to use the cluster default (linearizable, recommended for watch reconnection). - #[prost(enumeration = "ReadConsistencyPolicy", optional, tag = "3")] - pub consistency_policy: ::core::option::Option, } /// Response from a prefix scan. #[derive(serde::Serialize, serde::Deserialize)] diff --git a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs index a6c4056d..c7963679 100644 --- a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs +++ b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs @@ -101,6 +101,28 @@ pub struct RocksDBStateMachine { lease_enabled: bool, } +/// Returns the lexicographic successor of `prefix` for use as an iterator upper bound. +/// +/// Trims trailing 0xFF bytes (which would wrap to 0x00 on increment), then increments +/// the last remaining byte. Returns None when all bytes are 0xFF (no upper bound needed; +/// caller must use a starts_with guard instead). +/// +/// Examples: +/// [0x61, 0xFF] → Some([0x62]) — carry propagates past the 0xFF +/// [0xFF, 0xFF] → None — all-0xFF prefix, scan to end of keyspace +/// [0x2F] → Some([0x30]) — '/' (0x2F) → '0' (0x30), normal slash-prefix +fn prefix_successor(prefix: &[u8]) -> Option> { + let mut upper = prefix.to_vec(); + while upper.last() == Some(&0xFF) { + upper.pop(); + } + if upper.is_empty() { + return None; + } + *upper.last_mut().unwrap() += 1; + Some(upper) +} + impl RocksDBStateMachine { /// Opens (or creates) a dedicated RocksDB instance for state machine storage. /// @@ -595,20 +617,17 @@ impl RocksDBStateMachine { &self, prefix: &[u8], ) -> Result { - let mut upper = prefix.to_vec(); - match upper.last_mut() { - Some(b) => *b = b.wrapping_add(1), - None => { - let revision = self.last_applied_index.load(Ordering::SeqCst); - return Ok(ScanResult { - entries: vec![], - revision, - }); - } + if prefix.is_empty() { + let revision = self.last_applied_index.load(Ordering::SeqCst); + return Ok(ScanResult { entries: vec![], revision }); } let mut opts = ReadOptions::default(); - opts.set_iterate_upper_bound(upper); + // prefix_successor handles 0xFF carry: [0x61,0xFF] → [0x62], all-0xFF → None. + // When None, no upper bound is set and a starts_with guard terminates the scan. + if let Some(upper) = prefix_successor(prefix) { + opts.set_iterate_upper_bound(upper); + } let db = self.db.load(); let cf = db @@ -619,6 +638,9 @@ impl RocksDBStateMachine { let mut entries = Vec::new(); for item in iter { let (k, v) = item.map_err(|e| StorageError::DbError(e.to_string()))?; + if !k.starts_with(prefix) { + break; + } entries.push((Bytes::copy_from_slice(&k), Bytes::copy_from_slice(&v))); } diff --git a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs index a36e5698..e8cb48ac 100644 --- a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs +++ b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs @@ -267,6 +267,36 @@ async fn test_scan_prefix_revision_reflects_applied_index() { ); } +/// scan_prefix correctly handles a prefix whose last byte is 0xFF. +/// +/// The naive `last_byte + 1` implementation wraps 0xFF → 0x00, producing an upper +/// bound that is lexicographically *less than* the prefix itself — RocksDB's iterator +/// stops immediately and returns no rows. prefix_successor must propagate the carry: +/// [0x2F, 0xFF] (b"/\xFF") → [0x30] (b"0"), ensuring the scan covers all matching keys. +#[tokio::test] +async fn test_scan_prefix_0xff_suffix_carry_propagation() { + let dir = tempfile::TempDir::new().unwrap(); + let (_storage, sm) = RocksDBUnifiedEngine::open(dir.path()).unwrap(); + + // Key: b"/\xFF/node1" — prefix b"/\xFF/" ends with 0xFF then 0x2F + // Use a simpler case: prefix = b"a\xFF", key = b"a\xFFnode" + let prefix: &[u8] = b"\x61\xFF"; + let key: &[u8] = b"\x61\xFFnode"; + let decoy: &[u8] = b"\x62decoy"; // sits above [0x62], must not appear + + sm.apply_chunk(vec![ + encode_insert_at(key, b"value", 1), + encode_insert_at(decoy, b"must_not_appear", 2), + ]) + .await + .unwrap(); + + let result = sm.scan_prefix(prefix).unwrap(); + + assert_eq!(result.entries.len(), 1, "only key under 0xFF prefix should appear"); + assert_eq!(result.entries[0].0, Bytes::copy_from_slice(key)); +} + // ── Helpers ─────────────────────────────────────────────────────────────────── fn encode_insert( diff --git a/examples/service-discovery-standalone/watcher.rs b/examples/service-discovery-standalone/watcher.rs index 26f4753b..78cc26c1 100644 --- a/examples/service-discovery-standalone/watcher.rs +++ b/examples/service-discovery-standalone/watcher.rs @@ -160,11 +160,17 @@ async fn run_prefix_watch( println!("Events update the in-memory registry in real time.\n"); loop { - // Step 1: register watch FIRST — server buffers events from this moment - let mut stream = client - .watch_prefix(prefix) - .await - .map_err(|e| anyhow::anyhow!("Prefix watch failed: {e:?}"))?; + // Step 1: register watch FIRST — server buffers events from this moment. + // On transient errors (leader change, startup race) continue the loop rather + // than propagating — the reconnect pattern must be resilient to watch setup failure. + let mut stream = match client.watch_prefix(prefix).await { + Ok(s) => s, + Err(e) => { + eprintln!("[watch error] {e:?} — reconnecting in 1s"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + }; // Step 2: linearizable scan — any write before/during scan is in the snapshot let snapshot: ScanResult = match client.scan_prefix(prefix).await { From fa210f01b6e2622bb7243afc797a240d5941d48c Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Mon, 18 May 2026 15:25:53 +0800 Subject: [PATCH 4/7] test(scan) #378: add unit + integration tests for 100% scan_prefix coverage --- d-engine-client/src/grpc_client_test.rs | 78 +++++++++ d-engine-client/src/mock_rpc.rs | 9 +- d-engine-client/src/mock_rpc_service.rs | 29 ++++ .../src/raft_role/follower_state_test.rs | 27 +++ .../leader_state_test/client_read_test.rs | 79 +++++++++ .../src/storage/state_machine_test.rs | 160 ++++++++++++++++++ .../src/api/embedded_client_test.rs | 54 ++++++ .../network/grpc/grpc_raft_service_test.rs | 52 ++++++ .../adaptors/file/file_state_machine_test.rs | 150 ++++++++++++++++ .../adaptors/rocksdb/rocksdb_state_machine.rs | 5 +- .../rocksdb/rocksdb_state_machine_test.rs | 63 ++++++- 11 files changed, 703 insertions(+), 3 deletions(-) diff --git a/d-engine-client/src/grpc_client_test.rs b/d-engine-client/src/grpc_client_test.rs index 4df4f9ff..7b942ce7 100644 --- a/d-engine-client/src/grpc_client_test.rs +++ b/d-engine-client/src/grpc_client_test.rs @@ -1648,3 +1648,81 @@ mod watch_tests { ); } } + +// ============================================================================= +// scan_prefix tests +// ============================================================================= + +mod scan_tests { + use std::sync::Arc; + + use arc_swap::ArcSwap; + use bytes::Bytes; + use d_engine_core::ClientApi; + use tokio::sync::oneshot; + use tracing_test::traced_test; + + use crate::mock_rpc_service::MockNode; + use crate::{ClientConfig, ClientInner, ConnectionPool, GrpcClient}; + + async fn make_client(port: u16) -> GrpcClient { + let endpoints = vec![format!("http://localhost:{port}")]; + let config = ClientConfig::default(); + let pool = ConnectionPool::create(endpoints.clone(), config.clone()) + .await + .expect("Should create connection pool"); + GrpcClient::new(Arc::new(ArcSwap::from_pointee(ClientInner { + pool, + client_id: 42, + config, + endpoints, + }))) + } + + #[tokio::test] + #[traced_test] + async fn test_scan_prefix_success() { + use d_engine_proto::client::{KvEntry, ScanResponse}; + + let (_tx, rx) = oneshot::channel::<()>(); + let response = ScanResponse { + entries: vec![ + KvEntry { + key: Bytes::from("/services/node1"), + value: Bytes::from("10.0.0.1"), + }, + KvEntry { + key: Bytes::from("/services/node2"), + value: Bytes::from("10.0.0.2"), + }, + ], + revision: 7, + }; + let (_channel, port) = MockNode::simulate_scan_mock_server(rx, response).await.unwrap(); + + let client = make_client(port).await; + let result = client.scan_prefix(b"/services/").await.expect("scan should succeed"); + + assert_eq!(result.revision, 7); + assert_eq!(result.entries.len(), 2); + } + + #[tokio::test] + #[traced_test] + async fn test_scan_prefix_empty_result() { + use d_engine_proto::client::ScanResponse; + + let (_tx, rx) = oneshot::channel::<()>(); + let response = ScanResponse { + entries: vec![], + revision: 3, + }; + let (_channel, port) = MockNode::simulate_scan_mock_server(rx, response).await.unwrap(); + + let client = make_client(port).await; + let result = client.scan_prefix(b"/missing/").await.expect("scan should succeed"); + + assert_eq!(result.revision, 3); + assert!(result.entries.is_empty()); + } +} diff --git a/d-engine-client/src/mock_rpc.rs b/d-engine-client/src/mock_rpc.rs index 78f24bd4..5b7538be 100644 --- a/d-engine-client/src/mock_rpc.rs +++ b/d-engine-client/src/mock_rpc.rs @@ -42,6 +42,9 @@ pub struct MockRpcService { /// Membership watch stream snapshots to emit. None means return an error. pub expected_watch_membership_events: Option, tonic::Status>>, + + /// Scan response. None means return unimplemented. + pub expected_client_scan_response: Option>, } impl MockRpcService { pub fn with_metadata_response( @@ -180,6 +183,10 @@ impl RaftClientService for MockRpcService { &self, _request: tonic::Request, ) -> std::result::Result, tonic::Status> { - Err(tonic::Status::unimplemented("not used in mock")) + match &self.expected_client_scan_response { + Some(Ok(response)) => Ok(tonic::Response::new(response.clone())), + Some(Err(status)) => Err(status.clone()), + None => Err(tonic::Status::unimplemented("Scan not configured in mock")), + } } } diff --git a/d-engine-client/src/mock_rpc_service.rs b/d-engine-client/src/mock_rpc_service.rs index a991137b..ae1b8e12 100644 --- a/d-engine-client/src/mock_rpc_service.rs +++ b/d-engine-client/src/mock_rpc_service.rs @@ -341,4 +341,33 @@ impl MockNode { let channel = Self::mock_channel_with_port(port).await; Ok((channel, port)) } + + /// Simulate a scan server that returns the given ScanResponse. + pub(crate) async fn simulate_scan_mock_server( + rx: oneshot::Receiver<()>, + response: d_engine_proto::client::ScanResponse, + ) -> std::result::Result<(Channel, u16), tonic::Status> { + let builder = Box::new(|port: u16| { + Ok(ClusterMembership { + version: 1, + nodes: vec![NodeMeta { + id: 1, + role: NodeRole::Leader as i32, + address: format!("127.0.0.1:{port}"), + status: NodeStatus::Active.into(), + }], + current_leader_id: Some(1), + }) + }); + + let mock_service = crate::mock_rpc::MockRpcService { + expected_metadata_response: Some(Arc::new(builder)), + expected_client_scan_response: Some(Ok(response)), + ..Default::default() + }; + + let (port, _addr) = Self::mock_listener(mock_service, rx, true).await?; + let channel = Self::mock_channel_with_port(port).await; + Ok((channel, port)) + } } diff --git a/d-engine-core/src/raft_role/follower_state_test.rs b/d-engine-core/src/raft_role/follower_state_test.rs index 46fa7524..f1ac340f 100644 --- a/d-engine-core/src/raft_role/follower_state_test.rs +++ b/d-engine-core/src/raft_role/follower_state_test.rs @@ -1122,6 +1122,33 @@ async fn test_handle_client_write_request_redirects_to_leader() { assert!(err.message().contains("Not leader")); } +/// Test: FollowerState rejects ClientCmd::Scan with NotLeader error +/// +/// Scan reads require linearizable prefix enumeration from the leader. +/// Followers must reject immediately with FailedPrecondition so the client +/// can redirect to the current leader. +#[tokio::test] +async fn test_scan_cmd_follower_rejects_with_not_leader() { + use bytes::Bytes; + + let (_graceful_tx, graceful_rx) = watch::channel(()); + let (context, _temp_dir) = mock_raft_context_with_temp(graceful_rx, None); + + let mut state = + FollowerState::::new(1, context.node_config.clone(), None, None); + + let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new(); + let cmd = ClientCmd::Scan(Bytes::from("/services/"), resp_tx); + + state.push_client_cmd(cmd, &context); + + let result = resp_rx.recv().await.expect("channel should not be closed"); + assert!(result.is_err(), "Scan on follower should return error"); + let err = result.unwrap_err(); + assert_eq!(err.code(), tonic::Code::FailedPrecondition); + assert!(err.message().contains("Not leader")); +} + /// Test: FollowerState rejects ClientReadRequest with LinearizableRead policy /// /// Scenario: diff --git a/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs b/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs index 62049acf..f903b7e7 100644 --- a/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs +++ b/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs @@ -2212,3 +2212,82 @@ async fn test_linearizable_read_served_without_quorum_in_minority_partition() { (same as the LeaseRead expired path in process_lease_read)." ); } + +// ============================================================================ +// ClientCmd::Scan handler tests +// ============================================================================ + +/// Test: LeaderState::push_client_cmd with ClientCmd::Scan delivers the +/// state-machine result directly to the caller. +/// +/// Unlike reads, Scan is not buffered — the leader immediately calls +/// `state_machine().scan_prefix()` and sends the result back. This test +/// verifies the happy path: the mock SM returns entries and the caller +/// receives them. +#[tokio::test] +#[traced_test] +async fn test_scan_cmd_leader_delivers_result() { + use crate::ScanResult; + use crate::test_utils::mock_state_machine; + + let mut sm = mock_state_machine(); + sm.expect_scan_prefix().returning(|_prefix| { + Ok(ScanResult { + entries: vec![], + revision: 42, + }) + }); + + let (_graceful_tx, graceful_rx) = watch::channel(()); + let context = MockBuilder::new(graceful_rx) + .with_db_path("/tmp/test_scan_cmd_leader_delivers_result") + .with_state_machine(sm) + .build_context(); + + let mut state = LeaderState::::new(1, context.node_config.clone()); + + let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new(); + state.push_client_cmd( + ClientCmd::Scan(Bytes::from("/services/"), resp_tx), + &context, + ); + + let result = resp_rx.recv().await.expect("channel not closed").expect("scan succeeded"); + assert_eq!(result.revision, 42); + assert!(result.entries.is_empty()); +} + +/// Test: LeaderState::push_client_cmd with ClientCmd::Scan propagates state- +/// machine errors as tonic Status::internal to the caller. +#[tokio::test] +#[traced_test] +async fn test_scan_cmd_leader_propagates_sm_error() { + use crate::test_utils::mock_state_machine; + use crate::{Error, StorageError}; + + let mut sm = mock_state_machine(); + sm.expect_scan_prefix().returning(|_prefix| { + Err(Error::System(crate::SystemError::Storage( + StorageError::StateMachineError("scan failed".into()), + ))) + }); + + let (_graceful_tx, graceful_rx) = watch::channel(()); + let context = MockBuilder::new(graceful_rx) + .with_db_path("/tmp/test_scan_cmd_leader_propagates_sm_error") + .with_state_machine(sm) + .build_context(); + + let mut state = LeaderState::::new(1, context.node_config.clone()); + + let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new(); + state.push_client_cmd( + ClientCmd::Scan(Bytes::from("/services/"), resp_tx), + &context, + ); + + let err = resp_rx.recv().await.expect("channel not closed").unwrap_err(); + // The tonic Status is created from e.to_string() which renders as "Storage operation + // failed" (the outer SystemError::Storage variant). Verify it's Internal. + assert_eq!(err.code(), Code::Internal); +} diff --git a/d-engine-core/src/storage/state_machine_test.rs b/d-engine-core/src/storage/state_machine_test.rs index 44d39adc..b5971d66 100644 --- a/d-engine-core/src/storage/state_machine_test.rs +++ b/d-engine-core/src/storage/state_machine_test.rs @@ -931,3 +931,163 @@ fn create_cas_entry( }), } } + +// ============================================================================ +// Default scan_prefix implementation test +// ============================================================================ +// +// The default scan_prefix on the trait returns an error because not all state +// machines support prefix scans. This test exercises that code path through a +// minimal concrete type that delegates everything else but does NOT override +// scan_prefix, ensuring the default arm is reachable and exercised. +#[cfg(test)] +mod default_scan_prefix_tests { + use std::sync::atomic::{AtomicBool, Ordering}; + + use async_trait::async_trait; + use bytes::Bytes; + use d_engine_proto::common::{Entry, LogId}; + use d_engine_proto::server::storage::SnapshotMetadata; + + use crate::Error; + use crate::storage::state_machine::{ApplyResult, StateMachine}; + + struct MinimalSm { + running: AtomicBool, + } + + #[async_trait] + impl StateMachine for MinimalSm { + async fn start(&self) -> Result<(), Error> { + self.running.store(true, Ordering::SeqCst); + Ok(()) + } + + fn stop(&self) -> Result<(), Error> { + self.running.store(false, Ordering::SeqCst); + Ok(()) + } + + fn is_running(&self) -> bool { + self.running.load(Ordering::SeqCst) + } + + fn get( + &self, + _key_buffer: &[u8], + ) -> Result, Error> { + Ok(None) + } + + fn entry_term( + &self, + _entry_id: u64, + ) -> Option { + None + } + + async fn apply_chunk( + &self, + _chunk: Vec, + ) -> Result, Error> { + Ok(vec![]) + } + + fn len(&self) -> usize { + 0 + } + + fn update_last_applied( + &self, + _last_applied: LogId, + ) { + } + + fn last_applied(&self) -> LogId { + LogId::default() + } + + fn persist_last_applied( + &self, + _last_applied: LogId, + ) -> Result<(), Error> { + Ok(()) + } + + fn update_last_snapshot_metadata( + &self, + _snapshot_metadata: &SnapshotMetadata, + ) -> Result<(), Error> { + Ok(()) + } + + fn snapshot_metadata(&self) -> Option { + None + } + + fn persist_last_snapshot_metadata( + &self, + _snapshot_metadata: &SnapshotMetadata, + ) -> Result<(), Error> { + Ok(()) + } + + async fn apply_snapshot_from_file( + &self, + _metadata: &SnapshotMetadata, + _snapshot_path: std::path::PathBuf, + ) -> Result<(), Error> { + Ok(()) + } + + async fn generate_snapshot_data( + &self, + _new_snapshot_dir: std::path::PathBuf, + _last_included: LogId, + ) -> Result { + Ok(Bytes::new()) + } + + fn save_hard_state(&self) -> Result<(), Error> { + Ok(()) + } + + fn flush(&self) -> Result<(), Error> { + Ok(()) + } + + async fn flush_async(&self) -> Result<(), Error> { + Ok(()) + } + + async fn reset(&self) -> Result<(), Error> { + Ok(()) + } + // scan_prefix intentionally omitted → exercises the trait default implementation + } + + /// Test that the default scan_prefix returns an error. + /// + /// State machines that do not support prefix scans must return an error so + /// callers can surface a clear message instead of panicking. + #[tokio::test] + async fn test_scan_prefix_default_returns_error() { + let sm = MinimalSm { + running: AtomicBool::new(true), + }; + + let result = sm.scan_prefix(b"/test/prefix/"); + + assert!( + result.is_err(), + "default scan_prefix must return Err; got Ok unexpectedly" + ); + // The error type is StateMachineError — check the Debug representation which + // includes the nested message (Display only shows the outermost variant). + let err_debug = format!("{:?}", result.unwrap_err()); + assert!( + err_debug.contains("scan_prefix not supported"), + "error should mention 'scan_prefix not supported', got: {err_debug}" + ); + } +} diff --git a/d-engine-server/src/api/embedded_client_test.rs b/d-engine-server/src/api/embedded_client_test.rs index ee14cf01..f8ee5b55 100644 --- a/d-engine-server/src/api/embedded_client_test.rs +++ b/d-engine-server/src/api/embedded_client_test.rs @@ -320,4 +320,58 @@ max_batch_size = 1 engine.stop().await.expect("Failed to stop engine"); } } + + mod scan_prefix_operations { + use super::*; + + /// scan_prefix returns all keys that start with the prefix and their + /// values, plus the revision (last-applied-index) at scan time. + #[tokio::test] + async fn test_scan_prefix_returns_matching_entries() { + let (engine, _temp_dir) = create_test_engine().await; + let client = engine.client(); + + // Write three keys under /services/ and one decoy key + client.put(b"/services/node1".to_vec(), b"10.0.0.1".to_vec()).await.unwrap(); + client.put(b"/services/node2".to_vec(), b"10.0.0.2".to_vec()).await.unwrap(); + client.put(b"/services/node3".to_vec(), b"10.0.0.3".to_vec()).await.unwrap(); + client.put(b"/other/key".to_vec(), b"decoy".to_vec()).await.unwrap(); + + let result = client.scan_prefix(b"/services/").await.expect("scan should succeed"); + + assert_eq!( + result.entries.len(), + 3, + "exactly 3 /services/ keys expected" + ); + let keys: Vec<_> = result.entries.iter().map(|(k, _)| k.clone()).collect(); + assert!(keys.contains(&bytes::Bytes::from("/services/node1"))); + assert!(keys.contains(&bytes::Bytes::from("/services/node2"))); + assert!(keys.contains(&bytes::Bytes::from("/services/node3"))); + assert!( + result.revision >= 4, + "revision must reflect at least 4 applied entries" + ); + + engine.stop().await.expect("Failed to stop engine"); + } + + /// scan_prefix on a missing prefix returns an empty result, not an error. + #[tokio::test] + async fn test_scan_prefix_missing_namespace_returns_empty() { + let (engine, _temp_dir) = create_test_engine().await; + let client = engine.client(); + + client.put(b"/other/key".to_vec(), b"v".to_vec()).await.unwrap(); + + let result = client.scan_prefix(b"/missing/").await.expect("scan should succeed"); + + assert!( + result.entries.is_empty(), + "missing prefix must return empty entries" + ); + + engine.stop().await.expect("Failed to stop engine"); + } + } } diff --git a/d-engine-server/src/network/grpc/grpc_raft_service_test.rs b/d-engine-server/src/network/grpc/grpc_raft_service_test.rs index 2e19d7e9..987488b6 100644 --- a/d-engine-server/src/network/grpc/grpc_raft_service_test.rs +++ b/d-engine-server/src/network/grpc/grpc_raft_service_test.rs @@ -461,3 +461,55 @@ async fn test_watch_membership_yields_current_snapshot_then_sentinel_on_sender_d // Stream is exhausted after the sentinel. assert!(stream.next().await.is_none()); } + +// ============================================================================= +// handle_client_scan unit tests +// ============================================================================= + +/// Node not ready → handle_client_scan must return UNAVAILABLE immediately. +/// +/// The readiness guard fires before the request reaches the Raft command +/// channel, so the channel stays untouched and no goroutine is wasted. +#[tokio::test] +#[traced_test] +async fn test_handle_client_scan_not_ready_returns_unavailable() { + use d_engine_proto::client::ScanRequest; + + let (_graceful_tx, graceful_rx) = watch::channel(()); + let node = mock_node("/tmp/test_handle_client_scan_not_ready", graceful_rx, None); + node.set_rpc_ready(false); + + let result = node + .handle_client_scan(Request::new(ScanRequest { + client_id: 1, + prefix: bytes::Bytes::from("/services/"), + })) + .await; + + assert!(result.is_err()); + assert_eq!(result.unwrap_err().code(), Code::Unavailable); +} + +/// Node ready but no Raft loop → handle_client_scan must timeout. +/// +/// With no Raft actor draining the command channel the response never arrives, +/// so the RPC must return an error within the configured timeout rather than +/// blocking indefinitely. +#[tokio::test] +#[traced_test] +async fn test_handle_client_scan_timeout() { + use d_engine_proto::client::ScanRequest; + + let (_graceful_tx, graceful_rx) = watch::channel(()); + let node = mock_node("/tmp/test_handle_client_scan_timeout", graceful_rx, None); + node.set_rpc_ready(true); + + let result = node + .handle_client_scan(Request::new(ScanRequest { + client_id: 1, + prefix: bytes::Bytes::from("/services/"), + })) + .await; + + assert!(result.is_err(), "scan must error when Raft actor is absent"); +} diff --git a/d-engine-server/src/storage/adaptors/file/file_state_machine_test.rs b/d-engine-server/src/storage/adaptors/file/file_state_machine_test.rs index 090c928c..ce68a8cc 100644 --- a/d-engine-server/src/storage/adaptors/file/file_state_machine_test.rs +++ b/d-engine-server/src/storage/adaptors/file/file_state_machine_test.rs @@ -151,3 +151,153 @@ async fn test_wal_replay_after_crash() { Some(Bytes::from("value3")) ); } + +// ── scan_prefix tests ───────────────────────────────────────────────────────── + +/// scan_prefix returns only keys that start with the prefix, not all keys. +#[tokio::test] +async fn test_file_sm_scan_prefix_returns_matching_keys() { + use d_engine_core::StateMachine; + + let temp_dir = tempfile::tempdir().unwrap(); + let sm = FileStateMachine::new(temp_dir.path().to_path_buf()).await.unwrap(); + + let entries = vec![ + Entry { + index: 1, + term: 1, + payload: Some(EntryPayload { + payload: Some(Payload::Command( + WriteCommand { + operation: Some(Operation::Insert(Insert { + key: Bytes::from("/services/node1"), + value: Bytes::from("10.0.0.1"), + ttl_secs: 0, + })), + } + .encode_to_vec() + .into(), + )), + }), + }, + Entry { + index: 2, + term: 1, + payload: Some(EntryPayload { + payload: Some(Payload::Command( + WriteCommand { + operation: Some(Operation::Insert(Insert { + key: Bytes::from("/services/node2"), + value: Bytes::from("10.0.0.2"), + ttl_secs: 0, + })), + } + .encode_to_vec() + .into(), + )), + }), + }, + Entry { + index: 3, + term: 1, + payload: Some(EntryPayload { + payload: Some(Payload::Command( + WriteCommand { + operation: Some(Operation::Insert(Insert { + key: Bytes::from("/other/key"), + value: Bytes::from("must_not_appear"), + ttl_secs: 0, + })), + } + .encode_to_vec() + .into(), + )), + }), + }, + ]; + + sm.apply_chunk(entries).await.unwrap(); + + let result = sm.scan_prefix(b"/services/").unwrap(); + + assert_eq!( + result.entries.len(), + 2, + "only /services/ keys should appear" + ); + let keys: Vec<_> = result.entries.iter().map(|(k, _)| k.clone()).collect(); + assert!(keys.contains(&Bytes::from("/services/node1"))); + assert!(keys.contains(&Bytes::from("/services/node2"))); +} + +/// scan_prefix on a missing prefix returns empty entries, not an error. +#[tokio::test] +async fn test_file_sm_scan_prefix_empty_namespace() { + use d_engine_core::StateMachine; + + let temp_dir = tempfile::tempdir().unwrap(); + let sm = FileStateMachine::new(temp_dir.path().to_path_buf()).await.unwrap(); + + sm.apply_chunk(vec![Entry { + index: 1, + term: 1, + payload: Some(EntryPayload { + payload: Some(Payload::Command( + WriteCommand { + operation: Some(Operation::Insert(Insert { + key: Bytes::from("/other/key"), + value: Bytes::from("v"), + ttl_secs: 0, + })), + } + .encode_to_vec() + .into(), + )), + }), + }]) + .await + .unwrap(); + + let result = sm.scan_prefix(b"/missing/").unwrap(); + + assert!( + result.entries.is_empty(), + "missing prefix should return empty entries" + ); +} + +/// scan_prefix revision equals last_applied_index at scan time. +#[tokio::test] +async fn test_file_sm_scan_prefix_revision_reflects_applied_index() { + use d_engine_core::StateMachine; + + let temp_dir = tempfile::tempdir().unwrap(); + let sm = FileStateMachine::new(temp_dir.path().to_path_buf()).await.unwrap(); + + let entries: Vec = (1u64..=3) + .map(|i| Entry { + index: i, + term: 1, + payload: Some(EntryPayload { + payload: Some(Payload::Command( + WriteCommand { + operation: Some(Operation::Insert(Insert { + key: Bytes::from(format!("/s/{i}")), + value: Bytes::from(format!("{i}")), + ttl_secs: 0, + })), + } + .encode_to_vec() + .into(), + )), + }), + }) + .collect(); + + sm.apply_chunk(entries).await.unwrap(); + + let result = sm.scan_prefix(b"/s/").unwrap(); + + assert_eq!(result.entries.len(), 3); + assert_eq!(result.revision, 3, "revision must equal last_applied_index"); +} diff --git a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs index c7963679..a9fb8e1b 100644 --- a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs +++ b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine.rs @@ -619,7 +619,10 @@ impl RocksDBStateMachine { ) -> Result { if prefix.is_empty() { let revision = self.last_applied_index.load(Ordering::SeqCst); - return Ok(ScanResult { entries: vec![], revision }); + return Ok(ScanResult { + entries: vec![], + revision, + }); } let mut opts = ReadOptions::default(); diff --git a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs index e8cb48ac..384ce87f 100644 --- a/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs +++ b/d-engine-server/src/storage/adaptors/rocksdb/rocksdb_state_machine_test.rs @@ -293,7 +293,68 @@ async fn test_scan_prefix_0xff_suffix_carry_propagation() { let result = sm.scan_prefix(prefix).unwrap(); - assert_eq!(result.entries.len(), 1, "only key under 0xFF prefix should appear"); + assert_eq!( + result.entries.len(), + 1, + "only key under 0xFF prefix should appear" + ); + assert_eq!(result.entries[0].0, Bytes::copy_from_slice(key)); +} + +/// scan_prefix with an empty prefix returns an empty result immediately. +/// +/// An empty prefix would otherwise match every key in the database, which +/// is an unsafe operation for an arbitrarily large keyspace. The implementation +/// short-circuits and returns `ScanResult { entries: vec![], revision }` so +/// callers receive a well-defined result without iterating the whole DB. +#[tokio::test] +async fn test_scan_prefix_empty_prefix_returns_empty() { + let dir = tempfile::TempDir::new().unwrap(); + let (_storage, sm) = RocksDBUnifiedEngine::open(dir.path()).unwrap(); + + sm.apply_chunk(vec![encode_insert_at(b"/services/node1", b"v", 1)]) + .await + .unwrap(); + + let result = sm.scan_prefix(b"").unwrap(); + + assert!( + result.entries.is_empty(), + "empty prefix should return empty entries (not all keys)" + ); +} + +/// scan_prefix with an all-0xFF prefix has no successor, so no upper bound is +/// set and the iterator uses the starts_with guard to exclude non-matching keys. +/// +/// prefix_successor([0xFF, 0xFF]) returns None because all bytes are 0xFF — +/// there is no lexicographically larger byte string that could serve as an +/// upper bound. The scan must still return all keys that start with the prefix +/// and exclude everything else. +#[tokio::test] +async fn test_scan_prefix_all_0xff_no_upper_bound() { + let dir = tempfile::TempDir::new().unwrap(); + let (_storage, sm) = RocksDBUnifiedEngine::open(dir.path()).unwrap(); + + // Key that starts with [0xFF, 0xFF] — sits at the very top of the keyspace. + let prefix: &[u8] = b"\xFF\xFF"; + let key: &[u8] = b"\xFF\xFF\x01"; + let decoy: &[u8] = b"\xFE\xAA"; // lexicographically below the prefix + + sm.apply_chunk(vec![ + encode_insert_at(key, b"top_value", 1), + encode_insert_at(decoy, b"must_not_appear", 2), + ]) + .await + .unwrap(); + + let result = sm.scan_prefix(prefix).unwrap(); + + assert_eq!( + result.entries.len(), + 1, + "only key under all-0xFF prefix should appear" + ); assert_eq!(result.entries[0].0, Bytes::copy_from_slice(key)); } From 053289c38db669b3fb628d684c74eb84037df3c8 Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Mon, 18 May 2026 15:39:30 +0800 Subject: [PATCH 5/7] chore(test) #381: remove WIP regression test from #378 branch (keep local only) --- .../leader_state_test/client_read_test.rs | 163 ------------------ 1 file changed, 163 deletions(-) diff --git a/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs b/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs index f903b7e7..9710f0fb 100644 --- a/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs +++ b/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs @@ -2050,169 +2050,6 @@ async fn test_lease_read_empty_payload_verification_hangs_in_multi_node() { ); } -// ============================================================================ -// Regression test: Ticket #381 -// LinearizableRead must NOT be served without quorum in a multi-voter cluster -// ============================================================================ - -/// REGRESSION (#381): LinearizableRead served without quorum in minority partition. -/// -/// ## Why existing tests miss this bug -/// -/// The "quorum success" tests either: -/// - use empty voter lists (`single_voter = true`) — self is the quorum, safe to serve -/// - call `handle_apply_completed` directly — bypasses the missing quorum step entirely -/// -/// None test a multi-voter cluster where `flush_cmd_buffers` is called with a pure -/// read (no concurrent write) and no follower ever responds. -/// -/// ## Bug path in `execute_and_process_raft_rpc` -/// -/// ``` -/// Phase 3: if last_applied >= read_index { -/// execute_pending_reads(read_batch, ctx); // ← IMMEDIATE SERVE, no quorum check -/// } -/// Phase 4: if append_requests.is_empty() { return Ok(()); } // ← no peers → bail -/// Phase 5: // never reached -/// ``` -/// -/// For a pure read (no writes), `prepare_batch_requests` returns an empty -/// `PrepareResult`. Phase 4 bails immediately — no AppendEntries is sent to -/// any follower. Phase 3 has already responded to the client. -/// -/// ## Preconditions that trigger the fast-path -/// -/// - `noop_log_id = Some(0)` — noop guard (`is_none()` check) does not fire -/// - `commit_index = 0`, `last_applied = 0` (defaults) → `read_index = 0` -/// - `last_applied(0) >= read_index(0)` is **true** → fast-path executes -/// -/// ## Expected correct behavior (Raft §8) -/// -/// `flush_cmd_buffers` returns immediately (non-blocking) AND the response -/// channel stays **empty**. The read must be deferred until an AppendEntries -/// heartbeat is sent and a quorum ACK confirms the node is still the leader. -/// -/// ## Current behavior (bug) -/// -/// `resp_rx.try_recv()` returns `Ok(...)` immediately after `flush_cmd_buffers`. -/// No follower was contacted. A leader isolated by a network partition will -/// return data that may have been overwritten by the new leader's commits. -/// -/// **This test FAILS with the current code.** -/// It must PASS after the fix for Ticket #381. -#[tokio::test] -#[traced_test] -async fn test_linearizable_read_served_without_quorum_in_minority_partition() { - let (_graceful_tx, graceful_rx) = watch::channel(()); - - // Pure read, no log entries → prepare_batch_requests returns an empty PrepareResult. - // Empty PrepareResult means no append_requests are generated, so Phase 4 in - // execute_and_process_raft_rpc returns early: no AppendEntries is sent to any peer. - // This simulates a partitioned cluster where followers are silent. - let mut replication_handler = MockReplicationCore::new(); - replication_handler - .expect_prepare_batch_requests() - .times(1) - .returning(|_, _, _, _, _| Ok(crate::PrepareResult::default())); - - let mut node_config = RaftNodeConfig::default(); - node_config.raft.read_consistency.allow_client_override = true; - node_config.raft.batching.max_batch_size = 1; - node_config.raft.general_raft_timeout_duration_in_ms = 2000; - - let ctx = MockBuilder::new(graceful_rx) - .with_db_path("/tmp/test_linearizable_read_without_quorum_partition") - .with_replication_handler(replication_handler) - .with_node_config(node_config) - .build_context(); - - let mut state = LeaderState::::new(1, ctx.node_config.clone()); - - // Precondition 1: noop committed at index 0 (bypasses the LeaderNotReady guard). - // calculate_read_index() = max(commit_index=0, noop_index=0) = 0. - // Default mock SM returns last_applied().index = 0. - // Fast-path condition: last_applied(0) >= read_index(0) → TRUE. - state.noop_log_id = Some(0); - - // Precondition 2: multi-voter cluster — quorum requires peer ACKs. - // A single-voter leader is its own quorum and may safely serve immediately; - // the bug only manifests when leadership confirmation requires a majority of peers. - let mut membership = MockMembership::new(); - membership.expect_voters().returning(|| { - vec![ - NodeMeta { - id: 2, - address: "".into(), - status: NodeStatus::Active as i32, - role: Follower.into(), - }, - NodeMeta { - id: 3, - address: "".into(), - status: NodeStatus::Active as i32, - role: Follower.into(), - }, - ] - }); - membership.expect_replication_peers().returning(|| { - vec![ - NodeMeta { - id: 2, - address: "".into(), - status: NodeStatus::Active as i32, - role: Follower.into(), - }, - NodeMeta { - id: 3, - address: "".into(), - status: NodeStatus::Active as i32, - role: Follower.into(), - }, - ] - }); - state.init_cluster_metadata(&Arc::new(membership)).await.unwrap(); - assert!(!state.cluster_metadata.single_voter, "precondition: multi-voter cluster"); - - let (role_tx, _role_rx) = mpsc::unbounded_channel(); - - // Client sends a pure LinearizableRead (no concurrent write). - let req = ClientReadRequest { - client_id: 1, - keys: vec![safe_kv_bytes(1)], - consistency_policy: Some(ReadConsistencyPolicy::LinearizableRead as i32), - }; - let (resp_tx, mut resp_rx) = MaybeCloneOneshot::new(); - state.push_client_cmd(ClientCmd::Read(req, resp_tx), &ctx); - - // flush_cmd_buffers must return quickly — it must NOT block awaiting quorum. - tokio::time::timeout( - Duration::from_millis(200), - state.flush_cmd_buffers(&ctx, &role_tx), - ) - .await - .expect("flush_cmd_buffers must not block") - .expect("flush must succeed"); - - // ───────────────────────────────────────────────────────────────────────── - // CORE ASSERTION (Raft §8 requirement): - // - // The response channel must be EMPTY immediately after flush. - // In a multi-voter cluster, a LinearizableRead must be deferred until an - // AppendEntries heartbeat is sent and a quorum ACK confirms leadership. - // Only after `handle_append_result` confirms quorum may the read be served. - // - // CURRENT BUG: try_recv() returns Ok(...) — the read was served in Phase 3 - // before any follower was contacted. This test FAILS with the current code. - // ───────────────────────────────────────────────────────────────────────── - assert!( - resp_rx.try_recv().is_err(), - "BUG #381: LinearizableRead was served immediately in Phase 3 without any \ - quorum confirmation. A partitioned leader can return stale data. \ - The read must be deferred until handle_append_result confirms quorum \ - (same as the LeaseRead expired path in process_lease_read)." - ); -} - // ============================================================================ // ClientCmd::Scan handler tests // ============================================================================ From b6bf2fb7ce17128b2c88d514a62a9ac86ae523e0 Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Mon, 18 May 2026 15:57:41 +0800 Subject: [PATCH 6/7] fix(test) #381: remove unused imports left by RED test removal --- .../src/raft_role/leader_state_test/client_read_test.rs | 3 --- 1 file changed, 3 deletions(-) diff --git a/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs b/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs index 9710f0fb..357936d4 100644 --- a/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs +++ b/d-engine-core/src/raft_role/leader_state_test/client_read_test.rs @@ -13,12 +13,9 @@ use crate::test_utils::node_config; use crate::{Error, MockRaftLog, MockReplicationCore, RaftNodeConfig}; use bytes::Bytes; use d_engine_proto::client::ClientReadRequest; -use d_engine_proto::common::NodeRole::Follower; -use d_engine_proto::common::NodeStatus; use d_engine_proto::error::ErrorCode; use d_engine_proto::server::cluster::NodeMeta; use std::sync::Arc; -use std::time::Duration; use tokio::sync::mpsc; use tokio::sync::watch; use tokio::time::Instant; From 2a8e59a25d185b0cc5c9fb4e06a578b43c68d27c Mon Sep 17 00:00:00 2001 From: Joshua Chi Date: Mon, 18 May 2026 16:17:43 +0800 Subject: [PATCH 7/7] test(scan) #378: assert key/value content in scan_prefix tests --- d-engine-client/src/grpc_client_test.rs | 10 ++++++++++ d-engine-server/src/api/embedded_client_test.rs | 17 +++++++++++++---- 2 files changed, 23 insertions(+), 4 deletions(-) diff --git a/d-engine-client/src/grpc_client_test.rs b/d-engine-client/src/grpc_client_test.rs index 7b942ce7..4f27c690 100644 --- a/d-engine-client/src/grpc_client_test.rs +++ b/d-engine-client/src/grpc_client_test.rs @@ -1705,6 +1705,16 @@ mod scan_tests { assert_eq!(result.revision, 7); assert_eq!(result.entries.len(), 2); + + let map: std::collections::HashMap<_, _> = result.entries.into_iter().collect(); + assert_eq!( + map.get(&Bytes::from("/services/node1")), + Some(&Bytes::from("10.0.0.1")) + ); + assert_eq!( + map.get(&Bytes::from("/services/node2")), + Some(&Bytes::from("10.0.0.2")) + ); } #[tokio::test] diff --git a/d-engine-server/src/api/embedded_client_test.rs b/d-engine-server/src/api/embedded_client_test.rs index f8ee5b55..3a603bff 100644 --- a/d-engine-server/src/api/embedded_client_test.rs +++ b/d-engine-server/src/api/embedded_client_test.rs @@ -344,10 +344,19 @@ max_batch_size = 1 3, "exactly 3 /services/ keys expected" ); - let keys: Vec<_> = result.entries.iter().map(|(k, _)| k.clone()).collect(); - assert!(keys.contains(&bytes::Bytes::from("/services/node1"))); - assert!(keys.contains(&bytes::Bytes::from("/services/node2"))); - assert!(keys.contains(&bytes::Bytes::from("/services/node3"))); + let map: std::collections::HashMap<_, _> = result.entries.iter().cloned().collect(); + assert_eq!( + map.get(&bytes::Bytes::from("/services/node1")), + Some(&bytes::Bytes::from("10.0.0.1")) + ); + assert_eq!( + map.get(&bytes::Bytes::from("/services/node2")), + Some(&bytes::Bytes::from("10.0.0.2")) + ); + assert_eq!( + map.get(&bytes::Bytes::from("/services/node3")), + Some(&bytes::Bytes::from("10.0.0.3")) + ); assert!( result.revision >= 4, "revision must reflect at least 4 applied entries"