feat(scan) #378: add scan_prefix API for zero-race-window watch reconnection

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,28 @@ impl ClientApi for GrpcClient {
             Err(e) => Err(e),
         }
     }
+
+    async fn scan_prefix(
+        &self,
+        prefix: impl AsRef<[u8]> + Send,
+    ) -> ClientApiResult<ScanResult> {
+        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()),
+        };
+
+        let response = client
+            .handle_client_scan(request)
+            .await
+            .map_err(ClientApiError::from)?
+            .into_inner();
+
+        Ok(ScanResult {
+            entries: response.entries.into_iter().map(|e| (e.key, e.value)).collect(),
+            revision: response.revision,
+        })
+    }
 }

d-engine-client/src/grpc_client_test.rs

@@ -1648,3 +1648,91 @@ 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);
+
+        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]
+    #[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());
+    }
+}

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;
@@ -40,6 +42,9 @@ pub struct MockRpcService {
 
     /// Membership watch stream snapshots to emit. None means return an error.
     pub expected_watch_membership_events: Option<Result<Vec<MembershipSnapshot>, tonic::Status>>,
+
+    /// Scan response. None means return unimplemented.
+    pub expected_client_scan_response: Option<Result<ScanResponse, tonic::Status>>,
 }
 impl MockRpcService {
     pub fn with_metadata_response(
@@ -173,4 +178,15 @@ impl RaftClientService for MockRpcService {
             )),
         }
     }
+
+    async fn handle_client_scan(
+        &self,
+        _request: tonic::Request<ScanRequest>,
+    ) -> std::result::Result<tonic::Response<ScanResponse>, tonic::Status> {
+        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")),
+        }
+    }
 }

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))
+    }
 }

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<Option<Bytes>>;
+
+    /// 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<ScanResult>;
 }

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<std::result::Result<ClientResponse, Status>>,
     ),
+    Scan(
+        bytes::Bytes,
+        MaybeCloneOneshotSender<std::result::Result<ScanResult, Status>>,
+    ),
 }
 
 #[derive(Debug)]

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::<MockTypeConfig>::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:

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

@@ -1049,6 +1049,10 @@ impl<T: TypeConfig> RaftRoleState for LeaderState<T> {
                     }
                 }
             }
+            ClientCmd::Scan(prefix, sender) => {
+                let result = ctx.state_machine().scan_prefix(&prefix);
+                let _ = sender.send(result.map_err(|e| Status::internal(e.to_string())));
+            }
         }
     }
 

d-engine-core/src/raft_role/leader_state_test/client_read_test.rs

@@ -2046,3 +2046,82 @@ async fn test_lease_read_empty_payload_verification_hangs_in_multi_node() {
         "pending_client_writes must be empty"
     );
 }
+
+// ============================================================================
+// 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::<MockTypeConfig>::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::<MockTypeConfig>::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);
+}

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.