From 7370872b5b3cdfd34989aafed73cec97f288dd1e Mon Sep 17 00:00:00 2001
From: Paulo Cabral Sanz <paulo@railway.app>
Date: Fri, 14 Aug 2026 17:39:42 -0300
Subject: [PATCH 1/2] test: NATS resume must not wrap revision + 1

protocol::resume_window_ok saturates at u64::MAX. The three watch
call sites used wrapping addition and would seek at sequence 0.
---
 src/protocol.rs | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/src/protocol.rs b/src/protocol.rs
index 7424c62..e45a8e2 100644
--- a/src/protocol.rs
+++ b/src/protocol.rs
@@ -148,4 +148,13 @@ mod tests {
         assert!(!resume_window_ok(3, 5), "gap head-evicted: expired");
         assert!(resume_window_ok(u64::MAX, u64::MAX), "saturating boundary");
     }
+
+    #[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"
+        );
+    }
 }

From 37d1176393360c94dd08ee36d7e2444ecbcd1d13 Mon Sep 17 00:00:00 2001
From: Paulo Cabral Sanz <paulo@railway.app>
Date: Fri, 14 Aug 2026 17:40:25 -0300
Subject: [PATCH 2/2] fix: refuse watch resume at u64::MAX instead of wrapping
 to 0

resume_start_sequence is checked_add(1). The three NATS watch-from
sites share it so they cannot drift from the protocol kernel.
---
 src/nats.rs     | 14 +++++++++++---
 src/protocol.rs | 10 ++++++++++
 2 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/src/nats.rs b/src/nats.rs
index 16c466e..01e4b33 100644
--- a/src/nats.rs
+++ b/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()
         }))
diff --git a/src/protocol.rs b/src/protocol.rs
index e45a8e2..826820b 100644
--- a/src/protocol.rs
+++ b/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,6 +155,8 @@ 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]