handle new "no responders" error for JetStream consumers

Updated test cases to handle "no responders" errors consistently for streams and consumers. Added new tests to verify behavior when streams or consumers are deleted and re-created. Improved imports and error assertion patterns for better clarity and reliability.

Signed-off-by: Alberto Ricart <[email protected]>

Author: aricart

.github/workflows/test.yml

@@ -29,20 +29,12 @@ jobs:
         with:
           deno-version: ${{ matrix.deno-version }}
 
-      - name: Set NATS Server Version
-        run: echo "NATS_VERSION=v2.10.24" >> $GITHUB_ENV
-
-      # this here because dns seems to be wedged on gha
-#      - name: Add hosts to /etc/hosts
-#        run: |
-#          sudo echo "145.40.102.131 demo.nats.io" | sudo tee -a /etc/hosts
-
-      - name: Get nats-server
-        run: |
-          wget "https://github.com/nats-io/nats-server/releases/download/$NATS_VERSION/nats-server-$NATS_VERSION-linux-amd64.zip" -O tmp.zip
-          unzip tmp.zip
-          mv nats-server-$NATS_VERSION-linux-amd64 nats-server
-          rm nats-server/README.md LICENSE
+      - name: Install nats-server
+        uses: aricart/[email protected]
+        with:
+          repo: nats-io/nats-server
+          name: nats-server
+          cache: true
 
       - name: Lint Deno Module
         run: deno fmt --check --ignore=docs/

jetstream/consumer.ts

@@ -219,6 +219,12 @@ export enum ConsumerEvents {
    * the consumer is recreated. The argument is the name of the newly created consumer.
    */
   OrderedConsumerRecreated = "ordered_consumer_recreated",
+
+  /**
+   * This notification means that either both the stream and consumer were not
+   * found or that JetStream is not available.
+   */
+  NoResponders = "no_responders",
 }
 
 /**
@@ -304,6 +310,7 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
   resetHandler?: () => void;
   abortOnMissingResource?: boolean;
   bind: boolean;
+  inBackOff: boolean;
 
   // callback: ConsumerCallbackFn;
   constructor(
@@ -329,6 +336,7 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
     this.forOrderedConsumer = false;
     this.abortOnMissingResource = copts.abort_on_missing_resource === true;
     this.bind = copts.bind === true;
+    this.inBackOff = false;
 
     this.start();
   }
@@ -411,6 +419,17 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
                 this.stop(error);
                 return;
               }
+            } else if (code === 503 && description === "unknown") {
+              // this is a no responders - possibly the stream/consumer were
+              // deleted from under the client
+              this.notify(ConsumerEvents.NoResponders, `${code} No Responders`);
+              // in cases that we are in consume, the idle heartbeat will kick in
+              // which will do a reset, and possibly refine the error
+              if (!this.refilling || this.abortOnMissingResource) {
+                const error = new NatsError("no responders", `${code}`);
+                this.stop(error);
+                return;
+              }
             } else {
               this.notify(
                 ConsumerDebugEvents.DebugEvent,
@@ -563,9 +582,13 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
   }
 
   async resetPendingWithInfo(): Promise<boolean> {
+    if (this.inBackOff) {
+      // already failing to get stream or consumer
+      return false;
+    }
     let notFound = 0;
     let streamNotFound = 0;
-    const bo = backoff();
+    const bo = backoff([this.opts.expires]);
     let attempt = 0;
     while (true) {
       if (this.done) {
@@ -578,6 +601,8 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
       try {
         // check we exist
         await this.consumer.info();
+        this.inBackOff = false;
+
         notFound = 0;
         // we exist, so effectively any pending state is gone
         // so reset and re-pull
@@ -616,6 +641,7 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
           notFound = 0;
           streamNotFound = 0;
         }
+        this.inBackOff = true;
         const to = bo.backoff(attempt);
         // wait for delay or till the client closes
         const de = delay(to);
@@ -1069,6 +1095,7 @@ export class OrderedPullConsumerImpl implements Consumer {
   }
 
   async resetConsumer(seq = 0): Promise<ConsumerInfo> {
+    const id = nuid.next();
     const isNew = this.serial === 0;
     // try to delete the consumer
     this.consumer?.delete().catch(() => {});
@@ -1078,7 +1105,7 @@ export class OrderedPullConsumerImpl implements Consumer {
     const config = this.getConsumerOpts(seq);
     config.max_deliver = 1;
     config.mem_storage = true;
-    const bo = backoff();
+    const bo = backoff([this.opts?.expires || 30_000]);
     let ci;
     for (let i = 0;; i++) {
       try {
@@ -1098,12 +1125,14 @@ export class OrderedPullConsumerImpl implements Consumer {
             return Promise.reject(err);
           }
         }
-
+        console.log("caught: ", err.message);
         if (isNew && i >= this.maxInitialReset) {
           // consumer was never created, so we can fail this
           throw err;
         } else {
-          await delay(bo.backoff(i + 1));
+          const to = bo.backoff(i + 1);
+          console.log(id, "is waiting for", to);
+          await delay(to);
         }
       }
     }

jetstream/tests/consume_test.ts

@@ -310,7 +310,7 @@ Deno.test("consume - stream not found request abort", async () => {
       }
     },
     Error,
-    "stream not found",
+    "no responders",
   );
 
   await cleanup(ns, nc);
@@ -380,7 +380,7 @@ Deno.test("consume - consumer not found request abort", async () => {
       }
     },
     Error,
-    "consumer not found",
+    "no responders",
   );
 
   await cleanup(ns, nc);

jetstream/tests/consumers_ordered_test.ts

@@ -23,7 +23,9 @@ import {
 } from "https://deno.land/[email protected]/assert/mod.ts";
 import {
   ConsumerDebugEvents,
+  ConsumerEvents,
   ConsumerMessages,
+  ConsumerStatus,
   DeliverPolicy,
   JsMsg,
 } from "../mod.ts";
@@ -1073,3 +1075,107 @@ Deno.test("ordered consumers - initial creation fails, consumer fails", async ()
 
   await cleanup(ns, nc);
 });
+
+Deno.test("ordered consumers - no responders - stream deleted", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const jsm = await nc.jetstreamManager();
+
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+
+  // stream is deleted
+  const c = await jsm.jetstream().consumers.get("messages");
+  const iter = await c.consume({ expires: 10_000 });
+  // FIXME: this impl of the consumer has a bug in notification
+  //   that unless it has pulled, it has no way of returning events.
+  //   for a test we want to miss a few, and then recreate.
+  await jsm.streams.delete("messages");
+
+  const buf: ConsumerStatus[] = [];
+  const snfP = deferred();
+  (async () => {
+    const status = await iter.status();
+    for await (const s of status) {
+      console.log(s);
+      buf.push(s);
+      if (s.type === ConsumerEvents.HeartbeatsMissed) {
+        if (s.data === 5) {
+          snfP.resolve();
+        }
+      }
+    }
+  })().then();
+
+  const process = (async () => {
+    for await (const m of iter) {
+      if (m) {
+        break;
+      }
+    }
+  })();
+
+  await snfP;
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+
+  await nc.jetstream().publish("hello");
+  await deadline(process, 15_000);
+  await cleanup(ns, nc);
+});
+
+Deno.test("ordered consumers fetch - no responders - stream deleted", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const jsm = await nc.jetstreamManager();
+
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+
+  // stream is deleted
+  const c = await jsm.jetstream().consumers.get("messages");
+  await jsm.streams.delete("messages");
+  await assertRejects(
+    async () => {
+      const iter = await c.fetch({ expires: 10_000 });
+      for await (const _ of iter) {
+        // ignored
+      }
+    },
+    Error,
+    "stream not found",
+  );
+
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+  await nc.jetstream().publish("hello");
+
+  const d = deferred();
+  const iter = await c.fetch({ expires: 10_000, max_messages: 1 });
+  for await (const _ of iter) {
+    d.resolve();
+  }
+  await d;
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("ordered consumers next - no responders - stream deleted", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const jsm = await nc.jetstreamManager();
+
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+
+  // stream is deleted
+  const c = await jsm.jetstream().consumers.get("messages");
+  await jsm.streams.delete("messages");
+  await assertRejects(
+    () => {
+      return c.next({ expires: 10_000 });
+    },
+    Error,
+    "stream not found",
+  );
+
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+  await nc.jetstream().publish("hello");
+
+  const m = await c.next({ expires: 10_000 });
+  assertExists(m);
+
+  await cleanup(ns, nc);
+});