Merge pull request #689 from nats-io/fix-688

[FIX] Kv Watcher reliability during server restarts

jetstream/jsclient.ts

@@ -98,6 +98,7 @@ import {
   PullOptions,
   ReplayPolicy,
 } from "./jsapi_types.ts";
+import { nuid } from "../nats-base-client/nuid.ts";
 
 export enum PubHeaders {
   MsgIdHdr = "Nats-Msg-Id",
@@ -775,6 +776,7 @@ export class JetStreamSubscriptionImpl extends TypedSubscription<JsMsg>
     const nci = this.js.nc;
     nci._resub(this.sub, newDeliver);
     const info = this.info;
+    info.config.name = nuid.next();
     info.ordered_consumer_sequence.delivery_seq = 0;
     info.flow_control.heartbeat_count = 0;
     info.flow_control.fc_count = 0;
@@ -842,6 +844,7 @@ export class JetStreamSubscriptionImpl extends TypedSubscription<JsMsg>
         // reset the consumer
         const seq = this.info?.ordered_consumer_sequence?.stream_seq || 0;
         this._resetOrderedConsumer(seq + 1);
+        this.monitor?.restart();
         // if we are ordered, we will reset the consumer and keep
         // feeding the iterator or callback - we are not stopping
         return false;

jetstream/tests/kv_test.ts

@@ -2134,3 +2134,33 @@ Deno.test("kv - honors checkAPI option", async () => {
 
   await cleanup(ns, nc);
 });
+
+Deno.test("kv - watcher on server restart", async () => {
+  let { ns, nc } = await setup(
+    jetstreamServerConf({}),
+  );
+  const js = nc.jetstream();
+  const kv = await js.views.kv("A");
+  const iter = await kv.watch();
+  const d = deferred<KvEntry>();
+  (async () => {
+    for await (const e of iter) {
+      d.resolve(e);
+      break;
+    }
+  })().then();
+
+  ns = await ns.restart();
+  console.log("server restarted");
+  for (let i = 0; i < 10; i++) {
+    try {
+      await kv.put("hello", "world");
+      break;
+    } catch {
+      await delay(500);
+    }
+  }
+
+  await d;
+  await cleanup(ns, nc);
+});

nats-base-client/idleheartbeat_monitor.ts

@@ -78,6 +78,7 @@ export class IdleHeartbeatMonitor {
     }
     this.timer = 0;
     this.autoCancelTimer = 0;
+    this.missed = 0;
   }
 
   /**

src/deno_transport.ts

@@ -157,7 +157,7 @@ export class DenoTransport implements Transport {
     const sto = await (this.loadTlsOptions(hostname));
     this.conn = await Deno.startTls(
       //@ts-ignore: just the conn
-      this.conn,
+      this.conn as Deno.TcpConn,
       sto,
     );
     // this is necessary because the startTls process doesn't