Replace NDJSON envelope with length-prefixed streaming protocol (#2615)

## Summary

Closes #2614

- Replace `JSON.stringify` wrapping of HTML content in the Node→Ruby
streaming protocol with a length-prefixed format that sends raw content
bytes without escaping
- Auto-detect format on Ruby side for backward compatibility with older
node renderers
- Refactor console replay logging into a shared method to handle both
Hash and String chunks

## Problem

The current NDJSON protocol wraps every HTML chunk in a JSON envelope:
```json
{"html":"<div>...escaped HTML...</div>","consoleReplayScript":"","hasErrors":false,"isShellReady":true}\n
```

`JSON.stringify` escapes every `"`, `\`, and `\n` in the HTML content —
adding **~30% overhead** for typical payloads. Ruby then `JSON.parse`s
each line, undoing all the escaping. This serialize→deserialize
round-trip is pure waste for the bulk content (HTML is 99.9% of each
chunk by size).

## Solution

New wire format per chunk:
```
<metadata JSON>\t<content byte length hex>\n<raw content bytes>
```

Example:
```
{"consoleReplayScript":"","hasErrors":false,"isShellReady":true}\t00028A3C\n<div>...(raw HTML, zero escaping)...
```

- **Metadata** (~80 bytes): small JSON without the `html` field —
negligible escaping cost
- **Content** (bulk): raw bytes with length prefix — **zero** escaping
overhead
- **Ruby parser**: reads header line (until `\n`), extracts content
length, reads exactly that many raw bytes
- **Auto-detection**: if the header line contains `\t`, it's
length-prefixed; otherwise legacy NDJSON (backward compatible)

## Changes

| File | Change |
|------|--------|
| `streamingUtils.ts` | Write `metadata\tcontent_length\nraw_content`
instead of `JSON.stringify(envelope)\n` |
| `stream_request.rb` | Replace `loop_response_lines` with
`loop_response_chunks` (length-prefixed parser with auto-detection) |
| `ruby_embedded_java_script.rb` | Handle both Hash (new) and String
(legacy) chunks; extract shared `replay_console_to_rails_logger` |
| `streamServerRenderedReactComponent.test.jsx` | Update
`expectStreamChunk` to parse length-prefixed format |
| `removeRSCChunkStack.ts` | Update to parse length-prefixed format with
NDJSON fallback |

## Test plan

- [x] `streamServerRenderedReactComponent.test.jsx` — 16/16 pass (core
streaming tests)
- [x] `concurrentRSCPayloadGeneration.rsc.test.tsx` — pass (parallel RSC
rendering)
- [x] `RSCSerialization.rsc.test.tsx` — pass
- [x] `ReactOnRailsRSC.rsc.test.tsx` — pass
- [x] `serverRenderRSCReactComponent.rsc.test.tsx` — pass
- [x] `injectRSCPayload.test.ts` — 9/9 pass
- [ ] Integration tests with real node renderer
- [ ] E2E tests with dummy app

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Improvements**
* Switched streaming to a length‑prefixed protocol for atomic chunk
delivery, correct multibyte handling, and compatibility with legacy
newline‑delimited streams.
* More reliable stream parsing with improved server-side console
replay/logging, clearer error signaling, and robust end‑of‑stream
handling.

* **Tests**
* Added tests validating length‑prefixed parsing, including UTF‑8
multibyte content.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>
Co-authored-by: Justin Gordon <[email protected]>

Author: 3 people

packages/react-on-rails-pro-node-renderer/src/worker/vm.ts

@@ -415,8 +415,10 @@ export async function buildExecutionContext(
         return newStreamAfterHandlingError;
       }
       if (typeof result !== 'string') {
-        const objectResult = await result;
-        result = JSON.stringify(objectResult);
+        const resolvedResult = await result;
+        // If the resolved value is already a string (e.g., length-prefixed format from
+        // buildLengthPrefixedResult), use it directly. Only JSON.stringify objects.
+        result = typeof resolvedResult === 'string' ? resolvedResult : JSON.stringify(resolvedResult);
       }
       if (log.level === 'debug' && result) {
         log.debug(`result from JS:

packages/react-on-rails-pro-node-renderer/tests/concurrentHtmlStreaming.test.ts

@@ -2,11 +2,10 @@ import { randomUUID } from 'crypto';
 import { createClient } from 'redis';
 import parser from 'node-html-parser';
 
-// eslint-disable-next-line import/no-relative-packages
-import { RSCPayloadChunk } from '../../react-on-rails/lib/types';
 import buildApp from '../src/worker';
 import { createTestConfig } from './testingNodeRendererConfigs';
 import { makeRequest } from './httpRequestUtils';
+import { LengthPrefixedStreamParser } from './parseLengthPrefixedStream';
 
 const { config } = createTestConfig('concurrentHtmlStreaming');
 const app = buildApp(config);
@@ -32,11 +31,13 @@ const sendRedisItemValue = async (redisRequestId: string, itemIndex: number, val
   await sendRedisValue(redisRequestId, `Item${itemIndex}`, value);
 };
 
-const extractHtmlFromChunks = (chunks: string) => {
-  const html = chunks
-    .split('\n')
-    .map((chunk) => (chunk.trim().length > 0 ? (JSON.parse(chunk) as RSCPayloadChunk).html : chunk))
-    .join('');
+const extractHtmlFromChunks = (chunks: string, valuesToStrip: string[] = []) => {
+  const streamParser = new LengthPrefixedStreamParser();
+  streamParser.feed(chunks);
+  let html = streamParser.htmlChunks.join('');
+  valuesToStrip.forEach((value) => {
+    html = html.replace(new RegExp(value, 'g'), '');
+  });
   const parsedHtml = parser.parse(html);
   // TODO: investigate why ReactOnRails produces different RSC payload on each request
   parsedHtml.querySelectorAll('script').forEach((x) => x.remove());
@@ -55,16 +56,13 @@ const createParallelRenders = (size: number) => {
     });
   });
 
-  const expectNextChunk = async (expectedNextChunk: string) => {
+  const expectNextChunk = async (expectedNextChunkHtml: string) => {
     const nextChunks = await Promise.all(
       renderRequests.map((renderRequest) => renderRequest.waitForNextChunk()),
     );
     nextChunks.forEach((chunk, index) => {
       const redisRequestId = redisRequestIds[index]!;
-      const chunksAfterRemovingRequestId = chunk.replace(new RegExp(redisRequestId, 'g'), '');
-      expect(extractHtmlFromChunks(chunksAfterRemovingRequestId)).toEqual(
-        extractHtmlFromChunks(expectedNextChunk),
-      );
+      expect(extractHtmlFromChunks(chunk, [redisRequestId])).toEqual(expectedNextChunkHtml);
     });
   };
 
@@ -92,55 +90,55 @@ test('Happy Path', async () => {
   const parallelInstances = 50;
   expect.assertions(parallelInstances * 7 + 7);
   const redisRequestId = randomUUID();
+  const expectedChunks: string[] = [];
   const { waitForNextChunk, finishedPromise, getBuffer } = makeRequest(app, {
     componentName: 'RedisReceiver',
     props: { requestId: redisRequestId },
   });
-  const chunks: string[] = [];
   let chunk = await waitForNextChunk();
   expect(chunk).not.toContain('Unique Value');
-  chunks.push(chunk.replace(new RegExp(redisRequestId, 'g'), ''));
+  expectedChunks.push(extractHtmlFromChunks(chunk, [redisRequestId]));
 
   await sendRedisItemValue(redisRequestId, 0, 'First Unique Value');
   chunk = await waitForNextChunk();
   expect(chunk).toContain('First Unique Value');
-  chunks.push(chunk.replace(new RegExp(redisRequestId, 'g'), ''));
+  expectedChunks.push(extractHtmlFromChunks(chunk, [redisRequestId]));
 
   await sendRedisItemValue(redisRequestId, 4, 'Fifth Unique Value');
   chunk = await waitForNextChunk();
   expect(chunk).toContain('Fifth Unique Value');
-  chunks.push(chunk.replace(new RegExp(redisRequestId, 'g'), ''));
+  expectedChunks.push(extractHtmlFromChunks(chunk, [redisRequestId]));
 
   await sendRedisItemValue(redisRequestId, 2, 'Third Unique Value');
   chunk = await waitForNextChunk();
   expect(chunk).toContain('Third Unique Value');
-  chunks.push(chunk.replace(new RegExp(redisRequestId, 'g'), ''));
+  expectedChunks.push(extractHtmlFromChunks(chunk, [redisRequestId]));
 
   await sendRedisItemValue(redisRequestId, 1, 'Second Unique Value');
   chunk = await waitForNextChunk();
   expect(chunk).toContain('Second Unique Value');
-  chunks.push(chunk.replace(new RegExp(redisRequestId, 'g'), ''));
+  expectedChunks.push(extractHtmlFromChunks(chunk, [redisRequestId]));
 
   await sendRedisItemValue(redisRequestId, 3, 'Forth Unique Value');
   chunk = await waitForNextChunk();
   expect(chunk).toContain('Forth Unique Value');
-  chunks.push(chunk.replace(new RegExp(redisRequestId, 'g'), ''));
+  expectedChunks.push(extractHtmlFromChunks(chunk, [redisRequestId]));
 
   await finishedPromise;
   expect(getBuffer).toHaveLength(0);
 
   const { expectNextChunk, sendRedisItemValues, waitUntilFinished } =
     createParallelRenders(parallelInstances);
-  await expectNextChunk(chunks[0]!);
+  await expectNextChunk(expectedChunks[0]!);
   await sendRedisItemValues(0, 'First Unique Value');
-  await expectNextChunk(chunks[1]!);
+  await expectNextChunk(expectedChunks[1]!);
   await sendRedisItemValues(4, 'Fifth Unique Value');
-  await expectNextChunk(chunks[2]!);
+  await expectNextChunk(expectedChunks[2]!);
   await sendRedisItemValues(2, 'Third Unique Value');
-  await expectNextChunk(chunks[3]!);
+  await expectNextChunk(expectedChunks[3]!);
   await sendRedisItemValues(1, 'Second Unique Value');
-  await expectNextChunk(chunks[4]!);
+  await expectNextChunk(expectedChunks[4]!);
   await sendRedisItemValues(3, 'Forth Unique Value');
-  await expectNextChunk(chunks[5]!);
+  await expectNextChunk(expectedChunks[5]!);
   await waitUntilFinished();
 }, 50000);

packages/react-on-rails-pro-node-renderer/tests/handleRenderRequest.test.ts

@@ -405,7 +405,7 @@ describe(testName, () => {
     expect(result.response).toEqual({
       status: 200,
       headers: { 'Cache-Control': 'public, max-age=31536000' },
-      data: JSON.stringify('undefined'),
+      data: 'undefined',
     });
   });
 

packages/react-on-rails-pro-node-renderer/tests/htmlStreaming.test.js

@@ -3,6 +3,7 @@ import buildApp from '../src/worker';
 import { createTestConfig } from './testingNodeRendererConfigs';
 import * as errorReporter from '../src/shared/errorReporter';
 import { createForm, SERVER_BUNDLE_TIMESTAMP } from './httpRequestUtils';
+import { LengthPrefixedStreamParser } from './parseLengthPrefixedStream';
 
 const { config } = createTestConfig('htmlStreaming');
 const app = buildApp(config);
@@ -30,34 +31,16 @@ const makeRequest = async (options = {}) => {
   });
   request.setEncoding('utf8');
 
-  const chunks = [];
-  const jsonChunks = [];
+  const parser = new LengthPrefixedStreamParser();
   let firstByteTime;
   let status;
-  const decoder = new TextDecoder();
 
   request.on('response', (headers) => {
     status = headers[':status'];
   });
 
   request.on('data', (data) => {
-    // Sometimes, multiple chunks are merged into one.
-    // So, the server uses \n as a delimiter between chunks.
-    const decodedData = typeof data === 'string' ? data : decoder.decode(data, { stream: false });
-    const decodedChunksFromData = decodedData
-      .split('\n')
-      .map((chunk) => chunk.trim())
-      .filter((chunk) => chunk.length > 0);
-    chunks.push(...decodedChunksFromData);
-    jsonChunks.push(
-      ...decodedChunksFromData.map((chunk) => {
-        try {
-          return JSON.parse(chunk);
-        } catch (e) {
-          return { hasErrors: true, error: `JSON parsing failed: ${e.message}` };
-        }
-      }),
-    );
+    parser.feed(data);
     if (!firstByteTime) {
       firstByteTime = Date.now();
     }
@@ -80,6 +63,7 @@ const makeRequest = async (options = {}) => {
   });
 
   const endTime = Date.now();
+  const { htmlChunks: chunks, parsedChunks: jsonChunks } = parser;
   const fullBody = chunks.join('');
   const timeToFirstByte = firstByteTime - startTime;
   const streamingTime = endTime - firstByteTime;

packages/react-on-rails-pro-node-renderer/tests/httpRequestUtils.ts

@@ -122,9 +122,7 @@ export const makeRequest = (app: ReturnType<typeof buildApp>, options: Partial<R
     ':path': `/bundles/${usedBundleTimestamp}/render/454a82526211afdb215352755d36032c`,
     'content-type': `multipart/form-data; boundary=${form.getBoundary()}`,
   });
-  request.setEncoding('utf8');
-
-  const buffer: string[] = [];
+  const buffer: Buffer[] = [];
 
   const statusPromise = new Promise<number | undefined>((resolve) => {
     request.on('response', (headers) => {
@@ -142,15 +140,15 @@ export const makeRequest = (app: ReturnType<typeof buildApp>, options: Partial<R
     }
 
     resolveChunkPromiseTimeout = setTimeout(() => {
-      resolveChunksPromise?.(buffer.join(''));
+      resolveChunksPromise?.(Buffer.concat(buffer).toString('utf8'));
       resolveChunksPromise = undefined;
       rejectChunksPromise = undefined;
       buffer.length = 0;
     }, 1000);
   };
 
-  request.on('data', (data: Buffer) => {
-    buffer.push(data.toString());
+  request.on('data', (data: Buffer | string) => {
+    buffer.push(Buffer.isBuffer(data) ? data : Buffer.from(data, 'utf8'));
     if (resolveChunksPromise) {
       scheduleResolveChunkPromise();
     }

packages/react-on-rails-pro-node-renderer/tests/parseLengthPrefixedStream.ts

@@ -0,0 +1,34 @@
+/**
+ * Test utility that wraps the production LengthPrefixedStreamParser with a
+ * convenience API for collecting parsed chunks. Uses the same parser as
+ * production code — no duplicated parsing logic.
+ */
+
+// eslint-disable-next-line import/no-relative-packages
+import ProductionParser from '../../react-on-rails-pro/src/parseLengthPrefixedStream';
+
+export interface ParsedChunk {
+  html: string;
+  consoleReplayScript?: string;
+  hasErrors?: boolean;
+  isShellReady?: boolean;
+  renderingError?: { message: string; stack: string };
+  [key: string]: unknown;
+}
+
+export class LengthPrefixedStreamParser {
+  private parser = new ProductionParser();
+
+  readonly htmlChunks: string[] = [];
+
+  readonly parsedChunks: ParsedChunk[] = [];
+
+  feed(data: string | Buffer): void {
+    const buf = typeof data === 'string' ? Buffer.from(data, 'utf8') : data;
+    this.parser.feed(buf, (content, metadata) => {
+      const html = new TextDecoder().decode(content);
+      this.htmlChunks.push(html);
+      this.parsedChunks.push({ html, ...metadata } as ParsedChunk);
+    });
+  }
+}

packages/react-on-rails-pro-node-renderer/tests/streamErrorHang.test.ts

@@ -8,6 +8,7 @@
  */
 
 import fs from 'fs';
+import { LengthPrefixedStreamParser } from './parseLengthPrefixedStream';
 import path from 'path';
 import http2 from 'http2';
 import FormData from 'form-data';
@@ -69,7 +70,7 @@ const makeRequest = (renderingRequest: string, timeoutMs = 3000) =>
     });
     request.setEncoding('utf8');
 
-    const chunks: string[] = [];
+    const parser = new LengthPrefixedStreamParser();
     let status: number | undefined;
     let settled = false;
 
@@ -78,11 +79,7 @@ const makeRequest = (renderingRequest: string, timeoutMs = 3000) =>
     });
 
     request.on('data', (data: string) => {
-      const decoded = data
-        .split('\n')
-        .map((c) => c.trim())
-        .filter((c) => c.length > 0);
-      chunks.push(...decoded);
+      parser.feed(data);
     });
 
     form.pipe(request);
@@ -92,7 +89,7 @@ const makeRequest = (renderingRequest: string, timeoutMs = 3000) =>
       if (settled) return;
       settled = true;
       client.destroy();
-      resolve({ status, chunks, timedOut });
+      resolve({ status, chunks: parser.htmlChunks, timedOut });
     };
 
     const timeout = setTimeout(() => finish(true), timeoutMs);
@@ -112,12 +109,27 @@ const makeRequest = (renderingRequest: string, timeoutMs = 3000) =>
 // The bundle exposes `Readable` globally via `global.Readable = require('stream').Readable`.
 // ---------------------------------------------------------------------------
 
+// Helper: builds a length-prefixed chunk string for the mock rendering requests.
+// Format: <metadata JSON>\t<content byte length hex>\n<raw content bytes>
+function buildLengthPrefixedChunk(html: string, metadata: Record<string, unknown> = {}): string {
+  const meta = {
+    consoleReplayScript: '',
+    hasErrors: false,
+    isShellReady: true,
+    payloadType: 'string',
+    ...metadata,
+  };
+  const metaJson = JSON.stringify(meta);
+  const byteLen = Buffer.byteLength(html).toString(16).padStart(8, '0');
+  return `${metaJson}\t${byteLen}\n${html}`;
+}
+
 const RENDERING_REQUEST = {
   /** Pushes one chunk, then destroys the stream with an error. */
   errorMidStream: `(function() {
     var stream = new Readable({ read() {} });
     setTimeout(function() {
-      stream.push('{"html":"<div>partial</div>","consoleReplayScript":"","hasErrors":false,"isShellReady":true}\\n');
+      stream.push(${JSON.stringify(buildLengthPrefixedChunk('<div>partial</div>'))});
     }, 10);
     setTimeout(function() {
       stream.destroy(new Error('mid-stream rendering error'));
@@ -138,7 +150,7 @@ const RENDERING_REQUEST = {
   happyPath: `(function() {
     var stream = new Readable({ read() {} });
     setTimeout(function() {
-      stream.push('{"html":"<div>ok</div>","consoleReplayScript":"","hasErrors":false,"isShellReady":true}\\n');
+      stream.push(${JSON.stringify(buildLengthPrefixedChunk('<div>ok</div>'))});
       stream.push(null);
     }, 10);
     return stream;