fix(events): abort the reconnect loop on close

src/resources/events.ts

@@ -72,6 +72,11 @@ export class EventsResource {
   subscribe(options?: SubscribeOptions): TypedEventStream<WhatsAppEvent> {
     let lastCursor = options?.cursor;
     const stallTimeoutMs = options?.stallTimeoutMs ?? DEFAULT_STALL_TIMEOUT_MS;
+    // Closing the returned stream has to kill the reconnect loop underneath it.
+    // Closing alone only queues `return()` on the generator, which lands at a
+    // yield point — and a stream that fails before its first event never
+    // reaches one, so the loop would otherwise outlive every close.
+    const controller = new AbortController();
 
     const stream = withResumableReconnect<WhatsAppEvent>(
       () =>
@@ -106,10 +111,13 @@ export class EventsResource {
         }
       },
       () => lastCursor,
-      options?.reconnect
+      { ...options?.reconnect, signal: controller.signal }
     );
 
-    return new TypedEventStream(stream);
+    return new TypedEventStream(stream, () => {
+      controller.abort();
+      return Promise.resolve();
+    });
   }
 
   async fetchMissed(options: FetchMissedOptions): Promise<FetchMissedResult> {

src/streaming/reconnect.ts

@@ -24,7 +24,8 @@ interface ResolvedOptions {
   readonly maxAttempts: number;
   readonly maxDelay: number;
   readonly multiplier: number;
-  readonly onReconnect?: (attempt: number) => void;
+  readonly onReconnect?: (attempt: number, cause?: unknown) => void;
+  readonly signal?: AbortSignal;
 }
 
 // ---------------------------------------------------------------------------
@@ -38,6 +39,7 @@ function resolveOptions(options?: ReconnectOptions): ResolvedOptions {
     multiplier: options?.multiplier ?? 2,
     maxAttempts: options?.maxAttempts ?? Number.POSITIVE_INFINITY,
     onReconnect: options?.onReconnect,
+    signal: options?.signal,
   };
 }
 
@@ -60,19 +62,29 @@ async function* consumeStream<T>(
 
 async function backoff(
   state: BackoffState,
-  opts: ResolvedOptions
+  opts: ResolvedOptions,
+  cause?: unknown
 ): Promise<boolean> {
   state.consecutiveFailures++;
 
   if (state.consecutiveFailures > opts.maxAttempts) {
     return false;
   }
 
-  opts.onReconnect?.(state.consecutiveFailures);
+  // Checked before the callback so a stream closed mid-backoff goes quiet
+  // immediately rather than emitting one last reconnect notification.
+  if (opts.signal?.aborted) {
+    return false;
+  }
+
+  opts.onReconnect?.(state.consecutiveFailures, cause);
 
-  await sleep(state.delay);
+  // A failing stream spends nearly all of its life parked here, so this is the
+  // wait that has to be interruptible — a queued `return()` on the generator
+  // cannot land while it sleeps.
+  await sleep(state.delay, opts.signal);
   state.delay = Math.min(state.delay * opts.multiplier, opts.maxDelay);
-  return true;
+  return !opts.signal?.aborted;
 }
 
 // ---------------------------------------------------------------------------
@@ -92,13 +104,20 @@ export function withReconnect<T>(
     };
 
     for (;;) {
+      if (opts.signal?.aborted) {
+        return;
+      }
+
+      let cause: unknown;
       try {
         yield* consumeStream(createStream(), state, opts);
-      } catch {
-        // Stream errored — fall through to reconnect logic.
+      } catch (error) {
+        // Stream errored — reconnect, but keep the reason: it is the only
+        // explanation a caller's onReconnect can surface.
+        cause = error;
       }
 
-      if (!(await backoff(state, opts))) {
+      if (!(await backoff(state, opts, cause))) {
         return;
       }
     }
@@ -146,18 +165,27 @@ export function withResumableReconnect<T>(
     let isFirstConnect = true;
 
     for (;;) {
+      // Covers an abort that lands while the factory or gap-fill is in flight,
+      // where there is still no yield point for a queued `return()` to reach.
+      if (opts.signal?.aborted) {
+        return;
+      }
+
+      let cause: unknown;
       try {
         if (!isFirstConnect) {
           yield* gapFill(fetchMissed, getCursor);
         }
 
         isFirstConnect = false;
         yield* consumeStream(createStream(), state, opts);
-      } catch {
-        // Stream errored — fall through to reconnect logic.
+      } catch (error) {
+        // Stream errored — reconnect, but keep the reason: it is the only
+        // explanation a caller's onReconnect can surface.
+        cause = error;
       }
 
-      if (!(await backoff(state, opts))) {
+      if (!(await backoff(state, opts, cause))) {
         return;
       }
     }
@@ -170,6 +198,20 @@ export function withResumableReconnect<T>(
 // Helpers
 // ---------------------------------------------------------------------------
 
-function sleep(ms: number): Promise<void> {
-  return new Promise((resolve) => setTimeout(resolve, ms));
+function sleep(ms: number, signal?: AbortSignal): Promise<void> {
+  if (signal?.aborted) {
+    return Promise.resolve();
+  }
+  return new Promise((resolve) => {
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    const onAbort = (): void => {
+      clearTimeout(timer);
+      resolve();
+    };
+    timer = setTimeout(() => {
+      signal?.removeEventListener("abort", onAbort);
+      resolve();
+    }, ms);
+    signal?.addEventListener("abort", onAbort, { once: true });
+  });
 }

src/types/common.ts

@@ -20,8 +20,20 @@ export interface ReconnectOptions {
   readonly maxDelay?: number;
   /** Multiplier applied to the delay after each failed attempt. Default `2`. */
   readonly multiplier?: number;
-  /** Callback invoked before each reconnect attempt. */
-  readonly onReconnect?: (attempt: number) => void;
+  /**
+   * Invoked before each reconnect attempt. `cause` is the error that ended
+   * the previous attempt, and is undefined when the stream simply ended.
+   * Without it a caller logging this callback has no way to say *why* the
+   * stream is reconnecting.
+   */
+  readonly onReconnect?: (attempt: number, cause?: unknown) => void;
+  /**
+   * Aborts the reconnect loop. `TypedEventStream.close()` wires this up, so
+   * callers rarely set it directly. Without it a stream that keeps failing
+   * before its first event never reaches a yield point, and the loop survives
+   * `close()` for the life of the process.
+   */
+  readonly signal?: AbortSignal;
 }
 
 // ---------------------------------------------------------------------------

tests/unit/events.test.ts

@@ -403,3 +403,56 @@ describe("EventsResource.subscribe teardown robustness", () => {
     expect(ids).toEqual(["msg1", "msg2"]);
   });
 });
+
+describe("EventsResource.subscribe close cancels the reconnect loop", () => {
+  // The production leak, end to end: a line whose backend rejects every
+  // connect never reaches a yield point, so closing the stream could only
+  // queue a `return()` that never landed. The loop then reconnected on its
+  // own timer for the life of the process — one immortal WARN emitter per
+  // subscribe, accumulating on every token refresh.
+  it("stops reconnecting after close() on a stream that never yielded", async () => {
+    let subscribeCalls = 0;
+    let reconnects = 0;
+
+    const client = {
+      subscribeEvents: () => {
+        subscribeCalls++;
+        return (async function* () {
+          throw new Error("connect refused");
+          // biome-ignore lint/correctness/noUnreachable: generator must be async-iterable
+          yield undefined as unknown as SubscribeEventsResponse;
+        })();
+      },
+      fetchMissedEvents: async () => ({ events: [] }),
+    } as unknown as MessageServiceClient;
+
+    const events = new EventsResource(client);
+    const stream = events.subscribe({
+      reconnect: {
+        initialDelay: 5,
+        maxDelay: 5,
+        onReconnect: () => {
+          reconnects++;
+        },
+      },
+      stallTimeoutMs: 0,
+    });
+
+    const iterator = stream[Symbol.asyncIterator]();
+    const pending = iterator.next();
+
+    await new Promise((r) => setTimeout(r, 40));
+    expect(reconnects).toBeGreaterThan(0);
+
+    await stream.close();
+    const result = await pending;
+    expect(result.done).toBe(true);
+
+    // Frozen: no further backoff ticks, no further transport calls.
+    const reconnectsAtClose = reconnects;
+    const callsAtClose = subscribeCalls;
+    await new Promise((r) => setTimeout(r, 60));
+    expect(reconnects).toBe(reconnectsAtClose);
+    expect(subscribeCalls).toBe(callsAtClose);
+  });
+});

tests/unit/reconnect.test.ts

@@ -99,3 +99,121 @@ describe("withResumableReconnect", () => {
     expect(items.map((i) => i.value)).toEqual([1, 2, 3, 10]);
   });
 });
+
+describe("reconnect cancellation", () => {
+  // Reproduces the production leak: a line whose client is closed fails on
+  // every connect attempt, so the generator never reaches a yield point and a
+  // queued `return()` can never land. Before the signal existed this loop
+  // outlived close() forever, emitting one onReconnect per backoff until the
+  // process died.
+  it("terminates a loop that never yielded, and stops reconnecting", async () => {
+    const controller = new AbortController();
+    let attempts = 0;
+    let created = 0;
+
+    const stream = withResumableReconnect<number>(
+      () => {
+        created++;
+        return {
+          // biome-ignore lint/correctness/useYield: models a stream that always fails before its first event
+          async *[Symbol.asyncIterator]() {
+            throw new Error("connect failed");
+          },
+        };
+      },
+      () => Promise.resolve([]),
+      () => undefined,
+      {
+        initialDelay: 5,
+        maxDelay: 5,
+        onReconnect: () => {
+          attempts++;
+        },
+        signal: controller.signal,
+      }
+    );
+
+    const iterator = stream[Symbol.asyncIterator]();
+    const next = iterator.next();
+
+    // Let it spin through a few failed connects, then abort mid-backoff.
+    await new Promise((r) => setTimeout(r, 40));
+    const attemptsAtAbort = attempts;
+    expect(attemptsAtAbort).toBeGreaterThan(0);
+
+    controller.abort();
+
+    // The pending next() resolves as done rather than hanging: the loop
+    // returned instead of sleeping out its backoff.
+    const result = await next;
+    expect(result.done).toBe(true);
+
+    // And nothing keeps ticking afterwards.
+    const createdAtAbort = created;
+    await new Promise((r) => setTimeout(r, 40));
+    expect(attempts).toBe(attemptsAtAbort);
+    expect(created).toBe(createdAtAbort);
+  });
+
+  it("does not fire onReconnect when aborted before the callback", async () => {
+    const controller = new AbortController();
+    controller.abort();
+    let attempts = 0;
+
+    const stream = withReconnect<number>(
+      () => ({
+        // biome-ignore lint/correctness/useYield: models a stream that always fails before its first event
+        async *[Symbol.asyncIterator]() {
+          throw new Error("connect failed");
+        },
+      }),
+      {
+        initialDelay: 5,
+        onReconnect: () => {
+          attempts++;
+        },
+        signal: controller.signal,
+      }
+    );
+
+    const items: number[] = [];
+    for await (const item of stream) {
+      items.push(item);
+    }
+    expect(items).toEqual([]);
+    expect(attempts).toBe(0);
+  });
+});
+
+describe("onReconnect cause", () => {
+  it("forwards the error that ended the previous attempt", async () => {
+    const causes: unknown[] = [];
+    const controller = new AbortController();
+
+    const stream = withReconnect<number>(
+      () => ({
+        // biome-ignore lint/correctness/useYield: models a stream that always fails before its first event
+        async *[Symbol.asyncIterator]() {
+          throw new Error("boom");
+        },
+      }),
+      {
+        initialDelay: 5,
+        maxDelay: 5,
+        onReconnect: (_attempt, cause) => {
+          causes.push(cause);
+        },
+        signal: controller.signal,
+      }
+    );
+
+    const iterator = stream[Symbol.asyncIterator]();
+    const pending = iterator.next();
+    await new Promise((r) => setTimeout(r, 30));
+    controller.abort();
+    await pending;
+
+    expect(causes.length).toBeGreaterThan(0);
+    expect((causes[0] as Error).message).toBe("boom");
+  });
+});