fix(jetstream): more resilient ordered consumers and protocol message…

… handling (#104)

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

jetstream/deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/jetstream",
-  "version": "3.0.0-21",
+  "version": "3.0.0-22",
   "exports": {
     ".": "./src/mod.ts",
     "./internal": "./src/internal_mod.ts"

jetstream/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/jetstream",
-  "version": "3.0.0-21",
+  "version": "3.0.0-22",
   "files": [
     "lib/",
     "LICENSE",

jetstream/src/consumer.ts

@@ -166,8 +166,15 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
         }
         this.monitor?.work();
 
-        const isProtocol = msg.subject === this.inbox;
+        const isProtocol = this.consumer.ordered
+          ? msg.subject.indexOf(this?.inboxPrefix!) === 0
+          : msg.subject === this.inbox;
+
         if (isProtocol) {
+          if (msg.subject !== (this.sub as SubscriptionImpl).subject) {
+            // this is a stale message - was not sent to the current inbox
+            return;
+          }
           const status = new JetStreamStatus(msg);
 
           if (status.isIdleHeartbeat()) {
@@ -178,7 +185,6 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
           const description = status.description;
 
           const { msgsLeft, bytesLeft } = status.parseDiscard();
-          console.log("pending", msgsLeft, bytesLeft);
           if ((msgsLeft && msgsLeft > 0) || (bytesLeft && bytesLeft > 0)) {
             this.pending.msgs -= msgsLeft;
             this.pending.bytes -= bytesLeft;
@@ -842,15 +848,12 @@ export class PullConsumerImpl implements Consumer {
     return ci;
   }
 
-  info(cached = false): Promise<ConsumerInfo> {
+  async info(cached = false): Promise<ConsumerInfo> {
     if (cached) {
       return Promise.resolve(this._info);
     }
     const { stream_name, name } = this._info;
-    return this.api.info(stream_name, name)
-      .then((ci) => {
-        this._info = ci;
-        return this._info;
-      });
+    this._info = await this.api.info(stream_name, name);
+    return this._info;
   }
 }

jetstream/src/jsbaseclient_api.ts

@@ -119,8 +119,8 @@ export class BaseApiClientImpl {
       } catch (err) {
         const re = err instanceof RequestError ? err as RequestError : null;
         if (
-          err instanceof errors.TimeoutError ||
-          re?.isNoResponders() && i + 1 < retries
+          (err instanceof errors.TimeoutError || re?.isNoResponders()) &&
+          i + 1 < retries
         ) {
           await delay(bo.backoff(i));
         } else {

jetstream/src/jserrors.ts

@@ -60,7 +60,9 @@ export class JetStreamStatus {
 
   debug() {
     console.log({
-      message: this.description,
+      subject: this.msg.subject,
+      reply: this.msg.reply,
+      description: this.description,
       status: this.code,
       headers: this.msg.headers,
     });

jetstream/src/pushconsumer.ts

@@ -46,6 +46,7 @@ import type {
   QueuedIterator,
   Status,
   Subscription,
+  SubscriptionImpl,
 } from "@nats-io/nats-core/internal";
 import { JetStreamStatus } from "./jserrors.ts";
 
@@ -281,9 +282,26 @@ export class PushConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
         }
         this.monitor?.work();
 
-        const isProtocol = msg.subject === subject;
+        // need to make sure to catch all protocol messages even
+        const isProtocol = this.ordered
+          ? msg.subject.indexOf(this?.deliverPrefix!) === 0
+          : msg.subject === subject;
+
         if (isProtocol) {
+          if (msg.subject !== (this.sub as SubscriptionImpl).subject) {
+            // this is a stale message - was not sent to the current inbox
+            return;
+          }
+
           const status = new JetStreamStatus(msg);
+          if (status.isFlowControlRequest()) {
+            this._push(() => {
+              msg.respond();
+              this.notify(ConsumerDebugEvents.FlowControl, null);
+            });
+            return;
+          }
+
           if (status.isIdleHeartbeat()) {
             const natsLastConsumer = msg.headers?.get("Nats-Last-Consumer");
             const natsLastStream = msg.headers?.get("Nats-Last-Stream");
@@ -293,14 +311,6 @@ export class PushConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
             });
             return;
           }
-          if (status.isFlowControlRequest()) {
-            status.debug();
-            this._push(() => {
-              msg.respond();
-              this.notify(ConsumerDebugEvents.FlowControl, null);
-            });
-            return;
-          }
 
           const code = status.code;
           const description = status.description;

kv/deno.json

@@ -34,6 +34,6 @@
   },
   "imports": {
     "@nats-io/nats-core": "jsr:@nats-io/nats-core@~3.0.0-34",
-    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-21"
+    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-22"
   }
 }

kv/import_map.json

@@ -2,8 +2,8 @@
   "imports": {
     "@nats-io/nats-core": "jsr:@nats-io/nats-core@~3.0.0-34",
     "@nats-io/nats-core/internal": "jsr:@nats-io/nats-core@~3.0.0-34/internal",
-    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-21",
-    "@nats-io/jetstream/internal": "jsr:@nats-io/jetstream@~3.0.0-21/internal",
+    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-22",
+    "@nats-io/jetstream/internal": "jsr:@nats-io/jetstream@~3.0.0-22/internal",
     "test_helpers": "../test_helpers/mod.ts",
     "@nats-io/nkeys": "jsr:@nats-io/[email protected]",
     "@nats-io/nuid": "jsr:@nats-io/[email protected]",

kv/package.json

@@ -34,7 +34,7 @@
   },
   "description": "kv library - this library implements all the base functionality for NATS KV javascript clients",
   "dependencies": {
-    "@nats-io/jetstream": "3.0.0-21",
+    "@nats-io/jetstream": "3.0.0-22",
     "@nats-io/nats-core": "3.0.0-34"
   },
   "devDependencies": {

obj/deno.json

@@ -34,6 +34,6 @@
   },
   "imports": {
     "@nats-io/nats-core": "jsr:@nats-io/nats-core@~3.0.0-34",
-    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-21"
+    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-22"
   }
 }

obj/import_map.json

@@ -2,8 +2,8 @@
   "imports": {
     "@nats-io/nats-core": "jsr:@nats-io/nats-core@~3.0.0-34",
     "@nats-io/nats-core/internal": "jsr:@nats-io/nats-core@~3.0.0-34/internal",
-    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-21",
-    "@nats-io/jetstream/internal": "jsr:@nats-io/jetstream@~3.0.0-21/internal",
+    "@nats-io/jetstream": "jsr:@nats-io/jetstream@~3.0.0-22",
+    "@nats-io/jetstream/internal": "jsr:@nats-io/jetstream@~3.0.0-22/internal",
     "test_helpers": "../test_helpers/mod.ts",
     "@nats-io/nkeys": "jsr:@nats-io/[email protected]",
     "@nats-io/nuid": "jsr:@nats-io/[email protected]",

obj/package.json

@@ -34,7 +34,7 @@
   },
   "description": "obj library - this library implements all the base functionality for NATS objectstore for javascript clients",
   "dependencies": {
-    "@nats-io/jetstream": "3.0.0-21",
+    "@nats-io/jetstream": "3.0.0-22",
     "@nats-io/nats-core": "3.0.0-34"
   },
   "devDependencies": {

transport-node/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nats-io/transport-node",
-  "version": "3.0.0-19",
+  "version": "3.0.0-20",
   "description": "Node.js client for NATS, a lightweight, high-performance cloud native messaging system",
   "keywords": [
     "nats",
@@ -64,7 +64,7 @@
     "nats-jwt": "^0.0.9",
     "shx": "^0.3.3",
     "typescript": "5.6.3",
-    "@nats-io/jetstream": "3.0.0-21",
+    "@nats-io/jetstream": "3.0.0-22",
     "@nats-io/kv": "3.0.0-16",
     "@nats-io/obj": "3.0.0-17"
   }

transport-node/src/version.ts

@@ -1,2 +1,2 @@
 // This file is generated - do not edit
-export const version = "3.0.0-19";
+export const version = "3.0.0-20";