Refuse watch resume at u64::MAX instead of wrapping to 0

src/nats.rs

@@ -1174,6 +1174,12 @@ async fn stream_watch_floor_guarded(
 ///
 /// If these messages ever change, `cursor_expired_matches_known_nats_error_strings`
 /// is the canary that fails loudly on the next dependency bump.
+fn resume_start(revision: u64) -> Result<u64, KvError> {
+    crate::protocol::resume_start_sequence(revision).ok_or_else(|| {
+        KvError::WatchError("resume revision is u64::MAX; refusing wrap to 0".into())
+    })
+}
+
 fn is_cursor_expired_error(err: &str) -> bool {
     use std::sync::OnceLock;
     // One Aho-Corasick automaton over all needles: a single pass over the error
@@ -1359,7 +1365,8 @@ impl KvWatcher for NatsKvWatcher {
         };
         self.check_resume_window(revision).await?;
 
-        let watcher = match timed(self.kv.watch_all_from_revision(revision + 1)).await? {
+        let start = resume_start(revision)?;
+        let watcher = match timed(self.kv.watch_all_from_revision(start)).await? {
             Ok(w) => w,
             Err(e) => {
                 let err_str = e.to_string();
@@ -1397,7 +1404,8 @@ impl KvWatcher for NatsKvWatcher {
         self.check_resume_window(revision).await?;
 
         let nats_key = format!("{prefix}>");
-        let watcher = match timed(self.kv.watch_from_revision(&nats_key, revision + 1)).await? {
+        let start = resume_start(revision)?;
+        let watcher = match timed(self.kv.watch_from_revision(&nats_key, start)).await? {
             Ok(w) => w,
             Err(e) => {
                 let err_str = e.to_string();
@@ -1473,7 +1481,7 @@ impl KvWatcher for NatsKvWatcher {
             filter_subjects,
             replay_policy: ReplayPolicy::Instant,
             deliver_policy: DeliverPolicy::ByStartSequence {
-                start_sequence: revision + 1,
+                start_sequence: resume_start(revision)?,
             },
             ..Default::default()
         }))

src/protocol.rs

@@ -111,6 +111,14 @@ pub fn resume_window_ok(revision: u64, first_sequence: u64) -> bool {
     first_sequence <= revision.saturating_add(1)
 }
 
+/// Sequence the watch should start at after applying `revision`.
+///
+/// `None` at `u64::MAX`: wrapping to 0 would silently replay the bucket
+/// from the beginning. Callers must fail the watch rather than wrap.
+pub fn resume_start_sequence(revision: u64) -> Option<u64> {
+    revision.checked_add(1)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -147,5 +155,16 @@ mod tests {
         assert!(resume_window_ok(3, 1), "history intact");
         assert!(!resume_window_ok(3, 5), "gap head-evicted: expired");
         assert!(resume_window_ok(u64::MAX, u64::MAX), "saturating boundary");
+        assert_eq!(resume_start_sequence(3), Some(4));
+        assert_eq!(resume_start_sequence(u64::MAX), None);
+    }
+
+    #[test]
+    fn resume_start_does_not_wrap_in_nats() {
+        let src = include_str!("nats.rs");
+        assert!(
+            !src.contains("revision + 1"),
+            "watch resume must not wrap u64::MAX to 0; use resume_start_sequence"
+        );
     }
 }