nats backend services - remove sub on start to prevent potential subscription leak

Author: williamstein

src/packages/jupyter/kernel/nats-service.ts

@@ -108,7 +108,7 @@ export async function initNatsService({
       return { buffer64: bufferToBase64(buffer) };
     },
   };
-  return await createNatsJupyterService({
+   return await createNatsJupyterService({
     project_id,
     path,
     impl,

src/packages/jupyter/redux/actions.ts

@@ -28,7 +28,6 @@ const MAX_BLOB_STORE_SIZE = 100 * 1000000;
 
 declare const localStorage: any;
 
-import { reuseInFlight } from "@cocalc/util/reuse-in-flight";
 import * as immutable from "immutable";
 import { Actions } from "@cocalc/util/redux/Actions";
 import { three_way_merge } from "@cocalc/sync/editor/generic/util";
@@ -50,6 +49,7 @@ import type { Client } from "@cocalc/sync/client/types";
 import latexEnvs from "@cocalc/util/latex-envs";
 import { jupyterApiClient } from "@cocalc/nats/service/jupyter";
 import { type AKV, akv } from "@cocalc/nats/sync/akv";
+import { reuseInFlight } from "@cocalc/util/reuse-in-flight";
 
 const { close, required, defaults } = misc;
 

src/packages/jupyter/redux/project-actions.ts

@@ -33,6 +33,7 @@ import { removeJupyterRedux } from "@cocalc/jupyter/kernel";
 import { initNatsService } from "@cocalc/jupyter/kernel/nats-service";
 import { type DKV, dkv } from "@cocalc/nats/sync/dkv";
 import { computeServerManager } from "@cocalc/nats/compute/manager";
+import { reuseInFlight } from "@cocalc/util/reuse-in-flight";
 
 // see https://github.com/sagemathinc/cocalc/issues/8060
 const MAX_OUTPUT_SAVE_DELAY = 30000;
@@ -216,15 +217,20 @@ export class JupyterActions extends JupyterActions0 {
     );
   }
 
-  private initNatsApi = async () => {
-    const service = await initNatsService({
+  private natsService?;
+  private initNatsApi = reuseInFlight(async () => {
+    if (this.natsService != null) {
+      this.natsService.close();
+      this.natsService = null;
+    }
+    const service = (this.natsService = await initNatsService({
       project_id: this.project_id,
       path: this.path,
-    });
+    }));
     this.syncdb.on("closed", () => {
       service.close();
     });
-  };
+  });
 
   private _first_load = async () => {
     const dbg = this.dbg("_first_load");

src/packages/nats/service/service.ts

@@ -26,7 +26,7 @@ import { requestMany, respondMany } from "./many";
 import { encodeBase64, waitUntilConnected } from "@cocalc/nats/util";
 
 const DEFAULT_TIMEOUT = 10 * 1000;
-const MONITOR_INTERVAL = 90 * 1000;
+const MONITOR_INTERVAL = 45 * 1000;
 
 // switching this is awkward since it would have to be changed in projects
 // and frontends or things would hang. I'm making it toggleable just for
@@ -272,8 +272,18 @@ export class NatsService extends EventEmitter {
     try {
       await waitUntilConnected();
       this.emit("starting");
+
       this.log("starting service");
       const env = await getEnv();
+
+      // close any subscriptions by this client to the subject, which might be left from previous runs of this service.
+      // @ts-ignore
+      for (const sub of env.nc.protocol.subscriptions.subs) {
+        if (sub[1].subject == this.subject) {
+          sub[1].close();
+        }
+      }
+
       if (this.options.enableServiceFramework ?? ENABLE_SERVICE_FRAMEWORK) {
         const svcm = new Svcm(env.nc);
         const service = await svcm.add({

src/packages/util/smc-version.js

@@ -1,2 +1,2 @@
 /* autogenerated by the update_version script */
-exports.version=1745705764;
+exports.version=1745732279;