[bug] clean up resources during unhandled rejections from inside stream transforms

[narengogi]

this fixes #1206

to test, I simply modified the code inside a stream transform to throw an error

export const AnthropicChatCompleteStreamChunkTransform: (
  response: string,
  fallbackId: string,
  streamState: AnthropicStreamState,
  _strictOpenAiCompliance: boolean
) => string | undefined = (
  responseChunk,
  fallbackId,
  streamState,
  strictOpenAiCompliance
) => {
  throw new Error("unhandled error");
  let chunk = responseChunk.trim();
  if (
    chunk.startsWith('event: ping') ||
    chunk.startsWith('event: content_block_stop')
  ) {
    return;

[matter-code-review]

This PR adds important error handling and resource cleanup to prevent resource leaks during stream processing. The changes look good overall, but I have one suggestion to improve the error handling.

[matter-code-review]

Description

Summary By MatterAI

🔄 What Changed

This Pull Request introduces try...catch...finally blocks around the asynchronous stream processing loops within the handleStreamingMode function. This change applies to both readAWSStream and readStream paths.

🔍 Impact of the Change

The primary impact is improved resource management and robustness. By wrapping the stream iteration in a try...finally block, writer.close() is now guaranteed to be called even if an unhandled rejection or error occurs inside the stream transform functions (e.g., responseTransformer). This prevents potential resource leaks and ensures proper stream termination. Additionally, a catch block is added to log any errors to the console, aiding in debugging.

📁 Total Files Changed

1 file was changed: src/handlers/streamHandler.ts.

🧪 Test Added

Manual testing was performed by intentionally modifying code inside a stream transform to throw an error, verifying that the finally block correctly closes the writer and the error is caught and logged. No automated unit or integration tests were explicitly added in this PR.

🔒Security Vulnerabilities

No new security vulnerabilities were detected in this patch. The changes improve the stability and resource management of the application.

Motivation

This PR fixes issue #1206, which addresses the problem of resources not being properly cleaned up (specifically, the stream writer not being closed) when unhandled rejections occur from within stream transforms.

Type of Change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• Documentation update
• Refactoring (no functional changes)

How Has This Been Tested?

• Unit Tests
• Integration Tests
• Manual Testing

Screenshots (if applicable)

N/A

Checklist

• My code follows the style guidelines of this project
• I have performed a self-review of my own code
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• My changes generate no new warnings
• I have added tests that prove my fix is effective or that my feature works
• New and existing unit tests pass locally with my changes

Related Issues

#1206

Tip

Quality Recommendations

1. Consider more robust error handling beyond just console.error. Depending on the application's needs, errors from stream transforms might need to be propagated upstream, trigger specific fallback mechanisms, or be logged with more context (e.g., using a structured logger).

2. Add dedicated unit tests that specifically target the error-handling paths within readAWSStream and readStream to ensure that writer.close() is always called and that errors are logged as expected, even when exceptions occur inside the stream transform functions.

Sequence Diagram

sequenceDiagram
participant H as handleStreamingMode
participant R as Stream Reader
participant W as Stream Writer
participant E as Encoder
participant C as Console

H->>H: Check proxyProvider (BEDROCK or other)

alt proxyProvider === BEDROCK
    H->>+R: readAWSStream(reader, responseTransformer, fallbackChunkId, strictOpenAiCompliance, gatewayRequest)
    loop for await (chunk of stream)
        R-->>H: chunk
        H->>+E: encode(chunk)
        E-->>-H: encodedChunk
        H->>+W: write(encodedChunk)
        W-->>-H: writeSuccess
    end
    H->>H: Process stream chunks
    alt Error during stream processing
        H->>C: console.error(error)
    end
    H->>W: close()
    R-->>-H: Stream finished/closed
else proxyProvider !== BEDROCK
    H->>+R: readStream(reader, splitPattern, responseTransformer, isSleepTimeRequired, fallbackChunkId, strictOpenAiCompliance, gatewayRequest)
    loop for await (chunk of stream)
        R-->>H: chunk
        H->>+E: encode(chunk)
        E-->>-H: encodedChunk
        H->>+W: write(encodedChunk)
        W-->>-H: writeSuccess
    end
    H->>H: Process stream chunks
    alt Error during stream processing
        H->>C: console.error(error)
    end
    H->>W: close()
    R-->>-H: Stream finished/closed
end

[matter-code-review]

Important

PR Review Skipped

PR review skipped as per the configuration setting. Run a manually review by commenting /matter review

💡Tips to use Matter AI

Command List

• /matter summary: Generate AI Summary for the PR
• /matter review: Generate AI Reviews for the latest commit in the PR
• /matter review-full: Generate AI Reviews for the complete PR
• /matter release-notes: Generate AI release-notes for the PR
• /matter : Chat with your PR with Matter AI Agent
• /matter remember : Generate AI memories for the PR
• /matter explain: Get an explanation of the PR
• /matter help: Show the list of available commands and documentation
• Need help? Join our Discord server: https://discord.gg/fJU5DvanU3