[WIP] Make node renderer use ndjson format for communication

lib/react_on_rails_pro/request.rb

@@ -28,7 +28,12 @@ def render_code_as_stream(path, js_code, is_rsc_payload:)
         end
 
         ReactOnRailsPro::StreamRequest.create do |send_bundle|
-          form = form_with_code(js_code, send_bundle)
+          if send_bundle
+            Rails.logger.info { "[ReactOnRailsPro] Sending bundle to the node renderer" }
+            upload_assets
+          end
+
+          form = form_with_code(js_code, false)
           perform_request(path, form: form, stream: true)
         end
       end

lib/react_on_rails_pro/server_rendering_js_code.rb

@@ -36,6 +36,7 @@ def generate_rsc_payload_js_function(render_options)
           renderingRequest,
           rscBundleHash: '#{ReactOnRailsPro::Utils.rsc_bundle_hash}',
         }
+        const runOnOtherBundle = globalThis.runOnOtherBundle;
         if (typeof generateRSCPayload !== 'function') {
           globalThis.generateRSCPayload = function generateRSCPayload(componentName, props, railsContext) {
             const { renderingRequest, rscBundleHash } = railsContext.serverSideRSCPayloadParameters;

packages/node-renderer/src/shared/utils.ts

@@ -8,6 +8,7 @@ import * as errorReporter from './errorReporter';
 import { getConfig } from './configBuilder';
 import log from './log';
 import type { RenderResult } from '../worker/vm';
+import fileExistsAsync from './fileExistsAsync';
 
 export const TRUNCATION_FILLER = '\n... TRUNCATED ...\n';
 
@@ -168,3 +169,29 @@ export function getAssetPath(bundleTimestamp: string | number, filename: string)
   const bundleDirectory = getBundleDirectory(bundleTimestamp);
   return path.join(bundleDirectory, filename);
 }
+
+export async function validateBundlesExist(
+  bundleTimestamp: string | number,
+  dependencyBundleTimestamps?: (string | number)[],
+): Promise<ResponseResult | null> {
+  const missingBundles = (
+    await Promise.all(
+      [...(dependencyBundleTimestamps ?? []), bundleTimestamp].map(async (timestamp) => {
+        const bundleFilePath = getRequestBundleFilePath(timestamp);
+        const fileExists = await fileExistsAsync(bundleFilePath);
+        return fileExists ? null : timestamp;
+      }),
+    )
+  ).filter((timestamp) => timestamp !== null);
+
+  if (missingBundles.length > 0) {
+    const missingBundlesText = missingBundles.length > 1 ? 'bundles' : 'bundle';
+    log.info(`No saved ${missingBundlesText}: ${missingBundles.join(', ')}`);
+    return {
+      headers: { 'Cache-Control': 'no-cache, no-store, max-age=0, must-revalidate' },
+      status: 410,
+      data: 'No bundle uploaded',
+    };
+  }
+  return null;
+}

packages/node-renderer/src/worker.ts

@@ -13,10 +13,20 @@ import log, { sharedLoggerOptions } from './shared/log';
 import packageJson from './shared/packageJson';
 import { buildConfig, Config, getConfig } from './shared/configBuilder';
 import fileExistsAsync from './shared/fileExistsAsync';
-import type { FastifyInstance, FastifyReply, FastifyRequest } from './worker/types';
-import checkProtocolVersion from './worker/checkProtocolVersionHandler';
-import authenticate from './worker/authHandler';
-import { handleRenderRequest, type ProvidedNewBundle } from './worker/handleRenderRequest';
+import type { FastifyInstance, FastifyReply } from './worker/types';
+import { performRequestPrechecks } from './worker/requestPrechecks';
+import { AuthBody, authenticate } from './worker/authHandler';
+import {
+  handleRenderRequest,
+  type ProvidedNewBundle,
+  handleNewBundlesProvided,
+} from './worker/handleRenderRequest';
+import {
+  handleIncrementalRenderRequest,
+  type IncrementalRenderInitialRequest,
+  type IncrementalRenderSink,
+} from './worker/handleIncrementalRenderRequest';
+import { handleIncrementalRenderStream } from './worker/handleIncrementalRenderStream';
 import {
   errorResponseResult,
   formatExceptionMessage,
@@ -160,41 +170,11 @@ export default function run(config: Partial<Config>) {
     },
   });
 
-  const isProtocolVersionMatch = async (req: FastifyRequest, res: FastifyReply) => {
-    // Check protocol version
-    const protocolVersionCheckingResult = checkProtocolVersion(req);
-
-    if (typeof protocolVersionCheckingResult === 'object') {
-      await setResponse(protocolVersionCheckingResult, res);
-      return false;
-    }
-
-    return true;
-  };
-
-  const isAuthenticated = async (req: FastifyRequest, res: FastifyReply) => {
-    // Authenticate Ruby client
-    const authResult = authenticate(req);
-
-    if (typeof authResult === 'object') {
-      await setResponse(authResult, res);
-      return false;
-    }
-
-    return true;
-  };
-
-  const requestPrechecks = async (req: FastifyRequest, res: FastifyReply) => {
-    if (!(await isProtocolVersionMatch(req, res))) {
-      return false;
-    }
-
-    if (!(await isAuthenticated(req, res))) {
-      return false;
-    }
-
-    return true;
-  };
+  // Ensure NDJSON bodies are not buffered and are available as a stream immediately
+  app.addContentTypeParser('application/x-ndjson', (req, payload, done) => {
+    // Pass through the raw stream; the route will consume req.raw
+    done(null, payload);
+  });
 
   // See https://github.com/shakacode/react_on_rails_pro/issues/119 for why
   // the digest is part of the request URL. Yes, it's not used here, but the
@@ -209,7 +189,9 @@ export default function run(config: Partial<Config>) {
     // Can't infer from the route like Express can
     Params: { bundleTimestamp: string; renderRequestDigest: string };
   }>('/bundles/:bundleTimestamp/render/:renderRequestDigest', async (req, res) => {
-    if (!(await requestPrechecks(req, res))) {
+    const precheckResult = performRequestPrechecks(req.body);
+    if (precheckResult) {
+      await setResponse(precheckResult, res);
       return;
     }
 
@@ -251,7 +233,7 @@ export default function run(config: Partial<Config>) {
             providedNewBundles,
             assetsToCopy,
           });
-          await setResponse(result, res);
+          await setResponse(result.response, res);
         } catch (err) {
           const exceptionMessage = formatExceptionMessage(
             renderingRequest,
@@ -269,17 +251,124 @@ export default function run(config: Partial<Config>) {
     }
   });
 
+  // Streaming NDJSON incremental render endpoint
+  app.post<{
+    Params: { bundleTimestamp: string; renderRequestDigest: string };
+  }>('/bundles/:bundleTimestamp/incremental-render/:renderRequestDigest', async (req, res) => {
+    const { bundleTimestamp } = req.params;
+
+    // Stream parser state
+    let incrementalSink: IncrementalRenderSink | undefined;
+
+    try {
+      // Handle the incremental render stream
+      await handleIncrementalRenderStream({
+        request: req,
+        onRenderRequestReceived: async (obj: unknown) => {
+          // Build a temporary FastifyRequest shape for protocol/auth check
+          const tempReqBody = typeof obj === 'object' && obj !== null ? (obj as Record<string, unknown>) : {};
+
+          // Perform request prechecks
+          const precheckResult = performRequestPrechecks(tempReqBody);
+          if (precheckResult) {
+            return {
+              response: precheckResult,
+              shouldContinue: false,
+            };
+          }
+
+          // Extract data for incremental render request
+          const dependencyBundleTimestamps = extractBodyArrayField(
+            tempReqBody as WithBodyArrayField<Record<string, unknown>, 'dependencyBundleTimestamps'>,
+            'dependencyBundleTimestamps',
+          );
+
+          const initial: IncrementalRenderInitialRequest = {
+            renderingRequest: String((tempReqBody as { renderingRequest?: string }).renderingRequest ?? ''),
+            bundleTimestamp,
+            dependencyBundleTimestamps,
+          };
+
+          try {
+            const { response, sink } = await handleIncrementalRenderRequest(initial);
+            incrementalSink = sink;
+
+            return {
+              response,
+              shouldContinue: !!incrementalSink,
+            };
+          } catch (err) {
+            const errorResponse = errorResponseResult(
+              formatExceptionMessage(
+                'IncrementalRender',
+                err,
+                'Error while handling incremental render request',
+              ),
+            );
+            return {
+              response: errorResponse,
+              shouldContinue: false,
+            };
+          }
+        },
+
+        onUpdateReceived: (obj: unknown) => {
+          if (!incrementalSink) {
+            log.error({ msg: 'Unexpected update chunk received after rendering was aborted', obj });
+            return;
+          }
+
+          try {
+            incrementalSink.add(obj);
+          } catch (err) {
+            // Log error but don't stop processing
+            log.error({ err, msg: 'Error processing update chunk' });
+          }
+        },
+
+        onResponseStart: async (response: ResponseResult) => {
+          await setResponse(response, res);
+        },
+
+        onRequestEnded: () => {
+          // Do nothing
+        },
+      });
+    } catch (err) {
+      // If an error occurred during stream processing, send error response
+      const errorResponse = errorResponseResult(
+        formatExceptionMessage('IncrementalRender', err, 'Error while processing incremental render stream'),
+      );
+      await setResponse(errorResponse, res);
+    }
+  });
+
   // There can be additional files that might be required at the runtime.
   // Since the remote renderer doesn't contain any assets, they must be uploaded manually.
   app.post<{
     Body: WithBodyArrayField<Record<string, Asset>, 'targetBundles'>;
   }>('/upload-assets', async (req, res) => {
-    if (!(await requestPrechecks(req, res))) {
+    const precheckResult = performRequestPrechecks(req.body);
+    if (precheckResult) {
+      await setResponse(precheckResult, res);
       return;
     }
     let lockAcquired = false;
     let lockfileName: string | undefined;
-    const assets: Asset[] = Object.values(req.body).filter(isAsset);
+    const assets: Asset[] = [];
+
+    // Extract bundles that start with 'bundle_' prefix
+    const bundles: Array<{ timestamp: string; bundle: Asset }> = [];
+    Object.entries(req.body).forEach(([key, value]) => {
+      if (isAsset(value)) {
+        if (key.startsWith('bundle_')) {
+          const timestamp = key.replace('bundle_', '');
+          bundles.push({ timestamp, bundle: value });
+        } else {
+          assets.push(value);
+        }
+      }
+    });
 
     // Handle targetBundles as either a string or an array
     const targetBundles = extractBodyArrayField(req.body, 'targetBundles');
@@ -291,7 +380,9 @@ export default function run(config: Partial<Config>) {
     }
 
     const assetsDescription = JSON.stringify(assets.map((asset) => asset.filename));
-    const taskDescription = `Uploading files ${assetsDescription} to bundle directories: ${targetBundles.join(', ')}`;
+    const bundlesDescription =
+      bundles.length > 0 ? ` and bundles ${JSON.stringify(bundles.map((b) => b.bundle.filename))}` : '';
+    const taskDescription = `Uploading files ${assetsDescription}${bundlesDescription} to bundle directories: ${targetBundles.join(', ')}`;
 
     try {
       const { lockfileName: name, wasLockAcquired, errorMessage } = await lock('transferring-assets');
@@ -330,7 +421,24 @@ export default function run(config: Partial<Config>) {
 
           await Promise.all(assetCopyPromises);
 
-          // Delete assets from uploads directory
+          // Handle bundles using the existing logic from handleRenderRequest
+          if (bundles.length > 0) {
+            const providedNewBundles = bundles.map(({ timestamp, bundle }) => ({
+              timestamp,
+              bundle,
+            }));
+
+            // Use the existing bundle handling logic
+            // Note: handleNewBundlesProvided will handle deleting the uploaded bundle files
+            // Pass null for assetsToCopy since we handle assets separately in this endpoint
+            const bundleResult = await handleNewBundlesProvided('upload-assets', providedNewBundles, null);
+            if (bundleResult) {
+              await setResponse(bundleResult, res);
+              return;
+            }
+          }
+
+          // Delete assets from uploads directory (bundles are already handled by handleNewBundlesProvided)
           await deleteUploadedAssets(assets);
 
           await setResponse(
@@ -341,7 +449,7 @@ export default function run(config: Partial<Config>) {
             res,
           );
         } catch (err) {
-          const msg = 'ERROR when trying to copy assets';
+          const msg = 'ERROR when trying to copy assets and bundles';
           const message = `${msg}. ${err}. Task: ${taskDescription}`;
           log.error({
             msg,
@@ -373,7 +481,9 @@ export default function run(config: Partial<Config>) {
     Querystring: { filename: string };
     Body: WithBodyArrayField<Record<string, unknown>, 'targetBundles'>;
   }>('/asset-exists', async (req, res) => {
-    if (!(await isAuthenticated(req, res))) {
+    const authResult = authenticate(req.body as AuthBody);
+    if (authResult) {
+      await setResponse(authResult, res);
       return;
     }
 

packages/node-renderer/src/worker/authHandler.ts

@@ -6,13 +6,16 @@
  */
 // TODO: Replace with fastify-basic-auth per https://github.com/shakacode/react_on_rails_pro/issues/110
 
-import type { FastifyRequest } from './types';
 import { getConfig } from '../shared/configBuilder';
 
-export = function authenticate(req: FastifyRequest) {
+export interface AuthBody {
+  password?: string;
+}
+
+export function authenticate(body: AuthBody) {
   const { password } = getConfig();
 
-  if (password && password !== (req.body as { password?: string }).password) {
+  if (password && password !== body.password) {
     return {
       headers: { 'Cache-Control': 'no-cache, no-store, max-age=0, must-revalidate' },
       status: 401,
@@ -21,4 +24,4 @@ export = function authenticate(req: FastifyRequest) {
   }
 
   return undefined;
-};
+}

packages/node-renderer/src/worker/checkProtocolVersionHandler.ts

@@ -2,23 +2,26 @@
  * Logic for checking protocol version.
  * @module worker/checkProtocVersionHandler
  */
-import type { FastifyRequest } from './types';
 import packageJson from '../shared/packageJson';
 
-export = function checkProtocolVersion(req: FastifyRequest) {
-  const reqProtocolVersion = (req.body as { protocolVersion?: string }).protocolVersion;
+export interface ProtocolVersionBody {
+  protocolVersion?: string;
+}
+
+export function checkProtocolVersion(body: ProtocolVersionBody) {
+  const reqProtocolVersion = body.protocolVersion;
   if (reqProtocolVersion !== packageJson.protocolVersion) {
     return {
       headers: { 'Cache-Control': 'no-cache, no-store, max-age=0, must-revalidate' },
       status: 412,
       data: `Unsupported renderer protocol version ${
         reqProtocolVersion
           ? `request protocol ${reqProtocolVersion}`
-          : `MISSING with body ${JSON.stringify(req.body)}`
+          : `MISSING with body ${JSON.stringify(body)}`
       } does not match installed renderer protocol ${packageJson.protocolVersion} for version ${packageJson.version}.
 Update either the renderer or the Rails server`,
     };
   }
 
   return undefined;
-};
+}