Fix silent error absorption in RSC payload injection and streaming pipes

Errors in injectRSCPayload (htmlStream errors, startRSC catch, rscPromise
catch) were silently swallowed, preventing them from reaching errorReporter
and Sentry. This change emits those errors on resultStream so they propagate
through handleStreamError to the error reporting infrastructure without
breaking the pipe or causing Fastify to return error responses.

Also introduces a shared safePipe utility that replaces the scattered
pipe+close boilerplate from PR #2407 with a single function that handles
the Node.js pipe() gap (destination not ended on source destroy) and
provides an optional onError callback for non-fatal error reporting.

Closes #2450

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: AbanoubGhadban

packages/react-on-rails-pro-node-renderer/src/shared/utils.ts

@@ -2,7 +2,7 @@ import cluster from 'cluster';
 import path from 'path';
 import { MultipartFile } from '@fastify/multipart';
 import { createWriteStream, ensureDir, move, MoveOptions, copy, CopyOptions } from 'fs-extra';
-import { Readable, pipeline, PassThrough } from 'stream';
+import { Readable, Writable, pipeline, PassThrough } from 'stream';
 import { promisify } from 'util';
 import * as errorReporter from './errorReporter.js';
 import { getConfig } from './configBuilder.js';
@@ -128,26 +128,43 @@ export const isReadableStream = (stream: unknown): stream is Readable =>
   typeof (stream as Readable).pipe === 'function' &&
   typeof (stream as Readable).read === 'function';
 
-export const handleStreamError = (stream: Readable, onError: (error: Error) => void) => {
-  const newStreamAfterHandlingError = new PassThrough();
-  // Propagate errors for logging/reporting, but don't terminate — error is not the end of the stream.
-  // Non-fatal errors (e.g., emitError for throwJsErrors) emit 'error' without destroying
-  // the stream, and React may continue rendering.
-  stream.on('error', (error) => {
-    onError(error);
-  });
+/**
+ * Pipes source to destination with proper 'close' event handling.
+ *
+ * Node.js `pipe()` does NOT end the destination when the source is destroyed —
+ * it silently unpipes, leaving the destination open forever. This function fills
+ * that gap by listening for the 'close' event (which fires after both normal
+ * 'end' and `destroy()`) and ending the destination if needed.
+ *
+ * An optional `onError` callback provides observability for source stream errors
+ * without forwarding them to the destination (which would break the pipe).
+ */
+export const safePipe = <T extends Writable>(
+  source: Readable,
+  destination: T,
+  onError?: (err: Error) => void,
+): T => {
+  if (onError) {
+    // Propagate errors for logging/reporting, but don't terminate — error is not the
+    // end of the stream. Non-fatal errors (e.g., emitError for throwJsErrors) emit
+    // 'error' without destroying the stream, and React may continue rendering.
+    source.on('error', onError);
+  }
   // 'close' fires after both normal 'end' and destroy().
-  // On normal end, pipe() already forwards 'end' to the PassThrough — this is a no-op.
-  // On destroy, pipe() unpipes but does NOT end the PassThrough — we do it here.
-  stream.on('close', () => {
-    if (!newStreamAfterHandlingError.writableEnded) {
-      newStreamAfterHandlingError.end();
+  // On normal end, pipe() already forwards 'end' to the destination — this is a no-op.
+  // On destroy, pipe() unpipes but does NOT end the destination — we do it here.
+  source.on('close', () => {
+    if (!destination.writableEnded) {
+      destination.end();
     }
   });
-  stream.pipe(newStreamAfterHandlingError);
-  return newStreamAfterHandlingError;
+  source.pipe(destination);
+  return destination;
 };
 
+export const handleStreamError = (stream: Readable, onError: (error: Error) => void) =>
+  safePipe(stream, new PassThrough(), onError);
+
 export const isErrorRenderResult = (result: RenderResult): result is { exceptionMessage: string } =>
   typeof result === 'object' && !isReadableStream(result) && 'exceptionMessage' in result;
 

packages/react-on-rails-pro/src/injectRSCPayload.ts

@@ -12,11 +12,12 @@
  * https://github.com/shakacode/react_on_rails/blob/master/REACT-ON-RAILS-PRO-LICENSE.md
  */
 
-import { PassThrough, Readable } from 'stream';
+import { PassThrough } from 'stream';
 import { finished } from 'stream/promises';
 import { PipeableOrReadableStream } from 'react-on-rails/types';
 import { createRSCPayloadKey } from './utils.ts';
 import RSCRequestTracker from './RSCRequestTracker.ts';
+import safePipe from './safePipe.ts';
 
 // In JavaScript, when an escape sequence with a backslash (\) is followed by a character
 // that isn't a recognized escape character, the backslash is ignored, and the character
@@ -78,16 +79,7 @@ export default function injectRSCPayload(
   domNodeId: string | undefined,
 ) {
   const htmlStream = new PassThrough();
-  pipeableHtmlStream.pipe(htmlStream);
-  // When the source is destroyed, pipe() unpipes but does NOT end htmlStream.
-  // Listen for 'close' to ensure htmlStream ends, which triggers the cleanup chain.
-  if (typeof (pipeableHtmlStream as Readable).on === 'function') {
-    (pipeableHtmlStream as Readable).on('close', () => {
-      if (!htmlStream.writableEnded) {
-        htmlStream.end();
-      }
-    });
-  }
+  safePipe(pipeableHtmlStream, htmlStream);
   const decoder = new TextDecoder();
   let rscPromise: Promise<void> | null = null;
 
@@ -267,7 +259,8 @@ export default function injectRSCPayload(
 
       // Wait for HTML stream to complete, then wait for all RSC promises
       await finished(htmlStream).then(() => Promise.all(rscPromises));
-    } catch {
+    } catch (e) {
+      resultStream.emit('error', e instanceof Error ? e : new Error(String(e)));
       endResultStream();
     }
   };
@@ -297,10 +290,14 @@ export default function injectRSCPayload(
   });
 
   /**
-   * Prevent unhandled error crash. Error alone is not the end of the stream —
-   * termination is handled by the 'close' event below.
+   * Report errors on htmlStream by emitting them on resultStream, where they
+   * propagate to handleStreamError → errorReporter in the node renderer.
+   * Error alone is not the end of the stream — termination is handled by the
+   * 'close' event below.
    */
-  htmlStream.on('error', () => {});
+  htmlStream.on('error', (err) => {
+    resultStream.emit('error', err instanceof Error ? err : new Error(String(err)));
+  });
 
   /**
    * 'close' fires after both normal 'end' and destroy().
@@ -339,7 +336,10 @@ export default function injectRSCPayload(
       .finally(() => {
         rscRequestTracker.clear();
       })
-      .catch(() => endResultStream());
+      .catch((e: unknown) => {
+        resultStream.emit('error', e instanceof Error ? e : new Error(String(e)));
+        endResultStream();
+      });
   });
 
   return resultStream;

packages/react-on-rails-pro/src/safePipe.ts

@@ -0,0 +1,65 @@
+/*
+ * Copyright (c) 2025 Shakacode LLC
+ *
+ * This file is NOT licensed under the MIT (open source) license.
+ * It is part of the React on Rails Pro offering and is licensed separately.
+ *
+ * Unauthorized copying, modification, distribution, or use of this file,
+ * via any medium, is strictly prohibited without a valid license agreement
+ * from Shakacode LLC.
+ *
+ * For licensing terms, please see:
+ * https://github.com/shakacode/react_on_rails/blob/master/REACT-ON-RAILS-PRO-LICENSE.md
+ */
+
+import { Readable, Writable } from 'stream';
+import { PipeableOrReadableStream } from 'react-on-rails/types';
+
+/**
+ * Pipes source to destination with proper 'close' event handling.
+ *
+ * Node.js `pipe()` does NOT end the destination when the source is destroyed —
+ * it silently unpipes, leaving the destination open forever. This function fills
+ * that gap by listening for the 'close' event (which fires after both normal
+ * 'end' and `destroy()`) and ending the destination if the source didn't end
+ * normally.
+ *
+ * An optional `onError` callback provides observability for source stream errors
+ * without forwarding them to the destination (which would break the pipe).
+ *
+ * NOTE: `PipeableOrReadableStream` can be either a React `PipeableStream`
+ * (which only has `.pipe()` and `.abort()`, no `.on()`) or a Node.js
+ * `ReadableStream`. The 'close' and 'error' listeners are only attached when
+ * the source supports `.on()`.
+ *
+ * @param source - The source stream to pipe from
+ * @param destination - The destination stream to pipe into
+ * @param onError - Optional callback for source stream errors (for reporting, not forwarding)
+ * @returns The destination stream (for chaining)
+ */
+export default function safePipe<T extends Writable>(
+  source: PipeableOrReadableStream,
+  destination: T,
+  onError?: (err: Error) => void,
+): T {
+  source.pipe(destination);
+
+  if (typeof (source as Readable).on === 'function') {
+    const readableSource = source as Readable;
+
+    if (onError) {
+      readableSource.on('error', onError);
+    }
+
+    // 'close' fires after both normal 'end' and destroy().
+    // On normal end, pipe() already forwards 'end' to the destination — this is a no-op.
+    // On destroy, pipe() unpipes but does NOT end the destination — we do it here.
+    readableSource.on('close', () => {
+      if (!destination.writableEnded) {
+        destination.end();
+      }
+    });
+  }
+
+  return destination;
+}

packages/react-on-rails-pro/src/streamingUtils.ts

@@ -31,6 +31,7 @@ import {
 import * as ComponentRegistry from './ComponentRegistry.ts';
 import PostSSRHookTracker from './PostSSRHookTracker.ts';
 import RSCRequestTracker from './RSCRequestTracker.ts';
+import safePipe from './safePipe.ts';
 
 type BufferedEvent = {
   event: 'data' | 'error' | 'end';
@@ -133,28 +134,26 @@ export const transformRenderStreamChunksToResultObject = (renderState: StreamRen
     },
   });
 
-  let pipedStream: PipeableOrReadableStream | null = null;
-  const pipeToTransform = (pipeableStream: PipeableOrReadableStream) => {
-    pipeableStream.pipe(transformStream);
-    // 'close' fires after both normal 'end' and destroy().
-    // On normal end, pipe() already forwards 'end' to transformStream — this is a no-op.
-    // On destroy, pipe() unpipes but does NOT end transformStream — we do it here.
-    if (typeof (pipeableStream as Readable).on === 'function') {
-      (pipeableStream as Readable).on('close', () => {
-        if (!transformStream.writableEnded) {
-          transformStream.end();
-        }
-      });
-    }
-    pipedStream = pipeableStream;
-  };
   // We need to wrap the transformStream in a Readable stream to properly handle errors:
   // 1. If we returned transformStream directly, we couldn't emit errors into it externally
   // 2. If an error is emitted into the transformStream, it would cause the render to fail
   // 3. By wrapping in Readable.from(), we can explicitly emit errors into the readableStream without affecting the transformStream
   // Note: Readable.from can merge multiple chunks into a single chunk, so we need to ensure that we can separate them later
   const { stream: readableStream, emitError } = bufferStream(transformStream);
 
+  let pipedStream: PipeableOrReadableStream | null = null;
+  const pipeToTransform = (pipeableStream: PipeableOrReadableStream) => {
+    // safePipe handles the 'close' event to end transformStream when the source is destroyed.
+    // The onError callback forwards source errors to readableStream (via emitError), which
+    // propagates them to handleStreamError → errorReporter in the node renderer. Emitting on
+    // the source (not the destination) keeps the pipe intact so data continues flowing for
+    // non-fatal errors.
+    safePipe(pipeableStream, transformStream, (err) => {
+      emitError(err);
+    });
+    pipedStream = pipeableStream;
+  };
+
   const writeChunk = (chunk: string) => transformStream.write(chunk);
   const endStream = () => {
     transformStream.end();

packages/react-on-rails-pro/src/transformRSCNodeStream.ts

@@ -12,7 +12,8 @@
  * https://github.com/shakacode/react_on_rails/blob/master/REACT-ON-RAILS-PRO-LICENSE.md
  */
 
-import { Transform } from 'stream';
+import { Readable, Transform } from 'stream';
+import safePipe from './safePipe.ts';
 
 /**
  * Transforms an RSC Node.js stream for server-side processing.
@@ -55,5 +56,5 @@ export default function transformRSCStream(stream: NodeJS.ReadableStream): NodeJ
     },
   });
 
-  return stream.pipe(htmlExtractor);
+  return safePipe(stream as Readable, htmlExtractor);
 }