Use Redis for the communication between Server and Engine (#1172)

MarceloRGonc · web-flow · commit 0b6ae1214ec0 · 2025-08-27T10:46:03.000+01:00
Part of OPS-2396. Additional Notes - Save the requests to the engine in cache - Save the responses from the engine in cache - Save the run progress requests in cache
diff --git a/packages/engine/src/api-handler.ts b/packages/engine/src/api-handler.ts
@@ -1,6 +1,9 @@
+/* eslint-disable @typescript-eslint/no-explicit-any */
 import { FastifyPluginAsyncTypebox } from '@fastify/type-provider-typebox';
 import {
   blocksBuilder,
+  BodyAccessKeyRequest,
+  bodyConverterModule,
   logger,
   MAX_REQUEST_BODY_BYTES,
   runWithLogContext,
@@ -25,35 +28,38 @@ const engineController: FastifyPluginAsyncTypebox = async (fastify) => {
       },
     },
     async (request, reply) => {
-      const requestIdHeaders = request.headers;
-      const requestId = Array.isArray(requestIdHeaders)
-        ? requestIdHeaders[0]
-        : requestIdHeaders;
-
+      const requestBody = request.body;
       await runWithLogContext(
         {
           deadlineTimestamp: request.body.deadlineTimestamp.toString(),
-          requestId: requestId ?? request.id,
-          operationType: request.body.operationType,
+          operationType: requestBody.operationType,
+          requestId: requestBody.requestId,
         },
-        () => handleRequest(request, reply),
+        () => handleRequest(reply, requestBody),
       );
     },
   );
 };
 
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
-async function handleRequest(request: any, reply: any): Promise<void> {
+async function handleRequest(
+  reply: any,
+  requestBody: EngineRequest,
+): Promise<void> {
   try {
-    const { engineInput, operationType } = request.body as EngineRequest;
-
+    const { engineInput, operationType } = requestBody;
     logger.info(`Received request for operation [${operationType}]`, {
       operationType,
     });
 
-    const result = await executeEngine(engineInput, operationType);
+    const bodyAccessKey = await executeEngine(
+      requestBody.requestId,
+      engineInput,
+      operationType,
+    );
 
-    await reply.status(StatusCodes.OK).send(result);
+    await reply.status(StatusCodes.OK).send({
+      bodyAccessKey,
+    } as BodyAccessKeyRequest);
   } catch (error) {
     logger.error('Engine execution failed.', { error });
 
@@ -73,6 +79,9 @@ export const start = async (): Promise<void> => {
     setStopHandlers(app);
 
     await blocksBuilder();
+
+    await app.register(bodyConverterModule);
+
     await app.register(engineController);
 
     await app.listen({
diff --git a/packages/engine/src/engine-executor.ts b/packages/engine/src/engine-executor.ts
@@ -2,15 +2,17 @@ import {
   encryptionKeyInitializer,
   logger,
   networkUtls,
+  saveRequestBody,
 } from '@openops/server-shared';
-import { EngineOperationType, EngineResponse } from '@openops/shared';
+import { EngineOperationType } from '@openops/shared';
 import { execute } from './lib/operations';
 
 export async function executeEngine(
+  requestId: string,
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   engineInput: any,
   operationType: EngineOperationType,
-): Promise<EngineResponse<unknown>> {
+): Promise<string> {
   const startTime = performance.now();
 
   await encryptionKeyInitializer();
@@ -23,6 +25,8 @@ export async function executeEngine(
 
   const duration = Math.floor(performance.now() - startTime);
 
+  const key = await saveRequestBody(requestId, result);
+
   logger.info(
     `Finished operation [${operationType}] with status [${result.status}] in ${duration}ms`,
     {
@@ -31,5 +35,5 @@ export async function executeEngine(
     },
   );
 
-  return result;
+  return key;
 }
diff --git a/packages/engine/src/lambda-handler.ts b/packages/engine/src/lambda-handler.ts
@@ -1,7 +1,13 @@
-import { logger, runWithLogContext, sendLogs } from '@openops/server-shared';
+/* eslint-disable @typescript-eslint/no-explicit-any */
+import {
+  BodyAccessKeyRequest,
+  getRequestBody,
+  logger,
+  runWithLogContext,
+  sendLogs,
+} from '@openops/server-shared';
 import { EngineResponseStatus } from '@openops/shared';
 import { APIGatewayEvent, APIGatewayProxyResult, Context } from 'aws-lambda';
-import { nanoid } from 'nanoid';
 import { executeEngine } from './engine-executor';
 import { EngineRequest } from './main';
 
@@ -19,41 +25,55 @@ export async function lambdaHandler(
     };
   }
 
-  const data = await parseJson<EngineRequest>(event.body);
+  const data = await parseJson<BodyAccessKeyRequest>(event.body);
+
+  let requestBody: EngineRequest;
+  try {
+    requestBody = await getRequestBody<EngineRequest>(data.bodyAccessKey);
+  } catch (error) {
+    return {
+      statusCode: 400,
+      body: JSON.stringify({
+        status: EngineResponseStatus.ERROR,
+        message: (error as Error).message,
+      }),
+    };
+  }
 
   return runWithLogContext<APIGatewayProxyResult | undefined>(
     {
-      requestId: event.headers.requestId ?? nanoid(),
+      deadlineTimestamp: requestBody.deadlineTimestamp.toString(),
+      operationType: requestBody.operationType,
       awsRequestId: context.awsRequestId,
-      operationType: data.operationType,
+      requestId: requestBody.requestId,
     },
-    () => handleEvent(data),
+    () => handleEvent(requestBody),
   );
 }
 
 async function handleEvent(
-  data: EngineRequest,
+  requestBody: EngineRequest,
 ): Promise<APIGatewayProxyResult | undefined> {
   try {
-    logger.info('Request received in the engine.');
+    const { requestId, engineInput, operationType } = requestBody;
 
-    if (!data.engineInput) {
-      return {
-        statusCode: 400,
-        body: JSON.stringify({
-          status: EngineResponseStatus.ERROR,
-          message: 'The received input is not valid.',
-        }),
-      };
-    }
+    logger.info(`Received request for operation [${operationType}]`, {
+      operationType,
+    });
 
-    const result = await executeEngine(data.engineInput, data.operationType);
+    const bodyAccessKey = await executeEngine(
+      requestId,
+      engineInput,
+      operationType,
+    );
 
     await sendLogs();
 
     return {
       statusCode: 200,
-      body: JSON.stringify(result),
+      body: JSON.stringify({
+        bodyAccessKey,
+      } as BodyAccessKeyRequest),
     };
   } catch (error) {
     logger.error('Engine execution failed.', { error });
diff --git a/packages/engine/src/lib/services/progress.service.ts b/packages/engine/src/lib/services/progress.service.ts
@@ -1,5 +1,10 @@
 import { makeHttpRequest } from '@openops/common';
-import { hashUtils, logger } from '@openops/server-shared';
+import {
+  BodyAccessKeyRequest,
+  hashUtils,
+  logger,
+  saveRequestBody,
+} from '@openops/server-shared';
 import { UpdateRunProgressRequest } from '@openops/shared';
 import { Mutex } from 'async-mutex';
 import { AxiosHeaders } from 'axios';
@@ -53,14 +58,18 @@ const sendUpdateRunRequest = async (
   lastRequestHash = requestHash;
 
   try {
+    const resourceKey = await saveRequestBody(request.runId, request);
+
     await makeHttpRequest(
       'POST',
       url.toString(),
       new AxiosHeaders({
         'Content-Type': 'application/json',
         Authorization: `Bearer ${engineConstants.engineToken}`,
       }),
-      request,
+      {
+        bodyAccessKey: resourceKey,
+      } as BodyAccessKeyRequest,
       {
         retries: MAX_RETRIES,
         retryDelay: (retryCount: number) => {
diff --git a/packages/engine/src/main.ts b/packages/engine/src/main.ts
@@ -9,9 +9,10 @@ import { lambdaHandler } from './lambda-handler';
 import { EngineConstants } from './lib/handler/context/engine-constants';
 
 export const EngineRequest = Type.Object({
-  operationType: Type.Enum(EngineOperationType),
-  engineInput: Type.Object({}),
+  requestId: Type.String(),
+  engineInput: Type.Unknown(),
   deadlineTimestamp: Type.Number(),
+  operationType: Type.Enum(EngineOperationType),
 });
 
 export type EngineRequest = Static<typeof EngineRequest>;
diff --git a/packages/engine/test/services/progress.service.test.ts b/packages/engine/test/services/progress.service.test.ts
@@ -1,9 +1,10 @@
+const saveRequestBodyMock = jest.fn();
 import { progressService } from '../../src/lib/services/progress.service';
-import { logger } from '@openops/server-shared';
 import { throwIfExecutionTimeExceeded } from '../../src/lib/timeout-validator';
 
 jest.mock('@openops/server-shared', () => ({
   ...jest.requireActual('@openops/server-shared'),
+  saveRequestBody: saveRequestBodyMock,
   logger: {
     debug: jest.fn(),
     error: jest.fn(),
@@ -44,12 +45,12 @@ describe('Progress Service', () => {
 
   beforeEach(() => {
     jest.clearAllMocks();
-    
+
     // Reset the timeout mock to not throw by default
     mockThrowIfExecutionTimeExceeded.mockReset();
-    
+
     mockMakeHttpRequest.mockResolvedValue({});
-    
+
     // Reset the global lastRequestHash by calling with unique params
     // This ensures no deduplication issues between tests
     jest.clearAllMocks();
@@ -69,18 +70,23 @@ describe('Progress Service', () => {
         },
       };
 
+      saveRequestBodyMock.mockResolvedValue('request:test-run-id');
       await progressService.sendUpdate(successParams);
 
+      expect(saveRequestBodyMock).toHaveBeenCalledWith('test-run-id', expect.objectContaining({
+        executionCorrelationId: 'test-correlation-id-success',
+        runId: 'test-run-id',
+        workerHandlerId: 'test-handler-id',
+        progressUpdateType: 'WEBHOOK_RESPONSE',
+      }));
+
       expect(successParams.flowExecutorContext.toResponse).toHaveBeenCalled();
       expect(mockMakeHttpRequest).toHaveBeenCalledWith(
         'POST',
         'http://localhost:3000/v1/engine/update-run',
         expect.any(Object),
         expect.objectContaining({
-          executionCorrelationId: 'test-correlation-id-success',
-          runId: 'test-run-id',
-          workerHandlerId: 'test-handler-id',
-          progressUpdateType: 'WEBHOOK_RESPONSE',
+          bodyAccessKey: 'request:test-run-id',
         }),
         expect.objectContaining({
           retries: 3,
@@ -98,8 +104,22 @@ describe('Progress Service', () => {
         },
       };
 
+      saveRequestBodyMock.mockResolvedValue('request:test-run-id');
       await progressService.sendUpdate(uniqueParams);
 
+      expect(saveRequestBodyMock).toHaveBeenCalledWith('test-run-id', expect.objectContaining({
+        executionCorrelationId: 'test-correlation-id-payload',
+        runId: 'test-run-id',
+        workerHandlerId: 'test-handler-id',
+        progressUpdateType: 'WEBHOOK_RESPONSE',
+        runDetails: expect.objectContaining({
+          steps: {},
+          status: 'RUNNING',
+          duration: 1000,
+          tasks: 1,
+        }),
+      }));
+
       expect(mockMakeHttpRequest).toHaveBeenCalledTimes(1);
       const call = mockMakeHttpRequest.mock.calls[0];
       const [method, url, _, requestBody] = call;
@@ -108,16 +128,7 @@ describe('Progress Service', () => {
       expect(url).toBe('http://localhost:3000/v1/engine/update-run');
       expect(requestBody).toEqual(
         expect.objectContaining({
-          executionCorrelationId: 'test-correlation-id-payload',
-          runId: 'test-run-id',
-          workerHandlerId: 'test-handler-id',
-          progressUpdateType: 'WEBHOOK_RESPONSE',
-          runDetails: expect.objectContaining({
-            steps: {},
-            status: 'RUNNING',
-            duration: 1000,
-            tasks: 1,
-          }),
+          bodyAccessKey: 'request:test-run-id',
         })
       );
     });
@@ -132,13 +143,30 @@ describe('Progress Service', () => {
         },
       };
 
+      saveRequestBodyMock.mockResolvedValue('request:test-run-id');
       await progressService.sendUpdate(paramsWithoutHandlerId);
 
+      expect(saveRequestBodyMock).toHaveBeenCalledWith('test-run-id', expect.objectContaining({
+        executionCorrelationId: 'test-correlation-id-null-handler',
+        runId: 'test-run-id',
+        workerHandlerId: null,
+        progressUpdateType: 'WEBHOOK_RESPONSE',
+        runDetails: expect.objectContaining({
+          steps: {},
+          status: 'RUNNING',
+          duration: 1000,
+          tasks: 1,
+        }),
+      }));
+
       expect(mockMakeHttpRequest).toHaveBeenCalledTimes(1);
       const call = mockMakeHttpRequest.mock.calls[0];
       const [_, __, ___, requestBody] = call;
-      
-      expect(requestBody.workerHandlerId).toBe(null);
+
+      expect(requestBody).toEqual({
+          bodyAccessKey: 'request:test-run-id',
+        }
+      );
     });
 
     it('should deduplicate identical requests based on hash', async () => {
@@ -327,10 +355,10 @@ describe('Progress Service', () => {
       // Test the retry delay function
       const call = mockMakeHttpRequest.mock.calls[0];
       const [_, __, ___, ____, options] = call;
-      
+
       expect(options.retryDelay(0)).toBe(200); // 1st retry: 200ms
       expect(options.retryDelay(1)).toBe(400); // 2nd retry: 400ms
       expect(options.retryDelay(2)).toBe(600); // 3rd retry: 600ms
     });
   });
-}); 
+});
diff --git a/packages/server/api/src/app/app.ts b/packages/server/api/src/app/app.ts
diff --git a/packages/server/worker/src/lib/engine/call-engine.ts b/packages/server/worker/src/lib/engine/call-engine.ts
