Refactor incremental rendering logic and enhance bundle validation

- Introduced `validateAndGetBundlePaths` and `buildVMsForBundles` functions to streamline bundle validation and VM building processes.
- Updated `handleIncrementalRenderRequest` and `handleRenderRequest` to utilize the new validation and VM building functions, improving code clarity and maintainability.
- Enhanced error handling for rendering execution and added support for incremental updates using an EventEmitter.
- Created a new `sharedRenderUtils` module to encapsulate shared rendering logic, promoting code reuse and organization.

Author: AbanoubGhadban

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

@@ -34,11 +34,11 @@ import {
   getAssetPath,
   getBundleDirectory,
   deleteUploadedAssets,
-  validateBundlesExist,
 } from './shared/utils';
 import * as errorReporter from './shared/errorReporter';
 import { lock, unlock } from './shared/locks';
 import { startSsrRequestOptions, trace } from './shared/tracing';
+import { validateAndGetBundlePaths } from './worker/sharedRenderUtils';
 
 // Uncomment the below for testing timeouts:
 // import { delay } from './shared/utils';
@@ -323,15 +323,16 @@ export default function run(config: Partial<Config>) {
             };
           }
 
-          // Bundle validation
+          // Bundle validation using shared utility
           const dependencyBundleTimestamps = extractBodyArrayField(
             tempReqBody as WithBodyArrayField<Record<string, unknown>, 'dependencyBundleTimestamps'>,
             'dependencyBundleTimestamps',
           );
-          const missingBundleError = await validateBundlesExist(bundleTimestamp, dependencyBundleTimestamps);
-          if (missingBundleError) {
+          
+          const validationResult = await validateAndGetBundlePaths(bundleTimestamp, dependencyBundleTimestamps);
+          if (!validationResult.success) {
             return {
-              response: missingBundleError,
+              response: validationResult.error!,
               shouldContinue: false,
             };
           }

react_on_rails_pro/packages/node-renderer/src/worker/handleIncrementalRenderRequest.ts

@@ -1,5 +1,7 @@
 import { Readable } from 'stream';
+import { EventEmitter } from 'events';
 import type { ResponseResult } from '../shared/utils';
+import { validateAndGetBundlePaths, buildVMsForBundles, executeRenderInVM } from './sharedRenderUtils';
 
 export type IncrementalRenderSink = {
   /** Called for every subsequent NDJSON object after the first one */
@@ -22,36 +24,156 @@ export type IncrementalRenderResult = {
 };
 
 /**
- * Starts handling an incremental render request. This function is intended to:
- * - Initialize any resources needed to process the render
- * - Return both a stream that will be sent to the client and a sink for incoming chunks
- *
- * NOTE: This is intentionally left unimplemented. Tests should mock this.
+ * Starts handling an incremental render request. This function:
+ * - Creates an EventEmitter for handling updates
+ * - Builds the VM if needed
+ * - Executes the initial render request
+ * - Returns both a stream that will be sent to the client and a sink for incoming chunks
  */
-export function handleIncrementalRenderRequest(initial: IncrementalRenderInitialRequest): Promise<IncrementalRenderResult> {
-  // Empty placeholder implementation. Real logic will be added later.
-  return Promise.resolve({
+export async function handleIncrementalRenderRequest(
+  initial: IncrementalRenderInitialRequest,
+): Promise<IncrementalRenderResult> {
+  const { renderingRequest, bundleTimestamp, dependencyBundleTimestamps } = initial;
+
+  // Create event emitter for this specific request
+  const updateEmitter = new EventEmitter();
+
+  // Validate bundles and get paths
+  const validationResult = await validateAndGetBundlePaths(bundleTimestamp, dependencyBundleTimestamps);
+  if (!validationResult.success || !validationResult.bundleFilePath) {
+    return {
+      response: validationResult.error || {
+        status: 500,
+        headers: { 'Cache-Control': 'no-cache, no-store, max-age=0, must-revalidate' },
+        data: 'Bundle validation failed',
+      },
+      sink: {
+        add: () => {
+          /* no-op */
+        },
+        end: () => {
+          /* no-op */
+        },
+        abort: () => {
+          /* no-op */
+        },
+      },
+    };
+  }
+
+  // Build VMs
+  const vmBuildResult = await buildVMsForBundles(
+    validationResult.bundleFilePath,
+    validationResult.dependencyBundleFilePaths || [],
+  );
+  if (!vmBuildResult.success) {
+    return {
+      response: vmBuildResult.error || {
+        status: 500,
+        headers: { 'Cache-Control': 'no-cache, no-store, max-age=0, must-revalidate' },
+        data: 'VM building failed',
+      },
+      sink: {
+        add: () => {
+          /* no-op */
+        },
+        end: () => {
+          /* no-op */
+        },
+        abort: () => {
+          /* no-op */
+        },
+      },
+    };
+  }
+
+  // Create the response stream
+  const responseStream = new Readable({
+    read() {
+      // No-op - data will be pushed via events
+    },
+  });
+
+  // Set up event listeners for the response stream
+  updateEmitter.on('update', (data: unknown) => {
+    // Push update data to the response stream
+    responseStream.push(`${JSON.stringify(data)}\n`);
+  });
+
+  updateEmitter.on('end', () => {
+    // End the response stream
+    responseStream.push(null);
+  });
+
+  updateEmitter.on('error', (error: unknown) => {
+    // Handle error and end stream
+    const errorMessage = error instanceof Error ? error.message : String(error);
+    responseStream.push(`{"error":"${errorMessage}"}\n`);
+    responseStream.push(null);
+  });
+
+  // Execute the initial render request with the update emitter
+  const executionResult = await executeRenderInVM(
+    renderingRequest,
+    validationResult.bundleFilePath,
+    updateEmitter,
+  );
+
+  // Handle the render result
+  if (executionResult.success && executionResult.result) {
+    // Initial render completed successfully
+    if (executionResult.result.data) {
+      const dataString =
+        typeof executionResult.result.data === 'string'
+          ? executionResult.result.data
+          : JSON.stringify(executionResult.result.data);
+      responseStream.push(`${dataString}\n`);
+    }
+  } else {
+    // Render failed
+    const errorMessage =
+      typeof executionResult.error?.data === 'string' ? executionResult.error.data : 'Unknown render error';
+    responseStream.push(`{"error":"${errorMessage}"}\n`);
+    responseStream.push(null);
+    return {
+      response: executionResult.error || {
+        status: 500,
+        headers: { 'Cache-Control': 'no-cache, no-store, max-age=0, must-revalidate' },
+        data: 'Render execution failed',
+      },
+      sink: {
+        add: () => {
+          /* no-op */
+        },
+        end: () => {
+          /* no-op */
+        },
+        abort: () => {
+          /* no-op */
+        },
+      },
+    };
+  }
+
+  return {
     response: {
       status: 200,
       headers: { 'Cache-Control': 'no-cache, no-store, max-age=0, must-revalidate' },
-      stream: new Readable({
-        read() {
-          // No-op for now
-        },
-      }),
-    } as ResponseResult,
+      stream: responseStream,
+    },
     sink: {
-      add: () => {
-        /* no-op */
+      add: (chunk: unknown) => {
+        // Emit event when chunk arrives
+        updateEmitter.emit('update', chunk);
       },
       end: () => {
-        /* no-op */
+        updateEmitter.emit('end');
       },
-      abort: () => {
-        /* no-op */
+      abort: (error: unknown) => {
+        updateEmitter.emit('error', error);
       },
     },
-  });
+  };
 }
 
 export type { ResponseResult };

react_on_rails_pro/packages/node-renderer/src/worker/handleRenderRequest.ts

@@ -5,7 +5,6 @@
  * @module worker/handleRenderRequest
  */
 
-import cluster from 'cluster';
 import path from 'path';
 import { mkdir } from 'fs/promises';
 import { lock, unlock } from '../shared/locks';
@@ -19,15 +18,17 @@ import {
   copyUploadedAssets,
   ResponseResult,
   moveUploadedAsset,
-  isReadableStream,
-  isErrorRenderResult,
   getRequestBundleFilePath,
   deleteUploadedAssets,
-  validateBundlesExist,
 } from '../shared/utils';
 import { getConfig } from '../shared/configBuilder';
-import * as errorReporter from '../shared/errorReporter';
-import { buildVM, hasVMContextForBundle, runInVM } from './vm';
+import { hasVMContextForBundle } from './vm';
+import {
+  validateAndGetBundlePaths,
+  buildVMsForBundles,
+  executeRenderInVM,
+  createRenderErrorResponse,
+} from './sharedRenderUtils';
 
 export type ProvidedNewBundle = {
   timestamp: string | number;
@@ -39,36 +40,15 @@ async function prepareResult(
   bundleFilePathPerTimestamp: string,
 ): Promise<ResponseResult> {
   try {
-    const result = await runInVM(renderingRequest, bundleFilePathPerTimestamp, cluster);
-
-    let exceptionMessage = null;
-    if (!result) {
-      const error = new Error('INVALID NIL or NULL result for rendering');
-      exceptionMessage = formatExceptionMessage(renderingRequest, error, 'INVALID result for prepareResult');
-    } else if (isErrorRenderResult(result)) {
-      ({ exceptionMessage } = result);
-    }
-
-    if (exceptionMessage) {
-      return errorResponseResult(exceptionMessage);
-    }
+    const executionResult = await executeRenderInVM(renderingRequest, bundleFilePathPerTimestamp);
 
-    if (isReadableStream(result)) {
-      return {
-        headers: { 'Cache-Control': 'public, max-age=31536000' },
-        status: 200,
-        stream: result,
-      };
+    if (!executionResult.success || !executionResult.result) {
+      return executionResult.error || errorResponseResult('Unknown error during render execution');
     }
 
-    return {
-      headers: { 'Cache-Control': 'public, max-age=31536000' },
-      status: 200,
-      data: result,
-    };
+    return executionResult.result;
   } catch (err) {
-    const exceptionMessage = formatExceptionMessage(renderingRequest, err, 'Unknown error calling runInVM');
-    return errorResponseResult(exceptionMessage);
+    return createRenderErrorResponse(renderingRequest, err, 'Unknown error calling runInVM');
   }
 }
 
@@ -193,7 +173,6 @@ export async function handleRenderRequest({
   assetsToCopy?: Asset[] | null;
 }): Promise<ResponseResult> {
   try {
-    // const bundleFilePathPerTimestamp = getRequestBundleFilePath(bundleTimestamp);
     const allBundleFilePaths = Array.from(
       new Set([...(dependencyBundleTimestamps ?? []), bundleTimestamp].map(getRequestBundleFilePath)),
     );
@@ -222,25 +201,27 @@ export async function handleRenderRequest({
       }
     }
 
-    // Check if the bundle exists:
-    const missingBundleError = await validateBundlesExist(bundleTimestamp, dependencyBundleTimestamps);
-    if (missingBundleError) {
-      return missingBundleError;
+    // Validate bundles and get paths
+    const validationResult = await validateAndGetBundlePaths(bundleTimestamp, dependencyBundleTimestamps);
+    if (!validationResult.success || !validationResult.bundleFilePath) {
+      return validationResult.error || errorResponseResult('Bundle validation failed');
     }
 
-    // The bundle exists, but the VM has not yet been created.
-    // Another worker must have written it or it was saved during deployment.
-    log.info('Bundle %s exists. Building VM for worker %s.', entryBundleFilePath, workerIdLabel());
-    await Promise.all(allBundleFilePaths.map((bundleFilePath) => buildVM(bundleFilePath)));
+    // Build VMs
+    const vmBuildResult = await buildVMsForBundles(
+      validationResult.bundleFilePath,
+      validationResult.dependencyBundleFilePaths || [],
+    );
+    if (!vmBuildResult.success) {
+      return vmBuildResult.error || errorResponseResult('VM building failed');
+    }
 
     return await prepareResult(renderingRequest, entryBundleFilePath);
   } catch (error) {
-    const msg = formatExceptionMessage(
+    return createRenderErrorResponse(
       renderingRequest,
       error,
       'Caught top level error in handleRenderRequest',
     );
-    errorReporter.message(msg);
-    return Promise.reject(error as Error);
   }
 }