[IMPROVED] Add test checking KV TTL watcher updates (#1916)

Signed-off-by: Piotr Piotrowski <piotr@synadia.com>

jetstream/kv.go

@@ -254,6 +254,8 @@ type (
 
 		// LimitMarkerTTL is how long the bucket keeps markers when keys are
 		// removed by the TTL setting.
+		// It is required for per-key TTL to work and for watcher to notify
+		// about TTL expirations (both per key and per bucket)
 		LimitMarkerTTL time.Duration
 	}
 

jetstream/test/kv_test.go

@@ -113,6 +113,38 @@ func TestKeyValueBasics(t *testing.T) {
 	}
 }
 
+func TestKeyValueTTL(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx := context.Background()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "TEST", History: 5, TTL: 100 * time.Millisecond, LimitMarkerTTL: time.Second})
+	expectOk(t, err)
+
+	_, err = kv.Put(ctx, "name", []byte("pp"))
+	expectOk(t, err)
+
+	watcher, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
+	expectOk(t, err)
+
+	time.Sleep(500 * time.Millisecond)
+
+	_, err = kv.Get(ctx, "name")
+	expectErr(t, err, jetstream.ErrKeyNotFound)
+
+	select {
+	case v := <-watcher.Updates():
+		if v == nil || v.Key() != "name" || v.Operation() != jetstream.KeyValuePurge {
+			t.Fatalf("Expected purge operation for key 'name', got: %+v", v)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatalf("Expected update notification for key 'name', got nothing")
+	}
+}
+
 func TestCreateKeyValue(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer shutdownJSServerAndRemoveStorage(t, s)