feat(watch) #379: prev_kv + progress heartbeat

d-engine-client/src/grpc_client.rs

@@ -224,6 +224,7 @@ impl GrpcClient {
             client_id: client_inner.client_id,
             key: Bytes::copy_from_slice(key.as_ref()),
             prefix: false,
+            prev_kv: false,
         };
 
         // Watch can connect to any node (leader or follower)
@@ -255,6 +256,7 @@ impl GrpcClient {
             client_id: client_inner.client_id,
             key: Bytes::copy_from_slice(prefix.as_ref()),
             prefix: true,
+            prev_kv: false,
         };
 
         let mut client = self.make_client().await?;

d-engine-client/src/grpc_client_test.rs

@@ -1592,13 +1592,15 @@ mod watch_tests {
             WatchResponse {
                 key: Bytes::from("lock"),
                 value: Bytes::from("owner-a"),
+                prev_value: Bytes::new(),
                 event_type: WatchEventType::Put as i32,
                 error: 0,
                 revision: 0,
             },
             WatchResponse {
                 key: Bytes::from("lock"),
                 value: Bytes::new(),
+                prev_value: Bytes::new(),
                 event_type: WatchEventType::Delete as i32,
                 error: 0,
                 revision: 0,

d-engine-core/src/config/raft.rs

@@ -1253,6 +1253,19 @@ pub struct WatchConfig {
     /// **Default**: i64::MAX (effectively unlimited)
     #[serde(default = "default_max_watcher_count")]
     pub max_watcher_count: usize,
+
+    /// Heartbeat interval in milliseconds for progress notifications.
+    ///
+    /// The dispatcher broadcasts a `Progress` event to all active watchers on
+    /// each tick so clients can confirm the stream is alive even during quiet
+    /// periods.  Set to 0 to disable heartbeats entirely.
+    ///
+    /// Use milliseconds (not seconds) so tests can set short intervals (e.g. 50ms)
+    /// without sleeping for a full second.
+    ///
+    /// **Default**: 30_000 (30 seconds)
+    #[serde(default = "default_heartbeat_interval_ms")]
+    pub heartbeat_interval_ms: u64,
 }
 
 impl Default for WatchConfig {
@@ -1262,6 +1275,7 @@ impl Default for WatchConfig {
             watcher_buffer_size: default_watcher_buffer_size(),
             enable_metrics: default_enable_watch_metrics(),
             max_watcher_count: default_max_watcher_count(),
+            heartbeat_interval_ms: default_heartbeat_interval_ms(),
         }
     }
 }
@@ -1319,6 +1333,10 @@ const fn default_max_watcher_count() -> usize {
     9_223_372_036_854_775_807
 }
 
+const fn default_heartbeat_interval_ms() -> u64 {
+    30_000
+}
+
 /// Performance metrics configuration
 ///
 /// Controls emission of observability metrics. Disabling metrics reduces overhead

d-engine-core/src/state_machine_handler/default_state_machine_handler.rs

@@ -106,6 +106,11 @@ where
     /// StateMachine sends events here without blocking on watchers
     #[cfg(feature = "watch")]
     watch_event_tx: Option<tokio::sync::broadcast::Sender<d_engine_proto::client::WatchResponse>>,
+
+    /// Number of active watchers that opted in to prev_kv.
+    /// When > 0, apply_chunk reads old values before applying each write.
+    #[cfg(feature = "watch")]
+    prev_kv_watcher_count: Arc<std::sync::atomic::AtomicUsize>,
 }
 
 #[derive(Debug, PartialEq, Hash, Eq, Clone)]
@@ -217,6 +222,17 @@ where
 
         let sm = self.state_machine.clone();
 
+        // Read prev values before applying if any watcher requested prev_kv.
+        // Must happen BEFORE apply_chunk so old values are still in the state machine.
+        #[cfg(feature = "watch")]
+        let prev_values: Option<Vec<Option<bytes::Bytes>>> = if self.watch_event_tx.is_some()
+            && self.prev_kv_watcher_count.load(std::sync::atomic::Ordering::Relaxed) > 0
+        {
+            Some(Self::read_prev_values(&*sm, &chunk))
+        } else {
+            None
+        };
+
         // Apply the chunk and track errors
         let apply_result = sm.apply_chunk(chunk.clone()).await;
 
@@ -225,7 +241,7 @@ where
         if let Ok(ref results) = apply_result
             && let Some(ref tx) = self.watch_event_tx
         {
-            self.broadcast_watch_events(&chunk, results, tx);
+            self.broadcast_watch_events(&chunk, results, tx, prev_values.as_deref());
         }
 
         // Record latency and chunk size histogram *after* the operation
@@ -624,21 +640,92 @@ impl<T> DefaultStateMachineHandler<T>
 where
     T: TypeConfig,
 {
+    /// Read the current value for every write operation in `chunk`, before applying.
+    ///
+    /// Returns one `Option<Bytes>` per entry in the same order:
+    /// - `Some(value)` for Insert / Delete / CAS entries (empty Bytes if key didn't exist)
+    /// - `None` for Noop / Config entries (no prev_value needed)
+    #[cfg(feature = "watch")]
+    fn read_prev_values(
+        sm: &dyn crate::StateMachine,
+        chunk: &[Entry],
+    ) -> Vec<Option<bytes::Bytes>> {
+        use std::collections::HashMap;
+
+        use d_engine_proto::client::WriteCommand;
+        use d_engine_proto::client::write_command::Operation;
+        use d_engine_proto::common::entry_payload::Payload;
+        use prost::Message;
+
+        // Tracks the value each key would hold after each processed entry.
+        // None = key was deleted by an earlier entry in this batch.
+        // Ensures that two writes to the same key in one chunk produce correct
+        // prev_values: the second write sees the first write's value, not the
+        // pre-batch SM state.
+        let mut overlay: HashMap<bytes::Bytes, Option<bytes::Bytes>> = HashMap::new();
+        let mut result = Vec::with_capacity(chunk.len());
+
+        for entry in chunk {
+            let prev = (|| -> Option<bytes::Bytes> {
+                let payload = entry.payload.as_ref()?;
+                let Some(Payload::Command(cmd_bytes)) = &payload.payload else {
+                    return None;
+                };
+                let Ok(write_cmd) = WriteCommand::decode(cmd_bytes.as_ref()) else {
+                    return None;
+                };
+                let op = write_cmd.operation?;
+
+                let key: bytes::Bytes = match &op {
+                    Operation::Insert(ins) => ins.key.clone(),
+                    Operation::Delete(del) => del.key.clone(),
+                    Operation::CompareAndSwap(cas) => cas.key.clone(),
+                };
+
+                // Check overlay first (reflects mutations earlier in this batch),
+                // then fall back to the SM's pre-batch state.
+                let prev = if let Some(v) = overlay.get(&key) {
+                    v.clone()
+                } else {
+                    sm.get(&key).ok().flatten()
+                };
+
+                // Update overlay so later entries in this batch see the updated state.
+                // CAS outcome is unknown before apply, so we conservatively skip it;
+                // same-key CAS-pairs in one batch are rare and semantically ambiguous.
+                match op {
+                    Operation::Insert(ins) => {
+                        overlay.insert(key, Some(ins.value));
+                    }
+                    Operation::Delete(_) => {
+                        overlay.insert(key, None);
+                    }
+                    Operation::CompareAndSwap(_) => {}
+                }
+
+                Some(prev.unwrap_or_default())
+            })();
+            result.push(prev);
+        }
+        result
+    }
+
     /// Broadcast watch events for applied chunk entries (fire-and-forget)
     ///
-    /// Parses chunk entries and broadcasts WatchEvent via tokio::broadcast channel.
+    /// Parses chunk entries and broadcasts WatchResponse via tokio::broadcast channel.
     /// Non-blocking: if channel is full, oldest events are dropped (lagging receivers).
-    /// Decoupled from WatchManager: only sends signal, doesn't care who listens.
     ///
     /// `results` must have the same length as `chunk` (enforced by StateMachineHandler contract).
     /// CAS entries where `results[i].succeeded == false` are skipped — no mutation occurred.
+    /// `prev_values`: optional slice of pre-apply values (None = no prev_kv watcher active).
     #[cfg(feature = "watch")]
     #[inline]
     pub(super) fn broadcast_watch_events(
         &self,
         chunk: &[Entry],
         results: &[ApplyResult],
         tx: &tokio::sync::broadcast::Sender<d_engine_proto::client::WatchResponse>,
+        prev_values: Option<&[Option<bytes::Bytes>]>,
     ) {
         use d_engine_proto::client::WatchEventType;
         use d_engine_proto::client::WatchResponse;
@@ -652,17 +739,24 @@ where
                 && let Some(Payload::Command(bytes)) = &payload.payload
                 && let Ok(write_cmd) = WriteCommand::decode(bytes.as_ref())
             {
+                let prev_value = prev_values
+                    .and_then(|pv| pv.get(i))
+                    .and_then(|v| v.clone())
+                    .unwrap_or_default();
+
                 let event = match write_cmd.operation {
                     Some(Operation::Insert(insert)) => Some(WatchResponse {
                         key: insert.key,
                         value: insert.value,
+                        prev_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(),
+                        prev_value,
                         event_type: WatchEventType::Delete as i32,
                         error: 0,
                         revision: entry.index,
@@ -674,6 +768,7 @@ where
                             Some(WatchResponse {
                                 key: cas.key,
                                 value: cas.new_value,
+                                prev_value,
                                 event_type: WatchEventType::Put as i32,
                                 error: 0,
                                 revision: entry.index,
@@ -707,6 +802,9 @@ where
         #[cfg_attr(not(feature = "watch"), allow(unused_variables))] watch_event_tx: Option<
             tokio::sync::broadcast::Sender<d_engine_proto::client::WatchResponse>,
         >,
+        #[cfg_attr(not(feature = "watch"), allow(unused_variables))] prev_kv_watcher_count: Arc<
+            std::sync::atomic::AtomicUsize,
+        >,
     ) -> Self {
         let (applied_notify_tx, applied_notify_rx) =
             tokio::sync::watch::channel(last_applied_index);
@@ -729,12 +827,12 @@ where
             applied_notify_rx,
             #[cfg(feature = "watch")]
             watch_event_tx,
+            #[cfg(feature = "watch")]
+            prev_kv_watcher_count,
         }
     }
 
     /// Convenience constructor for tests without watch
-    ///
-    /// Backward-compatible wrapper that calls `new()` with None for watch_event_tx.
     #[cfg(test)]
     pub(crate) fn new_without_watch(
         node_id: u32,
@@ -749,7 +847,8 @@ where
             state_machine,
             snapshot_config,
             snapshot_policy,
-            None, // No watch event tx for tests
+            None,
+            Arc::new(std::sync::atomic::AtomicUsize::new(0)),
         )
     }