[ADDED] StartRevision option to KV watcher

jetstream/kv.go

@@ -184,6 +184,8 @@ type (
 		updatesOnly bool
 		// retrieve only the meta data of the entry
 		metaOnly bool
+		// resumeFromRevision is the revision to resume from.
+		resumeFromRevision uint64
 	}
 
 	KVDeleteOpt interface {
@@ -875,6 +877,9 @@ func (kv *kvs) Watch(ctx context.Context, keys string, opts ...WatchOpt) (KeyWat
 	if o.metaOnly {
 		subOpts = append(subOpts, nats.HeadersOnly())
 	}
+	if o.resumeFromRevision > 0 {
+		subOpts = append(subOpts, nats.StartSequence(o.resumeFromRevision))
+	}
 	subOpts = append(subOpts, nats.Context(ctx))
 	// Create the sub and rest of initialization under the lock.
 	// We want to prevent the race between this code and the

jetstream/options.go

@@ -402,6 +402,14 @@ func MetaOnly() WatchOpt {
 	})
 }
 
+// ResumeFromRevision instructs the key watcher to resume from a specific revision number.
+func ResumeFromRevision(revision uint64) WatchOpt {
+	return watchOptFn(func(opts *watchOpts) error {
+		opts.resumeFromRevision = revision
+		return nil
+	})
+}
+
 // DeleteMarkersOlderThan indicates that delete or purge markers older than that
 // will be deleted as part of PurgeDeletes() operation, otherwise, only the data
 // will be removed but markers that are recent will be kept.

jetstream/test/kv_test.go

@@ -404,6 +404,53 @@ func TestKeyValueWatch(t *testing.T) {
 		expectOk(t, err)
 		expectUpdate("t.age", "66", 12)
 	})
+
+	t.Run("watcher with start revision", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH"})
+		expectOk(t, err)
+
+		_, err = kv.Create(ctx, "name", []byte("derek"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "age", []byte("22"))
+		expectOk(t, err)
+
+		watcher, err := kv.WatchAll(ctx, jetstream.ResumeFromRevision(2))
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectUpdate := expectUpdateF(t, watcher)
+
+		// check that we get only updates after revision 2
+		expectUpdate("name", "rip", 2)
+		expectUpdate("age", "22", 3)
+
+		// stop first watcher
+		watcher.Stop()
+
+		_, err = kv.Put(ctx, "name2", []byte("ik"))
+		expectOk(t, err)
+
+		// create a new watcher with start revision 3
+		watcher, err = kv.WatchAll(ctx, jetstream.ResumeFromRevision(3))
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectUpdate = expectUpdateF(t, watcher)
+
+		// check that we get only updates after revision 3
+		expectUpdate("age", "22", 3)
+		expectUpdate("name2", "ik", 4)
+	})
 }
 
 func TestKeyValueWatchContext(t *testing.T) {