fixed kv tests, checks for subscription leaks were no counting before…

… underlying consumer was cancelling. Signed-off-by: Alberto Ricart <[email protected]>
nats-io · Oct 3, 2024 · 3671901 · 3671901
1 parent 870d34b
commit 3671901
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 39 deletions.
diff --git a/kv/src/kv.ts b/kv/src/kv.ts
@@ -852,7 +852,6 @@ export class Bucket implements KV, KvRemove {
     const ignoreDeletes = opts.ignoreDeletes === true;
 
     let fn = opts.initializedFn;
-    let count = 0;
 
     const cc = this._buildCC(k, content, co);
     cc.name = `KV_WATCHER_${nuid.next()}`;
@@ -862,8 +861,19 @@ export class Bucket implements KV, KvRemove {
     }
 
     const oc = await this.js.consumers.getPushConsumer(this.stream, cc);
-    qi._data = oc;
+    const info = await oc.info(true);
+    const count = info.num_pending;
+    if (count === 0 && fn) {
+      try {
+        fn();
+      } catch (_err) {
+        // ignoring
+      } finally {
+        fn = undefined;
+      }
+    }
 
+    qi._data = oc;
     const iter = await oc.consume({
       callback: (m) => {
         const e = this.jmToEntry(m);
@@ -884,38 +894,14 @@ export class Bucket implements KV, KvRemove {
       },
     });
 
-    // by the time we are here, likely the subscription got messages
-    if (fn) {
-      // get the info used on consumer create
-      const last = await oc.info(true);
-      // this doesn't sound correct - we should be looking for a seq number instead
-      // then if we see a greater one, we are done.
-      const expect = last.num_pending + last.delivered.consumer_seq;
-      // if the iterator already queued - the only issue is other modifications
-      // did happen like stream was pruned, and the ordered consumer reset, etc
-      // we won't get what we are expecting - so the notification will never fire
-      // the sentinel ought to be coming from the server
-      if (expect === 0 || qi.received >= expect) {
-        try {
-          fn();
-        } catch (err) {
-          // fail it - there's something wrong in the user callback
-          qi.stop(err);
-        } finally {
-          fn = undefined;
-        }
-      } else {
-        count = expect;
-      }
-    }
+    qi.iterClosed.then(() => {
+      iter.stop();
+    });
     iter.closed().then(() => {
       qi.push(() => {
         qi.stop();
       });
     });
-    qi.iterClosed.then(() => {
-      iter.stop();
-    });
 
     return qi;
   }
@@ -927,6 +913,12 @@ export class Bucket implements KV, KvRemove {
     });
 
     const oc = await this.js.consumers.getPushConsumer(this.stream, cc);
+    const info = await oc.info();
+    if (info.num_pending === 0) {
+      keys.stop();
+      return keys;
+    }
+
     keys._data = oc;
 
     const iter = await oc.consume({
@@ -942,11 +934,6 @@ export class Bucket implements KV, KvRemove {
       },
     });
 
-    // get the info used on consumer create
-    const last = await oc.info(true);
-    if (last.num_pending === 0) {
-      keys.stop();
-    }
     iter.closed().then(() => {
       keys.push(() => {
         keys.stop();

diff --git a/kv/tests/kv_test.ts b/kv/tests/kv_test.ts
@@ -345,6 +345,8 @@ Deno.test("kv - cleanups/empty", async () => {
   if (await notCompatible(ns, nc, "2.6.3")) {
     return;
   }
+  const nci = nc as NatsConnectionImpl;
+
   const n = nuid.next();
   const js = jetstream(nc);
   const bucket = await new Kvm(js).create(n);
@@ -356,9 +358,9 @@ Deno.test("kv - cleanups/empty", async () => {
   const keys = await collect(await bucket.keys());
   assertEquals(keys.length, 0);
 
-  const nci = nc as NatsConnectionImpl;
   // mux should be created
   const min = nci.protocol.subscriptions.getMux() ? 1 : 0;
+
   assertEquals(nci.protocol.subscriptions.subs.size, min);
 
   await cleanup(ns, nc);
@@ -443,6 +445,7 @@ Deno.test("kv - bucket watch", async () => {
   await b.put("x", Empty);
 
   await done;
+  await delay(0);
 
   assertEquals(iter.getProcessed(), 7);
   assertEquals(m.get("a"), "2");
@@ -497,6 +500,7 @@ Deno.test("kv - key watch", async () => {
   const js = jetstream(nc);
   const bucket = await new Kvm(js).create(nuid.next()) as Bucket;
   await keyWatch(bucket);
+  await delay(0);
 
   const nci = nc as NatsConnectionImpl;
   const min = nci.protocol.subscriptions.getMux() ? 1 : 0;
@@ -519,6 +523,7 @@ Deno.test("kv - codec key watch", async () => {
     },
   }) as Bucket;
   await keyWatch(bucket);
+  await delay(0);
 
   const nci = nc as NatsConnectionImpl;
   const min = nci.protocol.subscriptions.getMux() ? 1 : 0;
@@ -660,10 +665,6 @@ Deno.test("kv - complex key", async () => {
   const vvv = await dd;
   assertEquals(vvv.string(), "hello");
 
-  const nci = nc as NatsConnectionImpl;
-  const min = nci.protocol.subscriptions.getMux() ? 1 : 0;
-  assertEquals(nci.protocol.subscriptions.subs.size, min);
-
   await cleanup(ns, nc);
 });