Merge pull request #750 from nats-io/fix-no-responders

breaking: nats-server 2.10.26 emits no responder if stream/consumer are not found

Author: aricart

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) {
+              // 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,7 +1125,6 @@ export class OrderedPullConsumerImpl implements Consumer {
             return Promise.reject(err);
           }
         }
-
         if (isNew && i >= this.maxInitialReset) {
           // consumer was never created, so we can fail this
           throw err;

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);
+});

jetstream/tests/consumers_test.ts

@@ -14,12 +14,18 @@
  */
 import { initStream } from "./jstest_util.ts";
 import {
+  assert,
   assertEquals,
   assertExists,
   assertRejects,
   assertStringIncludes,
 } from "https://deno.land/[email protected]/assert/mod.ts";
-import { deferred, delay, nanos } from "../../nats-base-client/mod.ts";
+import {
+  deadline,
+  deferred,
+  delay,
+  nanos,
+} from "../../nats-base-client/mod.ts";
 import {
   AckPolicy,
   Consumer,
@@ -645,3 +651,122 @@ Deno.test("consumers - getPullConsumerFor non existing misses heartbeats", async
 
   await cleanup(ns, nc);
 });
+
+Deno.test("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"] });
+  await jsm.consumers.add("messages", {
+    name: "c",
+    deliver_policy: DeliverPolicy.All,
+  });
+
+  // stream is deleted
+  const c = await jsm.jetstream().consumers.get("messages", "c");
+  await jsm.streams.delete("messages");
+  const iter = await c.consume({ expires: 10_000 });
+  const buf: ConsumerStatus[] = [];
+  const hbmP = 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 === 3) {
+          hbmP.resolve();
+        }
+      }
+    }
+  })().then();
+
+  const process = (async () => {
+    for await (const m of iter) {
+      if (m) {
+        break;
+      }
+    }
+  })();
+
+  await hbmP;
+
+  const snfs = buf.filter((s) => {
+    return s.type === ConsumerEvents.StreamNotFound;
+  }).length;
+  assert(snfs > 0);
+
+  const nrs = buf.filter((s) => {
+    return s.type === ConsumerEvents.NoResponders;
+  }).length;
+  assertEquals(nrs, 1);
+
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+  await jsm.consumers.add("messages", {
+    name: "c",
+    deliver_policy: DeliverPolicy.All,
+  });
+
+  await nc.jetstream().publish("hello");
+  await deadline(process, 5_000);
+  await cleanup(ns, nc);
+});
+
+Deno.test("consumers - no responders - consumer deleted", async () => {
+  const { ns, nc } = await setup(jetstreamServerConf());
+  const jsm = await nc.jetstreamManager();
+
+  await jsm.streams.add({ name: "messages", subjects: ["hello"] });
+  await jsm.consumers.add("messages", {
+    name: "c",
+    deliver_policy: DeliverPolicy.All,
+  });
+
+  // stream is deleted
+  const c = await jsm.jetstream().consumers.get("messages", "c");
+  await jsm.consumers.delete("messages", "c");
+  const iter = await c.consume({ expires: 10_000 });
+  const buf: ConsumerStatus[] = [];
+  const hbmP = 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 === 3) {
+          hbmP.resolve();
+        }
+      }
+    }
+  })().then();
+
+  const process = (async () => {
+    for await (const m of iter) {
+      if (m) {
+        break;
+      }
+    }
+  })();
+
+  await hbmP;
+
+  const snfs = buf.filter((s) => {
+    return s.type === ConsumerEvents.ConsumerNotFound;
+  }).length;
+  assert(snfs > 0);
+
+  const nrs = buf.filter((s) => {
+    return s.type === ConsumerEvents.NoResponders;
+  }).length;
+  assertEquals(nrs, 1);
+
+  await jsm.consumers.add("messages", {
+    name: "c",
+    deliver_policy: DeliverPolicy.All,
+  });
+
+  await nc.jetstream().publish("hello");
+  await deadline(process, 5_000);
+  await cleanup(ns, nc);
+});