`StreamMessagesHandler` throws "Controller is already closed" errors during parallel streaming with abort

[hntrl]

Privileged issue

• I am a LangGraph.js maintainer, or was asked directly by a LangGraph.js maintainer to create an issue here.

Description

When streaming a graph with parallel LLM calls and the stream is aborted (or completes), a race condition causes numerous TypeError [ERR_INVALID_STATE]: Invalid state: Controller is already closed errors to be logged. In production scenarios with many parallel tasks/subagents, this can result in 500+ error messages flooding the console.

Reproduction

// Graph with 3 parallel nodes, each making a streaming LLM call
const graph = new StateGraph(StateAnnotation)
  .addNode("node1", node1)  // async LLM call
  .addNode("node2", node2)  // async LLM call  
  .addNode("node3", node3)  // async LLM call
  .addEdge("__start__", "node1")
  .addEdge("__start__", "node2")
  .addEdge("__start__", "node3")
  // ... fan-in edges
  .compile();

const abortController = new AbortController();
const stream = await graph.stream(input, {
  signal: abortController.signal,
  streamMode: ["messages", "values"],
});

// Abort mid-stream
setTimeout(() => abortController.abort(), 500);

for await (const chunk of stream) {
  // Process chunks...
}

Output:

Error in handler StreamMessagesHandler, handleLLMNewToken: TypeError [ERR_INVALID_STATE]: Invalid state: Controller is already closed
Error in handler StreamMessagesHandler, handleLLMNewToken: TypeError [ERR_INVALID_STATE]: Invalid state: Controller is already closed
Error in handler StreamMessagesHandler, handleLLMNewToken: TypeError [ERR_INVALID_STATE]: Invalid state: Controller is already closed
... (repeated for each in-flight token across all parallel LLM calls)

Root Cause

The issue is in IterableReadableWritableStream.push() in @langchain/langgraph-core/src/pregel/stream.ts:

push(chunk: StreamChunk) {
  this.passthroughFn?.(chunk);
  this.controller.enqueue(chunk);  // ← Throws if controller is closed
}

Race condition timeline:

Model streaming:  [sleep] → handleLLMNewToken(token) → check signal.aborted → [sleep] → ...
                                    ↑
User aborts:      ──────────────────┼─── stream.close() called
                                    │
                                    └── Token in-flight, push() throws

1. Parallel LLM calls are streaming tokens via handleLLMNewToken callbacks
2. User aborts (or stream completes naturally) → stream.close() is called
3. close() sets _closed = true and calls controller.close()
4. In-flight LLM calls wake up from their async operations and call handleLLMNewToken
5. StreamMessagesHandler._emit() calls streamFn() → stream.push()
6. push() calls controller.enqueue() on the closed controller → Error

The abort signal IS properly propagated to the models (they do check and stop), but there's an unavoidable race window between:

• Stream closing (synchronous)
• Models checking the abort signal (asynchronous, on next iteration)

[Axadali]

When streaming graphs with parallel LLM nodes, aborting or completing the stream can trigger repeated ERR_INVALID_STATE errors.

The root cause is a race between synchronous controller.close() and asynchronous LLM token callbacks. Even though the abort signal is correctly propagated, some in-flight callbacks may still attempt to enqueue after the stream has been closed. ReadableStreamDefaultController.enqueue() throws in this case, resulting in noisy but non-fatal errors.

The proposed change makes IterableReadableWritableStream.push() idempotent after close by returning early once the stream is closed (or by ignoring the expected ERR_INVALID_STATE). This aligns with standard stream semantics where late events from async producers are safely dropped.

This does not change public APIs or observable stream behavior and is backward compatible. It only hardens internal stream handling against expected async races in parallel streaming scenarios.

@hntrl -> If this approach looks good, I’m happy to implement the fix and open a PR.