Ordered consumer changes

nats-base-client/jsclient.ts

@@ -28,6 +28,7 @@ import {
   ConsumerInfo,
   ConsumerInfoable,
   ConsumerOpts,
+  CreateConsumerRequest,
   DeliverPolicy,
   Destroyable,
   Empty,
@@ -524,6 +525,8 @@ export class JetStreamClientImpl extends BaseApiClient
       jsi.config.flow_control = true;
       jsi.config.idle_heartbeat = jsi.config.idle_heartbeat || nanos(5000);
       jsi.config.ack_wait = nanos(22 * 60 * 60 * 1000);
+      jsi.config.mem_storage = true;
+      jsi.config.num_replicas = 1;
     }
 
     if (jsi.config.ack_policy === AckPolicy.NotSet) {
@@ -714,10 +717,19 @@ export class JetStreamSubscriptionImpl extends TypedSubscription<JsMsg>
     info.config.deliver_subject = newDeliver;
     info.config.deliver_policy = DeliverPolicy.StartSequence;
     info.config.opt_start_seq = sseq;
+    // put the stream name
+    const req = {} as CreateConsumerRequest;
+    req.stream_name = this.info.stream;
+    req.config = info.config;
 
     const subj = `${info.api.prefix}.CONSUMER.CREATE.${info.stream}`;
 
-    this.js._request(subj, this.info.config)
+    this.js._request(subj, req)
+      .then((v) => {
+        const ci = v as ConsumerInfo;
+        this.info!.config = ci.config;
+        this.info!.name = ci.name;
+      })
       .catch((err) => {
         // to inform the subscription we inject an error this will
         // be at after the last message if using an iterator.
@@ -751,9 +763,25 @@ export class JetStreamSubscriptionImpl extends TypedSubscription<JsMsg>
         `${Js409Errors.IdleHeartbeatMissed}: ${v}`,
         this.sub.subject,
       );
-      this.sub.callback(null, msg);
-      // if we are a handler, we'll continue reporting
-      // iterators will stop
+      const ordered = this.info?.ordered;
+      // non-ordered consumers are always notified of the condition
+      // as they need to try and recover
+      if (!ordered) {
+        this.sub.callback(null, msg);
+      } else {
+        if (!this.js.nc.protocol.connected) {
+          // we are not connected don't do anything
+          return false;
+        }
+        // reset the consumer
+        const seq = this.info?.ordered_consumer_sequence?.stream_seq || 0;
+        this._resetOrderedConsumer(seq + 1);
+        // if we are ordered, we will reset the consumer and keep
+        // feeding the iterator or callback - we are not stopping
+        return false;
+      }
+      // let the hb monitor know if we are stopping for callbacks
+      // we don't as we deliver the errors via the cb.
       return !sub.noIterator;
     };
     // this only applies for push subscriptions

tests/jetstream_test.ts

@@ -3612,7 +3612,6 @@ Deno.test("jetstream - duplicate message pub", async () => {
 
   ack = await js.publish(subj, Empty, { msgID: "x" });
   assertEquals(ack.duplicate, true);
-  console.log(ack);
 
   await cleanup(ns, nc);
 });
@@ -3784,3 +3783,51 @@ Deno.test("jetstream - kv and object store views reject in older servers", async
 
   await cleanup(ns, nc);
 });
+
+Deno.test("jetstream - ordered consumer reset", async () => {
+  let { ns, nc } = await setup(jetstreamServerConf({}));
+  const { subj } = await initStream(nc, "A");
+  const d = deferred<JsMsg>();
+  const js = nc.jetstream();
+  const opts = consumerOpts();
+  opts.orderedConsumer();
+  opts.callback((err, m) => {
+    if (err) {
+      fail(err.message);
+    }
+    c.unsubscribe();
+    d.resolve(m!);
+  });
+  const c = await js.subscribe(subj, opts);
+
+  // stop the server and wait until hbs are missed
+  await ns.stop();
+  while (true) {
+    const missed = (c as JetStreamSubscriptionImpl).monitor?.missed || 0;
+    const connected = (nc as NatsConnectionImpl).protocol.connected;
+    // we want to wait until after 2 because we want to have a cycle
+    // where we try to recreate the consumer, but skip it because we are
+    // not connected
+    if (!connected && missed >= 3) {
+      break;
+    }
+    await delay(300);
+  }
+  ns = await ns.restart();
+  let ack: PubAck;
+  while (true) {
+    try {
+      ack = await js.publish(subj);
+      break;
+    } catch (err) {
+      if (err.code !== ErrorCode.Timeout) {
+        fail(err.message);
+      }
+      await delay(1000);
+    }
+  }
+  await c.closed;
+
+  assertEquals((await d).seq, ack.seq);
+  await cleanup(ns, nc);
+});

tests/jsm_test.ts

@@ -1758,7 +1758,6 @@ Deno.test("jsm - discard_new_per_subject option", async () => {
   await kv.put("B", Empty);
   await assertRejects(
     async () => {
-      (nc as NatsConnectionImpl).options.debug = true;
       await kv.put("B", Empty);
     },
     Error,