diff --git a/.changeset/disarm-delete-timeout.md b/.changeset/disarm-delete-timeout.md new file mode 100644 index 000000000..71e35d72e --- /dev/null +++ b/.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. diff --git a/package.json b/package.json index f0a158ebb..a35d1e9bd 100644 --- a/package.json +++ b/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", diff --git a/packages/core/smoke/disarm-delete-timeout.smoke.ts b/packages/core/smoke/disarm-delete-timeout.smoke.ts new file mode 100644 index 000000000..af5220271 --- /dev/null +++ b/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 }).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); diff --git a/packages/core/smoke/mutations/disarm-delete-timeout.json b/packages/core/smoke/mutations/disarm-delete-timeout.json new file mode 100644 index 000000000..af982719b --- /dev/null +++ b/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" + } + ] +} diff --git a/packages/core/smoke/mutations/membership-feed-reconnect.json b/packages/core/smoke/mutations/membership-feed-reconnect.json index 12f56d569..19357eb47 100644 --- a/packages/core/smoke/mutations/membership-feed-reconnect.json +++ b/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": " 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." diff --git a/packages/core/src/endpoint.ts b/packages/core/src/endpoint.ts index 13b52c3e7..6b24a5237 100644 --- a/packages/core/src/endpoint.ts +++ b/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.