Merge pull request #1207 from narengogi/bug/stream-transform-unhandled-rejection-resource-cleanup

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

Author: VisargD

src/handlers/streamHandler.ts

@@ -310,31 +310,41 @@ export function handleStreamingMode(
 
   if (proxyProvider === BEDROCK) {
     (async () => {
-      for await (const chunk of readAWSStream(
-        reader,
-        responseTransformer,
-        fallbackChunkId,
-        strictOpenAiCompliance,
-        gatewayRequest
-      )) {
-        await writer.write(encoder.encode(chunk));
+      try {
+        for await (const chunk of readAWSStream(
+          reader,
+          responseTransformer,
+          fallbackChunkId,
+          strictOpenAiCompliance,
+          gatewayRequest
+        )) {
+          await writer.write(encoder.encode(chunk));
+        }
+      } catch (error) {
+        console.error(error);
+      } finally {
+        writer.close();
       }
-      writer.close();
     })();
   } else {
     (async () => {
-      for await (const chunk of readStream(
-        reader,
-        splitPattern,
-        responseTransformer,
-        isSleepTimeRequired,
-        fallbackChunkId,
-        strictOpenAiCompliance,
-        gatewayRequest
-      )) {
-        await writer.write(encoder.encode(chunk));
+      try {
+        for await (const chunk of readStream(
+          reader,
+          splitPattern,
+          responseTransformer,
+          isSleepTimeRequired,
+          fallbackChunkId,
+          strictOpenAiCompliance,
+          gatewayRequest
+        )) {
+          await writer.write(encoder.encode(chunk));
+        }
+      } catch (error) {
+        console.error(error);
+      } finally {
+        writer.close();
       }
-      writer.close();
     })();
   }