From 9cf88df67b5f825982c50436bb4d238eda215749 Mon Sep 17 00:00:00 2001
From: Angel98518 <daniel98518@gmail.com>
Date: Wed, 21 Jan 2026 00:19:47 +0800
Subject: [PATCH] Improve streamToBuffer error handling and memory safety
 (#17255)

## Description

This PR improves error handling and memory safety for the streamToBuffer
utility function.

### Changes
- Add proper cleanup of event listeners to prevent memory leaks
- Add checks for already-ended streams to handle edge cases gracefully
- Add protection against multiple resolve/reject calls with isResolved
flag
- Add handling for close events with appropriate error messages
- Improve robustness when streams are destroyed or closed externally

### Impact
This fixes potential memory leaks and race conditions when streams are
cancelled, destroyed, or closed before completion.

### Code Statistics
- 1 file changed
- ~53 lines added, 9 lines modified
- ~95% code changes (functional improvements)

---------

Co-authored-by: GitTensor Miner <miner@gittensor.io>
Co-authored-by: Etienne <45695613+etiennejouan@users.noreply.github.com>
---
 .../utils/__test__/stream-to-buffer.spec.ts   | 79 +++++++++++++++++++
 .../src/utils/stream-to-buffer.ts             | 68 +++++++++++++---
 2 files changed, 138 insertions(+), 9 deletions(-)
 create mode 100644 packages/twenty-server/src/utils/__test__/stream-to-buffer.spec.ts

diff --git a/packages/twenty-server/src/utils/__test__/stream-to-buffer.spec.ts b/packages/twenty-server/src/utils/__test__/stream-to-buffer.spec.ts
new file mode 100644
index 0000000000..131a2c647b
--- /dev/null
+++ b/packages/twenty-server/src/utils/__test__/stream-to-buffer.spec.ts
@@ -0,0 +1,79 @@
+import { PassThrough, Readable } from 'stream';
+
+import { streamToBuffer } from 'src/utils/stream-to-buffer';
+
+describe('streamToBuffer', () => {
+  describe('successful scenarios', () => {
+    it('should convert a stream with single chunk to buffer', async () => {
+      const testData = 'Hello, World!';
+      const stream = Readable.from([Buffer.from(testData)]);
+
+      const result = await streamToBuffer(stream);
+
+      expect(result.toString()).toBe(testData);
+    });
+
+    it('should convert a stream with multiple chunks to buffer', async () => {
+      const chunks = ['Hello, ', 'World', '!'];
+      const stream = Readable.from(chunks.map((chunk) => Buffer.from(chunk)));
+
+      const result = await streamToBuffer(stream);
+
+      expect(result.toString()).toBe('Hello, World!');
+    });
+
+    it('should handle empty stream', async () => {
+      const stream = Readable.from([]);
+
+      const result = await streamToBuffer(stream);
+
+      expect(result.length).toBe(0);
+      expect(result.toString()).toBe('');
+    });
+  });
+
+  describe('error scenarios', () => {
+    it('should reject when stream is already ended', async () => {
+      const stream = Readable.from([Buffer.from('test')]);
+
+      await streamToBuffer(stream);
+
+      await expect(streamToBuffer(stream)).rejects.toThrow(
+        'Stream has already ended',
+      );
+    });
+
+    it('should reject when stream is not readable (destroyed)', async () => {
+      const stream = new PassThrough();
+
+      stream.destroy();
+
+      await expect(streamToBuffer(stream)).rejects.toThrow(
+        'Stream is not readable',
+      );
+    });
+
+    it('should reject when stream emits an error', async () => {
+      const stream = new PassThrough();
+      const testError = new Error('Stream error');
+
+      const promise = streamToBuffer(stream);
+
+      stream.write(Buffer.from('partial'));
+      stream.emit('error', testError);
+
+      await expect(promise).rejects.toThrow('Stream error');
+    });
+
+    it('should reject when stream closes before end', async () => {
+      const stream = new PassThrough();
+
+      const promise = streamToBuffer(stream);
+
+      stream.write(Buffer.from('data'));
+      stream.destroy();
+
+      await expect(promise).rejects.toThrow('Stream closed before end');
+    });
+  });
+});
diff --git a/packages/twenty-server/src/utils/stream-to-buffer.ts b/packages/twenty-server/src/utils/stream-to-buffer.ts
index a6edd160b4..73310e301f 100644
--- a/packages/twenty-server/src/utils/stream-to-buffer.ts
+++ b/packages/twenty-server/src/utils/stream-to-buffer.ts
@@ -4,16 +4,66 @@ export const streamToBuffer = async (stream: Readable): Promise<Buffer> => {
   const chunks: Buffer[] = [];
 
   return new Promise((resolve, reject) => {
-    stream.on('data', (chunk: Buffer) => {
-      chunks.push(chunk);
-    });
+    if (stream.readableEnded) {
+      reject(new Error('Stream has already ended'));
 
-    stream.on('end', () => {
-      resolve(Buffer.concat(chunks));
-    });
+      return;
+    }
 
-    stream.on('error', (error) => {
-      reject(error);
-    });
+    if (!stream.readable) {
+      reject(new Error('Stream is not readable'));
+
+      return;
+    }
+
+    let isResolved = false;
+
+    const cleanup = () => {
+      stream.removeListener('data', onData);
+      stream.removeListener('end', onEnd);
+      stream.removeListener('error', onError);
+      stream.removeListener('close', onClose);
+    };
+
+    const onData = (chunk: Buffer) => {
+      if (!isResolved) {
+        chunks.push(chunk);
+      }
+    };
+
+    const onEnd = () => {
+      if (!isResolved) {
+        isResolved = true;
+        cleanup();
+        resolve(Buffer.concat(chunks));
+      }
+    };
+
+    const onError = (error: Error) => {
+      if (!isResolved) {
+        isResolved = true;
+        cleanup();
+        reject(error);
+      }
+    };
+
+    const onClose = () => {
+      if (!isResolved) {
+        if (stream.readableEnded) {
+          isResolved = true;
+          cleanup();
+          resolve(Buffer.concat(chunks));
+        } else {
+          isResolved = true;
+          cleanup();
+          reject(new Error('Stream closed before end'));
+        }
+      }
+    };
+
+    stream.on('data', onData);
+    stream.on('end', onEnd);
+    stream.on('error', onError);
+    stream.on('close', onClose);
   });
 };