feat #174: Implement lock-free Watch mechanism for real-time key monitoring

config/base/raft.toml

@@ -101,3 +101,31 @@ cluster_response = true
 
 # Client responses - disabling improves performance in VPC/LAN environments
 client_response = false
+
+# Watch mechanism configuration for monitoring key changes
+# Provides lock-free event notification with minimal write path overhead
+[raft.watch]
+# Enable or disable Watch feature
+# When false, Watch RPC calls return UNAVAILABLE (zero overhead)
+# Default: true
+enabled = true
+
+# Global event queue buffer size (shared across all watchers)
+# Tuning guidelines:
+#   - Low traffic (< 1K writes/sec): 500-1000
+#   - Medium traffic (1K-10K writes/sec): 1000-2000
+#   - High traffic (> 10K writes/sec): 2000-5000
+# Memory impact: ~24 bytes per slot, default 1000 ≈ 24KB
+event_queue_size = 1000
+
+# Per-watcher channel buffer size
+# Tuning guidelines:
+#   - Fast consumers (< 1ms processing): 5-10
+#   - Normal consumers (1-10ms processing): 10-20
+#   - Slow consumers (> 10ms processing): 20-50
+# Memory impact: ~240 bytes per slot per watcher
+watcher_buffer_size = 10
+
+# Enable detailed metrics and logging for watch operations
+# Useful for debugging but adds ~0.001% overhead
+enable_metrics = false

d-engine-client/src/mock_rpc.rs

@@ -1,6 +1,8 @@
 use d_engine_proto::client::ClientReadRequest;
 use d_engine_proto::client::ClientResponse;
 use d_engine_proto::client::ClientWriteRequest;
+use d_engine_proto::client::WatchRequest;
+use d_engine_proto::client::WatchResponse;
 use d_engine_proto::client::raft_client_service_server::RaftClientService;
 use d_engine_proto::server::cluster::ClusterConfChangeRequest;
 use d_engine_proto::server::cluster::ClusterConfUpdateResponse;
@@ -11,6 +13,8 @@ use d_engine_proto::server::cluster::LeaderDiscoveryRequest;
 use d_engine_proto::server::cluster::LeaderDiscoveryResponse;
 use d_engine_proto::server::cluster::MetadataRequest;
 use d_engine_proto::server::cluster::cluster_management_service_server::ClusterManagementService;
+use futures::Stream;
+use std::pin::Pin;
 use std::sync::Arc;
 
 #[derive(Clone, Default)]
@@ -99,6 +103,8 @@ impl ClusterManagementService for MockRpcService {
 
 #[tonic::async_trait]
 impl RaftClientService for MockRpcService {
+    type WatchStream = Pin<Box<dyn Stream<Item = Result<WatchResponse, tonic::Status>> + Send>>;
+
     async fn handle_client_write(
         &self,
         _request: tonic::Request<ClientWriteRequest>,
@@ -124,4 +130,14 @@ impl RaftClientService for MockRpcService {
             )),
         }
     }
+
+    async fn watch(
+        &self,
+        _request: tonic::Request<WatchRequest>,
+    ) -> std::result::Result<tonic::Response<Self::WatchStream>, tonic::Status> {
+        // Mock implementation - return empty stream
+        Err(tonic::Status::unimplemented(
+            "Watch not implemented in mock",
+        ))
+    }
 }

d-engine-client/src/proto/client_ext.rs

@@ -90,89 +90,3 @@ impl ClientResponseExt for ClientResponse {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use bytes::Bytes;
-    use d_engine_proto::client::{ReadResults, client_response::SuccessResult};
-
-    #[test]
-    fn test_into_read_results_success() {
-        let response = ClientResponse {
-            error: ErrorCode::Success as i32,
-            metadata: None,
-            success_result: Some(SuccessResult::ReadData(ReadResults {
-                results: vec![ClientResult {
-                    key: Bytes::from(vec![1, 2, 3]),
-                    value: Bytes::from(vec![4, 5, 6]),
-                }],
-            })),
-        };
-
-        let result = response.into_read_results();
-        assert!(result.is_ok());
-        let data = result.unwrap();
-        assert_eq!(data.len(), 1);
-        assert!(data[0].is_some());
-    }
-
-    #[test]
-    fn test_into_read_results_wrong_variant_write_ack() {
-        let response = ClientResponse {
-            error: ErrorCode::Success as i32,
-            metadata: None,
-            success_result: Some(SuccessResult::WriteAck(true)),
-        };
-
-        let result = response.into_read_results();
-        assert!(result.is_err());
-
-        if let Err(ClientApiError::Protocol { code, message, .. }) = result {
-            assert_eq!(code, ErrorCode::InvalidResponse);
-            assert!(message.contains("expected ReadData"));
-            assert!(message.contains("found WriteAck"));
-        } else {
-            panic!("Expected Protocol error");
-        }
-    }
-
-    #[test]
-    fn test_into_read_results_none_variant() {
-        let response = ClientResponse {
-            error: ErrorCode::Success as i32,
-            metadata: None,
-            success_result: None,
-        };
-
-        let result = response.into_read_results();
-        assert!(result.is_err());
-
-        if let Err(ClientApiError::Protocol { code, message, .. }) = result {
-            assert_eq!(code, ErrorCode::InvalidResponse);
-            assert!(message.contains("expected ReadData"));
-            assert!(message.contains("found None"));
-        } else {
-            panic!("Expected Protocol error");
-        }
-    }
-
-    #[test]
-    fn test_into_read_results_with_error_code() {
-        let response = ClientResponse {
-            error: ErrorCode::NotLeader as i32,
-            metadata: None,
-            success_result: Some(SuccessResult::ReadData(ReadResults { results: vec![] })),
-        };
-
-        let result = response.into_read_results();
-        assert!(result.is_err());
-
-        // Should fail at validate_error() before checking success_result
-        if let Err(ClientApiError::Business { code, .. }) = result {
-            assert_eq!(code, ErrorCode::NotLeader);
-        } else {
-            panic!("Expected Business error with NotLeader code");
-        }
-    }
-}

d-engine-client/src/proto/client_ext_test.rs

@@ -0,0 +1,86 @@
+use super::*;
+use crate::ClientApiError;
+use bytes::Bytes;
+use d_engine_proto::{
+    client::{ClientResponse, ClientResult, ReadResults, client_response::SuccessResult},
+    error::ErrorCode,
+};
+
+#[test]
+fn test_into_read_results_success() {
+    let response = ClientResponse {
+        error: ErrorCode::Success as i32,
+        metadata: None,
+        success_result: Some(SuccessResult::ReadData(ReadResults {
+            results: vec![ClientResult {
+                key: Bytes::from(vec![1, 2, 3]),
+                value: Bytes::from(vec![4, 5, 6]),
+            }],
+        })),
+    };
+
+    let result = response.into_read_results();
+    assert!(result.is_ok());
+    let data = result.unwrap();
+    assert_eq!(data.len(), 1);
+    assert!(data[0].is_some());
+}
+
+#[test]
+fn test_into_read_results_wrong_variant_write_ack() {
+    let response = ClientResponse {
+        error: ErrorCode::Success as i32,
+        metadata: None,
+        success_result: Some(SuccessResult::WriteAck(true)),
+    };
+
+    let result = response.into_read_results();
+    assert!(result.is_err());
+
+    if let Err(ClientApiError::Protocol { code, message, .. }) = result {
+        assert_eq!(code, ErrorCode::InvalidResponse);
+        assert!(message.contains("expected ReadData"));
+        assert!(message.contains("found WriteAck"));
+    } else {
+        panic!("Expected Protocol error");
+    }
+}
+
+#[test]
+fn test_into_read_results_none_variant() {
+    let response = ClientResponse {
+        error: ErrorCode::Success as i32,
+        metadata: None,
+        success_result: None,
+    };
+
+    let result = response.into_read_results();
+    assert!(result.is_err());
+
+    if let Err(ClientApiError::Protocol { code, message, .. }) = result {
+        assert_eq!(code, ErrorCode::InvalidResponse);
+        assert!(message.contains("expected ReadData"));
+        assert!(message.contains("found None"));
+    } else {
+        panic!("Expected Protocol error");
+    }
+}
+
+#[test]
+fn test_into_read_results_with_error_code() {
+    let response = ClientResponse {
+        error: ErrorCode::NotLeader as i32,
+        metadata: None,
+        success_result: Some(SuccessResult::ReadData(ReadResults { results: vec![] })),
+    };
+
+    let result = response.into_read_results();
+    assert!(result.is_err());
+
+    // Should fail at validate_error() before checking success_result
+    if let Err(ClientApiError::Business { code, .. }) = result {
+        assert_eq!(code, ErrorCode::NotLeader);
+    } else {
+        panic!("Expected Business error with NotLeader code");
+    }
+}

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

@@ -1,2 +1,5 @@
 mod client_ext;
 pub use client_ext::*;
+
+#[cfg(test)]
+mod client_ext_test;

d-engine-core/Cargo.toml

@@ -48,6 +48,8 @@ crc32fast = "1.4.2"
 # used in stream
 http-body = "1.0"
 http-body-util = "0.1.3"
+# Lock-free multi-producer multi-consumer channels for Watch event queue
+crossbeam-channel = "0.5"
 crossbeam-skiplist = "0.1"
 crossbeam = "0.8"
 mockall = { version = "0.12.1", optional = true }