[FIX] pull adds a status listener on every call, and has no mechanism…

… for removing it - this modifies it so that if the status iterator closes it is removed from the list

jetstream/consumer.ts

@@ -23,6 +23,7 @@ import {
   MsgHdrs,
   NatsError,
   QueuedIterator,
+  Status,
   Subscription,
 } from "../nats-base-client/core.ts";
 import * as core from "../nats-base-client/idleheartbeat.ts";
@@ -228,6 +229,7 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
   timeout: Timeout<unknown> | null;
   cleanupHandler?: () => void;
   listeners: QueuedIterator<ConsumerStatus>[];
+  statusIterator?: QueuedIteratorImpl<Status>;
 
   // callback: ConsumerCallbackFn;
   constructor(
@@ -400,7 +402,9 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
     // now if we disconnect, the consumer could be gone
     // or we were slow consumer'ed by the server
     (async () => {
-      for await (const s of c.api.nc.status()) {
+      const status = c.api.nc.status();
+      this.statusIterator = status as QueuedIteratorImpl<Status>;
+      for await (const s of status) {
         switch (s.type) {
           case Events.Disconnect:
             // don't spam hb errors if we are disconnected
@@ -552,6 +556,7 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
   stop(err?: Error) {
     this.sub?.unsubscribe();
     this.clearTimers();
+    this.statusIterator?.stop();
     //@ts-ignore: fn
     this._push(() => {
       super.stop(err);

jetstream/tests/consumers_test.ts

@@ -29,6 +29,7 @@ import {
   AckPolicy,
   Consumer,
   ConsumerMessages,
+  DeliverPolicy,
   nanos,
   PubAck,
   PullOptions,
@@ -48,7 +49,7 @@ import {
   ConsumerStatus,
   PullConsumerMessagesImpl,
 } from "../consumer.ts";
-import { deadline } from "../../nats-base-client/util.ts";
+import { deadline, delay } from "../../nats-base-client/util.ts";
 
 Deno.test("consumers - min supported server", async () => {
   const { ns, nc } = await setup(jetstreamServerConf({}));
@@ -943,3 +944,79 @@ Deno.test("consumers - consume drain", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("consumers - fetch listener leaks", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const jsm = await nc.jetstreamManager();
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+
+  const js = nc.jetstream();
+  await js.publish("hello");
+
+  await jsm.consumers.add("messages", {
+    durable_name: "myconsumer",
+    deliver_policy: DeliverPolicy.All,
+    ack_policy: AckPolicy.Explicit,
+    ack_wait: nanos(3000),
+    max_waiting: 500,
+  });
+
+  const nci = nc as NatsConnectionImpl;
+  const base = nci.protocol.listeners.length;
+
+  const consumer = await js.consumers.get("messages", "myconsumer");
+
+  let done = false;
+  while (!done) {
+    const iter = await consumer.fetch({
+      max_messages: 1,
+    }) as PullConsumerMessagesImpl;
+    for await (const m of iter) {
+      assertEquals(nci.protocol.listeners.length, base);
+      m?.nak();
+      if (m.info.redeliveryCount > 100) {
+        done = true;
+      }
+    }
+  }
+
+  assertEquals(nci.protocol.listeners.length, base);
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("consumers - next listener leaks", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const jsm = await nc.jetstreamManager();
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+
+  const js = nc.jetstream();
+  await js.publish("hello");
+
+  await jsm.consumers.add("messages", {
+    durable_name: "myconsumer",
+    deliver_policy: DeliverPolicy.All,
+    ack_policy: AckPolicy.Explicit,
+    ack_wait: nanos(3000),
+    max_waiting: 500,
+  });
+
+  const nci = nc as NatsConnectionImpl;
+  const base = nci.protocol.listeners.length;
+
+  const consumer = await js.consumers.get("messages", "myconsumer");
+
+  let done = false;
+  while (true) {
+    const m = await consumer.next();
+    if (m) {
+      m.nak();
+      if (m.info?.redeliveryCount > 100) {
+        break;
+      }
+    }
+  }
+  assertEquals(nci.protocol.listeners.length, base);
+
+  await cleanup(ns, nc);
+});

nats-base-client/nats.ts

@@ -452,6 +452,10 @@ export class NatsConnectionImpl implements NatsConnection {
 
   status(): AsyncIterable<Status> {
     const iter = new QueuedIteratorImpl<Status>();
+    iter.iterClosed.then(() => {
+      const idx = this.listeners.indexOf(iter);
+      this.listeners.splice(idx, 1);
+    });
     this.listeners.push(iter);
     return iter;
   }