fix(core): do not kill the process on a membership-watch delete timeout

.changeset/disarm-delete-timeout.md

@@ -0,0 +1,5 @@
+---
+"@cotal-ai/core": patch
+---
+
+Treat a CONSUMER.DELETE timeout during membership-watch disarm as best-effort cleanup: catch it, emit an endpoint error, and continue. A live observer over a slow link must not die because cleanup did not answer in time.

package.json

@@ -66,6 +66,7 @@
     "smoke:lang-differential": "tsx packages/lang/smoke/differential.smoke.ts",
     "smoke:orientation": "tsx extensions/connector-core/smoke/orientation.smoke.ts",
     "smoke:docs": "tsx extensions/connector-core/smoke/docs.smoke.ts",
+    "smoke:disarm-delete-timeout": "tsx packages/core/smoke/disarm-delete-timeout.smoke.ts",
     "smoke:pi": "tsx extensions/pi/pi.smoke.ts && tsx extensions/pi/pi-sdk.smoke.ts",
     "smoke:codex-args": "tsx extensions/connector-codex/smoke/codex-args.smoke.ts",
     "smoke:codex-host": "tsx extensions/connector-codex/smoke/codex-host.smoke.ts",

packages/core/smoke/disarm-delete-timeout.smoke.ts

@@ -0,0 +1,110 @@
+/**
+ * #1047: a CONSUMER.DELETE timeout during membership-watch disarm is cleanup, not a process kill.
+ *
+ * The live crash was an unhandled TimeoutError from consumer.delete() in disarmMembershipWatch
+ * on a live observer over a slow VPN. The JS-API request timed out; the endpoint was still
+ * usable; the broker reaps the consumer anyway.
+ *
+ * This suite injects that rejection at the delete boundary (no broker). A live TCP stall of
+ * CONSUMER.DELETE is a named gap: this box must not drive cotal web against the production
+ * mesh, and a stall proxy is the transport-liveness suite's shape, not required to prove the
+ * catch policy.
+ *
+ * Run: pnpm smoke:disarm-delete-timeout
+ */
+import { CotalEndpoint } from "../src/endpoint.js";
+import type { PushConsumer } from "@nats-io/jetstream";
+
+let pass = 0;
+let fail = 0;
+function check(name: string, cond: boolean, extra?: unknown): void {
+  if (cond) {
+    pass++;
+    console.log(`  ✓ ${name}`);
+  } else {
+    fail++;
+    console.log(`  ✗ FAIL: ${name}`, extra ?? "");
+  }
+}
+
+class TimeoutError extends Error {
+  constructor() {
+    super("timeout");
+    this.name = "TimeoutError";
+  }
+}
+
+const ep = new CotalEndpoint({
+  space: "disarm-timeout",
+  servers: "nats://127.0.0.1:1",
+  card: { name: "observer", kind: "agent" },
+  registerPresence: false,
+  watchPresence: false,
+  consume: false,
+});
+
+const emitted: Error[] = [];
+ep.on("error", (err) => { emitted.push(err as Error); });
+(ep as unknown as { nc: { isClosed(): boolean }; reconnecting: boolean }).nc = { isClosed: () => false };
+(ep as unknown as { reconnecting: boolean }).reconnecting = false;
+
+const watch = {
+  onChange: () => {},
+  stopped: false,
+  arm: Promise.resolve(),
+  consumerStream: "KV_membership",
+  consumerName: "ordered-watch",
+  consumer: {
+    delete: () => Promise.reject(new TimeoutError()),
+  } as unknown as PushConsumer,
+};
+
+const disarm = (ep as unknown as { disarmMembershipWatch(watch: typeof watch): Promise<void> }).disarmMembershipWatch.bind(ep);
+
+let threw = false;
+let thrown: unknown;
+try {
+  await disarm(watch);
+} catch (err) {
+  threw = true;
+  thrown = err;
+}
+
+check("a live delete timeout does not reject disarmMembershipWatch", threw === false, thrown);
+check("the timeout is surfaced as an endpoint error event", emitted.length === 1 && emitted[0]?.name === "TimeoutError" && emitted[0]?.message === "timeout", emitted);
+check("consumer identity is kept for a later cleanup retry", watch.consumerStream === "KV_membership" && watch.consumerName === "ordered-watch", watch);
+
+const authWatch = {
+  onChange: () => {},
+  stopped: false,
+  arm: Promise.resolve(),
+  consumer: {
+    delete: () => Promise.reject(Object.assign(new Error("permissions violation for subscription"), { code: 503 })),
+  } as unknown as PushConsumer,
+};
+let authThrew = false;
+try {
+  await disarm(authWatch);
+} catch {
+  authThrew = true;
+}
+check("a non-timeout delete failure still throws", authThrew === true);
+
+const missingWatch = {
+  onChange: () => {},
+  stopped: false,
+  arm: Promise.resolve(),
+  consumerStream: "KV_membership",
+  consumerName: "gone",
+  consumer: {
+    delete: () => Promise.reject(Object.assign(new Error("consumer not found"), { code: 404 })),
+  } as unknown as PushConsumer,
+};
+await disarm(missingWatch);
+check("a 404 delete still clears consumer identity", missingWatch.consumerStream === undefined && missingWatch.consumerName === undefined, missingWatch);
+
+const EXPECTED = 5;
+check(`every cell ran - ${EXPECTED} expected`, pass + fail === EXPECTED, `${pass + fail} cells reported`);
+
+console.log(`DISARM-DELETE-TIMEOUT SMOKE ${fail === 0 ? "OK" : "FAILED"}`);
+process.exit(fail === 0 ? 0 : 1);

packages/core/smoke/mutations/disarm-delete-timeout.json

@@ -0,0 +1,29 @@
+{
+  "suite": "packages/core/smoke/disarm-delete-timeout.smoke.ts",
+  "guard": "a CONSUMER.DELETE timeout during membership-watch disarm is best-effort cleanup, not a process kill",
+  "command": "pnpm smoke:disarm-delete-timeout",
+  "proveWith": "node scripts/mutation-proof.mjs --config packages/core/smoke/mutations/disarm-delete-timeout.json",
+  "completionMarker": "DISARM-DELETE-TIMEOUT SMOKE",
+  "why": [
+    "The live crash was TimeoutError from consumer.delete in disarmMembershipWatch on a live",
+    "observer. The old catch rethrew any timeout while the connection was still open.",
+    "",
+    "A1 restores that live throw. A2 swallows the timeout without emitting the error event."
+  ],
+  "mutations": [
+    {
+      "name": "A1 live delete timeout still throws, the measured crash",
+      "file": "packages/core/src/endpoint.ts",
+      "find": "          if (timeout || closedEpoch || dyingEpochTimeout) {\n            this.emit(\"error\", err as Error);\n          } else {\n            throw err;\n          }",
+      "replace": "          if (!closedEpoch && !dyingEpochTimeout) throw err;",
+      "expectRed": "a live delete timeout does not reject disarmMembershipWatch"
+    },
+    {
+      "name": "A2 swallow timeout without surfacing an endpoint error",
+      "file": "packages/core/src/endpoint.ts",
+      "find": "          if (timeout || closedEpoch || dyingEpochTimeout) {\n            this.emit(\"error\", err as Error);\n          } else {\n            throw err;\n          }",
+      "replace": "          if (timeout || closedEpoch || dyingEpochTimeout) {\n            /* swallowed */\n          } else {\n            throw err;\n          }",
+      "expectRed": "the timeout is surfaced as an endpoint error event"
+    }
+  ]
+}

packages/core/smoke/mutations/membership-feed-reconnect.json

@@ -31,8 +31,8 @@
     {
       "name": "closed epoch delete rejects public stop instead of deferring cleanup",
       "file": "packages/core/src/endpoint.ts",
-      "find": "          const closedEpoch = (err as Error).name === \"ClosedConnectionError\" || /^closed connection$/i.test((err as Error).message);\n          const dyingEpochTimeout = /timeout/i.test((err as Error).message) && (this.reconnecting || !this.nc || this.nc.isClosed());\n          if (!closedEpoch && !dyingEpochTimeout) throw err;\n",
-      "replace": "          const closedEpoch = false;\n          const dyingEpochTimeout = false;\n          throw err;\n",
+      "find": "          const closedEpoch = (err as Error).name === \"ClosedConnectionError\" || /^closed connection$/i.test((err as Error).message);\n          const timeout = (err as Error).name === \"TimeoutError\" || /timeout/i.test((err as Error).message);\n          const dyingEpochTimeout = timeout && (this.reconnecting || !this.nc || this.nc.isClosed());\n          // Cleanup of an ordered consumer: a delete timeout means the broker did not answer in time,\n          // not that the endpoint is unusable. The broker reaps an idle/ephemeral consumer anyway.\n          // Throwing here killed a live observer over a slow VPN (#1047). Catch, surface, continue.\n          if (timeout || closedEpoch || dyingEpochTimeout) {\n            this.emit(\"error\", err as Error);\n          } else {\n            throw err;\n          }\n",
+      "replace": "          throw err;\n",
       "expectRed": "public stop concurrent with terminal close resolves after fresh cleanup",
       "cell": "public stop concurrent with terminal close resolves after fresh cleanup",
       "note": "A closed epoch is deferred cleanup, not deletion success and not a caller-visible stop failure."

packages/core/src/endpoint.ts

@@ -1062,7 +1062,7 @@ export class CotalEndpoint extends EventEmitter {
     this.channelConfigs.clear();
     this.channelDefaults = {};
     for (const watch of this.membershipFeedWatches)
-      watch.arm = watch.arm.catch(() => {}).then(() => this.disarmMembershipWatch(watch));
+      watch.arm = watch.arm.catch(() => {}).then(() => this.disarmMembershipWatch(watch)).catch((err) => { this.emit("error", err as Error); });
   }
 
   /** If stop() ran during a rebuild's `await connectAndBind`, the just-bound connection +
@@ -2307,10 +2307,17 @@ export class CotalEndpoint extends EventEmitter {
           watch.consumerStream = undefined;
           watch.consumerName = undefined;
         } else {
-          // A timeout is deferred only for an epoch that is actually closing/rebuilding; live timeouts stay loud.
           const closedEpoch = (err as Error).name === "ClosedConnectionError" || /^closed connection$/i.test((err as Error).message);
-          const dyingEpochTimeout = /timeout/i.test((err as Error).message) && (this.reconnecting || !this.nc || this.nc.isClosed());
-          if (!closedEpoch && !dyingEpochTimeout) throw err;
+          const timeout = (err as Error).name === "TimeoutError" || /timeout/i.test((err as Error).message);
+          const dyingEpochTimeout = timeout && (this.reconnecting || !this.nc || this.nc.isClosed());
+          // Cleanup of an ordered consumer: a delete timeout means the broker did not answer in time,
+          // not that the endpoint is unusable. The broker reaps an idle/ephemeral consumer anyway.
+          // Throwing here killed a live observer over a slow VPN (#1047). Catch, surface, continue.
+          if (timeout || closedEpoch || dyingEpochTimeout) {
+            this.emit("error", err as Error);
+          } else {
+            throw err;
+          }
         }
         // A terminal close leaves stream/name intact. The endpoint-owned stopped intent is retried
         // through the fresh JetStream manager before its public stop promise may resolve.