[FIX] ackAck() now provides a timeout option, and inherits default timeout from jetstream context.

Author: aricart

jetstream/consumer.ts

@@ -206,7 +206,7 @@ export class PullConsumerMessagesImpl extends QueuedIteratorImpl<JsMsg>
           }
         } else {
           // push the user message
-          this._push(toJsMsg(msg));
+          this._push(toJsMsg(msg, this.consumer.api.timeout));
           this.received++;
           if (this.pending.msgs) {
             this.pending.msgs--;

jetstream/jsclient.ts

@@ -323,7 +323,7 @@ export class JetStreamClientImpl extends BaseApiClientImpl
     if (err) {
       throw err;
     }
-    return toJsMsg(msg);
+    return toJsMsg(msg, this.timeout);
   }
 
   /*
@@ -436,7 +436,7 @@ export class JetStreamClientImpl extends BaseApiClientImpl
           // if we are doing heartbeats, message resets
           monitor?.work();
           qi.received++;
-          qi.push(toJsMsg(msg));
+          qi.push(toJsMsg(msg, this.timeout));
         }
       },
     });
@@ -700,7 +700,7 @@ export class JetStreamClientImpl extends BaseApiClientImpl
     jsi: JetStreamSubscriptionInfo,
   ): TypedSubscriptionOptions<JsMsg> {
     const so = {} as TypedSubscriptionOptions<JsMsg>;
-    so.adapter = msgAdapter(jsi.callbackFn === undefined);
+    so.adapter = msgAdapter(jsi.callbackFn === undefined, this.timeout);
     so.ingestionFilterFn = JetStreamClientImpl.ingestionFn(jsi.ordered);
     so.protocolFilterFn = (jm, ingest = false): boolean => {
       const jsmi = jm as JsMsgImpl;
@@ -1026,44 +1026,48 @@ class JetStreamPullSubscriptionImpl extends JetStreamSubscriptionImpl
   }
 }
 
-function msgAdapter(iterator: boolean): MsgAdapter<JsMsg> {
+function msgAdapter(iterator: boolean, ackTimeout: number): MsgAdapter<JsMsg> {
   if (iterator) {
-    return iterMsgAdapter;
+    return iterMsgAdapter(ackTimeout);
   } else {
-    return cbMsgAdapter;
+    return cbMsgAdapter(ackTimeout);
   }
 }
 
-function cbMsgAdapter(
-  err: NatsError | null,
-  msg: Msg,
-): [NatsError | null, JsMsg | null] {
-  if (err) {
-    return [err, null];
-  }
-  err = checkJsError(msg);
-  if (err) {
-    return [err, null];
-  }
-  // assuming that the protocolFilterFn is set!
-  return [null, toJsMsg(msg)];
+function cbMsgAdapter(ackTimeout: number): MsgAdapter<JsMsg> {
+  return (
+    err: NatsError | null,
+    msg: Msg,
+  ): [NatsError | null, JsMsg | null] => {
+    if (err) {
+      return [err, null];
+    }
+    err = checkJsError(msg);
+    if (err) {
+      return [err, null];
+    }
+    // assuming that the protocolFilterFn is set!
+    return [null, toJsMsg(msg, ackTimeout)];
+  };
 }
 
-function iterMsgAdapter(
-  err: NatsError | null,
-  msg: Msg,
-): [NatsError | null, JsMsg | null] {
-  if (err) {
-    return [err, null];
-  }
-  // iterator will close if we have an error
-  // check for errors that shouldn't close it
-  const ne = checkJsError(msg);
-  if (ne !== null) {
-    return [hideNonTerminalJsErrors(ne), null];
-  }
-  // assuming that the protocolFilterFn is set
-  return [null, toJsMsg(msg)];
+function iterMsgAdapter(ackTimeout: number): MsgAdapter<JsMsg> {
+  return (
+    err: NatsError | null,
+    msg: Msg,
+  ): [NatsError | null, JsMsg | null] => {
+    if (err) {
+      return [err, null];
+    }
+    // iterator will close if we have an error
+    // check for errors that shouldn't close it
+    const ne = checkJsError(msg);
+    if (ne !== null) {
+      return [hideNonTerminalJsErrors(ne), null];
+    }
+    // assuming that the protocolFilterFn is set
+    return [null, toJsMsg(msg, ackTimeout)];
+  };
 }
 
 function hideNonTerminalJsErrors(ne: NatsError): NatsError | null {

jetstream/jsmsg.ts

@@ -117,8 +117,11 @@ export interface JsMsg {
    * Indicate to the JetStream server that the message was processed
    * successfully and that the JetStream server should acknowledge back
    * that the acknowledgement was received.
+   * @param opts are optional options (currently only a timeout value
+   * if not specified uses the timeout specified in the JetStreamOptions
+   * when creating the JetStream context.
    */
-  ackAck(): Promise<boolean>;
+  ackAck(opts?: Partial<{ timeout: number }>): Promise<boolean>;
 
   /**
    * Convenience method to parse the message payload as JSON. This method
@@ -133,8 +136,8 @@ export interface JsMsg {
   string(): string;
 }
 
-export function toJsMsg(m: Msg): JsMsg {
-  return new JsMsgImpl(m);
+export function toJsMsg(m: Msg, ackTimeout = 5000): JsMsg {
+  return new JsMsgImpl(m, ackTimeout);
 }
 
 export function parseInfo(s: string): DeliveryInfo {
@@ -172,10 +175,12 @@ export class JsMsgImpl implements JsMsg {
   msg: Msg;
   di?: DeliveryInfo;
   didAck: boolean;
+  timeout: number;
 
-  constructor(msg: Msg) {
+  constructor(msg: Msg, timeout: number) {
     this.msg = msg;
     this.didAck = false;
+    this.timeout = timeout;
   }
 
   get subject(): string {
@@ -228,16 +233,18 @@ export class JsMsgImpl implements JsMsg {
 
   // this has to dig into the internals as the message has access
   // to the protocol but not the high-level client.
-  async ackAck(): Promise<boolean> {
+  async ackAck(opts?: Partial<{ timeout: number }>): Promise<boolean> {
     const d = deferred<boolean>();
     if (!this.didAck) {
       this.didAck = true;
       if (this.msg.reply) {
+        opts = opts || {};
+        opts.timeout = opts.timeout || this.timeout;
         const mi = this.msg as MsgImpl;
         const proto = mi.publisher as unknown as ProtocolHandler;
         const trace = !(proto.options?.noAsyncTraces || false);
         const r = new RequestOne(proto.muxSubscriptions, this.msg.reply, {
-          timeout: 1000,
+          timeout: this.timeout,
         }, trace);
         proto.request(r);
         try {

jetstream/tests/jetstream_test.ts

@@ -62,6 +62,7 @@ import type {
 import { defaultJsOptions } from "../jsbaseclient_api.ts";
 import {
   _setup,
+  assertBetween,
   cleanup,
   jetstreamServerConf,
   Lock,
@@ -1104,9 +1105,9 @@ Deno.test("jetstream - backoff", async () => {
     return v - start;
   });
 
-  assert(delta[1] > 250 && delta[1] < 1000, `250 < ${delta[1]} < 1000`);
-  assert(delta[2] > 1250 && delta[2] < 1500, `1250 < ${delta[2]} < 1500`);
-  assert(delta[3] > 4250 && delta[3] < 4500, `4250 < ${delta[3]} < 4500`);
+  assertBetween(delta[1], 250, 1000);
+  assertBetween(delta[2], 1250, 1500);
+  assertBetween(delta[3], 4250, 4500);
 
   await cleanup(ns, nc);
 });

jetstream/tests/jsmsg_test.ts

@@ -29,6 +29,7 @@ import type { JsMsgImpl } from "../jsmsg.ts";
 import { parseInfo, toJsMsg } from "../jsmsg.ts";
 import {
   _setup,
+  assertBetween,
   cleanup,
   jetstreamServerConf,
 } from "../../test_helpers/mod.ts";
@@ -213,13 +214,80 @@ Deno.test("jsmsg - explicit consumer ackAck timeout", async () => {
   // change the subject
   ((jm as JsMsgImpl).msg as MsgImpl)._reply = "xxxx";
   nc.subscribe("xxxx");
+  const start = Date.now();
+  await assertRejects(
+    (): Promise<boolean> => {
+      return jm!.ackAck({ timeout: 1500 });
+    },
+    Error,
+    "TIMEOUT",
+  );
+  assertBetween(Date.now() - start, 1300, 1700);
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("jsmsg - ackAck js options timeout", async () => {
+  const { ns, nc } = await _setup(connect, jetstreamServerConf());
+  const jsm = await jetstreamManager(nc) as JetStreamManagerImpl;
+  await jsm.streams.add({
+    name: "A",
+    subjects: ["a.>"],
+    storage: StorageType.Memory,
+    allow_direct: true,
+  });
+
+  // default is 5000
+  const js = jetstream(nc, { timeout: 1500 });
+  await js.publish("a.a");
+
+  await jsm.consumers.add("A", { durable_name: "a" });
+  const c = await js.consumers.get("A", "a");
+  const jm = await c.next();
+  // change the subject
+  ((jm as JsMsgImpl).msg as MsgImpl)._reply = "xxxx";
+  nc.subscribe("xxxx");
+  const start = Date.now();
+  await assertRejects(
+    (): Promise<boolean> => {
+      return jm!.ackAck();
+    },
+    Error,
+    "TIMEOUT",
+  );
+  assertBetween(Date.now() - start, 1300, 1700);
+
+  await cleanup(ns, nc);
+});
+
+Deno.test("jsmsg - ackAck legacy timeout", async () => {
+  const { ns, nc } = await _setup(connect, jetstreamServerConf());
+  const jsm = await jetstreamManager(nc) as JetStreamManagerImpl;
+  await jsm.streams.add({
+    name: "A",
+    subjects: ["a.>"],
+    storage: StorageType.Memory,
+    allow_direct: true,
+  });
+
+  // default is 5000
+  const js = jetstream(nc, { timeout: 1500 });
+  await js.publish("a.a");
+
+  await jsm.consumers.add("A", { durable_name: "a" });
+  const jm = await js.pull("A", "a");
+  // change the subject
+  ((jm as JsMsgImpl).msg as MsgImpl)._reply = "xxxx";
+  nc.subscribe("xxxx");
+  const start = Date.now();
   await assertRejects(
     (): Promise<boolean> => {
       return jm!.ackAck();
     },
     Error,
     "TIMEOUT",
   );
+  assertBetween(Date.now() - start, 1300, 1700);
 
   await cleanup(ns, nc);
 });