bug: Run subscriptions: aborting after break throws uncaught TypeError: Controller is already closed #5030

Provide environment information

Versions: @trigger.dev/core 4.5.7 and 4.7.4 (latest), Node 24.14, Chrome 153

Describe the bug

If a run subscription's signal is aborted after the consumer has stopped iterating, createAsyncIterableReadable (packages/core/src/v3/streams/asyncIterableStream.ts) throws an uncaught TypeError. In Node this crashes the process. In Chrome it shows up as an uncaught Failed to execute 'close' on 'ReadableStreamDefaultController' error.

Reproduction repo

N/A

To reproduce

Steps to reproduce

1. npm i @trigger.dev/core@4.7.4
2. Save this as repro.mjs. It stubs fetch, so no Trigger account is needed.

import { ApiClient } from "@trigger.dev/core/v3";

// Stub the realtime endpoint: one executing run, then a live long-poll that waits until aborted.
const now = new Date().toISOString().slice(0, -1); // Postgres-style timestamp (no "Z")
let served = false;
globalThis.fetch = async (_url, init) => {
  if (served) return new Promise((_, reject) => init?.signal?.addEventListener("abort", () => reject(init.signal.reason)));
  served = true;
  const run = { id: "1", friendlyId: "run_1", taskIdentifier: "t", status: "EXECUTING", createdAt: now, updatedAt: now };
  return Response.json([{ key: "1", headers: { operation: "insert" }, value: run }, { headers: { control: "up-to-date" } }], {
    headers: { "electric-offset": "0_0", "electric-handle": "h", "electric-schema": "{}", "electric-up-to-date": "true" },
  });
};

const abort = new AbortController();
for await (const run of new ApiClient("https://api.trigger.dev", "tr_test").subscribeToRun("run_1", { signal: abort.signal })) {
  console.log("received", run.id, run.status);
  break; // stop iterating once we have what we need
}
abort.abort(); // later teardown, e.g. a component unmount
console.log("aborted");

3. Run node repro.mjs.

Expected: prints received run_1 EXECUTING and aborted, then exits 0.
Actual: prints both lines, then crashes with exit 1:

TypeError [ERR_INVALID_STATE]: Invalid state: Controller is already closed

Suggested fix

We run a variant of this as a local patch. It fixes everything above except the polling.

• Track a closed flag, and only call controller.close()/enqueue() while it's false.
• Register the listener with { once: true }, handle an already-aborted signal, catch reader.cancel() rejections, and remove the listener in finally.
• Add a cancel() to the returned stream that cancels the inner reader (and returns early if already closed).

To stop the polling on break, cancelling a RunSubscription would also need to abort its internal controller.

Additional information

Cause

The abort listener runs queueMicrotask(() => { reader.cancel(); controller.close(); }) without checking the stream's state. If the consumer already cancelled the stream (break/return), or the stream already ended or errored, close() throws inside the microtask, where nothing can catch it.

The same function causes three related problems, all reproduced on 4.7.4:

• The source errors, then the signal aborts: reader.cancel() also rejects, and nothing handles the rejection.
• The signal aborts while runs are buffered (for example, a subscribeToRunsWithTag snapshot): the read loop's own close() throws and errors the stream. The consumer loses the buffered runs and gets the TypeError instead of an end-of-stream.
• The consumer breaks without aborting: the returned stream has no cancel(), so for a still-running run, Electric keeps polling until the signal aborts (40 requests in 2 s in our test).

Related request: forward fetchClient from subscribeToRun*

runShapeStream() already passes fetchClient to Electric's ShapeStream, but ApiClient.subscribeToRun, subscribeToRunsWithTag and subscribeToBatch (packages/core/src/v3/apiClient/index.ts) don't accept or forward it.

We need it because Electric's backoff (while (true), capped at 10 s) retries network errors and 5xx responses indefinitely, and only reports non-retryable 4xx errors to onError. So onFetchError never fires while a client is offline: we saw 11 retries in 5 s and no onFetchError calls. A custom fetch would let clients see the first failed attempt and close the subscription.

The change is to add fetchClient?: typeof fetch to those options and pass fetchClient: options?.fetchClient to runShapeStream().