langchain-ai/langgraphjs
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraphjs.git synced 2026-07-23 01:26:57 -04:00
Code Issues 108 Packages Projects Releases 20 Wiki Activity

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

New Issue
Open
opened 2026-02-15 18:16:25 -05:00 by yindo · 1 comment
yindo commented 2026-02-15 18:16:25 -05:00
Owner
Copy Link
Copy Source

Originally created by @hntrl on GitHub (Jan 16, 2026).

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)
Originally created by @hntrl on GitHub (Jan 16, 2026). ### Privileged issue - [x] 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 ```typescript // 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`: ```typescript 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)
yindo commented 2026-02-15 18:16:26 -05:00
Author
Owner
Copy Link
Copy Source

@Axadali commented on GitHub (Feb 2, 2026):

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.

@Axadali commented on GitHub (Feb 2, 2026): 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.
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
bug
checkpointer
docs-needed
documentation
enhancement
good first issue
good writeup
help wanted
invalid
langgraph-supervisor
open-swe-auto
platform
prebuilt
pull-request
Mirrored from GitHub Pull Request
 question
No Label
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraphjs#395
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?