diff --git a/CHANGELOG.md b/CHANGELOG.md index 96de3fe5..f689ddec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,15 @@ All notable changes to this project will be documented in this file. - Stream terminates with `UNAVAILABLE` on server shutdown — callers reconnect and resubscribe - `MembershipSnapshot` carries `members`, `learners`, and `committed_index` (idempotency key for deduplication after reconnect) +- **Watch: prefix watch and watcher limits** (#300): + - `watch_prefix(prefix)` API for both embedded and gRPC clients — one watcher covers an entire key namespace (e.g. `/services/payment/`) without registering a per-key watcher + - Prefix semantics: prefix must start and end with `/`; matching uses slash-boundary decomposition so `/services/payment/node1` matches `/services/payment/` but not `/services/payment` + - `WatchConfig.max_watcher_count` — hard cap on total active watchers (exact + prefix); `register()` / `watch_prefix()` return `LimitExceeded` error once reached (default: effectively unlimited) + - `WatchConfig.watcher_buffer_size` default raised from 10 → 256, reducing spurious `CANCELED` events under burst load + - Every `WatchEvent` carries `revision: u64` (Raft applied index) — use it to detect missed events after reconnection + - Architecture: two separate `DashMap`s (exact and prefix) — O(1) exact lookup, O(depth) prefix dispatch; no linear scan + - Proto: added `prefix: bool` to `WatchRequest`; added `revision: uint64` to `WatchResponse` + - **Unified RocksDB option** (#295): opt-in via `storage.unified_db = true`. Uses a single RocksDB instance with 4 column families instead of two separate instances. - Reduces memory RSS and file descriptor usage diff --git a/d-engine-client/src/grpc_client.rs b/d-engine-client/src/grpc_client.rs index e2593d8c..60513f0d 100644 --- a/d-engine-client/src/grpc_client.rs +++ b/d-engine-client/src/grpc_client.rs @@ -221,6 +221,7 @@ impl GrpcClient { let request = WatchRequest { client_id: client_inner.client_id, key: Bytes::copy_from_slice(key.as_ref()), + prefix: false, }; // Watch can connect to any node (leader or follower) @@ -237,6 +238,36 @@ impl GrpcClient { } } } + + /// Watch all keys under a path prefix. + /// + /// `prefix` must start with '/' and end with '/', e.g. `b"/services/"`. + /// Returns a stream of events for any key whose path begins with the prefix. + pub async fn watch_prefix( + &self, + prefix: impl AsRef<[u8]>, + ) -> ClientApiResult> { + let client_inner = self.client_inner.load(); + + let request = WatchRequest { + client_id: client_inner.client_id, + key: Bytes::copy_from_slice(prefix.as_ref()), + prefix: true, + }; + + let mut client = self.make_client().await?; + + match client.watch(request).await { + Ok(response) => { + debug!("Prefix watch stream established"); + Ok(response.into_inner()) + } + Err(status) => { + error!("Prefix watch request failed: {:?}", status); + Err(status.into()) + } + } + } } // ==================== Core ClientApi Trait Implementation ==================== diff --git a/d-engine-client/src/grpc_client_test.rs b/d-engine-client/src/grpc_client_test.rs index 8ffcca11..4df4f9ff 100644 --- a/d-engine-client/src/grpc_client_test.rs +++ b/d-engine-client/src/grpc_client_test.rs @@ -1594,12 +1594,14 @@ mod watch_tests { value: Bytes::from("owner-a"), event_type: WatchEventType::Put as i32, error: 0, + revision: 0, }, WatchResponse { key: Bytes::from("lock"), value: Bytes::new(), event_type: WatchEventType::Delete as i32, error: 0, + revision: 0, }, ]; diff --git a/d-engine-client/src/lib.rs b/d-engine-client/src/lib.rs index 2ce85e34..ca05eebb 100644 --- a/d-engine-client/src/lib.rs +++ b/d-engine-client/src/lib.rs @@ -74,6 +74,7 @@ mod utils; pub use builder::*; pub use config::*; pub use d_engine_core::client::{ClientApi, ClientApiError, ClientApiResult}; +pub use d_engine_proto::error::ErrorCode; pub use grpc_client::*; pub(crate) use pool::*; diff --git a/d-engine-core/src/config/raft.rs b/d-engine-core/src/config/raft.rs index 45410983..101460d6 100644 --- a/d-engine-core/src/config/raft.rs +++ b/d-engine-core/src/config/raft.rs @@ -1190,8 +1190,9 @@ fn default_client_compression() -> bool { /// ```toml /// [raft.watch] /// event_queue_size = 10240 -/// watcher_buffer_size = 10 +/// watcher_buffer_size = 256 /// enable_metrics = false +/// max_watcher_count = 10000 /// ``` #[derive(Debug, Serialize, Deserialize, Clone)] pub struct WatchConfig { @@ -1242,6 +1243,16 @@ pub struct WatchConfig { /// **Default**: false (minimal overhead in production) #[serde(default = "default_enable_watch_metrics")] pub enable_metrics: bool, + + /// Hard cap on total active watchers (exact + prefix combined). + /// + /// `register()` and `register_prefix()` return `WatchError::LimitExceeded` + /// once this count is reached. Leave at the default (effectively unlimited) + /// or set to a finite value to prevent runaway watcher growth. + /// + /// **Default**: i64::MAX (effectively unlimited) + #[serde(default = "default_max_watcher_count")] + pub max_watcher_count: usize, } impl Default for WatchConfig { @@ -1250,6 +1261,7 @@ impl Default for WatchConfig { event_queue_size: default_event_queue_size(), watcher_buffer_size: default_watcher_buffer_size(), enable_metrics: default_enable_watch_metrics(), + max_watcher_count: default_max_watcher_count(), } } } @@ -1294,13 +1306,19 @@ const fn default_event_queue_size() -> usize { } const fn default_watcher_buffer_size() -> usize { - 10 + 256 } const fn default_enable_watch_metrics() -> bool { false } +const fn default_max_watcher_count() -> usize { + // i64::MAX — the config crate uses signed 64-bit internally, so usize::MAX overflows it. + // This value is effectively unlimited for any realistic deployment. + 9_223_372_036_854_775_807 +} + /// Performance metrics configuration /// /// Controls emission of observability metrics. Disabling metrics reduces overhead diff --git a/d-engine-core/src/state_machine_handler/default_state_machine_handler.rs b/d-engine-core/src/state_machine_handler/default_state_machine_handler.rs index 79fc1294..3d2a166b 100644 --- a/d-engine-core/src/state_machine_handler/default_state_machine_handler.rs +++ b/d-engine-core/src/state_machine_handler/default_state_machine_handler.rs @@ -658,12 +658,14 @@ where value: insert.value, event_type: WatchEventType::Put as i32, error: 0, + revision: entry.index, }), Some(Operation::Delete(delete)) => Some(WatchResponse { key: delete.key, value: bytes::Bytes::new(), event_type: WatchEventType::Delete as i32, error: 0, + revision: entry.index, }), Some(Operation::CompareAndSwap(cas)) => { // Only broadcast if CAS actually mutated the value. @@ -674,6 +676,7 @@ where value: cas.new_value, event_type: WatchEventType::Put as i32, error: 0, + revision: entry.index, }) } else { None diff --git a/d-engine-core/src/watch/manager.rs b/d-engine-core/src/watch/manager.rs index 2e55e65b..0f0cd397 100644 --- a/d-engine-core/src/watch/manager.rs +++ b/d-engine-core/src/watch/manager.rs @@ -7,10 +7,12 @@ //! apply_chunk() -> broadcast_watch_events() -> broadcast::send(WatchEvent) [fire-and-forget] //! ↓ //! WatchDispatcher (spawned in Builder): -//! broadcast::subscribe() -> match key in DashMap -> mpsc::send(per-watcher) +//! broadcast::subscribe() -> +//! exact: DashMap> O(1) lookup per event +//! prefix: DashMap> O(depth) decomposition + O(1) lookup per segment //! ↓ //! Watchers: -//! Embedded: mpsc::Receiver +//! Embedded: mpsc::Receiver //! Standalone: mpsc::Receiver -> gRPC stream (protobuf conversion) //! ``` //! @@ -19,13 +21,13 @@ //! - **No hidden resource allocation**: All tokio::spawn calls are explicit in Builder //! - **Minimal abstraction**: Only essential data structures, no unnecessary wrappers //! - **Composable**: Registry and Dispatcher are independent, composed in Builder +//! - **O(depth) prefix dispatch**: Decompose event key into slash-terminated path segments, +//! O(1) DashMap lookup per segment — dispatch cost is independent of prefix watcher count use std::sync::Arc; -use std::sync::atomic::AtomicU64; -use std::sync::atomic::Ordering; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use bytes::Bytes; -// Re-export protobuf types for watch events pub use d_engine_proto::client::{WatchEventType, WatchResponse as WatchEvent}; use dashmap::DashMap; use tokio::sync::broadcast; @@ -34,14 +36,60 @@ use tracing::debug; use tracing::trace; use tracing::warn; -/// Handle for a registered watcher +/// Errors returned by WatchRegistry registration methods. +#[derive(Debug)] +pub enum WatchError { + /// Active watcher count has reached the configured max_watcher_count cap. + LimitExceeded(usize), + /// Prefix must start with '/' and end with '/'. + InvalidPrefix, +} + +impl std::fmt::Display for WatchError { + fn fmt( + &self, + f: &mut std::fmt::Formatter<'_>, + ) -> std::fmt::Result { + match self { + WatchError::LimitExceeded(n) => write!(f, "watcher limit ({n}) exceeded"), + WatchError::InvalidPrefix => { + write!(f, "prefix must start with '/' and end with '/'") + } + } + } +} + +impl std::error::Error for WatchError {} + +/// Decompose a key into all its slash-terminated path prefix candidates. +/// +/// The dispatcher calls this on every event key to perform O(depth) prefix +/// lookup via O(1) DashMap gets — one per path segment — instead of scanning +/// all registered prefixes linearly. +/// +/// Examples: +/// "/config/db/host" → ["/", "/config/", "/config/db/"] +/// "/" → ["/"] +/// "/config" → ["/"] +/// "/config/" → ["/", "/config/"] +pub(crate) fn prefix_segments(key: &Bytes) -> Vec { + key.iter() + .enumerate() + .filter(|&(_, &b)| b == b'/') + .map(|(i, _)| key.slice(0..i + 1)) + .collect() +} + +/// Handle for a registered watcher. /// /// When dropped, the watcher is automatically unregistered (if unregister_tx is Some). pub struct WatcherHandle { /// Unique identifier id: u64, - /// Key being watched + /// Key being watched (exact key) or prefix being watched (prefix watcher) key: Bytes, + /// True when registered via register_prefix() + is_prefix: bool, /// Channel receiver for watch events receiver: mpsc::Receiver, /// Unregister channel (None if cleanup disabled via into_receiver) @@ -49,24 +97,26 @@ pub struct WatcherHandle { } impl WatcherHandle { - /// Get the unique identifier for this watcher pub fn id(&self) -> u64 { self.id } - /// Get the key being watched pub fn key(&self) -> &Bytes { &self.key } - /// Get a mutable reference to the receiver + /// True if this handle was registered via register_prefix(). + pub fn is_prefix(&self) -> bool { + self.is_prefix + } + pub fn receiver_mut(&mut self) -> &mut mpsc::Receiver { &mut self.receiver } - /// Consume the handle and return the event receiver + /// Consume the handle and return the event receiver. /// - /// Disables automatic unregistration. The watcher will remain active until + /// Disables automatic unregistration. The watcher remains active until /// the receiver is dropped (causing send failures that trigger cleanup). /// /// Use this for long-lived streams (e.g., gRPC) where the receiver lifetime @@ -74,15 +124,10 @@ impl WatcherHandle { pub fn into_receiver(mut self) -> (u64, Bytes, mpsc::Receiver) { let id = self.id; let key = self.key.clone(); - - // Clear unregister_tx to disable Drop cleanup self.unregister_tx = None; - - // Create dummy receiver to satisfy Rust's move checker let (dummy_tx, dummy_rx) = mpsc::channel(1); - drop(dummy_tx); // Close immediately + drop(dummy_tx); let receiver = std::mem::replace(&mut self.receiver, dummy_rx); - (id, key, receiver) } } @@ -90,9 +135,13 @@ impl WatcherHandle { impl Drop for WatcherHandle { fn drop(&mut self) { if let Some(ref tx) = self.unregister_tx { - // Send unregister request (ignore errors if dispatcher stopped) let _ = tx.send((self.id, self.key.clone())); - trace!(watcher_id = self.id, key = ?self.key, "Watcher unregistered"); + trace!( + watcher_id = self.id, + key = ?self.key, + is_prefix = self.is_prefix, + "Watcher unregistered" + ); } } } @@ -100,129 +149,180 @@ impl Drop for WatcherHandle { /// Internal watcher state #[derive(Debug)] struct Watcher { - /// Unique identifier id: u64, - /// Channel sender for events sender: mpsc::Sender, } -/// Watch registry - manages watcher registration (Arc-shareable) +/// Watch registry — manages watcher registration (Arc-shareable). /// -/// This is the shared state that both Builder (for registration) and -/// WatchDispatcher (for event dispatch) can access concurrently. +/// Two independent DashMaps keep exact and prefix watchers separate so +/// dispatch can use different lookup strategies for each without coupling. pub struct WatchRegistry { - /// Watchers grouped by key (lock-free concurrent HashMap) - watchers: DashMap>, - - /// Next watcher ID (monotonically increasing) + /// Exact-match watchers: event_key → watchers + exact: DashMap>, + /// Prefix watchers: prefix → watchers (prefix must start and end with '/') + prefix: DashMap>, + /// Next watcher ID (monotonically increasing, globally unique) next_id: AtomicU64, - + /// Total active watchers across exact + prefix (for limit enforcement) + total_count: AtomicUsize, /// Per-watcher channel buffer size watcher_buffer_size: usize, - + /// Hard cap on total active watchers; register() returns LimitExceeded when reached + max_watcher_count: usize, /// Unregister channel sender (cloned for each WatcherHandle) unregister_tx: mpsc::UnboundedSender<(u64, Bytes)>, } impl WatchRegistry { - /// Create a new watch registry - /// - /// # Arguments - /// * `watcher_buffer_size` - Buffer size for per-watcher mpsc channels - /// * `unregister_tx` - Channel for receiving unregister requests + /// Create a new registry with no watcher count limit. pub fn new( watcher_buffer_size: usize, unregister_tx: mpsc::UnboundedSender<(u64, Bytes)>, + ) -> Self { + Self::new_with_limits(watcher_buffer_size, usize::MAX, unregister_tx) + } + + /// Create a new registry with a hard watcher count cap. + /// + /// `register()` and `register_prefix()` return `WatchError::LimitExceeded` + /// once `max_watcher_count` active watchers are registered. + pub fn new_with_limits( + watcher_buffer_size: usize, + max_watcher_count: usize, + unregister_tx: mpsc::UnboundedSender<(u64, Bytes)>, ) -> Self { Self { - watchers: DashMap::new(), + exact: DashMap::new(), + prefix: DashMap::new(), next_id: AtomicU64::new(1), + total_count: AtomicUsize::new(0), watcher_buffer_size, + max_watcher_count, unregister_tx, } } - /// Register a new watcher for a specific key + /// Register an exact-key watcher. /// - /// Returns a handle that receives watch events via an mpsc channel. - /// The watcher is automatically unregistered when the handle is dropped. + /// Returns `WatchError::LimitExceeded` if max_watcher_count is reached. pub fn register( &self, key: Bytes, - ) -> WatcherHandle { + ) -> Result { + self.do_register(key, false) + } + + /// Register a prefix watcher. + /// + /// `prefix` must start with '/' and end with '/'. + /// E.g. "/config/" watches all keys under /config/. + /// + /// Returns `WatchError::InvalidPrefix` if format is wrong. + /// Returns `WatchError::LimitExceeded` if max_watcher_count is reached. + pub fn register_prefix( + &self, + prefix: Bytes, + ) -> Result { + if !prefix.starts_with(b"/") || !prefix.ends_with(b"/") { + return Err(WatchError::InvalidPrefix); + } + self.do_register(prefix, true) + } + + fn do_register( + &self, + key: Bytes, + is_prefix: bool, + ) -> Result { + // Reserve the slot first, then check. This eliminates the TOCTOU window + // between a load-check and a separate fetch_add under concurrent registration. + let prev = self.total_count.fetch_add(1, Ordering::Relaxed); + if prev >= self.max_watcher_count { + self.total_count.fetch_sub(1, Ordering::Relaxed); + return Err(WatchError::LimitExceeded(self.max_watcher_count)); + } + let id = self.next_id.fetch_add(1, Ordering::Relaxed); - // +1 reserves one slot for the CANCELED sentinel so it can always be - // delivered even when all watcher_buffer_size normal slots are full. + // +1 reserves one slot for the CANCELED sentinel (always deliverable even when full) let (sender, receiver) = mpsc::channel(self.watcher_buffer_size + 1); - let watcher = Watcher { id, sender }; - // Insert into DashMap (lock-free) - self.watchers.entry(key.clone()).or_default().push(watcher); - - trace!(watcher_id = id, key = ?key, "Watcher registered"); + if is_prefix { + self.prefix.entry(key.clone()).or_default().push(watcher); + } else { + self.exact.entry(key.clone()).or_default().push(watcher); + } + trace!(watcher_id = id, key = ?key, is_prefix, "Watcher registered"); - WatcherHandle { + Ok(WatcherHandle { id, key, + is_prefix, receiver, unregister_tx: Some(self.unregister_tx.clone()), - } + }) } - /// Unregister a watcher fn unregister( &self, id: u64, key: &Bytes, ) { - self.watchers.remove_if_mut(key, |_key, watchers| { + // Watcher IDs are globally unique — try exact first, then prefix. + // Cold path: called only on drop or after overflow, not on every event. + let mut found = false; + self.exact.remove_if_mut(key, |_, watchers| { + let before = watchers.len(); watchers.retain(|w| w.id != id); + found = watchers.len() < before; watchers.is_empty() }); + if !found { + self.prefix.remove_if_mut(key, |_, watchers| { + let before = watchers.len(); + watchers.retain(|w| w.id != id); + found = watchers.len() < before; + watchers.is_empty() + }); + } + if found { + self.total_count.fetch_sub(1, Ordering::Relaxed); + } } - /// Get the number of active watchers for a specific key (for testing) #[cfg(test)] pub(crate) fn watcher_count( &self, key: &Bytes, ) -> usize { - self.watchers.get(key).map(|w| w.len()).unwrap_or(0) + self.exact.get(key).map(|w| w.len()).unwrap_or(0) } - /// Get the total number of watched keys (for testing) #[cfg(test)] pub(crate) fn watched_key_count(&self) -> usize { - self.watchers.len() + self.exact.len() + } + + #[cfg(test)] + pub(crate) fn prefix_watcher_count( + &self, + prefix: &Bytes, + ) -> usize { + self.prefix.get(prefix).map(|w| w.len()).unwrap_or(0) } } -/// Watch dispatcher - distributes events to watchers (background task) +/// Watch dispatcher — distributes events to watchers (background task). /// -/// This is spawned explicitly in NodeBuilder::build() to make resource -/// allocation visible. It continuously: -/// 1. Receives events from broadcast channel -/// 2. Matches keys in the registry -/// 3. Dispatches to matching watchers +/// Spawned explicitly in NodeBuilder::build() to make resource allocation visible. pub struct WatchDispatcher { - /// Shared registry (same instance held by Node) registry: Arc, - - /// Broadcast receiver for global events broadcast_rx: broadcast::Receiver, - - /// Unregister channel receiver unregister_rx: mpsc::UnboundedReceiver<(u64, Bytes)>, } impl WatchDispatcher { - /// Create a new watch dispatcher - /// - /// # Arguments - /// * `registry` - Shared registry for looking up watchers - /// * `broadcast_rx` - Receiver for watch events from StateMachine - /// * `unregister_rx` - Receiver for unregister requests from WatcherHandles pub fn new( registry: Arc, broadcast_rx: broadcast::Receiver, @@ -235,28 +335,18 @@ impl WatchDispatcher { } } - /// Run the dispatcher event loop - /// - /// This should be spawned as a tokio task in NodeBuilder::build(). - /// It will run until the broadcast channel is closed. pub async fn run(mut self) { debug!("WatchDispatcher started"); - loop { tokio::select! { biased; - - // Handle unregister requests first (cleanup priority) + // Cleanup first so dead watchers don't receive the next event Some((id, key)) = self.unregister_rx.recv() => { self.registry.unregister(id, &key); } - - // Receive broadcast event result = self.broadcast_rx.recv() => { match result { - Ok(event) => { - self.dispatch_event(event).await; - } + Ok(event) => self.dispatch_event(event).await, Err(broadcast::error::RecvError::Lagged(n)) => { warn!("WatchDispatcher lagged {} events (slow watchers)", n); } @@ -268,25 +358,41 @@ impl WatchDispatcher { } } } - debug!("WatchDispatcher stopped"); } - /// Dispatch an event to all watchers of a specific key async fn dispatch_event( &self, event: WatchEvent, ) { - if let Some(watchers) = self.registry.watchers.get(&event.key) { + // Step 1: exact match — O(1) DashMap lookup + self.dispatch_to_map(&self.registry.exact, &event.key, &event).await; + + // Step 2: prefix match — O(depth) where depth = number of '/' in key + for prefix in prefix_segments(&event.key) { + self.dispatch_to_map(&self.registry.prefix, &prefix, &event).await; + } + } + + /// Dispatch an event to all watchers under `lookup_key` in `map`. + /// + /// Handles overflow detection and dead watcher cleanup identically for + /// both the exact and prefix maps, keeping the two dispatch paths DRY. + async fn dispatch_to_map( + &self, + map: &DashMap>, + lookup_key: &Bytes, + event: &WatchEvent, + ) { + if let Some(watchers) = map.get(lookup_key) { let mut dead_watchers = Vec::new(); for watcher in watchers.iter() { let available = watcher.sender.capacity(); if available <= 1 { - // capacity == 1: exactly the reserved cancel slot remains. - // capacity == 0: defensive — shouldn't happen in normal flow - // since we check before every normal send, but handle it. + // capacity == 1: only the reserved cancel slot remains → overflow + // capacity == 0: defensive, shouldn't happen in normal flow if available == 1 { warn!( watcher_id = watcher.id, @@ -303,22 +409,26 @@ impl WatchDispatcher { continue; } - // Normal send: capacity > 1, reserved slot is untouched. + // Normal send: reserved slot untouched if let Err(mpsc::error::TrySendError::Closed(_)) = watcher.sender.try_send(event.clone()) { - // Receiver dropped: silent cleanup, no cancel needed. + // Receiver dropped: silent cleanup, no cancel needed dead_watchers.push(watcher.id); } } - // Cleanup dead watchers drop(watchers); for id in dead_watchers { - self.registry.unregister(id, &event.key); + self.registry.unregister(id, lookup_key); } - trace!(key = ?event.key, event_type = ?event.event_type, "Event dispatched"); + trace!( + event_key = ?event.key, + lookup_key = ?lookup_key, + event_type = ?event.event_type, + "Event dispatched" + ); } } } diff --git a/d-engine-core/src/watch/manager_test.rs b/d-engine-core/src/watch/manager_test.rs index c9d702d7..15afaa3c 100644 --- a/d-engine-core/src/watch/manager_test.rs +++ b/d-engine-core/src/watch/manager_test.rs @@ -13,6 +13,8 @@ use bytes::Bytes; use tokio::sync::broadcast; use tokio::sync::mpsc; use tokio::time::Duration; + +use crate::watch::manager::prefix_segments; use tokio::time::timeout; use super::*; @@ -43,7 +45,7 @@ async fn test_register_single_watcher() { let (_, registry, _dispatcher_handle) = setup_watch_system(10); let key = Bytes::from("test_key"); - let _handle = registry.register(key.clone()); + let _handle = registry.register(key.clone()).unwrap(); assert_eq!(registry.watcher_count(&key), 1); assert_eq!(registry.watched_key_count(), 1); @@ -55,9 +57,9 @@ async fn test_register_multiple_watchers_same_key() { let key = Bytes::from("shared_key"); - let _handle1 = registry.register(key.clone()); - let _handle2 = registry.register(key.clone()); - let _handle3 = registry.register(key.clone()); + let _handle1 = registry.register(key.clone()).unwrap(); + let _handle2 = registry.register(key.clone()).unwrap(); + let _handle3 = registry.register(key.clone()).unwrap(); assert_eq!(registry.watcher_count(&key), 3); assert_eq!(registry.watched_key_count(), 1); // Only 1 unique key @@ -70,7 +72,7 @@ async fn test_watcher_auto_cleanup_on_drop() { let key = Bytes::from("cleanup_key"); { - let _handle = registry.register(key.clone()); + let _handle = registry.register(key.clone()).unwrap(); assert_eq!(registry.watcher_count(&key), 1); // Handle dropped here } @@ -89,7 +91,7 @@ async fn test_dispatcher_dispatch_to_matching_watcher() { let key = Bytes::from("test_key"); let value = Bytes::from("test_value"); - let mut handle = registry.register(key.clone()); + let mut handle = registry.register(key.clone()).unwrap(); // Small delay to ensure dispatcher is ready tokio::time::sleep(Duration::from_millis(50)).await; @@ -100,6 +102,7 @@ async fn test_dispatcher_dispatch_to_matching_watcher() { value: value.clone(), event_type: WatchEventType::Put as i32, error: 0, + revision: 0, }; broadcast_tx.send(event).unwrap(); @@ -121,7 +124,7 @@ async fn test_dispatcher_ignores_non_matching_key() { let watched_key = Bytes::from("key1"); let other_key = Bytes::from("key2"); - let mut handle = registry.register(watched_key.clone()); + let mut handle = registry.register(watched_key.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; @@ -131,6 +134,7 @@ async fn test_dispatcher_ignores_non_matching_key() { value: Bytes::from("value"), event_type: WatchEventType::Put as i32, error: 0, + revision: 0, }; broadcast_tx.send(event).unwrap(); @@ -149,9 +153,9 @@ async fn test_multiple_watchers_all_receive_event() { let key = Bytes::from("shared_key"); let value = Bytes::from("shared_value"); - let mut handle1 = registry.register(key.clone()); - let mut handle2 = registry.register(key.clone()); - let mut handle3 = registry.register(key.clone()); + let mut handle1 = registry.register(key.clone()).unwrap(); + let mut handle2 = registry.register(key.clone()).unwrap(); + let mut handle3 = registry.register(key.clone()).unwrap(); assert_eq!(registry.watcher_count(&key), 3); @@ -163,6 +167,7 @@ async fn test_multiple_watchers_all_receive_event() { value: value.clone(), event_type: WatchEventType::Put as i32, error: 0, + revision: 0, }; broadcast_tx.send(event).unwrap(); @@ -183,7 +188,7 @@ async fn test_watcher_delete_event() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(10); let key = Bytes::from("test_key"); - let mut handle = registry.register(key.clone()); + let mut handle = registry.register(key.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; @@ -193,6 +198,7 @@ async fn test_watcher_delete_event() { value: Bytes::new(), event_type: WatchEventType::Delete as i32, error: 0, + revision: 0, }; broadcast_tx.send(event).unwrap(); @@ -210,7 +216,7 @@ async fn test_multiple_events_sequential() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(10); let key = Bytes::from("test_key"); - let mut handle = registry.register(key.clone()); + let mut handle = registry.register(key.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; @@ -221,6 +227,7 @@ async fn test_multiple_events_sequential() { value: Bytes::from("value1"), event_type: WatchEventType::Put as i32, error: 0, + revision: 0, }) .unwrap(); @@ -230,6 +237,7 @@ async fn test_multiple_events_sequential() { value: Bytes::from("value2"), event_type: WatchEventType::Put as i32, error: 0, + revision: 0, }) .unwrap(); @@ -239,6 +247,7 @@ async fn test_multiple_events_sequential() { value: Bytes::new(), event_type: WatchEventType::Delete as i32, error: 0, + revision: 0, }) .unwrap(); @@ -259,9 +268,9 @@ async fn test_watcher_count_after_partial_cleanup() { let key = Bytes::from("count_key"); - let handle1 = registry.register(key.clone()); - let _handle2 = registry.register(key.clone()); - let _handle3 = registry.register(key.clone()); + let handle1 = registry.register(key.clone()).unwrap(); + let _handle2 = registry.register(key.clone()).unwrap(); + let _handle3 = registry.register(key.clone()).unwrap(); assert_eq!(registry.watcher_count(&key), 3); @@ -278,8 +287,8 @@ async fn test_different_keys_isolated() { let key1 = Bytes::from("key1"); let key2 = Bytes::from("key2"); - let mut handle1 = registry.register(key1.clone()); - let mut handle2 = registry.register(key2.clone()); + let mut handle1 = registry.register(key1.clone()).unwrap(); + let mut handle2 = registry.register(key2.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; @@ -290,6 +299,7 @@ async fn test_different_keys_isolated() { value: Bytes::from("value1"), event_type: WatchEventType::Put as i32, error: 0, + revision: 0, }) .unwrap(); @@ -329,6 +339,7 @@ fn put_event( value: Bytes::from(value.to_owned()), event_type: WatchEventType::Put as i32, error: 0, + revision: 0, } } @@ -340,7 +351,7 @@ fn put_event( async fn test_watcher_buffer_overflow_sends_cancel_event() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(2); let key = Bytes::from("overflow_key"); - let mut handle = registry.register(key.clone()); + let mut handle = registry.register(key.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(20)).await; @@ -384,7 +395,7 @@ async fn test_watcher_buffer_overflow_sends_cancel_event() { async fn test_cancel_event_has_correct_fields() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(1); let key = Bytes::from("field_check_key"); - let mut handle = registry.register(key.clone()); + let mut handle = registry.register(key.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(20)).await; @@ -411,7 +422,7 @@ async fn test_cancel_event_has_correct_fields() { async fn test_overflow_removes_watcher_from_registry() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(1); let key = Bytes::from("registry_key"); - let _handle = registry.register(key.clone()); + let _handle = registry.register(key.clone()).unwrap(); assert_eq!(registry.watcher_count(&key), 1); @@ -430,7 +441,7 @@ async fn test_overflow_removes_watcher_from_registry() { async fn test_normal_events_precede_cancel_in_channel() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(2); let key = Bytes::from("order_key"); - let mut handle = registry.register(key.clone()); + let mut handle = registry.register(key.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(20)).await; @@ -466,7 +477,7 @@ async fn test_normal_events_precede_cancel_in_channel() { async fn test_closed_receiver_cleaned_up_silently() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(10); let key = Bytes::from("closed_key"); - let handle = registry.register(key.clone()); + let handle = registry.register(key.clone()).unwrap(); assert_eq!(registry.watcher_count(&key), 1); @@ -489,8 +500,8 @@ async fn test_slow_watcher_overflow_does_not_affect_healthy_watcher() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(2); let key = Bytes::from("isolation_key"); - let mut slow = registry.register(key.clone()); // never consumed - let mut fast = registry.register(key.clone()); // consumed between dispatches + let mut slow = registry.register(key.clone()).unwrap(); // never consumed + let mut fast = registry.register(key.clone()).unwrap(); // consumed between dispatches tokio::time::sleep(Duration::from_millis(20)).await; @@ -545,7 +556,7 @@ async fn test_slow_watcher_overflow_does_not_affect_healthy_watcher() { async fn test_healthy_watcher_no_false_cancel() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(2); let key = Bytes::from("healthy_key"); - let mut handle = registry.register(key.clone()); + let mut handle = registry.register(key.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(20)).await; @@ -577,7 +588,7 @@ async fn test_healthy_watcher_no_false_cancel() { async fn test_buffer_size_1_overflows_on_second_event() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(1); let key = Bytes::from("tiny_buffer_key"); - let mut handle = registry.register(key.clone()); + let mut handle = registry.register(key.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(20)).await; @@ -615,7 +626,7 @@ async fn test_buffer_size_1_overflows_on_second_event() { async fn test_events_not_delivered_after_overflow() { let (broadcast_tx, registry, _dispatcher_handle) = setup_watch_system(1); let key = Bytes::from("post_overflow_key"); - let _handle = registry.register(key.clone()); + let _handle = registry.register(key.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(20)).await; @@ -641,7 +652,7 @@ async fn test_into_receiver_disables_cleanup() { let (_, registry, _dispatcher_handle) = setup_watch_system(10); let key = Bytes::from("test_key"); - let handle = registry.register(key.clone()); + let handle = registry.register(key.clone()).unwrap(); assert_eq!(registry.watcher_count(&key), 1); @@ -697,7 +708,7 @@ async fn test_dispatcher_shutdown_on_broadcast_close() { let (broadcast_tx, registry, dispatcher_handle) = setup_watch_system(10); let key = Bytes::from("test_key"); - let _handle = registry.register(key.clone()); + let _handle = registry.register(key.clone()).unwrap(); tokio::time::sleep(Duration::from_millis(50)).await; @@ -711,3 +722,470 @@ async fn test_dispatcher_shutdown_on_broadcast_close() { "Dispatcher should exit within 2s on channel close" ); } + +// ============================================================================= +// #300: Prefix Watch + Revision +// ============================================================================= + +fn setup_watch_system_with_max( + buffer_size: usize, + max_watcher_count: usize, +) -> ( + broadcast::Sender, + Arc, + tokio::task::JoinHandle<()>, +) { + let (broadcast_tx, broadcast_rx) = broadcast::channel(1000); + let (unregister_tx, unregister_rx) = mpsc::unbounded_channel(); + let registry = Arc::new(WatchRegistry::new_with_limits( + buffer_size, + max_watcher_count, + unregister_tx, + )); + let dispatcher = WatchDispatcher::new(Arc::clone(®istry), broadcast_rx, unregister_rx); + let handle = tokio::spawn(async move { dispatcher.run().await }); + (broadcast_tx, registry, handle) +} + +// --- prefix_segments() unit tests (pure function) --- + +/// #300: prefix_segments is the core of O(depth) dispatch. +/// Verifies that a multi-level path is decomposed into all slash-terminated +/// prefix candidates so the dispatcher can do O(1) DashMap lookups per level +/// instead of scanning every registered prefix. +#[test] +fn test_prefix_segments_decomposes_path_correctly() { + let key = Bytes::from("/config/db/host"); + let segments = prefix_segments(&key); + assert_eq!( + segments, + vec![ + Bytes::from("/"), + Bytes::from("/config/"), + Bytes::from("/config/db/"), + ] + ); +} + +/// #300: "/" is both a valid key and the root prefix. +/// prefix_segments must return ["/"] so that a watcher registered on "/" +/// is notified when the root key itself changes. +#[test] +fn test_prefix_segments_root_key() { + let key = Bytes::from("/"); + let segments = prefix_segments(&key); + assert_eq!(segments, vec![Bytes::from("/")]); +} + +/// #300: A single-level key like "/config" (no trailing slash) has exactly +/// one ancestor prefix: "/". Ensures prefix watchers on "/" are notified +/// for top-level keys even when those keys are not directories. +#[test] +fn test_prefix_segments_single_level() { + let key = Bytes::from("/config"); + let segments = prefix_segments(&key); + assert_eq!(segments, vec![Bytes::from("/")]); +} + +/// #300: A key ending with "/" (directory-style) must include itself as a +/// prefix segment. This matters for service-discovery patterns where both +/// the namespace key and its children can be watched. +#[test] +fn test_prefix_segments_trailing_slash_included() { + let key = Bytes::from("/config/"); + let segments = prefix_segments(&key); + assert_eq!(segments, vec![Bytes::from("/"), Bytes::from("/config/")]); +} + +// --- prefix watch: basic dispatch --- + +/// #300: Core use case — a client watching an entire namespace receives +/// events for any key that falls under that namespace. Without prefix watch, +/// clients must register one watcher per key, which is impractical for +/// dynamic service-discovery or config namespaces. +#[tokio::test] +async fn test_prefix_watch_matches_child_key() { + let (broadcast_tx, registry, _handle) = setup_watch_system(10); + let mut watcher = registry.register_prefix(Bytes::from("/config/")).unwrap(); + + tokio::time::sleep(Duration::from_millis(50)).await; + + broadcast_tx + .send(WatchEvent { + key: Bytes::from("/config/db/host"), + value: Bytes::from("10.0.0.1"), + event_type: WatchEventType::Put as i32, + error: 0, + revision: 1, + }) + .unwrap(); + + let received = timeout(Duration::from_millis(100), watcher.receiver_mut().recv()) + .await + .expect("timeout") + .expect("closed"); + assert_eq!(received.key, Bytes::from("/config/db/host")); +} + +/// #300: Prefix isolation — a watcher on "/config/" must not receive events +/// for unrelated namespaces like "/dns/". Noisy cross-namespace delivery +/// would break multi-tenant deployments where each tenant owns a prefix. +#[tokio::test] +async fn test_prefix_watch_does_not_match_unrelated_key() { + let (broadcast_tx, registry, _handle) = setup_watch_system(10); + let mut watcher = registry.register_prefix(Bytes::from("/config/")).unwrap(); + + tokio::time::sleep(Duration::from_millis(50)).await; + + broadcast_tx + .send(WatchEvent { + key: Bytes::from("/dns/xyz"), + value: Bytes::from("1.1.1.1"), + event_type: WatchEventType::Put as i32, + error: 0, + revision: 1, + }) + .unwrap(); + + let result = timeout(Duration::from_millis(100), watcher.receiver_mut().recv()).await; + assert!(result.is_err(), "/config/ prefix must not match /dns/xyz"); +} + +/// #300: Slash boundary — "/config/" is a namespace prefix; "/config" is a +/// distinct key at the parent level. A watcher on the namespace must not +/// fire for the parent key. This matches etcd semantics and prevents +/// accidental cross-boundary notifications. +#[tokio::test] +async fn test_prefix_watch_slash_boundary_not_matched() { + let (broadcast_tx, registry, _handle) = setup_watch_system(10); + let mut watcher = registry.register_prefix(Bytes::from("/config/")).unwrap(); + + tokio::time::sleep(Duration::from_millis(50)).await; + + broadcast_tx + .send(WatchEvent { + key: Bytes::from("/config"), + value: Bytes::from("val"), + event_type: WatchEventType::Put as i32, + error: 0, + revision: 1, + }) + .unwrap(); + + let result = timeout(Duration::from_millis(100), watcher.receiver_mut().recv()).await; + assert!( + result.is_err(), + "/config/ prefix must not match key /config" + ); +} + +/// #300: The root prefix "/" acts as a global watch — useful for audit +/// logging or debugging where a single watcher must see all key changes +/// across all namespaces. +#[tokio::test] +async fn test_root_prefix_matches_child_keys() { + let (broadcast_tx, registry, _handle) = setup_watch_system(10); + let mut watcher = registry.register_prefix(Bytes::from("/")).unwrap(); + + tokio::time::sleep(Duration::from_millis(50)).await; + + broadcast_tx + .send(WatchEvent { + key: Bytes::from("/services/api"), + value: Bytes::from("up"), + event_type: WatchEventType::Put as i32, + error: 0, + revision: 1, + }) + .unwrap(); + + let received = timeout(Duration::from_millis(100), watcher.receiver_mut().recv()) + .await + .expect("timeout") + .expect("closed"); + assert_eq!(received.key, Bytes::from("/services/api")); +} + +/// #300: Edge case fix — when the event key IS "/" (root key itself), +/// prefix_segments("/") = ["/"], so a watcher on prefix "/" must be notified. +/// Without the fix (i < key.len()-1 guard), this event was silently dropped. +#[tokio::test] +async fn test_root_prefix_matches_root_key_itself() { + let (broadcast_tx, registry, _handle) = setup_watch_system(10); + let mut watcher = registry.register_prefix(Bytes::from("/")).unwrap(); + + tokio::time::sleep(Duration::from_millis(50)).await; + + broadcast_tx + .send(WatchEvent { + key: Bytes::from("/"), + value: Bytes::from("root"), + event_type: WatchEventType::Put as i32, + error: 0, + revision: 1, + }) + .unwrap(); + + let received = timeout(Duration::from_millis(100), watcher.receiver_mut().recv()) + .await + .expect("root prefix must match root key") + .expect("closed"); + assert_eq!(received.key, Bytes::from("/")); +} + +/// #300: Exact and prefix watchers are independent and must both fire for +/// the same event. A client watching a specific key for fast reaction AND a +/// monitoring client watching the whole namespace must not block each other. +#[tokio::test] +async fn test_exact_and_prefix_both_notified_on_same_event() { + let (broadcast_tx, registry, _handle) = setup_watch_system(10); + let key = Bytes::from("/config/db/host"); + + let mut exact_watcher = registry.register(key.clone()).unwrap(); + let mut prefix_watcher = registry.register_prefix(Bytes::from("/config/")).unwrap(); + + tokio::time::sleep(Duration::from_millis(50)).await; + + broadcast_tx + .send(WatchEvent { + key: key.clone(), + value: Bytes::from("10.0.0.1"), + event_type: WatchEventType::Put as i32, + error: 0, + revision: 5, + }) + .unwrap(); + + let exact_ev = timeout( + Duration::from_millis(100), + exact_watcher.receiver_mut().recv(), + ) + .await + .expect("exact timeout") + .expect("exact closed"); + let prefix_ev = timeout( + Duration::from_millis(100), + prefix_watcher.receiver_mut().recv(), + ) + .await + .expect("prefix timeout") + .expect("prefix closed"); + + assert_eq!(exact_ev.key, key); + assert_eq!(prefix_ev.key, key); +} + +// --- register_prefix: validation --- + +/// #300: Prefix format is enforced at registration time to prevent silent +/// mismatches at dispatch time. A prefix without a leading slash would never +/// match any key (all keys start with "/"). A prefix without a trailing slash +/// is semantically ambiguous with an exact key — reject early with a clear error. +#[test] +fn test_register_prefix_rejects_missing_trailing_slash() { + let (unregister_tx, _) = mpsc::unbounded_channel(); + let registry = WatchRegistry::new(10, unregister_tx); + + assert!( + registry.register_prefix(Bytes::from("/config")).is_err(), + "/config must be rejected — no trailing slash" + ); + assert!( + registry.register_prefix(Bytes::from("config/")).is_err(), + "config/ must be rejected — no leading slash" + ); +} + +// --- prefix watcher: overflow sends CANCELED --- + +/// #300: Prefix watchers are subject to the same overflow protection as exact +/// watchers (#294). A slow consumer watching a high-traffic namespace must +/// receive CANCELED (not silently die) so it can re-sync via the Read API. +/// Without this, the client has no signal that it missed events. +#[tokio::test] +async fn test_prefix_watcher_overflow_sends_cancel() { + let (broadcast_tx, registry, _handle) = setup_watch_system(1); // buffer_size=1 + let mut watcher = registry.register_prefix(Bytes::from("/config/")).unwrap(); + + tokio::time::sleep(Duration::from_millis(20)).await; + + // 2 events fill buffer + trigger overflow (capacity = buffer_size+1 = 2) + broadcast_tx + .send(WatchEvent { + key: Bytes::from("/config/key1"), + value: Bytes::from("v1"), + event_type: WatchEventType::Put as i32, + error: 0, + revision: 1, + }) + .unwrap(); + broadcast_tx + .send(WatchEvent { + key: Bytes::from("/config/key2"), + value: Bytes::from("v2"), + event_type: WatchEventType::Put as i32, + error: 0, + revision: 2, + }) + .unwrap(); + tokio::time::sleep(Duration::from_millis(50)).await; + + let _normal = watcher.receiver_mut().recv().await.unwrap(); + + let cancel = timeout(Duration::from_millis(100), watcher.receiver_mut().recv()) + .await + .expect("timeout") + .expect("closed"); + assert_eq!(cancel.event_type, WatchEventType::Canceled as i32); + assert_eq!(cancel.error, ErrorCode::WatchBufferOverflow as i32); +} + +// --- prefix watcher: auto-cleanup on drop --- + +/// #300: Prefix watcher handles follow the same RAII cleanup contract as +/// exact watchers. A leaked prefix watcher would continue consuming dispatch +/// CPU on every matching event even after the client is gone. +#[tokio::test] +async fn test_prefix_watcher_cleanup_on_drop() { + let (_, registry, _handle) = setup_watch_system(10); + let prefix = Bytes::from("/config/"); + + { + let _watcher = registry.register_prefix(prefix.clone()).unwrap(); + assert_eq!(registry.prefix_watcher_count(&prefix), 1); + } + + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(registry.prefix_watcher_count(&prefix), 0); +} + +// --- revision field --- + +/// #300: The revision field (= Raft applied index) is the client's only +/// mechanism to detect event gaps after a CANCELED notification. Without it, +/// the client cannot tell the Read API "give me the state as of revision N" +/// and must perform a full blind re-read with no consistency anchor. +#[tokio::test] +async fn test_revision_field_passed_through_to_watcher() { + let (broadcast_tx, registry, _handle) = setup_watch_system(10); + let key = Bytes::from("/config/host"); + let mut watcher = registry.register(key.clone()).unwrap(); + + tokio::time::sleep(Duration::from_millis(50)).await; + + broadcast_tx + .send(WatchEvent { + key: key.clone(), + value: Bytes::from("val"), + event_type: WatchEventType::Put as i32, + error: 0, + revision: 42, + }) + .unwrap(); + + let received = timeout(Duration::from_millis(100), watcher.receiver_mut().recv()) + .await + .expect("timeout") + .expect("closed"); + assert_eq!( + received.revision, 42, + "revision must be passed through unchanged" + ); +} + +// --- max_watcher_count limit --- + +/// #300: The dispatcher runs in a single task and does O(N) work per event +/// across all matched watchers. Without a hard cap, an unbounded number of +/// watchers degrades write latency for all clients. The limit forces operators +/// to make an explicit capacity decision rather than discovering the ceiling +/// in production under load. +#[tokio::test] +async fn test_max_watcher_count_rejects_registration_when_exceeded() { + let (_, registry, _handle) = setup_watch_system_with_max(10, 2); + let key = Bytes::from("/config/host"); + + let _h1 = registry.register(key.clone()).unwrap(); + let _h2 = registry.register(key.clone()).unwrap(); + let h3 = registry.register(key.clone()); + + assert!( + h3.is_err(), + "register must fail when max_watcher_count=2 is reached" + ); +} + +/// #300: prefix watchers consume the same shared cap as exact watchers. +/// Verifies that `register_prefix` is rejected once `max_watcher_count` is hit. +#[tokio::test] +async fn test_prefix_watcher_cap_enforced() { + let (_, registry, _handle) = setup_watch_system_with_max(10, 2); + + let _h1 = registry.register_prefix(Bytes::from("/a/")).unwrap(); + let _h2 = registry.register_prefix(Bytes::from("/b/")).unwrap(); + let h3 = registry.register_prefix(Bytes::from("/c/")); + + assert!( + h3.is_err(), + "register_prefix must fail when max_watcher_count=2 is reached" + ); +} + +/// #300: exact and prefix watchers share a single cap counter. +/// After 1 exact + 1 prefix fill the cap, a third registration of either +/// kind must be rejected. +#[tokio::test] +async fn test_mixed_exact_and_prefix_watchers_share_cap() { + let (_, registry, _handle) = setup_watch_system_with_max(10, 2); + + let _h1 = registry.register(Bytes::from("/exact/key")).unwrap(); + let _h2 = registry.register_prefix(Bytes::from("/prefix/")).unwrap(); + + let h3_exact = registry.register(Bytes::from("/exact/other")); + assert!( + h3_exact.is_err(), + "exact register must fail after mixed cap is reached" + ); + + let h3_prefix = registry.register_prefix(Bytes::from("/prefix2/")); + assert!( + h3_prefix.is_err(), + "prefix register must fail after mixed cap is reached" + ); +} + +/// #300 / fix: the TOCTOU fix in do_register must prevent the watcher count +/// from ever exceeding max_watcher_count under concurrent registration. +/// +/// The old load-check-then-fetch_add pattern lets N threads all pass the check +/// before any increment, so all N insert and the count overshoots. +/// The reserve-first pattern (fetch_add → check → rollback if over) is race-free. +#[tokio::test] +async fn test_concurrent_registrations_never_exceed_cap() { + let max = 5usize; + let (_, registry, _handle) = setup_watch_system_with_max(10, max); + let registry = Arc::clone(®istry); + + // Launch 4× more tasks than the cap; only exactly `max` must succeed. + let handles: Vec<_> = (0..20) + .map(|i| { + let registry = Arc::clone(®istry); + tokio::spawn(async move { + let key = Bytes::from(format!("/key/{i}")); + registry.register(key).is_ok() + }) + }) + .collect(); + + let mut success_count = 0usize; + for handle in handles { + if handle.await.unwrap() { + success_count += 1; + } + } + + assert_eq!( + success_count, max, + "exactly max_watcher_count registrations must succeed under concurrent load, got {success_count}" + ); +} diff --git a/d-engine-core/src/watch/mod.rs b/d-engine-core/src/watch/mod.rs index 259b369d..d78a11fb 100644 --- a/d-engine-core/src/watch/mod.rs +++ b/d-engine-core/src/watch/mod.rs @@ -83,10 +83,10 @@ //! use d_engine_core::watch::WatchConfig; //! //! let config = WatchConfig { -//! enabled: true, //! event_queue_size: 2000, -//! watcher_buffer_size: 20, +//! watcher_buffer_size: 256, //! enable_metrics: true, +//! max_watcher_count: 5000, //! }; //! ``` //! @@ -123,6 +123,7 @@ mod manager; mod manager_test; pub use manager::WatchDispatcher; +pub use manager::WatchError; pub use manager::WatchEvent; pub use manager::WatchEventType; pub use manager::WatchRegistry; @@ -143,5 +144,6 @@ pub(crate) fn make_cancel_event(key: bytes::Bytes) -> WatchEvent { value: bytes::Bytes::new(), event_type: WatchEventType::Canceled as i32, error: ErrorCode::WatchBufferOverflow as i32, + revision: 0, } } diff --git a/d-engine-proto/proto/client/client_api.proto b/d-engine-proto/proto/client/client_api.proto index d37ff1b9..9e6981d6 100644 --- a/d-engine-proto/proto/client/client_api.proto +++ b/d-engine-proto/proto/client/client_api.proto @@ -134,15 +134,19 @@ enum WatchEventType { WATCH_EVENT_TYPE_CANCELED = 2; } -// Request to watch for changes on a specific key -// -// In v1, only exact key matching is supported. -// Prefix watching may be added in future versions. +// Request to watch for changes on a key or key prefix. message WatchRequest { uint32 client_id = 1; - // Exact key to watch for changes + // Key to watch. + // prefix=false: exact key match. + // prefix=true: prefix match — key must start with "/" and end with "/". + // e.g. "/config/" watches all keys under /config/. bytes key = 2; + + // When true, key is treated as a path prefix (must end with "/"). + // Default false preserves backwards-compatible exact-match behaviour. + bool prefix = 3; } // Response containing a watch event notification @@ -158,6 +162,11 @@ message WatchResponse { // Error information if watch failed d_engine.error.ErrorCode error = 4; + + // Raft applied index at the time this event was produced. + // Monotonically increasing. Clients use this as an anchor after receiving + // CANCELED: re-read state, then re-register watching from revision+1. + uint64 revision = 5; } service RaftClientService { diff --git a/d-engine-proto/src/generated/d_engine.client.rs b/d-engine-proto/src/generated/d_engine.client.rs index 3ed70528..124640a1 100644 --- a/d-engine-proto/src/generated/d_engine.client.rs +++ b/d-engine-proto/src/generated/d_engine.client.rs @@ -143,18 +143,22 @@ pub struct MembershipSnapshot { #[prost(uint64, tag = "3")] pub committed_index: u64, } -/// Request to watch for changes on a specific key -/// -/// In v1, only exact key matching is supported. -/// Prefix watching may be added in future versions. +/// Request to watch for changes on a key or key prefix. #[derive(serde::Serialize, serde::Deserialize)] #[derive(Clone, PartialEq, ::prost::Message)] pub struct WatchRequest { #[prost(uint32, tag = "1")] pub client_id: u32, - /// Exact key to watch for changes + /// Key to watch. + /// prefix=false: exact key match. + /// prefix=true: prefix match — key must start with "/" and end with "/". + /// e.g. "/config/" watches all keys under /config/. #[prost(bytes = "bytes", tag = "2")] pub key: ::prost::bytes::Bytes, + /// When true, key is treated as a path prefix (must end with "/"). + /// Default false preserves backwards-compatible exact-match behaviour. + #[prost(bool, tag = "3")] + pub prefix: bool, } /// Response containing a watch event notification #[derive(serde::Serialize, serde::Deserialize)] @@ -172,6 +176,11 @@ pub struct WatchResponse { /// Error information if watch failed #[prost(enumeration = "super::error::ErrorCode", tag = "4")] pub error: i32, + /// Raft applied index at the time this event was produced. + /// Monotonically increasing. Clients use this as an anchor after receiving + /// CANCELED: re-read state, then re-register watching from revision+1. + #[prost(uint64, tag = "5")] + pub revision: u64, } /// Read consistency policy for controlling read operation guarantees /// diff --git a/d-engine-server/Cargo.toml b/d-engine-server/Cargo.toml index d8847061..64c6049e 100644 --- a/d-engine-server/Cargo.toml +++ b/d-engine-server/Cargo.toml @@ -26,6 +26,9 @@ rocksdb = ["dep:rocksdb"] # Experimental: Watch API for key change notifications # WARNING: This feature is under active development and not production-ready watch = ["d-engine-core/watch"] +# Re-exports StateMachine and StorageEngine test suites via `d_engine::state_machine_test` / +# `d_engine::storage_engine_test`. Enable this in `[dev-dependencies]` only. +__test_support = ["d-engine-core/__test_support"] [dependencies] d-engine-proto = { workspace = true } diff --git a/d-engine-server/benches/state_machine.rs b/d-engine-server/benches/state_machine.rs index 5b4c22cd..da4ad775 100644 --- a/d-engine-server/benches/state_machine.rs +++ b/d-engine-server/benches/state_machine.rs @@ -87,7 +87,7 @@ fn register_watchers( let mut handles = Vec::with_capacity(count); for i in 0..count { let key = format!("{key_prefix}{i}"); - handles.push(registry.register(key.into())); + handles.push(registry.register(key.into()).expect("register watcher")); } handles } @@ -353,6 +353,7 @@ fn bench_apply_with_1_watcher(c: &mut Criterion) { value: insert.value.clone(), event_type: d_engine_proto::client::WatchEventType::Put as i32, error: 0, + revision: 0, }; let _ = broadcast_tx.send(event); } @@ -362,6 +363,7 @@ fn bench_apply_with_1_watcher(c: &mut Criterion) { value: bytes::Bytes::new(), event_type: d_engine_proto::client::WatchEventType::Delete as i32, error: 0, + revision: 0, }; let _ = broadcast_tx.send(event); } @@ -371,6 +373,7 @@ fn bench_apply_with_1_watcher(c: &mut Criterion) { value: cas.new_value.clone(), event_type: d_engine_proto::client::WatchEventType::Put as i32, error: 0, + revision: 0, }; let _ = broadcast_tx.send(event); } @@ -414,6 +417,7 @@ fn bench_apply_with_10_watchers(c: &mut Criterion) { value: insert.value.clone(), event_type: d_engine_proto::client::WatchEventType::Put as i32, error: 0, + revision: 0, }; let _ = broadcast_tx.send(event); } @@ -423,6 +427,7 @@ fn bench_apply_with_10_watchers(c: &mut Criterion) { value: bytes::Bytes::new(), event_type: d_engine_proto::client::WatchEventType::Delete as i32, error: 0, + revision: 0, }; let _ = broadcast_tx.send(event); } @@ -432,6 +437,7 @@ fn bench_apply_with_10_watchers(c: &mut Criterion) { value: cas.new_value.clone(), event_type: d_engine_proto::client::WatchEventType::Put as i32, error: 0, + revision: 0, }; let _ = broadcast_tx.send(event); } @@ -475,6 +481,7 @@ fn bench_apply_with_100_watchers(c: &mut Criterion) { value: insert.value.clone(), event_type: d_engine_proto::client::WatchEventType::Put as i32, error: 0, + revision: 0, }; let _ = broadcast_tx.send(event); } @@ -484,6 +491,7 @@ fn bench_apply_with_100_watchers(c: &mut Criterion) { value: bytes::Bytes::new(), event_type: d_engine_proto::client::WatchEventType::Delete as i32, error: 0, + revision: 0, }; let _ = broadcast_tx.send(event); } @@ -493,6 +501,7 @@ fn bench_apply_with_100_watchers(c: &mut Criterion) { value: cas.new_value.clone(), event_type: d_engine_proto::client::WatchEventType::Put as i32, error: 0, + revision: 0, }; let _ = broadcast_tx.send(event); } @@ -519,7 +528,7 @@ fn bench_watch_e2e_latency(c: &mut Criterion) { let value = Bytes::from("test_value"); // Register a watcher - let mut watcher = registry.register(key.clone()); + let mut watcher = registry.register(key.clone()).expect("register watcher"); // Measure time from broadcast to receive let start = tokio::time::Instant::now(); @@ -529,7 +538,8 @@ fn bench_watch_e2e_latency(c: &mut Criterion) { key: key.clone(), value: value.clone(), event_type: d_engine_proto::client::WatchEventType::Put as i32, - error: 0, // No error + error: 0, + revision: 0, }; let _ = broadcast_tx.send(event); diff --git a/d-engine-server/src/api/embedded_client.rs b/d-engine-server/src/api/embedded_client.rs index 3662eae2..97cad936 100644 --- a/d-engine-server/src/api/embedded_client.rs +++ b/d-engine-server/src/api/embedded_client.rs @@ -470,7 +470,34 @@ impl EmbeddedClient { })?; let key_bytes = Bytes::copy_from_slice(key.as_ref()); - Ok(registry.register(key_bytes)) + registry.register(key_bytes).map_err(|e| ClientApiError::Business { + code: ErrorCode::Uncategorized, + message: e.to_string(), + required_action: None, + }) + } + + /// Register a prefix watcher — notified on every key under the given path prefix. + /// + /// `prefix` must start with '/' and end with '/'. E.g. `b"/services/"` watches + /// all keys whose path begins with `/services/`. + #[cfg(feature = "watch")] + pub fn watch_prefix( + &self, + prefix: impl AsRef<[u8]>, + ) -> ClientApiResult { + let registry = self.watch_registry.as_ref().ok_or_else(|| ClientApiError::Business { + code: ErrorCode::Uncategorized, + message: "Watch feature disabled (WatchRegistry not initialized)".to_string(), + required_action: None, + })?; + + let prefix_bytes = Bytes::copy_from_slice(prefix.as_ref()); + registry.register_prefix(prefix_bytes).map_err(|e| ClientApiError::Business { + code: ErrorCode::Uncategorized, + message: e.to_string(), + required_action: None, + }) } /// Internal helper: Get cluster membership via ClusterConf event diff --git a/d-engine-server/src/lib.rs b/d-engine-server/src/lib.rs index ef011ab1..25f0524a 100644 --- a/d-engine-server/src/lib.rs +++ b/d-engine-server/src/lib.rs @@ -142,6 +142,10 @@ pub use d_engine_core::client::{ClientApi, ClientApiError, ClientApiResult}; /// /// Implement this trait to create your own storage engine. pub use d_engine_core::{StateMachine, StorageEngine}; +/// Error code discriminator returned by [`ClientApiError::code()`]. +/// Use this to pattern-match error categories without inspecting message strings: +/// `e.code() == ErrorCode::NotLeader` +pub use d_engine_proto::error::ErrorCode; pub use node::Node; pub use node::NodeBuilder; // Re-export storage implementations @@ -152,6 +156,16 @@ pub use d_engine_core::ApplyResult; #[cfg(feature = "rocksdb")] pub use storage::{RocksDBStateMachine, RocksDBStorageEngine, RocksDBUnifiedEngine}; +// -------------------- Watch API Types -------------------- +/// Watch event types and handles — available when the `watch` feature is enabled. +/// +/// Use these when implementing embedded watch handlers: +/// ```rust,ignore +/// use d_engine::{WatchEventType, WatcherHandle, WatchEvent}; +/// ``` +#[cfg(feature = "watch")] +pub use d_engine_core::{WatchError, WatchEvent, WatchEventType, WatcherHandle}; + // -------------------- Data Types -------------------- /// Common Raft protocol types @@ -183,6 +197,26 @@ mod utils; // ==================== Test Utilities ==================== +/// Standardized test suite for custom [`StateMachine`] implementations. +/// +/// Enable the `__test_support` feature in your `[dev-dependencies]` to access this module: +/// ```toml +/// [dev-dependencies] +/// d-engine = { version = "...", features = ["server", "__test_support"] } +/// ``` +#[cfg(feature = "__test_support")] +pub use d_engine_core::state_machine_test; + +/// Standardized test suite for custom [`StorageEngine`] implementations. +/// +/// Enable the `__test_support` feature in your `[dev-dependencies]` to access this module: +/// ```toml +/// [dev-dependencies] +/// d-engine = { version = "...", features = ["server", "__test_support"] } +/// ``` +#[cfg(feature = "__test_support")] +pub use d_engine_core::storage_engine_test; + /// Test utilities for d-engine-server /// /// This module is only available when running tests. 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 67965871..660b2617 100644 --- a/d-engine-server/src/network/grpc/grpc_raft_service.rs +++ b/d-engine-server/src/network/grpc/grpc_raft_service.rs @@ -12,6 +12,8 @@ use d_engine_core::RaftEvent; use d_engine_core::RaftOneshot; use d_engine_core::StreamResponseSender; use d_engine_core::TypeConfig; +#[cfg(feature = "watch")] +use d_engine_core::WatchError; use d_engine_proto::client::ClientReadRequest; use d_engine_proto::client::ClientResponse; use d_engine_proto::client::ClientWriteRequest; @@ -486,14 +488,23 @@ where Status::unavailable("Watch feature is disabled in server configuration") })?; + let is_prefix = watch_request.prefix; info!( node_id = self.node_id, key = ?key, + is_prefix, "Registering watch for key" ); - // Register watcher and get receiver - let handle = registry.register(key); + // Register watcher (exact or prefix) and get receiver + let handle = if is_prefix { + registry.register_prefix(key).map_err(|e| match e { + WatchError::LimitExceeded(_) => Status::resource_exhausted(e.to_string()), + WatchError::InvalidPrefix => Status::invalid_argument(e.to_string()), + })? + } else { + registry.register(key).map_err(|e| Status::resource_exhausted(e.to_string()))? + }; let (_watcher_id, _key, receiver) = handle.into_receiver(); // Convert mpsc::Receiver -> Boxed Stream with sentinel error on close diff --git a/d-engine-server/src/node/builder.rs b/d-engine-server/src/node/builder.rs index c45e01a3..32025ddb 100644 --- a/d-engine-server/src/node/builder.rs +++ b/d-engine-server/src/node/builder.rs @@ -351,8 +351,9 @@ where let (unregister_tx, unregister_rx) = mpsc::unbounded_channel(); // Create shared registry - let registry = Arc::new(WatchRegistry::new( + let registry = Arc::new(WatchRegistry::new_with_limits( node_config.raft.watch.watcher_buffer_size, + node_config.raft.watch.max_watcher_count, unregister_tx, )); diff --git a/d-engine-server/tests/watch_and_subscriptions/watch_events_embedded.rs b/d-engine-server/tests/watch_and_subscriptions/watch_events_embedded.rs index 5e552e10..360cc7da 100644 --- a/d-engine-server/tests/watch_and_subscriptions/watch_events_embedded.rs +++ b/d-engine-server/tests/watch_and_subscriptions/watch_events_embedded.rs @@ -419,3 +419,222 @@ watcher_buffer_size = 10 Ok(()) } + +// ==================== #300: Prefix watch integration tests ==================== + +/// #300: Prefix watcher must fire for every child key under the prefix, regardless +/// of which specific child key is written. Clients use this to watch an entire +/// namespace (e.g. "/services/") without registering per-key watchers. +#[tokio::test] +async fn test_embedded_prefix_watch_receives_child_key_events() +-> Result<(), Box> { + let (engine, _temp_dir) = setup_engine().await?; + + let mut watcher = engine.client().watch_prefix(b"/config/")?; + + let handle = tokio::spawn(async move { + let mut events = Vec::new(); + for _ in 0..2 { + if let Some(event) = watcher.receiver_mut().recv().await { + events.push(event); + } + } + events + }); + + sleep(Duration::from_millis(50)).await; + + engine.client().put(b"/config/host", b"localhost").await?; + engine.client().put(b"/config/port", b"8080").await?; + + let events = tokio::time::timeout(Duration::from_secs(5), handle).await??; + assert_eq!( + events.len(), + 2, + "prefix watcher must receive both child key events" + ); + + let keys: Vec<&[u8]> = events.iter().map(|e| e.key.as_ref()).collect(); + assert!( + keys.contains(&b"/config/host".as_ref()), + "must see /config/host" + ); + assert!( + keys.contains(&b"/config/port".as_ref()), + "must see /config/port" + ); + + engine.stop().await?; + Ok(()) +} + +/// #300: A key that shares the prefix string but is not a child (e.g. "/config" with no +/// trailing slash) must NOT be delivered to a "/config/" prefix watcher. The slash +/// boundary is the contract boundary for namespace isolation. +#[tokio::test] +async fn test_embedded_prefix_watch_does_not_fire_for_parent_key() +-> Result<(), Box> { + let (engine, _temp_dir) = setup_engine().await?; + + let mut prefix_watcher = engine.client().watch_prefix(b"/config/")?; + let mut exact_watcher = engine.client().watch(b"/config")?; + + sleep(Duration::from_millis(50)).await; + + engine.client().put(b"/config", b"v").await?; + + // Exact watcher must fire + let exact_event = + tokio::time::timeout(Duration::from_secs(3), exact_watcher.receiver_mut().recv()).await?; + assert!(exact_event.is_some(), "exact watcher must fire for /config"); + + // Prefix watcher must NOT fire within a short window + let no_event = tokio::time::timeout( + Duration::from_millis(200), + prefix_watcher.receiver_mut().recv(), + ) + .await; + assert!( + no_event.is_err(), + "prefix watcher /config/ must not fire for key /config (no trailing slash)" + ); + + engine.stop().await?; + Ok(()) +} + +/// #300: Root prefix "/" must match every key written (since all paths start with "/"). +/// Useful for administrative observers that need a full-namespace view. +#[tokio::test] +async fn test_embedded_root_prefix_watch_receives_all_keys() +-> Result<(), Box> { + let (engine, _temp_dir) = setup_engine().await?; + + let mut watcher = engine.client().watch_prefix(b"/")?; + + let handle = tokio::spawn(async move { + let mut events = Vec::new(); + for _ in 0..3 { + if let Some(event) = watcher.receiver_mut().recv().await { + events.push(event); + } + } + events + }); + + sleep(Duration::from_millis(50)).await; + + engine.client().put(b"/a", b"1").await?; + engine.client().put(b"/b/c", b"2").await?; + engine.client().put(b"/x/y/z", b"3").await?; + + let events = tokio::time::timeout(Duration::from_secs(5), handle).await??; + assert_eq!( + events.len(), + 3, + "root prefix watcher must see all three writes" + ); + + engine.stop().await?; + Ok(()) +} + +/// #300: `revision` in each WatchEvent must be a positive Raft applied index — +/// it increases monotonically with each committed write. Clients depend on this +/// to detect gaps and decide whether to re-sync. +#[tokio::test] +async fn test_embedded_watch_event_revision_is_positive_and_increases() +-> Result<(), Box> { + let (engine, _temp_dir) = setup_engine().await?; + + let mut watcher = engine.client().watch(b"/rev-test")?; + + let handle = tokio::spawn(async move { + let mut revisions = Vec::new(); + for _ in 0..2 { + if let Some(event) = watcher.receiver_mut().recv().await { + revisions.push(event.revision); + } + } + revisions + }); + + sleep(Duration::from_millis(50)).await; + + engine.client().put(b"/rev-test", b"first").await?; + engine.client().put(b"/rev-test", b"second").await?; + + let revisions = tokio::time::timeout(Duration::from_secs(5), handle).await??; + assert_eq!(revisions.len(), 2); + assert!( + revisions[0] > 0, + "revision must be a positive Raft index, got 0" + ); + assert!( + revisions[1] > revisions[0], + "revision must increase monotonically: {} -> {}", + revisions[0], + revisions[1] + ); + + engine.stop().await?; + Ok(()) +} + +/// #300: Prefix watcher handle must be automatically cleaned up when dropped, just +/// like exact-key watchers. No zombie entries left in the registry. +#[tokio::test] +async fn test_embedded_prefix_watch_handle_drop_cleanup() -> Result<(), Box> +{ + let (engine, _temp_dir) = setup_engine().await?; + + { + let _watcher = engine.client().watch_prefix(b"/services/")?; + } // ← drop triggers unregister + + sleep(Duration::from_millis(100)).await; + + // After drop, a fresh registration must receive events normally. + let mut fresh = engine.client().watch_prefix(b"/services/")?; + let handle = tokio::spawn(async move { fresh.receiver_mut().recv().await }); + + sleep(Duration::from_millis(50)).await; + + engine.client().put(b"/services/api", b"running").await?; + + let event = tokio::time::timeout(Duration::from_secs(3), handle).await??; + assert!( + event.is_some(), + "fresh prefix watcher must receive event after stale handle dropped" + ); + + engine.stop().await?; + Ok(()) +} + +/// #300 / fix: watch_prefix must reject prefixes that don't start and end with '/'. +/// `InvalidPrefix` maps to an immediate `Err` at the embedded-client layer; +/// no partial state must be left in the registry. +#[tokio::test] +async fn test_embedded_watch_prefix_invalid_format_returns_error() +-> Result<(), Box> { + let (engine, _temp_dir) = setup_engine().await?; + + // Missing trailing slash + let r1 = engine.client().watch_prefix(b"/config"); + assert!(r1.is_err(), "prefix without trailing slash must fail"); + + // Missing leading slash + let r2 = engine.client().watch_prefix(b"config/"); + assert!(r2.is_err(), "prefix without leading slash must fail"); + + // Empty string + let r3 = engine.client().watch_prefix(b""); + assert!(r3.is_err(), "empty prefix must fail"); + + // Valid prefix must still work (sanity check no side-effects from errors above) + let _valid = engine.client().watch_prefix(b"/config/")?; + + engine.stop().await?; + Ok(()) +} diff --git a/d-engine-server/tests/watch_and_subscriptions/watch_events_grpc_standalone.rs b/d-engine-server/tests/watch_and_subscriptions/watch_events_grpc_standalone.rs index d77ee95b..5a3358bf 100644 --- a/d-engine-server/tests/watch_and_subscriptions/watch_events_grpc_standalone.rs +++ b/d-engine-server/tests/watch_and_subscriptions/watch_events_grpc_standalone.rs @@ -13,6 +13,7 @@ use crate::common::start_node; use d_engine_client::ClientBuilder; use d_engine_core::ClientApi; use d_engine_proto::client::WatchResponse; +use d_engine_proto::error::ErrorCode; use d_engine_server::FileStateMachine; use d_engine_server::FileStorageEngine; use futures::StreamExt; @@ -348,3 +349,45 @@ async fn test_watch_node_crash_standalone_mode() -> Result<(), Box Result<(), Box> { + let (client, test_ctx, _temp_dir) = setup_standalone_cluster().await?; + + // Missing trailing slash — server must reject with INVALID_ARGUMENT + let result = client.watch_prefix(b"/config").await; + assert!( + result.is_err(), + "watch_prefix without trailing slash must fail" + ); + assert_eq!( + result.unwrap_err().code(), + ErrorCode::InvalidRequest, + "server must return INVALID_ARGUMENT for malformed prefix" + ); + + // Missing leading slash + let result2 = client.watch_prefix(b"config/").await; + assert!( + result2.is_err(), + "watch_prefix without leading slash must fail" + ); + assert_eq!( + result2.unwrap_err().code(), + ErrorCode::InvalidRequest, + "server must return INVALID_ARGUMENT for malformed prefix" + ); + + // A valid prefix must still work + let stream = client.watch_prefix(b"/config/").await; + assert!(stream.is_ok(), "valid prefix must succeed"); + drop(stream); + + test_ctx.shutdown().await?; + Ok(()) +} diff --git a/d-engine-server/tests/watch_and_subscriptions/watch_membership_embedded.rs b/d-engine-server/tests/watch_and_subscriptions/watch_membership_embedded.rs index 78150617..6c66499e 100644 --- a/d-engine-server/tests/watch_and_subscriptions/watch_membership_embedded.rs +++ b/d-engine-server/tests/watch_and_subscriptions/watch_membership_embedded.rs @@ -448,10 +448,10 @@ async fn test_watch_membership_zombie_warns_without_removal() // AppendEntries being sent to the stopped node 3, which is what feeds the // failure counter and eventually fires the "Zombie detected" warning. // - // Use 20s here instead of ELECTION_TIMEOUT (8s): with election_timeout_max=3000ms, - // split-vote livelock between nodes 1 and 2 can produce 3-4 election rounds - // (~9-12s worst case) before one node wins a stable majority on slow CI machines. - let deadline = tokio::time::Instant::now() + Duration::from_secs(20); + // Use 35s here: with election_timeout_max=3000ms, split-vote livelock between + // nodes 1 and 2 can produce multiple election rounds (~9-12s typical, up to ~25s + // observed on slow CI machines). 35s provides a safe upper bound. + let deadline = tokio::time::Instant::now() + Duration::from_secs(35); loop { if engine1.is_leader() || engine2.is_leader() { break; diff --git a/d-engine/Cargo.toml b/d-engine/Cargo.toml index af34f190..f406a994 100644 --- a/d-engine/Cargo.toml +++ b/d-engine/Cargo.toml @@ -33,6 +33,9 @@ client = ["dep:d-engine-client"] server = ["dep:d-engine-server"] rocksdb = ["server", "d-engine-server/rocksdb"] watch = ["server", "d-engine-server/watch"] +# Enables `d_engine::state_machine_test` and `d_engine::storage_engine_test`. +# Add to `[dev-dependencies]` when implementing and testing custom state machines or storage engines. +__test_support = ["server", "d-engine-server/__test_support"] [dependencies] d-engine-client = { workspace = true, optional = true } diff --git a/d-engine/src/docs/server_guide/customize-state-machine.md b/d-engine/src/docs/server_guide/customize-state-machine.md index 2b0b87d5..faeb9e5c 100644 --- a/d-engine/src/docs/server_guide/customize-state-machine.md +++ b/d-engine/src/docs/server_guide/customize-state-machine.md @@ -154,7 +154,7 @@ async fn test_custom_state_machine() -> Result<(), Error> { Use `run_performance_tests()` to detect performance regressions: ```rust,ignore -use d_engine_core::state_machine_test::{StateMachineBuilder, StateMachineTestSuite}; +use d_engine::state_machine_test::{StateMachineBuilder, StateMachineTestSuite}; #[tokio::test] #[ignore] // Run with: cargo test -- --ignored diff --git a/d-engine/src/docs/server_guide/watch-feature.md b/d-engine/src/docs/server_guide/watch-feature.md index 2fdb1f4c..8608db69 100644 --- a/d-engine/src/docs/server_guide/watch-feature.md +++ b/d-engine/src/docs/server_guide/watch-feature.md @@ -11,8 +11,9 @@ Edit `config/base/raft.toml`: ```toml [raft.watch] event_queue_size = 10240 # Global broadcast channel buffer -watcher_buffer_size = 10 # Per-watcher buffer +watcher_buffer_size = 256 # Per-watcher buffer (raised from 10 in v0.2.4) enable_metrics = false # Detailed logging (default: false) +max_watcher_count = 10000 # Hard cap on total active watchers (optional) ``` > **Note**: Watch feature is controlled by the `watch` feature flag in `Cargo.toml` at compile time. If compiled with the feature, it's always enabled. @@ -22,7 +23,7 @@ enable_metrics = false # Detailed logging (default: false) **Key Insight**: Watch works on **any node** (Leader or Follower) because events trigger after Raft consensus → `StateMachine.apply()`. ```rust,ignore -use d_engine_client::Client; +use d_engine::{Client, ClientApi}; use futures::StreamExt; // Multi-node connection (recommended for high availability) @@ -34,19 +35,27 @@ let client = Client::builder(vec![ .build() .await?; -// Watch connects to any available node -let mut stream = client.watch("my_key").await?; +// Exact-key watch — reacts to changes on a single key +let mut stream = client.watch("/config/timeout").await?; -// Receive events while let Some(event) = stream.next().await { match event { Ok(event) => { - println!("Event: {:?} Key: {:?} Value: {:?}", - event.event_type, event.key, event.value); + println!("revision={} type={:?} key={:?} value={:?}", + event.revision, event.event_type, event.key, event.value); } Err(e) => eprintln!("Watch error: {:?}", e), } } + +// Prefix watch — reacts to any key under a namespace +// prefix must start and end with '/' +let mut stream = client.watch_prefix("/services/payment/").await?; + +while let Some(event) = stream.next().await { + // event.key is the specific child key that changed, e.g. "/services/payment/node1" + // event.revision is the Raft applied index — use it to detect gaps on reconnect +} ``` ### Embedded Usage @@ -71,6 +80,74 @@ tokio::spawn(async move { // Or use into_receiver() for long-lived streams (disables auto-cleanup) let (id, key, receiver) = watcher.into_receiver(); + +// Prefix watch (embedded) — same API as exact-key, but registers on a namespace +let prefix_watcher = engine.client().watch_prefix(b"/services/payment/")?; +let (_, _, mut prefix_rx) = prefix_watcher.into_receiver(); +tokio::spawn(async move { + while let Some(event) = prefix_rx.recv().await { + // event.key is the specific child key that changed + // event.revision is the Raft applied index + println!("child key changed: {:?} revision={}", event.key, event.revision); + } +}); +``` + +## Prefix Watch + +Prefix watch fires for every key whose path begins with a given prefix. +The prefix must start and end with `/`, e.g. `/services/payment/`. + +### Slash-boundary semantics + +```text +prefix = "/services/payment/" + +FIRES for: + /services/payment/node1 ✅ child key + /services/payment/node2 ✅ child key + /services/payment/node1/sub ✅ deeper child + +DOES NOT fire for: + /services/payment ❌ missing trailing slash - exact key, not a child + /services/ ❌ different prefix level + /services/paymentextra/node ❌ different namespace +``` + +### Reconnection pattern + +On disconnect or `CANCELED` (buffer overflow), re-sync current state before re-watching: + +```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 + let mut watcher = client.watch_prefix("/services/payment/").await?; + + 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 + } + } + tokio::time::sleep(Duration::from_millis(500)).await; +} +``` + +### Revision field + +Every `WatchEvent` carries `revision: u64` — the Raft applied index when the write was committed. +Use it to detect gaps after reconnection: + +```rust,ignore +let last_revision = event.revision; +// After reconnect: if new events arrive with revision > last_revision + 1, +// events were dropped — trigger a full re-sync. ``` ## Configuration @@ -89,13 +166,19 @@ let (id, key, receiver) = watcher.into_receiver(); ### `watcher_buffer_size` - **Type**: usize -- **Default**: `10` +- **Default**: `256` (raised from 10 in v0.2.4 — reduces spurious CANCELED events under burst load) - **Description**: Per-watcher channel buffer. Each watcher gets its own mpsc channel. -- **Memory**: ~240 bytes per slot per watcher (10 slots × 100 watchers ≈ 240KB) +- **Memory**: ~240 bytes per slot per watcher (256 slots × 100 watchers ≈ 6MB) - **Tuning**: - - Fast consumers (< 1ms): 5-10 - - Normal consumers (1-10ms): 10-20 - - Slow consumers (> 10ms): 20-50 + - Fast consumers (< 1ms): 64-256 + - Normal consumers (1-10ms): 256-512 + - Slow consumers (> 10ms): 512-1024 + +### `max_watcher_count` + +- **Type**: usize +- **Default**: `i64::MAX` (effectively unlimited) +- **Description**: Hard cap on total active watchers (exact + prefix combined). `register()` and `watch_prefix()` return an error once this limit is reached. Set to a finite value to prevent runaway watcher growth in multi-tenant deployments. ### `enable_metrics` @@ -111,7 +194,7 @@ let (id, key, receiver) = watcher.into_receiver(); - **Order**: FIFO per key (events from StateMachine apply order) - **Dropped Events**: When a per-watcher buffer overflows, the watcher receives a `CANCELED` sentinel event and is forcibly unregistered - **Lagging**: Broadcast channel drops oldest events when receivers are slow (global buffer) -- **CANCELED Event**: `WatchEventType::Canceled` with `ErrorCode::WATCH_BUFFER_OVERFLOW` signals that the watcher was forcibly terminated. Client must re-sync via Read API and re-register. +- **CANCELED Event**: `WatchEventType::CANCELED` with `ErrorCode::WATCH_BUFFER_OVERFLOW` signals that the watcher was forcibly terminated. Client must re-sync via Read API and re-register. ### Lifecycle @@ -119,11 +202,13 @@ let (id, key, receiver) = watcher.into_receiver(); - **Cleanup**: Watcher is automatically unregistered when handle is dropped - **No History**: Watch only receives future events (not historical changes) -### Limitations (v1) +### Limitations -- **Exact Key Match Only**: Prefix/range watch not supported -- **No Persistence**: Watchers lost on server restart -- **No Replay**: Cannot receive historical events +- **No Persistence**: Watchers lost on server restart — re-register on reconnect +- **No Replay**: Watch delivers future events only; use the Read API to re-sync current state after reconnect +- **Prefix format**: Prefix must start and end with `/` — arbitrary byte-range watch is not supported +- **No prev_kv**: Events do not include the previous value (planned for a future release) +- **No heartbeat**: The stream is silent when no events occur — implement an application-level keepalive if needed (planned for a future release) ## Performance @@ -207,7 +292,7 @@ watch(b"config:feature_flags") ### `WATCH_BUFFER_OVERFLOW` (ErrorCode 5001) - **Cause**: Per-watcher channel full — the client is consuming events too slowly -- **Detection**: Receive an event where `event_type == WatchEventType::Canceled` and `error == ErrorCode::WATCH_BUFFER_OVERFLOW` +- **Detection**: Receive an event where `event_type == WatchEventType::CANCELED` and `error == ErrorCode::WATCH_BUFFER_OVERFLOW` - **Effect**: The watcher is forcibly unregistered server-side; no further events will be delivered on this stream - **Solution**: 1. Re-sync state via the Read API to get the current value @@ -221,7 +306,7 @@ watch(b"config:feature_flags") 3. **Idempotent Handlers**: Handle duplicate events gracefully (at-most-once delivery) 4. **Buffer Tuning**: Monitor lagged events and adjust buffer sizes accordingly 5. **Reconnect Logic**: Implement automatic reconnection on stream errors -6. **Handle CANCELED**: Always check for `WatchEventType::Canceled` — treat it as a forced disconnect. Re-sync via Read API then re-register the watch before processing further events. +6. **Handle CANCELED**: Always check for `WatchEventType::CANCELED` — treat it as a forced disconnect. Re-sync via Read API then re-register the watch before processing further events. ## Watch Reliability and Reconnection @@ -468,7 +553,7 @@ while let Some(event) = stream.next().await { ```text ┌─────────────┐ -│ StateMachine│ +│ StateMachine│ revision = Raft applied index │ apply() │ └──────┬──────┘ │ broadcast::send (fire-and-forget, <10ns) @@ -481,13 +566,16 @@ while let Some(event) = stream.next().await { └──────┬──────────────────┘ │ subscribe ▼ -┌─────────────────────────┐ -│ WatchDispatcher │ -│ (tokio task) │ -│ - Recv from broadcast │ -│ - Match key in DashMap │ -│ - Send to mpsc │ -└──────┬──────────────────┘ +┌──────────────────────────────────────────────┐ +│ WatchDispatcher (single tokio task) │ +│ │ +│ 1. exact DashMap> │ +│ O(1) lookup per event │ +│ │ +│ 2. prefix DashMap> │ +│ prefix_segments(key) → O(depth) lookups │ +│ depth = number of '/' in key │ +└──────┬───────────────────────────────────────┘ │ ├─► Per-Watcher mpsc Channel ──► Embedded Client (Rust struct) │ @@ -497,8 +585,9 @@ while let Some(event) = stream.next().await { **Key Design Points**: - **StateMachine Decoupling**: Broadcast is fire-and-forget, never blocks apply -- **Unified Path**: Embedded and Standalone share same WatchRegistry/Dispatcher -- **Lock-Free**: DashMap for concurrent registration, no global mutex +- **Two DashMaps**: Exact and prefix watchers are stored separately for O(1) exact lookup and O(depth) prefix dispatch — no linear scan over all registered prefixes +- **Unified Path**: Embedded and Standalone share the same WatchRegistry/Dispatcher +- **Lock-Free**: DashMap sharding, no global mutex ## See Also diff --git a/examples/quick-start-standalone/go.mod b/examples/quick-start-standalone/go.mod index bb49aeba..79b1ef9c 100644 --- a/examples/quick-start-standalone/go.mod +++ b/examples/quick-start-standalone/go.mod @@ -1,20 +1,18 @@ module github.com/deventlab/d-engine/examples/quick-start-standalone -go 1.24.0 - -toolchain go1.24.5 +go 1.25.0 replace github.com/deventlab/d-engine/proto => ../../d-engine-proto/go require ( github.com/deventlab/d-engine/proto v0.0.0 - google.golang.org/grpc v1.78.0 + google.golang.org/grpc v1.80.0 ) require ( - golang.org/x/net v0.47.0 // indirect - golang.org/x/sys v0.38.0 // indirect - golang.org/x/text v0.31.0 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20251029180050-ab9386a59fda // indirect - google.golang.org/protobuf v1.36.10 // indirect + golang.org/x/net v0.53.0 // indirect + golang.org/x/sys v0.43.0 // indirect + golang.org/x/text v0.36.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260406210006-6f92a3bedf2d // indirect + google.golang.org/protobuf v1.36.11 // indirect ) diff --git a/examples/quick-start-standalone/go.sum b/examples/quick-start-standalone/go.sum index 8a84e18a..adb3ad1c 100644 --- a/examples/quick-start-standalone/go.sum +++ b/examples/quick-start-standalone/go.sum @@ -1,3 +1,5 @@ +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= @@ -10,27 +12,27 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= -go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8= -go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM= -go.opentelemetry.io/otel/metric v1.38.0 h1:Kl6lzIYGAh5M159u9NgiRkmoMKjvbsKtYRwgfrA6WpA= -go.opentelemetry.io/otel/metric v1.38.0/go.mod h1:kB5n/QoRM8YwmUahxvI3bO34eVtQf2i4utNVLr9gEmI= -go.opentelemetry.io/otel/sdk v1.38.0 h1:l48sr5YbNf2hpCUj/FoGhW9yDkl+Ma+LrVl8qaM5b+E= -go.opentelemetry.io/otel/sdk v1.38.0/go.mod h1:ghmNdGlVemJI3+ZB5iDEuk4bWA3GkTpW+DOoZMYBVVg= -go.opentelemetry.io/otel/sdk/metric v1.38.0 h1:aSH66iL0aZqo//xXzQLYozmWrXxyFkBJ6qT5wthqPoM= -go.opentelemetry.io/otel/sdk/metric v1.38.0/go.mod h1:dg9PBnW9XdQ1Hd6ZnRz689CbtrUp0wMMs9iPcgT9EZA= -go.opentelemetry.io/otel/trace v1.38.0 h1:Fxk5bKrDZJUH+AMyyIXGcFAPah0oRcT+LuNtJrmcNLE= -go.opentelemetry.io/otel/trace v1.38.0/go.mod h1:j1P9ivuFsTceSWe1oY+EeW3sc+Pp42sO++GHkg4wwhs= -golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= -golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= -golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= -golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= -golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= -gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= -gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= -google.golang.org/genproto/googleapis/rpc v0.0.0-20251029180050-ab9386a59fda h1:i/Q+bfisr7gq6feoJnS/DlpdwEL4ihp41fvRiM3Ork0= -google.golang.org/genproto/googleapis/rpc v0.0.0-20251029180050-ab9386a59fda/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= -google.golang.org/grpc v1.78.0 h1:K1XZG/yGDJnzMdd/uZHAkVqJE+xIDOcmdSFZkBUicNc= -google.golang.org/grpc v1.78.0/go.mod h1:I47qjTo4OKbMkjA/aOOwxDIiPSBofUtQUI5EfpWvW7U= -google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= -google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48= +go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8= +go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0= +go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs= +go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18= +go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE= +go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8= +go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= +go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI= +go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA= +golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA= +golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs= +golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= +golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg= +golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260406210006-6f92a3bedf2d h1:wT2n40TBqFY6wiwazVK9/iTWbsQrgk5ZfCSVFLO9LQA= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260406210006-6f92a3bedf2d/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM= +google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= diff --git a/examples/service-discovery-embedded/Cargo.toml b/examples/service-discovery-embedded/Cargo.toml index a03c1960..f46d9f9e 100644 --- a/examples/service-discovery-embedded/Cargo.toml +++ b/examples/service-discovery-embedded/Cargo.toml @@ -10,7 +10,6 @@ path = "server.rs" [dependencies] d-engine = { path = "../../d-engine", features = ["watch", "rocksdb"] } -d-engine-core = { path = "../../d-engine-core", features = ["watch"] } tokio = { version = "1.36.0", features = ["full"] } tracing = "0.1" diff --git a/examples/service-discovery-embedded/README.md b/examples/service-discovery-embedded/README.md index 7ba0ef31..408f5ccb 100644 --- a/examples/service-discovery-embedded/README.md +++ b/examples/service-discovery-embedded/README.md @@ -1,71 +1,125 @@ -# Service Discovery - Embedded Mode +# Service Discovery — Embedded Mode -Demonstrates d-engine's **Watch API in embedded mode** — ultra-low latency change notifications through in-process communication. +Demonstrates d-engine's **Watch API in embedded mode** through two real-world patterns: +in-process exact-key config watch and a prefix-based live load-balancer registry. **What you'll learn:** -- Using Watch API with EmbeddedClient (no gRPC overhead) -- Sub-millisecond read and watch latency -- Embedded d-engine architecture patterns +- Demo 1 — Exact-key watch: sidecar-style config reload with zero network overhead +- Demo 2 — Prefix watch: API gateway routing table that updates in real time as nodes join and leave +- `EmbeddedEngine` lifecycle and `wait_ready()` +- Shared state between a background watch task and the main thread (`Arc>`) + +## Architecture + +```text +┌──────────────────────────────────────────────────────────┐ +│ Application Process │ +│ │ +│ ┌─────────────────────────────────────────────────┐ │ +│ │ EmbeddedEngine (single-node for demo) │ │ +│ │ ┌─────────────┐ ┌───────────────────────┐ │ │ +│ │ │EmbeddedClient│ │ WatchRegistry │ │ │ +│ │ │ put/delete │ │ exact DashMap O(1) │ │ │ +│ │ │ <0.1ms │ │ prefix DashMap O(d) │ │ │ +│ │ └─────────────┘ └──────────┬────────────┘ │ │ +│ └───────────────────────────────│─────────────────┘ │ +│ │ mpsc (μs latency) │ +│ ┌────────────────────────┴────────────────┐ │ +│ │ │ │ +│ ┌──────▼──────┐ ┌─────────────▼──┐ │ +│ │ exact-watch │ │ prefix-watch │ │ +│ │ tokio task │ │ tokio task │ │ +│ │ │ │ │ │ +│ │ /config/ │ │ Arc> │ │ +│ └─────────────┘ └────────────────┘ │ +└──────────────────────────────────────────────────────────┘ +``` -## Architecture (Consul-style) +## Quick Start +```bash +cargo run --bin server ``` -┌─────────────────────────────────────────────────────────┐ -│ Application Instance A (Leader) │ -│ ┌─────────────────────────────────────────────────┐ │ -│ │ d-engine (embedded) │ │ -│ │ ┌───────────────┐ ┌────────────────────────┐ │ │ -│ │ │EmbeddedClient │ │ Watch (in-process) │ │ │ -│ │ │ (<0.1ms read) │ │ (μs latency events) │ │ │ -│ │ └───────────────┘ └────────────────────────┘ │ │ -│ └─────────────────────────────────────────────────┘ │ -└─────────────────────────────────────────────────────────┘ - │ Raft Replication - ▼ -┌─────────────────────────────────────────────────────────┐ -│ Application Instance B (Follower) │ -│ ┌─────────────────────────────────────────────────┐ │ -│ │ d-engine (embedded) - same structure │ │ -│ └─────────────────────────────────────────────────┘ │ -└─────────────────────────────────────────────────────────┘ + +No cluster setup needed — the demo starts a single-node embedded engine automatically. + +## What the Demo Does + +### Demo 1 — Exact-key watch (config reload pattern) + +A background task watches `/config/payment-service/timeout`. +The main thread writes two updates; the task prints each change as it arrives. + +```text +[exact-watch] config changed: /config/payment-service/timeout = 30s (revision=1) +[exact-watch] config changed: /config/payment-service/timeout = 60s (revision=2) ``` -## Key Differences from Standalone +**Real-world use case**: a sidecar process that watches its own config key and reloads +settings without a restart — no polling, no config file reload loop. -| Aspect | Standalone (gRPC) | Embedded | -| ------------- | ----------------- | --------------- | -| Read latency | 1-5ms | <0.1ms | -| Watch latency | ~1ms | ~μs | -| Deployment | Separate cluster | App = Raft node | -| Complexity | Simpler | More complex | +### Demo 2 — Prefix watch (live load-balancer registry) -## Quick Start +A background task watches `/services/payment/` and maintains a `HashMap`. +The main thread simulates three nodes registering, one updating its endpoint (rolling +restart), and one deregistering (crash). The routing table is printed after every event. -```bash -cargo run --bin server +```text +[prefix-watch] node up: /services/payment/node1 → 10.0.0.1:8080 (revision=3) + routing table (1 nodes): + /services/payment/node1 → 10.0.0.1:8080 +... +[prefix-watch] node down: /services/payment/node1 (revision=6) + routing table (2 nodes): + /services/payment/node2 → 10.0.0.2:9090 + /services/payment/node3 → 10.0.0.3:8080 + +Final routing table (2 nodes): + /services/payment/node2 → 10.0.0.2:9090 + /services/payment/node3 → 10.0.0.3:8080 ``` -The demo will: +**Real-world use case**: an API gateway keeps a live routing table for a service namespace. +One prefix watcher replaces N per-node exact-key watchers — no matter how many nodes join +or leave, the cost stays constant. -1. Start embedded d-engine in single node -2. Register a service endpoint -3. Show how to read with EmbeddedClient -4. Demonstrate in-process watch notifications +## Key Concepts -## Key Concepts Demonstrated +| Concept | Exact-key | Prefix | +|---|---|---| +| API | `engine.client().watch(key)?` | `engine.client().watch_prefix(prefix)?` | +| Fires for | One specific key | All keys under the prefix | +| Prefix constraint | — | Must start and end with `/` | +| Typical use | Config reload, distributed lock | Load balancer, DNS-like registry | -### Primary: Embedded Watch API +### `into_receiver()` vs `receiver_mut()` + +```rust,ignore +// Short-lived: auto-cleanup when handle is dropped +let mut watcher = engine.client().watch(key)?; +watcher.receiver_mut().recv().await; + +// Long-lived background task: disable auto-cleanup, own the channel +let (id, key, mut rx) = watcher.into_receiver(); +tokio::spawn(async move { while let Some(e) = rx.recv().await { ... } }); +``` -- **EmbeddedClient** — In-process KV operations with <0.1ms latency -- **Watch notifications** — Microsecond-level event delivery -- **Zero serialization** — Direct memory access, no gRPC overhead +### `CANCELED` event -### Secondary: Embedded Architecture +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. -- Each app instance embeds d-engine (similar to Consul) -- Raft replication between embedded nodes -- Service discovery as a practical use case +## Key Differences from Standalone + +| Aspect | Standalone (gRPC) | Embedded | +| ------------- | ----------------- | --------------- | +| Watch latency | ~1ms | ~μs | +| Serialization | Protobuf | None (direct struct) | +| Deployment | Separate cluster | App = Raft node | +| Suitable for | Any language | Rust apps only | ## When to Use Embedded Mode @@ -75,4 +129,5 @@ The demo will: ## Related -- [Standalone Example](../service-discovery-standalone/) — gRPC client mode +- [Standalone Example](../service-discovery-standalone/) — gRPC client with `--prefix` flag +- [Watch Feature Guide](https://docs.rs/d-engine/latest/d_engine/docs/server_guide/watch_feature/index.html) diff --git a/examples/service-discovery-embedded/server.rs b/examples/service-discovery-embedded/server.rs index a2405083..5b1abbd7 100644 --- a/examples/service-discovery-embedded/server.rs +++ b/examples/service-discovery-embedded/server.rs @@ -2,19 +2,21 @@ //! //! Demonstrates using d-engine in embedded mode for service discovery. //! Features: -//! - In-process Watch API (zero network overhead) -//! - EmbeddedClient (zero serialization overhead) +//! - Exact-key Watch: react to changes on a single specific key +//! - Prefix Watch: maintain a live registry of all nodes in a service namespace +//! - In-process Watch API (zero network overhead, zero serialization overhead) //! - Automatic lifecycle management -use d_engine::EmbeddedEngine; -use d_engine_core::watch::WatchEventType; -use std::error::Error; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; use std::time::Duration; + +use d_engine::EmbeddedEngine; +use d_engine::WatchEventType; use tokio::signal; #[tokio::main] -async fn main() -> Result<(), Box> { - // Initialize logging +async fn main() -> Result<(), Box> { tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::from_default_env() @@ -24,120 +26,177 @@ async fn main() -> Result<(), Box> { println!("Starting embedded d-engine for service discovery...\n"); - // Start embedded engine with explicit config file - // This automatically handles node startup, storage creation, and background tasks let engine = EmbeddedEngine::start_with("d-engine.toml").await?; - // Wait for leader election (single-node cluster elects itself immediately) let leader = engine.wait_ready(Duration::from_secs(5)).await?; println!( - "✓ Cluster is ready: leader {} (term {})", + "✓ Cluster ready: leader {} (term {})\n", leader.leader_id, leader.term ); - // Get local client (zero-overhead) let client = engine.client(); - // --- DEMO: In-process Watch --- + // ----------------------------------------------------------------------- + // DEMO 1: Exact-key watch + // + // Use case: a sidecar process watching its own config key. + // One watcher, one key, reacts to PUT and DELETE on that specific key. + // ----------------------------------------------------------------------- - let service_key = "services/payment-service/node1"; - println!("\n=== Starting In-process Watcher ==="); - println!("Watching key: {service_key}"); + let config_key = b"/config/payment-service/timeout"; + println!("=== Demo 1: Exact-key Watch ==="); + println!("Key: {}", String::from_utf8_lossy(config_key)); - // Register watcher directly on the engine - let watcher = engine.client().watch(service_key)?; + let exact_watcher = engine.client().watch(config_key)?; + let (_, _, mut exact_rx) = exact_watcher.into_receiver(); - // Spawn a background task to process watch events - // This simulates the "Watcher" component running inside the same process tokio::spawn(async move { - // Get the event receiver from the handle - let (_, _, mut receiver) = watcher.into_receiver(); - - while let Some(event) = receiver.recv().await { - let value = String::from_utf8_lossy(&event.value); + while let Some(event) = exact_rx.recv().await { + let key = String::from_utf8_lossy(&event.key); match event.event_type { e if e == WatchEventType::Put as i32 => { + let value = String::from_utf8_lossy(&event.value); println!( - "\n[WATCHER] Service Updated: {} -> {}", - String::from_utf8_lossy(&event.key), - value + "[exact-watch] config changed: {key} = {value} (revision={})", + event.revision ); } e if e == WatchEventType::Delete as i32 => { println!( - "\n[WATCHER] Service Removed: {}", - String::from_utf8_lossy(&event.key) + "[exact-watch] config deleted: {key} (revision={})", + event.revision ); } - _ => { - eprintln!("[WATCHER] Unknown event type: {}", event.event_type); - } + _ => {} } } }); - // --- DEMO: Service Registration (Write) --- + tokio::time::sleep(Duration::from_millis(50)).await; - println!("\n=== Registering Service (Write) ==="); - let endpoint = "10.0.0.5:8080"; - println!("Registering: {service_key} -> {endpoint}"); - - // Perform local write (direct to Raft core) - client.put(service_key, endpoint).await?; - - // Verify write with linearizable read (guarantees we see the latest write) - match client.get_linearizable(service_key).await? { - Some(value) => { - let stored = String::from_utf8_lossy(&value); - println!("✓ Verified: {service_key} -> {stored}"); - assert_eq!(stored, endpoint, "Read value must match written value"); - } - None => { - panic!("❌ BUG: get_linearizable returned None immediately after PUT!"); - } - } + client.put(config_key, b"30s").await?; + tokio::time::sleep(Duration::from_millis(100)).await; - // Give time for watcher to print + client.put(config_key, b"60s").await?; tokio::time::sleep(Duration::from_millis(100)).await; - // --- DEMO: Service Update --- + // ----------------------------------------------------------------------- + // DEMO 2: Prefix watch — live load-balancer registry + // + // Use case: an API Gateway needs a live view of all payment-service nodes. + // When any node registers, updates its endpoint, or deregisters, the + // gateway's in-memory routing table updates instantly — no polling. + // + // One prefix watcher on "/services/payment/" replaces N per-node watchers. + // The registry is a shared HashMap updated by the background task and read + // by the main routing loop. + // ----------------------------------------------------------------------- + + let service_prefix = b"/services/payment/"; + println!("\n=== Demo 2: Prefix Watch — Live Load-Balancer Registry ==="); + println!("Namespace: {}", String::from_utf8_lossy(service_prefix)); + + // Shared registry: key → endpoint, readable from any thread + let registry: Arc, String>>> = Arc::new(Mutex::new(HashMap::new())); + let registry_bg = Arc::clone(®istry); + + let prefix_watcher = engine.client().watch_prefix(service_prefix)?; + let (_, _, mut prefix_rx) = prefix_watcher.into_receiver(); - println!("\n=== Updating Service ==="); - let new_endpoint = "10.0.0.5:9090"; - println!("Updating: {service_key} -> {new_endpoint}"); - client.put(service_key, new_endpoint).await?; + tokio::spawn(async move { + while let Some(event) = prefix_rx.recv().await { + let key_str = String::from_utf8_lossy(&event.key).to_string(); + let mut reg = registry_bg.lock().unwrap(); - // Verify update with linearizable read - match client.get_linearizable(service_key).await? { - Some(value) => { - let stored = String::from_utf8_lossy(&value); - println!("✓ Verified: {service_key} -> {stored}"); - assert_eq!(stored, new_endpoint, "Read value must match updated value"); - } - None => { - panic!("❌ BUG: get_linearizable returned None immediately after UPDATE!"); + match event.event_type { + e if e == WatchEventType::Put as i32 => { + let value = String::from_utf8_lossy(&event.value).to_string(); + println!( + "[prefix-watch] node up: {key_str} → {value} (revision={})", + event.revision + ); + reg.insert(event.key.to_vec(), value); + } + e if e == WatchEventType::Delete as i32 => { + println!( + "[prefix-watch] node down: {key_str} (revision={})", + event.revision + ); + reg.remove(event.key.as_ref()); + } + e if e == WatchEventType::Canceled as i32 => { + // Buffer overflow: registry may be stale. + // Production code: re-scan the prefix then re-watch (ticket #301). + println!( + "[prefix-watch] CANCELED — buffer overflow on {key_str}; registry may be stale" + ); + reg.clear(); + } + _ => {} + } + + // Print current routing table after every change + print_registry(®); } - } + }); + tokio::time::sleep(Duration::from_millis(50)).await; + + // Simulate three nodes coming online + println!("\n--- Three nodes registering ---"); + client + .put(b"/services/payment/node1", b"10.0.0.1:8080") + .await?; + client + .put(b"/services/payment/node2", b"10.0.0.2:8080") + .await?; + client + .put(b"/services/payment/node3", b"10.0.0.3:8080") + .await?; + tokio::time::sleep(Duration::from_millis(200)).await; + + // Simulate node2 updating its endpoint (rolling restart) + println!("\n--- node2 rolling restart: endpoint changes ---"); + client + .put(b"/services/payment/node2", b"10.0.0.2:9090") + .await?; tokio::time::sleep(Duration::from_millis(100)).await; - // --- DEMO: Service Unregistration --- - - println!("\n=== Unregistering Service ==="); - client.delete(service_key).await?; - + // Simulate node1 crashing (TTL expiry or explicit deregister) + println!("\n--- node1 deregisters (crash / graceful shutdown) ---"); + client.delete(b"/services/payment/node1").await?; tokio::time::sleep(Duration::from_millis(100)).await; - println!("\n=== Demo Complete ==="); - println!("Press Ctrl+C to exit..."); + // Show final registry state via a direct read (what routing would use) + { + let reg = registry.lock().unwrap(); + println!("\n✓ Final routing table ({} nodes):", reg.len()); + let mut entries: Vec<_> = reg.iter().collect(); + entries.sort_by_key(|(k, _)| (*k).clone()); + for (k, v) in entries { + println!(" {} → {}", String::from_utf8_lossy(k), v); + } + } - // Wait for shutdown signal + println!("\n=== Demo complete. Press Ctrl+C to exit. ==="); signal::ctrl_c().await?; - // Graceful shutdown println!("\nShutting down..."); engine.stop().await?; println!("Done"); Ok(()) } + +fn print_registry(reg: &HashMap, String>) { + if reg.is_empty() { + println!(" routing table: (empty)"); + } else { + println!(" routing table ({} nodes):", reg.len()); + let mut entries: Vec<_> = reg.iter().collect(); + entries.sort_by_key(|(k, _)| (*k).clone()); + for (k, v) in entries { + println!(" {} → {}", String::from_utf8_lossy(k), v); + } + } +} diff --git a/examples/service-discovery-standalone/Cargo.toml b/examples/service-discovery-standalone/Cargo.toml index 32ada4a5..86bde861 100644 --- a/examples/service-discovery-standalone/Cargo.toml +++ b/examples/service-discovery-standalone/Cargo.toml @@ -13,7 +13,7 @@ name = "watcher" path = "watcher.rs" [dependencies] -d-engine-client = { path = "../../d-engine-client", features = ["watch"] } +d-engine = { path = "../../d-engine", features = ["client", "watch"] } clap = { version = "4.0", features = ["derive"] } tokio = { version = "1.36.0", features = ["full"] } diff --git a/examples/service-discovery-standalone/Makefile b/examples/service-discovery-standalone/Makefile index afacdc25..c0219311 100644 --- a/examples/service-discovery-standalone/Makefile +++ b/examples/service-discovery-standalone/Makefile @@ -1,27 +1,55 @@ -.PHONY: help run-watcher register unregister check-cluster +.PHONY: help check-cluster \ + run-watcher run-watcher-prefix \ + register register-node2 register-node3 \ + unregister unregister-node2 help: ## Show this help message @echo 'Usage: make [target]' @echo '' - @echo 'Targets:' - @echo ' run-watcher Start the watcher client (reads & watches)' - @echo ' register Register a demo service (writes)' - @echo ' unregister Unregister the demo service (deletes)' - @echo ' check-cluster Check if d-engine cluster is reachable' - @echo '' @echo 'Prerequisites:' - @echo ' Ensure d-engine cluster is running (cd ../three-nodes-standalone && make start-cluster)' + @echo ' Start a cluster: cd ../three-nodes-standalone && make start-cluster' + @echo '' + @echo '--- Mode 1: exact-key watch ---' + @echo ' run-watcher Watch /services/payment/node1 (Terminal 1)' + @echo ' register Register payment/node1 -> 10.0.0.1:8080' + @echo ' unregister Deregister payment/node1' + @echo '' + @echo '--- Mode 2: prefix watch ---' + @echo ' run-watcher-prefix Watch all /services/payment/* (Terminal 1)' + @echo ' register-node2 Register payment/node2 -> 10.0.0.2:8080' + @echo ' register-node3 Register payment/node3 -> 10.0.0.3:8080' + @echo ' unregister-node2 Deregister payment/node2' check-cluster: @echo "Checking connectivity to 127.0.0.1:9081..." @nc -z 127.0.0.1 9081 || (echo "❌ Cluster not reachable. Please start three-nodes-standalone first." && exit 1) @echo "✅ Cluster reachable" -run-watcher: check-cluster ## Start the watcher - cargo run --bin watcher -- --key "services/api-gateway/node1" +# --------------------------------------------------------------------------- +# Mode 1 — exact-key watch +# --------------------------------------------------------------------------- + +run-watcher: check-cluster ## Watch /services/payment/node1 (exact key) + cargo run --bin watcher -- --key "/services/payment/node1" + +register: check-cluster ## Register payment/node1 -> 10.0.0.1:8080 + cargo run --bin admin -- register --name payment --instance node1 --endpoint "10.0.0.1:8080" + +unregister: check-cluster ## Deregister payment/node1 + cargo run --bin admin -- unregister --name payment --instance node1 + +# --------------------------------------------------------------------------- +# Mode 2 — prefix watch +# --------------------------------------------------------------------------- + +run-watcher-prefix: check-cluster ## Watch all /services/payment/* (prefix) + cargo run --bin watcher -- --key "/services/payment/" --prefix + +register-node2: check-cluster ## Register payment/node2 -> 10.0.0.2:8080 + cargo run --bin admin -- register --name payment --instance node2 --endpoint "10.0.0.2:8080" -register: check-cluster ## Register service: api-gateway -> 192.168.1.10:8080 - cargo run --bin admin -- register --name api-gateway --instance node1 --endpoint "192.168.1.10:8080" +register-node3: check-cluster ## Register payment/node3 -> 10.0.0.3:8080 + cargo run --bin admin -- register --name payment --instance node3 --endpoint "10.0.0.3:8080" -unregister: check-cluster ## Unregister service: api-gateway - cargo run --bin admin -- unregister --name api-gateway --instance node1 +unregister-node2: check-cluster ## Deregister payment/node2 + cargo run --bin admin -- unregister --name payment --instance node2 diff --git a/examples/service-discovery-standalone/README.md b/examples/service-discovery-standalone/README.md index 3522c6e0..42822329 100644 --- a/examples/service-discovery-standalone/README.md +++ b/examples/service-discovery-standalone/README.md @@ -1,82 +1,137 @@ -# Service Discovery Example +# Service Discovery — Standalone Mode -Demonstrates d-engine's **Watch API** through a practical service discovery use case. +Demonstrates d-engine's **Watch API** through a practical service discovery use case, +using the gRPC client against a running d-engine cluster. **What you'll learn:** -- How to use the Watch API for real-time change notifications -- Implementing the read-then-watch pattern -- Building reactive systems without polling +- Exact-key watch: react to changes on a single specific key +- Prefix watch: maintain a live in-memory service registry without per-node watchers +- Read-then-watch pattern: avoid the race between "current state" and "future events" +- Reconnection with `CANCELED` (buffer overflow) handling ## Architecture -``` +```text ┌────────────────────────────────────────────────────────┐ │ d-engine Cluster │ -│ (stores service registry: /services/{name}/{instance})│ -└────────────────────┬───────────────────────────────────┘ - │ - ┌──────────┴──────────┐ - │ │ - ┌─────▼─────┐ ┌─────▼─────┐ - │ Admin │ │ Watcher │ - │ (writes) │ │ (reads) │ - │ │ │ │ - │ Register/ │ │ Watch + │ - │ Unregister│ │ Cache │ - └───────────┘ └───────────┘ +│ (stores service registry: /services/{name}/{node}) │ +└──────────────────────┬─────────────────────────────────┘ + │ gRPC + ┌────────────┴────────────┐ + │ │ + ┌─────▼─────┐ ┌─────▼──────────────────┐ + │ admin │ │ watcher │ + │ (writes) │ │ │ + │ │ │ --key /services/payment/node1 │ + │ register │ │ exact-key watch │ + │ unregister│ │ ── or ── │ + └───────────┘ │ --key /services/payment/ │ + │ --prefix │ + │ prefix watch │ + │ (live registry HashMap)│ + └─────────────────────────┘ ``` ## Prerequisites -**Start the d-engine server cluster first** using the three-nodes-standalone example: +Start a d-engine node first: ```bash -# In another terminal, start at least one node cd examples/three-nodes-standalone -make start-cluster # or: make start-node-1 +make start-cluster # or: make start-node-1 ``` ## Quick Start +### Mode 1 — Exact-key watch + +Watches a single key. Prints current value on startup, then streams PUT/DELETE events. + ```bash -# Terminal 1: Start watcher (waits for changes) -cargo run --bin watcher -- --key "services/api-gateway/node1" +# Terminal 1: watch one specific node +make run-watcher + +# Terminal 2: register / update / deregister +make register +make unregister +``` -# Terminal 2: Register service -cargo run --bin admin -- register --name api-gateway --instance node1 --endpoint "192.168.1.10:8080" +Expected watcher output: -# Watcher terminal will show: [PUT] services/api-gateway/node1 = 192.168.1.10:8080 +```text +=== Current State === + /services/payment/node1 = (not found) -# Unregister service -cargo run --bin admin -- unregister --name api-gateway --instance node1 +=== Watching for Changes (Ctrl+C to exit) === -# Watcher terminal will show: [DELETE] services/api-gateway/node1 +[PUT ] /services/payment/node1 = 10.0.0.1:8080 (revision=3) +[DELETE] /services/payment/node1 (revision=4) ``` -## Components +### Mode 2 — Prefix watch (namespace observer) -| Binary | Purpose | -| --------- | ----------------------------------------- | -| `admin` | Registers/unregisters service endpoints | -| `watcher` | Watches for service changes via Watch API | +One watcher covers the entire `/services/payment/` namespace. +Every node that registers, updates, or deregisters fires an event. +The watcher maintains a live `HashMap` registry and prints it after each change. -## Key Concepts Demonstrated +```bash +# Terminal 1: watch the whole payment namespace +make run-watcher-prefix + +# Terminal 2: simulate cluster churn +make register # node1 +make register-node2 # node2 +make register-node3 # node3 +make unregister-node2 # node2 crashes +``` -### Primary: Watch API Usage +Expected watcher output: + +```text +[PUT ] /services/payment/node1 = 10.0.0.1:8080 (revision=3) + registry (1 nodes): + /services/payment/node1 → 10.0.0.1:8080 +[PUT ] /services/payment/node2 = 10.0.0.2:8080 (revision=4) + registry (2 nodes): + /services/payment/node1 → 10.0.0.1:8080 + /services/payment/node2 → 10.0.0.2:8080 +... +[DELETE] /services/payment/node2 (revision=6) + registry (2 nodes): + /services/payment/node1 → 10.0.0.1:8080 + /services/payment/node3 → 10.0.0.3:8080 +``` + +## Key Concepts + +| Concept | Exact-key (`watch`) | Prefix (`watch_prefix`) | +|---|---|---| +| 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) | + +### `revision` field -1. **Watch API** — Real-time change notifications without polling -2. **Read-then-Watch Pattern** — Read current state, then watch for changes -3. **Event Handling** — Process PUT/DELETE events reactively +Every event carries `revision: u64` — the Raft applied index when the write committed. +After reconnect, compare the first new event's revision to your last known revision: +if `new_revision > last_revision + 1`, events were dropped — trigger a full re-sync. -### Secondary: Service Discovery Pattern +### `CANCELED` event + +When a watcher's buffer overflows (server writes faster than client consumes), +d-engine sends a `CANCELED` sentinel and forcibly unregisters the watcher. +The prefix watch loop in this example handles it: clear the local registry and reconnect. + +## Components -- Using d-engine as a service registry -- EventualConsistency reads for fast lookups -- Key-value schema design for services +| Binary | Purpose | +|-----------|-------------------------------------------------| +| `admin` | Registers/unregisters service endpoints | +| `watcher` | Watches for service changes (exact or prefix) | -## Related Documentation +## Related -- Service Discovery Pattern (coming soon on docs.rs) +- [Embedded Mode Example](../service-discovery-embedded/) — in-process watch, zero gRPC overhead - [Watch Feature Guide](https://docs.rs/d-engine/latest/d_engine/docs/server_guide/watch_feature/index.html) -- [Read Consistency Guide](https://docs.rs/d-engine/latest/d_engine/docs/client_guide/read_consistency/index.html) diff --git a/examples/service-discovery-standalone/admin.rs b/examples/service-discovery-standalone/admin.rs index 4093352a..f02d23fe 100644 --- a/examples/service-discovery-standalone/admin.rs +++ b/examples/service-discovery-standalone/admin.rs @@ -5,8 +5,8 @@ use anyhow::Result; use clap::{Parser, Subcommand}; -use d_engine_client::Client; -use d_engine_client::ClientApi; +use d_engine::Client; +use d_engine::ClientApi; use std::time::Duration; #[derive(Parser)] @@ -75,8 +75,8 @@ async fn main() -> Result<()> { instance, endpoint, } => { - // Key format: services/{service_name}/{instance_id} - let key = format!("services/{name}/{instance}"); + // Key format: /services/{service_name}/{instance_id} + let key = format!("/services/{name}/{instance}"); client .put(&key, &endpoint) @@ -87,7 +87,7 @@ async fn main() -> Result<()> { } Commands::Unregister { name, instance } => { - let key = format!("services/{name}/{instance}"); + let key = format!("/services/{name}/{instance}"); client.delete(&key).await.map_err(|e| anyhow::anyhow!("Delete failed: {e:?}"))?; @@ -102,7 +102,7 @@ async fn main() -> Result<()> { println!("Workaround: Maintain an index key like 'services/{name}_index'"); // Try to read index if exists - let index_key = format!("services/{name}_index"); + let index_key = format!("/services/{name}_index"); let result = client.get(&index_key).await.map_err(|e| anyhow::anyhow!("Get failed: {e:?}"))?; if let Some(result) = result { diff --git a/examples/service-discovery-standalone/watcher.rs b/examples/service-discovery-standalone/watcher.rs index b064125b..c5af664a 100644 --- a/examples/service-discovery-standalone/watcher.rs +++ b/examples/service-discovery-standalone/watcher.rs @@ -1,32 +1,63 @@ //! Watcher client for service discovery example //! -//! Demonstrates the Watch API for receiving real-time service updates. -//! This is the "read path" in service discovery pattern. +//! Demonstrates both the exact-key Watch API and the prefix Watch API. +//! +//! # Exact-key watch (default) +//! +//! ```bash +//! cargo run --bin watcher -- --key /services/payment/node1 +//! ``` +//! +//! Watches a single specific key. Use this when you own exactly one resource +//! and need to react to its changes (e.g. your own config entry). +//! +//! # Prefix watch (--prefix) +//! +//! ```bash +//! cargo run --bin watcher -- --key /services/payment/ --prefix +//! ``` +//! +//! Watches an entire namespace. Every key that starts with the prefix fires an +//! event — new nodes joining, existing nodes updating their endpoint, nodes +//! deregistering. Use this to maintain a live service registry without +//! registering per-node watchers. +//! +//! # 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). + +use std::collections::HashMap; +use std::time::Duration; use anyhow::Result; use clap::Parser; -use d_engine_client::Client; -use d_engine_client::ClientApi; -use d_engine_client::protocol::{WatchEventType, WatchResponse}; +use d_engine::protocol::{WatchEventType, WatchResponse}; +use d_engine::{Client, ClientApi}; use futures::StreamExt; -use std::time::Duration; #[derive(Parser)] #[command(name = "watcher")] -#[command(about = "Watch for service changes in d-engine")] +#[command(about = "Watch for service changes in d-engine (exact key or prefix namespace)")] struct Cli { /// d-engine server endpoint #[arg(short, long, default_value = "http://127.0.0.1:9081")] endpoint: String, - /// Key to watch (e.g., "services/api-gateway/node1") + /// Key to watch (exact) or prefix to watch (with --prefix). + /// Prefix must start and end with '/', e.g. /services/payment/ #[arg(short, long)] key: String, + + /// Watch all keys under this prefix instead of an exact key. + /// When set, --key is treated as a path prefix. + #[arg(short, long, default_value_t = false)] + prefix: bool, } #[tokio::main] async fn main() -> Result<()> { - // Initialize logging tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::from_default_env() @@ -37,9 +68,14 @@ async fn main() -> Result<()> { let cli = Cli::parse(); println!("Connecting to: {}", cli.endpoint); - println!("Watching key: {}\n", cli.key); + if cli.prefix { + println!("Mode: prefix watch"); + println!("Namespace: {}\n", cli.key); + } else { + println!("Mode: exact key watch"); + println!("Key: {}\n", cli.key); + } - // Connect to d-engine cluster let client = Client::builder(vec![cli.endpoint.clone()]) .connect_timeout(Duration::from_secs(5)) .request_timeout(Duration::from_secs(30)) @@ -47,32 +83,45 @@ async fn main() -> Result<()> { .await .map_err(|e| anyhow::anyhow!("Failed to connect: {e:?}"))?; + if cli.prefix { + run_prefix_watch(&client, &cli.key).await + } else { + run_exact_watch(&client, &cli.key).await + } +} + +// --------------------------------------------------------------------------- +// Exact-key watch +// --------------------------------------------------------------------------- + +async fn run_exact_watch( + client: &Client, + key: &str, +) -> Result<()> { // Pattern: Read-then-Watch - // 1. First, read current value (if any) + // Read current value first so the caller sees the starting state before + // any events arrive. This avoids the race between "what is it now" and + // "what did it just change to". println!("=== Current State ==="); let current = client - .get_eventual(&cli.key) + .get_eventual(key) .await .map_err(|e| anyhow::anyhow!("Read failed: {e:?}"))?; - if let Some(result) = current { - println!(" {} = {}", cli.key, String::from_utf8_lossy(&result)); - } else { - println!(" {} = (not found)", cli.key); + match current { + Some(v) => println!(" {key} = {}", String::from_utf8_lossy(&v)), + None => println!(" {key} = (not found)"), } - // 2. Start watching for future changes println!("\n=== Watching for Changes (Ctrl+C to exit) ===\n"); let mut stream = client - .watch(&cli.key) + .watch(key) .await .map_err(|e| anyhow::anyhow!("Watch failed: {e:?}"))?; while let Some(event_result) = stream.next().await { match event_result { - Ok(response) => { - print_watch_event(&cli.key, &response); - } + Ok(response) => print_event(&response), Err(e) => { eprintln!("Watch error: {e:?}"); break; @@ -84,26 +133,139 @@ async fn main() -> Result<()> { Ok(()) } -fn print_watch_event( - key: &str, +// --------------------------------------------------------------------------- +// Prefix watch — live service registry (namespace observer pattern) +// --------------------------------------------------------------------------- +// +// Business scenario: an API Gateway needs a live view of all nodes belonging +// to /services/payment/. When node1 registers, updates its endpoint, or +// crashes (delete), the gateway reacts immediately — no polling required. +// +// 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). + +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 { + let mut stream = client + .watch_prefix(prefix) + .await + .map_err(|e| anyhow::anyhow!("Prefix watch failed: {e:?}"))?; + + println!("[connected] watching {prefix}"); + + while let Some(event_result) = stream.next().await { + match event_result { + Ok(response) => { + 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; + } + } + Err(e) => { + eprintln!("[stream error] {e:?} — reconnecting in 1s"); + tokio::time::sleep(Duration::from_secs(1)).await; + break; + } + } + } + + println!("[reconnecting] re-establishing prefix watch for {prefix}\n"); + tokio::time::sleep(Duration::from_millis(500)).await; + } +} + +/// Returns `true` if the event was `CANCELED` (caller should break and reconnect). +fn apply_event_to_registry( response: &WatchResponse, -) { + registry: &mut HashMap, String>, +) -> bool { + let key = response.key.to_vec(); + let key_str = String::from_utf8_lossy(&response.key); + let event_type = WatchEventType::try_from(response.event_type).ok(); + + match event_type { + Some(WatchEventType::Put) => { + let value = String::from_utf8_lossy(&response.value).to_string(); + println!( + "[PUT ] {key_str} = {value} (revision={})", + response.revision + ); + registry.insert(key, value); + false + } + Some(WatchEventType::Delete) => { + println!("[DELETE] {key_str} (revision={})", response.revision); + registry.remove(&key); + false + } + Some(WatchEventType::Canceled) => { + // Buffer overflow: missed events after this revision. + // Return true → caller breaks inner loop → reconnects. + println!( + "[CANCELED] buffer overflow — missed events after revision {}; reconnecting", + response.revision + ); + registry.clear(); + true + } + None => { + println!("[UNKNOWN] {key_str} (event_type={})", response.event_type); + false + } + } +} + +fn print_registry(registry: &HashMap, String>) { + if registry.is_empty() { + println!(" registry: (empty — waiting for registrations or reconnecting)"); + } else { + println!(" registry ({} nodes):", registry.len()); + let mut entries: Vec<_> = registry.iter().collect(); + entries.sort_by_key(|(k, _)| (*k).clone()); + for (k, v) in entries { + println!(" {} → {}", String::from_utf8_lossy(k), v); + } + } +} + +fn print_event(response: &WatchResponse) { let event_type = WatchEventType::try_from(response.event_type).ok(); + let key = String::from_utf8_lossy(&response.key); match event_type { Some(WatchEventType::Put) => { let value = String::from_utf8_lossy(&response.value); - println!("[PUT] {key} = {value}"); + println!("[PUT ] {key} = {value} (revision={})", response.revision); } Some(WatchEventType::Delete) => { - println!("[DELETE] {key}"); + println!("[DELETE] {key} (revision={})", response.revision); } Some(WatchEventType::Canceled) => { - println!("[CANCELED] {key} — watcher forcibly terminated (buffer overflow); re-sync and re-register"); + println!( + "[CANCELED] {key} — buffer overflow; re-sync and re-register" + ); } None => { - let event_type = response.event_type; - println!("[UNKNOWN] {key} (event_type={event_type})"); + println!("[UNKNOWN] {key} (event_type={})", response.event_type); } } } diff --git a/examples/sled-cluster/Cargo.toml b/examples/sled-cluster/Cargo.toml index 0f6712f7..4a827ef9 100644 --- a/examples/sled-cluster/Cargo.toml +++ b/examples/sled-cluster/Cargo.toml @@ -5,7 +5,6 @@ edition = "2024" [dependencies] d-engine = { path = "../../d-engine", features = ["server"], default-features = false } -d-engine-proto = { path = "../../d-engine-proto" } # d-engine = "0.2" sled = { version = "0.34.7", features = [ diff --git a/examples/sled-cluster/src/sled_engine_test.rs b/examples/sled-cluster/src/sled_engine_test.rs index eca53951..16698d38 100644 --- a/examples/sled-cluster/src/sled_engine_test.rs +++ b/examples/sled-cluster/src/sled_engine_test.rs @@ -1,11 +1,11 @@ use super::*; use bytes::{Bytes, BytesMut}; use d_engine::{LogStore, Result, StorageEngine}; -use d_engine_proto::client::{ +use d_engine::client::{ WriteCommand, write_command::{Insert, Operation}, }; -use d_engine_proto::common::{Entry, EntryPayload, entry_payload::Payload}; +use d_engine::common::{Entry, EntryPayload, entry_payload::Payload}; use prost::Message; use std::sync::Arc; use tempfile::TempDir; diff --git a/examples/three-nodes-embedded/README.md b/examples/three-nodes-embedded/README.md index 6608a6eb..25255a59 100644 --- a/examples/three-nodes-embedded/README.md +++ b/examples/three-nodes-embedded/README.md @@ -111,21 +111,40 @@ Ports are customizable via CLI arguments: ### Business API -**Write operation:** +**Writes must go to the leader.** Find it first via the health check endpoints: ```bash -curl -X POST http://localhost:8081/kv \ +# Find the leader (health port returns 200 only on the leader node) +for port in 10001 10002 10003; do + code=$(curl -s -o /dev/null -w "%{http_code}" http://localhost:$port/primary) + echo "health:$port → $code" +done +# health:10001 → 503 +# health:10002 → 200 ← leader is on business port 8082 +# health:10003 → 503 +``` + +**Write operation (to the leader):** + +```bash +curl -X POST http://localhost:8082/kv \ -H "Content-Type: application/json" \ -d '{"key": "username", "value": "alice"}' +# Response: {} ``` -**Read operation:** +Writing to a follower returns `503 Service Unavailable` with `{"error":"not_leader"}`. + +**Read operation (any node):** ```bash curl http://localhost:8081/kv/username # Response: {"value":"alice"} ``` +Reads use `get_eventual()` — they serve from the local state machine. A freshly +replicated write may take a few milliseconds to appear on follower nodes. + ### Health Check API **Leader check:** @@ -174,7 +193,8 @@ done ### Test Write Operations ```bash -# Write to any node (will forward internally if not leader) +# Must write to leader — follower writes return 503 {"error":"not_leader"} +# Use GET /primary on port 8091/8092/8093 to find the current leader first. curl -X POST http://localhost:8081/kv \ -H "Content-Type: application/json" \ -d '{"key": "test", "value": "hello"}' diff --git a/examples/three-nodes-embedded/src/main.rs b/examples/three-nodes-embedded/src/main.rs index 0c7b04f9..2e3ebeb0 100644 --- a/examples/three-nodes-embedded/src/main.rs +++ b/examples/three-nodes-embedded/src/main.rs @@ -8,7 +8,7 @@ use axum::{ http::StatusCode, routing::{get, post}, }; -use d_engine::EmbeddedEngine; +use d_engine::{ClientApiError, EmbeddedEngine, ErrorCode}; use serde::{Deserialize, Serialize}; use tokio::sync::watch; @@ -198,13 +198,27 @@ async fn start_business_server( .expect("Business server failed"); } +fn is_not_leader(e: &ClientApiError) -> bool { + e.code() == ErrorCode::NotLeader +} + async fn handle_put( State(engine): State>, Json(req): Json, -) -> StatusCode { +) -> (StatusCode, Json) { match engine.client().put(req.key.into_bytes(), req.value.into_bytes()).await { - Ok(_) => StatusCode::OK, - Err(_) => StatusCode::INTERNAL_SERVER_ERROR, + Ok(_) => (StatusCode::OK, Json(serde_json::json!({}))), + Err(e) if is_not_leader(&e) => ( + StatusCode::SERVICE_UNAVAILABLE, + Json(serde_json::json!({ + "error": "not_leader", + "hint": "Use GET /primary on the health port to find the leader, then retry the write there." + })), + ), + Err(e) => ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "error": format!("{e}") })), + ), } }