[Flight] Fix detached ArrayBuffer error when streaming typed arrays

When streaming typed arrays, the underlying `ArrayBuffer` may be
detached by downstream consumers (e.g. compression streams). This can
corrupt the output or cause a "detached ArrayBuffer" error.

To avoid this issue, we now clone chunks that are larger than the view
size to ensure they have their own backing store. We already did this
in the Edge runtime to allow transferring large chunks, and now we
also do it in the Node.js runtime to avoid the downstream issues.

Author: unstubbable

packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMEdge-test.js

@@ -15,6 +15,8 @@ global.ReadableStream =
   require('web-streams-polyfill/ponyfill/es6').ReadableStream;
 global.WritableStream =
   require('web-streams-polyfill/ponyfill/es6').WritableStream;
+global.TransformStream =
+  require('web-streams-polyfill/ponyfill/es6').TransformStream;
 global.TextEncoder = require('util').TextEncoder;
 global.TextDecoder = require('util').TextDecoder;
 global.Blob = require('buffer').Blob;
@@ -796,6 +798,73 @@ describe('ReactFlightDOMEdge', () => {
     expect(result).toEqual(buffers);
   });
 
+  it('handles detached ArrayBuffer during streaming of typed-array views', async () => {
+    const VIEW_SIZE = 2048;
+    const length = VIEW_SIZE + 50;
+    const underlying = new ArrayBuffer(length + 20);
+    const source = new Uint8Array(underlying, 10, length);
+    for (let i = 0; i < source.length; i++) source[i] = i % 251;
+
+    // Create a copy of the source data to compare against later, because the
+    // underlying buffer will be detached by the time we read the result.
+    const expected = source.slice();
+
+    const {MessageChannel} = require('worker_threads');
+
+    // Detaches an ArrayBuffer by transferring it via MessageChannel
+    function detachBuffer(arrayBuffer) {
+      const {port1, port2} = new MessageChannel();
+      port2.postMessage(arrayBuffer, [arrayBuffer]);
+      port1.close();
+      port2.close();
+    }
+
+    // Transform stream that detaches the original buffer on first binary chunk,
+    // simulating a scenario where the backing store becomes unavailable
+    // mid-stream.
+    let detached = false;
+    const detachingTransformStream = new TransformStream({
+      transform(chunk, controller) {
+        try {
+          if (typeof chunk !== 'string') {
+            if (!detached) {
+              detached = true;
+              detachBuffer(underlying);
+            }
+            // Re-wrap the chunk (common downstream pattern) - will throw if
+            // detached.
+            controller.enqueue(
+              new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength),
+            );
+          } else {
+            controller.enqueue(chunk);
+          }
+        } catch (err) {
+          controller.error(err);
+        }
+      },
+    });
+
+    const stream = await serverAct(() =>
+      passThrough(
+        ReactServerDOMServer.renderToReadableStream(source).pipeThrough(
+          detachingTransformStream,
+        ),
+      ),
+    );
+
+    const promise = ReactServerDOMClient.createFromReadableStream(stream, {
+      serverConsumerManifest: {
+        moduleMap: {},
+        moduleLoading: webpackModuleLoading,
+      },
+    });
+
+    const result = await promise;
+
+    expect(expected).toEqual(result);
+  });
+
   it('should be able to serialize a blob', async () => {
     const bytes = new Uint8Array([
       123, 4, 10, 5, 100, 255, 244, 45, 56, 67, 43, 124, 67, 89, 100, 20,

packages/react-server-dom-webpack/src/__tests__/ReactFlightDOMNode-test.js

@@ -331,6 +331,76 @@ describe('ReactFlightDOMNode', () => {
     expect(result).toEqual(buffers);
   });
 
+  it('handles detached ArrayBuffer during streaming of typed-array views', async () => {
+    const VIEW_SIZE = 2048;
+    const length = VIEW_SIZE + 50;
+    const underlying = new ArrayBuffer(length + 20);
+    const source = new Uint8Array(underlying, 10, length);
+    for (let i = 0; i < source.length; i++) source[i] = i % 251;
+
+    // Create a copy of the source data to compare against later, because the
+    // underlying buffer will be detached by the time we read the result.
+    const expected = source.slice();
+
+    const stream = await serverAct(() =>
+      ReactServerDOMServer.renderToPipeableStream(source),
+    );
+
+    const {MessageChannel} = require('worker_threads');
+
+    // Detaches an ArrayBuffer by transferring it via MessageChannel
+    function detachBuffer(arrayBuffer) {
+      const {port1, port2} = new MessageChannel();
+      port2.postMessage(arrayBuffer, [arrayBuffer]);
+      port1.close();
+      port2.close();
+    }
+
+    // Transform stream that detaches the original buffer on first binary chunk,
+    // simulating a scenario where the backing store becomes unavailable
+    // mid-stream.
+    const detachingTransform = new Stream.Transform({
+      ...streamOptions,
+      transform(chunk, _enc, callback) {
+        try {
+          if (typeof chunk !== 'string') {
+            if (!this.detached) {
+              this.detached = true;
+              detachBuffer(underlying);
+            }
+            // Re-wrap the chunk (common downstream pattern) - will throw if
+            // detached.
+            this.push(
+              new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength),
+            );
+          } else {
+            this.push(chunk);
+          }
+          callback();
+        } catch (err) {
+          callback(err);
+        }
+      },
+    });
+
+    const readable = new Stream.PassThrough(streamOptions);
+    const promise = ReactServerDOMClient.createFromNodeStream(readable, {
+      moduleMap: {},
+      moduleLoading: webpackModuleLoading,
+    });
+
+    await new Promise((resolve, reject) => {
+      detachingTransform.on('error', reject);
+      readable.on('error', reject);
+      readable.on('finish', resolve);
+      stream.pipe(detachingTransform).pipe(readable);
+    });
+
+    const result = await promise;
+
+    expect(expected).toEqual(result);
+  });
+
   it('should allow accept a nonce option for Flight preinitialized scripts', async () => {
     function ClientComponent() {
       return <span>Client Component</span>;

packages/react-server/src/ReactServerStreamConfigEdge.js

@@ -153,7 +153,10 @@ export function typedArrayToBinaryChunk(
     content.byteLength,
   );
   // We clone large chunks so that we can transfer them when we write them.
-  // Others get copied into the target buffer.
+  // Others get copied into the target buffer. This also avoids downstream
+  // issues where a shared backing ArrayBuffer might be retained, mutated, or
+  // detached by streams/compression/structured-clone, which can corrupt output
+  // or trigger “detached ArrayBuffer” errors.
   return content.byteLength > VIEW_SIZE ? buffer.slice() : buffer;
 }
 

packages/react-server/src/ReactServerStreamConfigNode.js

@@ -213,7 +213,17 @@ export function typedArrayToBinaryChunk(
   content: $ArrayBufferView,
 ): BinaryChunk {
   // Convert any non-Uint8Array array to Uint8Array. We could avoid this for Uint8Arrays.
-  return new Uint8Array(content.buffer, content.byteOffset, content.byteLength);
+  const buffer = new Uint8Array(
+    content.buffer,
+    content.byteOffset,
+    content.byteLength,
+  );
+  // We clone large chunks so that we can transfer them when we write them.
+  // Others get copied into the target buffer. This also avoids downstream
+  // issues where a shared backing ArrayBuffer might be retained, mutated, or
+  // detached by streams/compression/structured-clone, which can corrupt output
+  // or trigger “detached ArrayBuffer” errors.
+  return content.byteLength > VIEW_SIZE ? buffer.slice() : buffer;
 }
 
 export function byteLengthOfChunk(chunk: Chunk | PrecomputedChunk): number {