From 030dbe9a7e6a09070e4027bebe8dd4c5a5091126 Mon Sep 17 00:00:00 2001 From: wahaha1223 <0668001153@xydigit.com> Date: Thu, 30 Jul 2026 00:49:11 +0800 Subject: [PATCH] fix(tlon): stop SSE reconnects after client close (#108168) * fix(tlon): stop reconnecting after SSE client close * fix(tlon): cancel SSE reconnect timers when monitoring stops Preserve the original Tlon SSE reconnect fix while canceling both retry timers at stopReceiving(), the real monitor shutdown boundary. Verified against an authenticated, SSRF-guarded loopback SSE server; both reconnect waits settle immediately without changing replay, close cleanup, or uninterrupted reconnection. Co-authored-by: wahaha1223 <0668001153@xydigit.com> * fix(tlon): prepare conflict-free reconnect refresh Prepare a conflict-free refresh of the original Tlon SSE shutdown correction without changing its ownership or losing the contributor commit. Co-authored-by: wahaha1223 <0668001153@xydigit.com> * fix(tlon): cancel reconnect timers when monitoring stops Cancel both Tlon SSE reconnect waits at stopReceiving(), the monitor-owned shutdown boundary, while preserving guarded HTTP, durable same-channel replay, and the original contributor's fix. Verified with authenticated loopback SSE, both real retry delays, normal reconnection, all 252 Tlon tests, extension type checks, lint, and focused owner regressions. Co-authored-by: wahaha1223 <0668001153@xydigit.com> --------- Co-authored-by: Peter Steinberger --- .../urbit/sse-client.reconnect-close.test.ts | 234 ++++++++++++++++++ .../sse-client.reconnect-close.unit.test.ts | 43 ++++ extensions/tlon/src/urbit/sse-client.ts | 29 ++- 3 files changed, 298 insertions(+), 8 deletions(-) create mode 100644 extensions/tlon/src/urbit/sse-client.reconnect-close.test.ts create mode 100644 extensions/tlon/src/urbit/sse-client.reconnect-close.unit.test.ts diff --git a/extensions/tlon/src/urbit/sse-client.reconnect-close.test.ts b/extensions/tlon/src/urbit/sse-client.reconnect-close.test.ts new file mode 100644 index 000000000000..2f83a8af7778 --- /dev/null +++ b/extensions/tlon/src/urbit/sse-client.reconnect-close.test.ts @@ -0,0 +1,234 @@ +import { createServer, type Server, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; +import type { LookupFn } from "openclaw/plugin-sdk/ssrf-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { UrbitSSEClient } from "./sse-client.js"; + +const proofCookie = "urbauth-~zod=synthetic-reconnect-proof"; +const lookupLoopback = (async () => [{ address: "127.0.0.1", family: 4 }]) as unknown as LookupFn; +const runningServers: Server[] = []; + +type UrbitChannelProof = { + baseUrl: string; + requests: string[]; + unauthorizedRequests: number; +}; + +async function startUrbitChannelServer(options: { holdStream?: boolean } = {}) { + const proof: UrbitChannelProof = { + baseUrl: "", + requests: [], + unauthorizedRequests: 0, + }; + const heldStreams = new Set(); + const server = createServer((request, response) => { + if (!(request.headers.cookie ?? "").includes(proofCookie)) { + proof.unauthorizedRequests += 1; + response.writeHead(401).end(); + return; + } + proof.requests.push(`${request.method ?? "GET"} ${request.url ?? "/"}`); + if (request.method === "GET") { + response.writeHead(200, { + "Cache-Control": "no-cache", + "Content-Type": "text/event-stream", + }); + if (options.holdStream) { + heldStreams.add(response); + response.once("close", () => heldStreams.delete(response)); + response.write(": connected\n\n"); + } else { + response.end(); + } + return; + } + response.writeHead(204).end(); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + runningServers.push(server); + const address = server.address() as AddressInfo; + proof.baseUrl = `http://127.0.0.1:${address.port}`; + return proof; +} + +async function waitFor(predicate: () => boolean, timeoutMs = 2_000): Promise { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() >= deadline) { + throw new Error("Timed out waiting for the real Tlon SSE reconnect lifecycle"); + } + await new Promise((resolve) => { + setTimeout(resolve, 5); + }); + } +} + +async function expectPromptReconnectSettlement(reconnect: Promise) { + let deadline: ReturnType | undefined; + try { + const outcome = await Promise.race([ + reconnect.then(() => "settled" as const), + new Promise<"timed out">((resolve) => { + deadline = setTimeout(() => resolve("timed out"), 250); + }), + ]); + expect(outcome).toBe("settled"); + } finally { + if (deadline !== undefined) { + clearTimeout(deadline); + } + } +} + +afterEach(async () => { + vi.restoreAllMocks(); + await Promise.all( + runningServers.splice(0).map( + (server) => + new Promise((resolve) => { + server.close(() => resolve()); + server.closeAllConnections(); + }), + ), + ); +}); + +describe("UrbitSSEClient real reconnect shutdown lifecycle", () => { + it("settles a real SSE reconnect when the monitor stops receiving", async () => { + const proof = await startUrbitChannelServer(); + const logs: string[] = []; + const onReconnect = vi.fn(); + const client = new UrbitSSEClient(proof.baseUrl, proofCookie, { + ship: "zod", + ssrfPolicy: { allowPrivateNetwork: true }, + lookupFn: lookupLoopback, + reconnectDelay: 400, + logger: { log: (message) => logs.push(message) }, + onReconnect, + }); + const reconnectSpy = vi.spyOn(client, "attemptReconnect"); + + try { + await client.connect(); + await waitFor(() => logs.some((message) => message.includes("in 400ms"))); + const reconnect = reconnectSpy.mock.results[0]?.value as Promise | undefined; + if (!reconnect) { + throw new Error("The real SSE stream did not enter its reconnect backoff"); + } + + const requestsBeforeStop = proof.requests.length; + client.stopReceiving(); + await expectPromptReconnectSettlement(reconnect); + + expect(onReconnect).not.toHaveBeenCalled(); + expect(proof.requests).toHaveLength(requestsBeforeStop); + expect(proof.requests.some((request) => request.startsWith("GET /~/channel/"))).toBe(true); + + const unauthorizedResponse = await fetch(`${proof.baseUrl}/unauthorized-control`); + expect(unauthorizedResponse.status).toBe(401); + expect(proof.unauthorizedRequests).toBe(1); + } finally { + await client.close(); + } + }); + + it("settles the real ten-second retry cooldown when the monitor stops receiving", async () => { + const proof = await startUrbitChannelServer({ holdStream: true }); + const logs: string[] = []; + const onReconnect = vi.fn(); + const client = new UrbitSSEClient(proof.baseUrl, proofCookie, { + ship: "zod", + ssrfPolicy: { allowPrivateNetwork: true }, + lookupFn: lookupLoopback, + maxReconnectAttempts: 1, + reconnectDelay: 20, + logger: { log: (message) => logs.push(message) }, + onReconnect, + }); + + try { + await client.connect(); + client.reconnectAttempts = client.maxReconnectAttempts; + const reconnect = client.attemptReconnect(); + expect(logs.some((message) => message.includes("Waiting 10s"))).toBe(true); + + const requestsBeforeStop = proof.requests.length; + client.stopReceiving(); + await expectPromptReconnectSettlement(reconnect); + + expect(onReconnect).not.toHaveBeenCalled(); + expect(proof.requests).toHaveLength(requestsBeforeStop); + expect(logs.some((message) => message.includes("reset, resuming"))).toBe(false); + } finally { + await client.close(); + } + }); + + it("settles a real pending reconnect when the public client closes", async () => { + const proof = await startUrbitChannelServer(); + const logs: string[] = []; + const onReconnect = vi.fn(); + const client = new UrbitSSEClient(proof.baseUrl, proofCookie, { + ship: "zod", + ssrfPolicy: { allowPrivateNetwork: true }, + lookupFn: lookupLoopback, + reconnectDelay: 400, + logger: { log: (message) => logs.push(message) }, + onReconnect, + }); + const reconnectSpy = vi.spyOn(client, "attemptReconnect"); + + try { + await client.connect(); + await waitFor(() => logs.some((message) => message.includes("in 400ms"))); + const reconnect = reconnectSpy.mock.results[0]?.value as Promise | undefined; + if (!reconnect) { + throw new Error("The real SSE stream did not enter its reconnect backoff"); + } + + await client.close(); + const requestsAfterClose = proof.requests.length; + await expectPromptReconnectSettlement(reconnect); + + expect(onReconnect).not.toHaveBeenCalled(); + expect(proof.requests).toHaveLength(requestsAfterClose); + expect(proof.requests.some((request) => request.startsWith("DELETE /~/channel/"))).toBe(true); + } finally { + await client.close(); + } + }); + + it("still reconnects an uninterrupted authenticated SSE stream", async () => { + const proof = await startUrbitChannelServer(); + let reconnects = 0; + const client = new UrbitSSEClient(proof.baseUrl, proofCookie, { + ship: "zod", + ssrfPolicy: { allowPrivateNetwork: true }, + lookupFn: lookupLoopback, + reconnectDelay: 25, + onReconnect: (reconnectingClient) => { + reconnects += 1; + reconnectingClient.autoReconnect = false; + }, + }); + + try { + await client.connect(); + await waitFor(() => reconnects === 1); + await waitFor( + () => + proof.requests.filter((request) => request.startsWith("GET /~/channel/")).length === 2, + ); + expect(reconnects).toBe(1); + expect( + proof.requests.filter((request) => request.startsWith("GET /~/channel/")), + ).toHaveLength(2); + expect(proof.unauthorizedRequests).toBe(0); + } finally { + await client.close(); + } + }); +}); diff --git a/extensions/tlon/src/urbit/sse-client.reconnect-close.unit.test.ts b/extensions/tlon/src/urbit/sse-client.reconnect-close.unit.test.ts new file mode 100644 index 000000000000..94b70945502c --- /dev/null +++ b/extensions/tlon/src/urbit/sse-client.reconnect-close.unit.test.ts @@ -0,0 +1,43 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { UrbitSSEClient } from "./sse-client.js"; + +afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); +}); + +describe("UrbitSSEClient owned reconnect timers", () => { + it.each([ + { name: "ordinary reconnect", exhausted: false, delayMs: 1_000 }, + { name: "exhausted ten-second cooldown", exhausted: true, delayMs: 10_000 }, + ])("clears the $name timer immediately when the monitor stops receiving", async (params) => { + vi.useFakeTimers(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const onReconnect = vi.fn(); + const logger = { log: vi.fn() }; + const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=synthetic", { + maxReconnectAttempts: 1, + reconnectDelay: 1_000, + onReconnect, + logger, + }); + if (params.exhausted) { + client.reconnectAttempts = client.maxReconnectAttempts; + } + + const reconnect = client.attemptReconnect(); + + expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), params.delayMs); + expect(vi.getTimerCount()).toBe(1); + client.stopReceiving(); + + await expect(reconnect).resolves.toBeUndefined(); + expect(vi.getTimerCount()).toBe(0); + expect(onReconnect).not.toHaveBeenCalled(); + if (params.exhausted) { + expect(logger.log).not.toHaveBeenCalledWith( + expect.stringContaining("reset, resuming reconnection"), + ); + } + }); +}); diff --git a/extensions/tlon/src/urbit/sse-client.ts b/extensions/tlon/src/urbit/sse-client.ts index 2689d41786e7..28ce5cc456a1 100644 --- a/extensions/tlon/src/urbit/sse-client.ts +++ b/extensions/tlon/src/urbit/sse-client.ts @@ -3,6 +3,7 @@ import { randomUUID } from "node:crypto"; import { Readable } from "node:stream"; import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime"; import { readResponseTextLimited } from "openclaw/plugin-sdk/provider-http"; +import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; import type { LookupFn, SsrFPolicy } from "openclaw/plugin-sdk/ssrf-runtime"; import { ensureUrbitChannelOpen, pokeUrbitChannel, scryUrbitPath } from "./channel-ops.js"; import { getUrbitContext, normalizeUrbitCookie } from "./context.js"; @@ -89,6 +90,9 @@ export class UrbitSSEClient { private lastHeardEventId = -1; private lastAcknowledgedEventId = -1; private readonly ackThreshold = 20; + // The monitor stops ingress before it closes the channel; cancel both retry + // delays at that first shutdown edge so stale work cannot outlive its owner. + private readonly reconnectAbortController = new AbortController(); constructor(url: string, cookie: string, options: UrbitSseOptions = {}) { const ctx = getUrbitContext(url, options.ship); @@ -472,9 +476,9 @@ export class UrbitSSEClient { ); // Wait 10 seconds before resetting and trying again const extendedBackoff = 10000; // 10 seconds - await new Promise((resolve) => { - setTimeout(resolve, extendedBackoff); - }); + if (!(await this.waitForReconnectDelay(extendedBackoff)) || this.aborted) { + return; + } this.reconnectAttempts = 0; // Reset counter to continue trying this.logger.log?.("[SSE] Reconnection attempts reset, resuming reconnection..."); } @@ -489,11 +493,7 @@ export class UrbitSSEClient { `[SSE] Reconnection attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts} in ${delay}ms...`, ); - await new Promise((resolve) => { - setTimeout(resolve, delay); - }); - - if (this.aborted || !this.autoReconnect) { + if (!(await this.waitForReconnectDelay(delay)) || this.aborted || !this.autoReconnect) { return; } @@ -528,6 +528,7 @@ export class UrbitSSEClient { stopReceiving(): void { this.aborted = true; this.isConnected = false; + this.reconnectAbortController.abort(); this.streamController?.abort(); } @@ -586,6 +587,18 @@ export class UrbitSSEClient { } } + private async waitForReconnectDelay(delayMs: number): Promise { + try { + await sleepWithAbort(delayMs, this.reconnectAbortController.signal); + return true; + } catch (error) { + if (this.reconnectAbortController.signal.aborted) { + return false; + } + throw error; + } + } + private async putChannelPayload( payload: unknown, params: { timeoutMs: number; auditContext: string },