d_engine_core/watch/
mod.rs

1//! Watch mechanism for monitoring key changes
2//!
3//! This module provides a high-performance, lock-free watch system that allows
4//! clients to monitor specific keys for changes with minimal overhead on the write path.
5//!
6//! # Architecture Overview
7//!
8//! The watch system is composed of two main components created explicitly in NodeBuilder:
9//!
10//! 1. **WatchRegistry** (Shared State): Lock-free registration via DashMap
11//! 2. **WatchDispatcher** (Background Task): Spawned explicitly in Builder, distributes events
12//!
13//! ```text
14//! ┌─────────────┐
15//! │ StateMachine│
16//! │  apply()    │
17//! └──────┬──────┘
18//!        │ notify_change() [~10ns]
19//!        ▼
20//! ┌─────────────────┐
21//! │  Event Queue    │ (crossbeam-channel, lock-free)
22//! │  (bounded 1000) │
23//! └──────┬──────────┘
24//!        │
25//!        ▼
26//! ┌─────────────────┐
27//! │   Dispatcher    │ (background thread)
28//! │   Thread        │
29//! └──────┬──────────┘
30//!        │ lookup in DashMap (lock-free)
31//!        ▼
32//! ┌─────────────────┐
33//! │ Per-Watcher     │ (tokio mpsc, bounded 10)
34//! │ Channels        │
35//! └──────┬──────────┘
36//!        │
37//!        ▼
38//! ┌─────────────────┐
39//! │ gRPC Stream     │
40//! │ to Client       │
41//! └─────────────────┘
42//! ```
43//!
44//! # Performance Characteristics
45//!
46//! - **Write overhead**: < 0.01% with 100+ watchers (target: <10ns per notify)
47//! - **Event latency**: Typically < 100μs end-to-end
48//! - **Memory per watcher**: ~2.4KB with default buffer size (10)
49//! - **Scalability**: Tested with 1000+ concurrent watchers
50//!
51//! # Usage Example
52//!
53//! ```ignore
54//! use d_engine_core::watch::{WatchRegistry, WatchDispatcher};
55//! use bytes::Bytes;
56//!
57//! # tokio::runtime::Runtime::new().unwrap().block_on(async {
58//! // Create components (typically done in NodeBuilder)
59//! let (unregister_tx, unregister_rx) = mpsc::unbounded_channel();
60//! let registry = Arc::new(WatchRegistry::new(10, unregister_tx));
61//! let dispatcher = WatchDispatcher::new(registry.clone(), broadcast_rx, unregister_rx);
62//!
63//! // Explicitly spawn dispatcher (visible resource allocation)
64//! tokio::spawn(async move {
65//!     dispatcher.run().await;
66//! });
67//!
68//! // Register a watcher
69//! let key = Bytes::from("mykey");
70//! let handle = registry.register(key);
71//! # });
72//! ```
73//!
74//! # Configuration
75//!
76//! The watch system is configurable via [`WatchConfig`]:
77//!
78//! - `event_queue_size`: Global event queue buffer (default: 1000)
79//! - `watcher_buffer_size`: Per-watcher channel buffer (default: 10)
80//! - `enable_metrics`: Enable detailed logging and metrics (default: false)
81//!
82//! ```
83//! use d_engine_core::watch::WatchConfig;
84//!
85//! let config = WatchConfig {
86//!     enabled: true,
87//!     event_queue_size: 2000,
88//!     watcher_buffer_size: 20,
89//!     enable_metrics: true,
90//! };
91//! ```
92//!
93//! # Error Handling
94//!
95//! The watch system prioritizes write performance over guaranteed event delivery:
96//!
97//! - When the global event queue is full, new events are **dropped**
98//! - When a per-watcher channel is full, events for that watcher are **dropped**
99//! - Clients should use the Read API to re-sync if they detect missing events
100//!
101//! This design ensures that a slow or stuck watcher never blocks the write path.
102//!
103//! # Thread Safety
104//!
105//! All types in this module are thread-safe and can be safely shared across threads:
106//!
107//! - [`WatchRegistry`] is shared via `Arc` between Node and Dispatcher
108//! - All operations are lock-free (DashMap + AtomicU64)
109//! - The dispatcher runs as an independent tokio task
110//!
111//! # Design Principles
112//!
113//! - **No hidden resource allocation**: All tokio::spawn calls are explicit in Builder
114//! - **Minimal abstraction**: Only essential data structures, no unnecessary wrappers
115//! - **Composable**: Registry and Dispatcher are independent, composed in Builder
116//! - Uses `DashMap` for the watcher registry (lock-free concurrent HashMap)
117//! - Uses `tokio::sync::mpsc` for per-watcher channels (async-friendly)
118//! - Watchers are automatically cleaned up when dropped (RAII pattern)
119
120mod manager;
121
122#[cfg(test)]
123mod manager_test;
124
125pub use manager::WatchDispatcher;
126pub use manager::WatchEvent;
127pub use manager::WatchEventType;
128pub use manager::WatchRegistry;
129pub use manager::WatcherHandle;
130
131// Re-export WatchConfig from the unified config system
132pub use crate::config::WatchConfig;
d_engine_core/watch/mod.rs

d_engine_core/watch/
mod.rs
