fix: connection abort

Author: tilyupo

src/index.ts

@@ -685,15 +685,18 @@ async function fetchWithTimeout(
   if (!timeoutMs) return fetchFn(input, init);
   const timeoutController = new AbortController();
   const t = setTimeout(() => timeoutController.abort(), timeoutMs);
-  const { signal, cleanUpSignals } = combineAbortSignals([init.signal, timeoutController.signal]);
+  // Note: We intentionally don't call cleanUpSignals() here because for streaming
+  // responses (like SSE), the connection remains open after the response headers
+  // are received. The abort signal needs to remain connected so that close() can
+  // propagate the abort through the signal chain.
+  const { signal } = combineAbortSignals([init.signal, timeoutController.signal]);
   try {
     return await fetchFn(input, {
       ...init,
       signal,
     });
   } finally {
     clearTimeout(t);
-    cleanUpSignals();
   }
 }
 
@@ -703,7 +706,7 @@ async function* fetchSse(params: {
   fetchFn: typeof fetch;
   url: string;
   timeoutMs: number;
-  body?: unknown;
+  body?: string;
   headers?: Record<string, string>;
   method?: string;
   signal?: AbortSignal;
@@ -719,7 +722,7 @@ async function* fetchSse(params: {
       {
         method: params.method ?? "GET",
         headers: { Accept: "text/event-stream", ...(params.headers ?? {}) },
-        body: params.body ? JSON.stringify(params.body) : undefined,
+        body: params.body,
         signal,
       },
       params.timeoutMs,

tests/index.spec.ts

@@ -1508,6 +1508,11 @@ describe("createReplaneClient", () => {
     it("should stop receiving updates after close", async () => {
       clientPromise = createClient();
       const connection = await mockServer.acceptConnection();
+
+      // Verify signal is passed to the mock
+      expect(connection.hasSignal).toBe(true);
+      expect(connection.aborted).toBe(false);
+
       await connection.push({
         type: "init",
         configs: [{ name: "config1", overrides: [], version: 1, value: "initial" }],
@@ -1520,6 +1525,9 @@ describe("createReplaneClient", () => {
       client.close();
       await sync();
 
+      // Verify signal was aborted after close
+      expect(connection.aborted).toBe(true);
+
       // This shouldn't update the config
       await connection.push({
         type: "config_change",

tests/utils.ts

@@ -72,7 +72,7 @@ export function createReplaneServerMock(handler: ReplaneServerMockHandler) {
       });
     }
 
-    const body = JSON.parse(await req.json()) as StartReplicationStreamBody;
+    const body = (await req.json()) as StartReplicationStreamBody;
 
     if (!Array.isArray(body.currentConfigs)) {
       return new Response("Invalid request", { status: 400 });
@@ -198,48 +198,68 @@ export class MockReplaneServerController {
 export class MockReplaneServerConnection {
   private readonly _events = new Channel<ReplicationStreamRecord>();
   private readonly _signal?: AbortSignal;
+  private _closed = false;
 
   constructor(
-    private readonly body: StartReplicationStreamBody,
+    private readonly _body: StartReplicationStreamBody,
     signal?: AbortSignal
   ) {
     this._signal = signal;
 
     // Close the channel when the signal is aborted
     if (signal) {
-      signal.addEventListener("abort", () => {
+      if (signal.aborted) {
+        this._closed = true;
         this._events.close();
-      });
+      } else {
+        signal.addEventListener("abort", () => {
+          this._closed = true;
+          this._events.close();
+        });
+      }
     }
   }
 
   get signal(): AbortSignal | undefined {
     return this._signal;
   }
 
+  get hasSignal(): boolean {
+    return this._signal !== undefined;
+  }
+
   get aborted(): boolean {
-    return this._signal?.aborted ?? false;
+    return this._closed || (this._signal?.aborted ?? false);
+  }
+
+  get closed(): boolean {
+    return this._closed;
   }
 
   get events(): AsyncIterable<ReplicationStreamRecord> {
     return this._events;
   }
 
+  get requestBody(): StartReplicationStreamBody {
+    return this._body;
+  }
+
   async push(event: ReplicationStreamRecord) {
-    if (this._signal?.aborted) {
+    if (this._closed || this._signal?.aborted) {
       return;
     }
     await this._events.push(event);
   }
 
   async throw(error: Error) {
-    if (this._signal?.aborted) {
+    if (this._closed || this._signal?.aborted) {
       return;
     }
     await this._events.throw(error);
   }
 
   close() {
+    this._closed = true;
     this._events.close();
   }
 }