Fix Issue: StreamMessagesHandler throws "Controller is already closed" errors during parallel streaming with abort #1908

[Axadali]

Fix Race Condition in IterableReadableWritableStream.push()

Summary

This PR addresses a critical race condition in IterableReadableWritableStream.push() that was causing TypeError [ERR_INVALID_STATE]: Invalid state: Controller is already closed errors when streaming a graph with multiple parallel LLM nodes and aborting the stream (or when it completes naturally).

Problem

Users were experiencing numerous ERR_INVALID_STATE errors in production when:

• Streaming graphs with multiple parallel LLM nodes
• Aborting the stream or when it completes naturally
• In-flight token callbacks from LLMs are asynchronous and may still execute after stream closure
• Calling enqueue() on a closed controller throws the error
• This race condition causes console flooding in production environments

Solution

The fix implements a robust two-layer approach to handle the race condition:

1. Pre-checking State: Checking this._closed and this.controller existence before attempting enqueue
2. Try-Catch Protection: Wrapping the enqueue operation to catch any remaining race conditions
3. Specific Error Handling: Only suppressing the "Controller is already closed" error while allowing other errors to propagate
4. Maintaining Semantics: Preserving all existing functionality while preventing the problematic errors

Code Changes

Modified push() method in libs/langgraph-core/src/pregel/stream.ts:

push(chunk: StreamChunk) {
  // Prevent pushing to a closed stream to avoid race condition errors
  if (this._closed || !this.controller) {
    // Silently drop chunks when stream is closed - this is expected behavior
    // when async operations try to push after stream termination
    return;
  }

  try {
    // Forward chunk to passthrough function if provided
    this.passthroughFn?.(chunk);

    // Attempt to enqueue the chunk to the underlying stream
    this.controller.enqueue(chunk);
  } catch (error) {
    // Handle the specific case where controller was closed between check and enqueue
    // This race condition can occur with parallel async operations
    if (error instanceof TypeError && 
        error.message.includes('Controller is already closed')) {
      // Silently ignore - this is expected during stream closure with concurrent pushes
      return;
    }
    // Re-throw any other unexpected errors to maintain proper error reporting
    throw error;
  }
}

Enhanced Test Coverage

Added comprehensive test scenarios covering:

• Basic race condition handling
• Concurrent pushes during closure
• Rapid successive operations
• Passthrough function integration during race conditions
• Multiple close calls
• Parallel node simulation mimicking the original issue

Backwards Compatibility

• ✅ No changes to public API
• ✅ No changes to stream behavior during normal operation
• ✅ Maintains all existing functionality
• ✅ Only affects the error case (pushes after stream closure)

Testing

The fix has been validated across multiple scenarios:

• Basic race condition scenarios
• Concurrent operations during stream closure
• Multiple parallel nodes simulating the original issue
• Edge cases with multiple close calls
• Passthrough function integration

Impact

• ✅ Eliminates console flooding with ERR_INVALID_STATE errors
• ✅ Improves stability in production environments with parallel streaming
• ✅ Maintains performance and functionality of normal stream operations
• ✅ Safe for use with multiple parallel LLM nodes

This fix resolves the race condition issue while maintaining full backward compatibility and following best practices for error handling in async environments.

[changeset-bot]

🦋 Changeset detected

Latest commit: 26400bd

The changes in this PR will be included in the next version bump.

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[pkg-pr-new]

Open in StackBlitz

@langchain/langgraph-checkpoint

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-checkpoint@1959

@langchain/langgraph-checkpoint-mongodb

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-checkpoint-mongodb@1959

@langchain/langgraph-checkpoint-postgres

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-checkpoint-postgres@1959

@langchain/langgraph-checkpoint-redis

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-checkpoint-redis@1959

@langchain/langgraph-checkpoint-sqlite

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-checkpoint-sqlite@1959

@langchain/langgraph-checkpoint-validation

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-checkpoint-validation@1959

create-langgraph

npm i https://pkg.pr.new/langchain-ai/langgraphjs/create-langgraph@1959

@langchain/langgraph-api

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-api@1959

@langchain/langgraph-cli

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-cli@1959

@langchain/langgraph

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph@1959

@langchain/langgraph-cua

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-cua@1959

@langchain/langgraph-supervisor

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-supervisor@1959

@langchain/langgraph-swarm

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-swarm@1959

@langchain/langgraph-ui

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-ui@1959

@langchain/langgraph-sdk

npm i https://pkg.pr.new/langchain-ai/langgraphjs/@langchain/langgraph-sdk@1959

commit: c60d111