[FIX] [CONSUMERS] fixed subscription leaks

nats-io · May 18, 2023 · fa87854 · fa87854
1 parent 17d3a76
commit fa87854
Show file tree

Hide file tree

Showing 4 changed files with 157 additions and 3 deletions.
diff --git a/nats-base-client/consumer.ts b/nats-base-client/consumer.ts
@@ -109,6 +109,7 @@ export class PullConsumerImpl implements Consumer {
             (s.data as number) >= 2
           ) {
             d.reject(new Error("consumer missed heartbeats"));
+            break;
           }
         }
       })().catch();

diff --git a/nats-base-client/consumermessages.ts b/nats-base-client/consumermessages.ts
@@ -100,6 +100,11 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
       }
     });
 
+    const { sub } = this;
+    if (sub) {
+      sub.unsubscribe();
+    }
+
     this.sub = c.api.nc.subscribe(this.inbox, {
       callback: (err, msg) => {
         if (err) {
@@ -333,7 +338,6 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
     if (bytesLeft) {
       discard.bytesLeft = parseInt(bytesLeft);
     }
-    // FIXME: batch complete header goes here...
 
     return discard;
   }
@@ -363,6 +367,7 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
   }
 
   stop(err?: Error) {
+    this.sub?.unsubscribe();
     this.clearTimers();
     //@ts-ignore: fn
     this._push(() => {

diff --git a/tests/consumers_test.ts b/tests/consumers_test.ts
@@ -836,3 +836,76 @@ Deno.test("consumers - next", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("consumers - sub leaks next()", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const { stream } = await initStream(nc);
+
+  const jsm = await nc.jetstreamManager();
+  await jsm.consumers.add(stream, {
+    durable_name: stream,
+    ack_policy: AckPolicy.Explicit,
+  });
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  const js = nc.jetstream();
+  const c = await js.consumers.get(stream, stream);
+  await c.next({ expires: 1000 });
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  await cleanup(ns, nc);
+});
+
+Deno.test("consumers - sub leaks fetch()", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const { stream } = await initStream(nc);
+
+  const jsm = await nc.jetstreamManager();
+  await jsm.consumers.add(stream, {
+    durable_name: stream,
+    ack_policy: AckPolicy.Explicit,
+  });
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  const js = nc.jetstream();
+  const c = await js.consumers.get(stream, stream);
+  const iter = await c.fetch({ expires: 1000 });
+  const done = (async () => {
+    for await (const _m of iter) {
+      // nothing
+    }
+  })().then();
+  await done;
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  await cleanup(ns, nc);
+});
+
+Deno.test("consumers - sub leaks consume()", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const { stream } = await initStream(nc);
+
+  const jsm = await nc.jetstreamManager();
+  await jsm.consumers.add(stream, {
+    durable_name: stream,
+    ack_policy: AckPolicy.Explicit,
+  });
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  const js = nc.jetstream();
+  const c = await js.consumers.get(stream, stream);
+  const iter = await c.consume({ expires: 30000 });
+  const done = (async () => {
+    for await (const _m of iter) {
+      // nothing
+    }
+  })().then();
+  setTimeout(() => {
+    iter.close();
+  }, 1000);
+
+  await done;
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  await cleanup(ns, nc);
+});
diff --git a/tests/consumersordered_test.ts b/tests/consumersordered_test.ts
@@ -13,14 +13,19 @@
  * limitations under the License.
  */
 
-import { cleanup, jetstreamServerConf, setup } from "./jstest_util.ts";
+import {
+  cleanup,
+  initStream,
+  jetstreamServerConf,
+  setup,
+} from "./jstest_util.ts";
 import {
   assertEquals,
   assertExists,
   assertRejects,
 } from "https://deno.land/std@0.125.0/testing/asserts.ts";
 import { OrderedPullConsumerImpl } from "../nats-base-client/consumer.ts";
-import { DeliverPolicy, JsMsg } from "../nats-base-client/types.ts";
+import { AckPolicy, DeliverPolicy, JsMsg } from "../nats-base-client/types.ts";
 import { deferred } from "../nats-base-client/mod.ts";
 import { notCompatible } from "./helpers/mod.ts";
 import { delay } from "../nats-base-client/util.ts";
@@ -608,3 +613,73 @@ Deno.test("ordered - next", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("ordered - sub leaks next()", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const { ns, nc } = await setup(jetstreamServerConf());
+  if (await notCompatible(ns, nc, "2.10.0")) {
+    return;
+  }
+  const { stream } = await initStream(nc);
+
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  const js = nc.jetstream();
+  const c = await js.consumers.get(stream);
+  await c.next({ expires: 1000 });
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  await cleanup(ns, nc);
+});
+
+Deno.test("ordered - sub leaks fetch()", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const { ns, nc } = await setup(jetstreamServerConf());
+  if (await notCompatible(ns, nc, "2.10.0")) {
+    return;
+  }
+  const { stream } = await initStream(nc);
+
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  const js = nc.jetstream();
+  const c = await js.consumers.get(stream);
+  const iter = await c.fetch({ expires: 1000 });
+  const done = (async () => {
+    for await (const _m of iter) {
+      // nothing
+    }
+  })().then();
+  await done;
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  await cleanup(ns, nc);
+});
+
+Deno.test("ordered - sub leaks consume()", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const { ns, nc } = await setup(jetstreamServerConf());
+  if (await notCompatible(ns, nc, "2.10.0")) {
+    return;
+  }
+  const { stream } = await initStream(nc);
+
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  const js = nc.jetstream();
+  const c = await js.consumers.get(stream);
+  const iter = await c.consume({ expires: 30000 });
+  const done = (async () => {
+    for await (const _m of iter) {
+      // nothing
+    }
+  })().then();
+  setTimeout(() => {
+    iter.close();
+  }, 1000);
+
+  await done;
+  //@ts-ignore: test
+  assertEquals(nc.protocol.subscriptions.size(), 1);
+  await cleanup(ns, nc);
+});