Fix engine progress calls (#923)

alexandrudanpop · web-flow · commit 19c26ccef887 · 2025-07-08T15:07:35.000+03:00
diff --git a/packages/engine/src/lib/handler/flow-executor.ts b/packages/engine/src/lib/handler/flow-executor.ts
@@ -1,4 +1,4 @@
-import { logger, runWithTemporaryContext } from '@openops/server-shared';
+import { runWithTemporaryContext } from '@openops/server-shared';
 import {
   Action,
   ActionType,
@@ -57,7 +57,7 @@ export const flowExecutor = {
         ? output.setVerdict(ExecutionVerdict.SUCCEEDED, output.verdictResponse)
         : output;
 
-    sendProgress(newContext, constants);
+    await sendProgress(newContext, constants);
 
     return newContext.toResponse();
   },
@@ -103,7 +103,7 @@ export const flowExecutor = {
         duration: stepEndTime - stepStartTime,
       });
 
-      sendProgress(flowExecutionContext, constants);
+      await sendProgress(flowExecutionContext, constants);
 
       if (flowExecutionContext.verdict !== ExecutionVerdict.RUNNING) {
         break;
@@ -121,13 +121,9 @@ export const flowExecutor = {
 function sendProgress(
   flowExecutionContext: FlowExecutorContext,
   constants: EngineConstants,
-): void {
-  progressService
-    .sendUpdate({
-      engineConstants: constants,
-      flowExecutorContext: flowExecutionContext,
-    })
-    .catch((error) => {
-      logger.error('Error sending progress update', error);
-    });
+): Promise<void> {
+  return progressService.sendUpdate({
+    engineConstants: constants,
+    flowExecutorContext: flowExecutionContext,
+  });
 }
diff --git a/packages/engine/src/lib/services/progress.service.ts b/packages/engine/src/lib/services/progress.service.ts
@@ -1,12 +1,15 @@
+import { makeHttpRequest } from '@openops/common';
 import { hashUtils, logger } from '@openops/server-shared';
 import { UpdateRunProgressRequest } from '@openops/shared';
 import { Mutex } from 'async-mutex';
+import { AxiosHeaders } from 'axios';
 import { EngineConstants } from '../handler/context/engine-constants';
 import { FlowExecutorContext } from '../handler/context/flow-execution-context';
 import { throwIfExecutionTimeExceeded } from '../timeout-validator';
 
-let lastRequestHash: string | undefined = undefined;
+const MAX_RETRIES = 3;
 
+let lastRequestHash: string | undefined = undefined;
 const lock = new Mutex();
 
 export const progressService = {
@@ -42,6 +45,10 @@ const sendUpdateRunRequest = async (
     progressUpdateType: engineConstants.progressUpdateType,
   };
 
+  logger.debug(
+    `Sending progress update for ${request.runId} ${request.runDetails.status}`,
+  );
+
   // Request deduplication using hash comparison
   const requestHash = hashUtils.hashObject(request, (key, value) => {
     if (key === 'duration') return undefined;
@@ -54,14 +61,28 @@ const sendUpdateRunRequest = async (
 
   lastRequestHash = requestHash;
 
-  await fetch(url.toString(), {
-    method: 'POST',
-    headers: {
-      'Content-Type': 'application/json',
-      Authorization: `Bearer ${engineConstants.engineToken}`,
-    },
-    body: JSON.stringify(request),
-  });
+  try {
+    await makeHttpRequest(
+      'POST',
+      url.toString(),
+      new AxiosHeaders({
+        'Content-Type': 'application/json',
+        Authorization: `Bearer ${engineConstants.engineToken}`,
+      }),
+      request,
+      {
+        retries: MAX_RETRIES,
+        retryDelay: (retryCount: number) => {
+          return (retryCount + 1) * 200; // 200ms, 400ms, 600ms
+        },
+      },
+    );
+  } catch (error) {
+    logger.error(
+      `Progress update failed after ${MAX_RETRIES} retries for status ${request.runDetails.status} on run ${request.runId}`,
+      { error },
+    );
+  }
 };
 
 type UpdateStepProgressParams = {
diff --git a/packages/engine/test/services/progress.service.test.ts b/packages/engine/test/services/progress.service.test.ts
@@ -15,9 +15,12 @@ jest.mock('../../src/lib/timeout-validator', () => ({
   throwIfExecutionTimeExceeded: jest.fn(),
 }));
 
-const mockThrowIfExecutionTimeExceeded = throwIfExecutionTimeExceeded as jest.MockedFunction<typeof throwIfExecutionTimeExceeded>;
+jest.mock('@openops/common', () => ({
+  makeHttpRequest: jest.fn(),
+}));
 
-global.fetch = jest.fn();
+const mockThrowIfExecutionTimeExceeded = throwIfExecutionTimeExceeded as jest.MockedFunction<typeof throwIfExecutionTimeExceeded>;
+const mockMakeHttpRequest = require('@openops/common').makeHttpRequest as jest.MockedFunction<any>;
 
 describe('Progress Service', () => {
   const mockParams = {
@@ -45,10 +48,11 @@ describe('Progress Service', () => {
     // Reset the timeout mock to not throw by default
     mockThrowIfExecutionTimeExceeded.mockReset();
     
-    (global.fetch as jest.Mock).mockResolvedValue({
-      ok: true,
-      status: 200,
-    });
+    mockMakeHttpRequest.mockResolvedValue({});
+    
+    // Reset the global lastRequestHash by calling with unique params
+    // This ensures no deduplication issues between tests
+    jest.clearAllMocks();
   });
 
   afterEach(() => {
@@ -68,14 +72,19 @@ describe('Progress Service', () => {
       await progressService.sendUpdate(successParams);
 
       expect(successParams.flowExecutorContext.toResponse).toHaveBeenCalled();
-      expect(global.fetch).toHaveBeenCalledWith(
+      expect(mockMakeHttpRequest).toHaveBeenCalledWith(
+        'POST',
         'http://localhost:3000/v1/engine/update-run',
+        expect.any(Object),
         expect.objectContaining({
-          method: 'POST',
-          headers: {
-            'Content-Type': 'application/json',
-            Authorization: 'Bearer test-token',
-          },
+          executionCorrelationId: 'test-correlation-id-success',
+          runId: 'test-run-id',
+          workerHandlerId: 'test-handler-id',
+          progressUpdateType: 'WEBHOOK_RESPONSE',
+        }),
+        expect.objectContaining({
+          retries: 3,
+          retryDelay: expect.any(Function),
         })
       );
     });
@@ -109,11 +118,12 @@ describe('Progress Service', () => {
 
       await progressService.sendUpdate(uniqueParams);
 
-      expect(global.fetch).toHaveBeenCalledTimes(1);
-      const fetchCall = (global.fetch as jest.Mock).mock.calls[0];
-      const [url, options] = fetchCall;
-      const requestBody = JSON.parse(options.body);
+      expect(mockMakeHttpRequest).toHaveBeenCalledTimes(1);
+      const call = mockMakeHttpRequest.mock.calls[0];
+      const [method, url, _, requestBody] = call;
 
+      expect(method).toBe('POST');
+      expect(url).toBe('http://localhost:3000/v1/engine/update-run');
       expect(requestBody).toEqual(
         expect.objectContaining({
           executionCorrelationId: 'test-correlation-id-payload',
@@ -142,10 +152,9 @@ describe('Progress Service', () => {
 
       await progressService.sendUpdate(paramsWithoutHandlerId);
 
-      expect(global.fetch).toHaveBeenCalledTimes(1);
-      const fetchCall = (global.fetch as jest.Mock).mock.calls[0];
-      const [url, options] = fetchCall;
-      const requestBody = JSON.parse(options.body);
+      expect(mockMakeHttpRequest).toHaveBeenCalledTimes(1);
+      const call = mockMakeHttpRequest.mock.calls[0];
+      const [_, __, ___, requestBody] = call;
       
       expect(requestBody.workerHandlerId).toBe(null);
     });
@@ -162,7 +171,7 @@ describe('Progress Service', () => {
       await progressService.sendUpdate(duplicateParams);
       await progressService.sendUpdate(duplicateParams);
 
-      expect(global.fetch).toHaveBeenCalledTimes(1);
+      expect(mockMakeHttpRequest).toHaveBeenCalledTimes(1);
     });
 
     it('should send different requests when content changes', async () => {
@@ -185,7 +194,7 @@ describe('Progress Service', () => {
       await progressService.sendUpdate(params1);
       await progressService.sendUpdate(params2);
 
-      expect(global.fetch).toHaveBeenCalledTimes(2);
+      expect(mockMakeHttpRequest).toHaveBeenCalledTimes(2);
     });
 
     it('should deduplicate requests with different durations but same content', async () => {
@@ -224,7 +233,7 @@ describe('Progress Service', () => {
       await progressService.sendUpdate(params1);
       await progressService.sendUpdate(params2);
 
-      expect(global.fetch).toHaveBeenCalledTimes(1);
+      expect(mockMakeHttpRequest).toHaveBeenCalledTimes(1);
       expect(params1.flowExecutorContext.toResponse).toHaveBeenCalledTimes(1);
       expect(params2.flowExecutorContext.toResponse).toHaveBeenCalledTimes(1);
     });
@@ -265,47 +274,11 @@ describe('Progress Service', () => {
       await progressService.sendUpdate(params1);
       await progressService.sendUpdate(params2);
 
-      expect(global.fetch).toHaveBeenCalledTimes(2);
+      expect(mockMakeHttpRequest).toHaveBeenCalledTimes(2);
       expect(params1.flowExecutorContext.toResponse).toHaveBeenCalledTimes(1);
       expect(params2.flowExecutorContext.toResponse).toHaveBeenCalledTimes(1);
     });
 
-    it('should use mutex for thread safety', async () => {
-      const concurrentParams = {
-        ...mockParams,
-        engineConstants: {
-          ...mockParams.engineConstants,
-          executionCorrelationId: 'test-correlation-id-concurrent',
-        },
-      };
-
-      // Make multiple concurrent requests
-      const promises = [
-        progressService.sendUpdate(concurrentParams),
-        progressService.sendUpdate(concurrentParams),
-        progressService.sendUpdate(concurrentParams),
-      ];
-
-      await Promise.all(promises);
-
-      // Due to mutex locking and request deduplication, should only make one request
-      expect(global.fetch).toHaveBeenCalledTimes(1);
-    });
-
-    it('should handle fetch errors gracefully', async () => {
-      const errorParams = {
-        ...mockParams,
-        engineConstants: {
-          ...mockParams.engineConstants,
-          executionCorrelationId: 'test-correlation-id-error',
-        },
-      };
-
-      (global.fetch as jest.Mock).mockRejectedValue(new Error('Network error'));
-
-      await expect(progressService.sendUpdate(errorParams)).rejects.toThrow('Network error');
-    });
-
     it('should construct correct URL', async () => {
       const paramsWithDifferentUrl = {
         ...mockParams,
@@ -318,13 +291,15 @@ describe('Progress Service', () => {
 
       await progressService.sendUpdate(paramsWithDifferentUrl);
 
-      expect(global.fetch).toHaveBeenCalledWith(
+      expect(mockMakeHttpRequest).toHaveBeenCalledWith(
+        'POST',
         'https://api.example.com/v1/engine/update-run',
+        expect.any(Object),
+        expect.any(Object),
         expect.any(Object)
       );
     });
 
-
     it('should throw error when execution time is exceeded', async () => {
       const timeoutParams = {
         ...mockParams,
@@ -341,8 +316,39 @@ describe('Progress Service', () => {
 
       await expect(progressService.sendUpdate(timeoutParams)).rejects.toThrow('Execution time exceeded');
       expect(mockThrowIfExecutionTimeExceeded).toHaveBeenCalledTimes(1);
-      expect(global.fetch).not.toHaveBeenCalled();
+      expect(mockMakeHttpRequest).not.toHaveBeenCalled();
       expect(timeoutParams.flowExecutorContext.toResponse).not.toHaveBeenCalled();
     });
+
+    it('should use correct retry configuration', async () => {
+      const retryParams = {
+        ...mockParams,
+        engineConstants: {
+          ...mockParams.engineConstants,
+          executionCorrelationId: 'test-correlation-id-retry',
+        },
+      };
+
+      await progressService.sendUpdate(retryParams);
+
+      expect(mockMakeHttpRequest).toHaveBeenCalledWith(
+        'POST',
+        'http://localhost:3000/v1/engine/update-run',
+        expect.any(Object),
+        expect.any(Object),
+        expect.objectContaining({
+          retries: 3,
+          retryDelay: expect.any(Function),
+        })
+      );
+
+      // 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/workers/engine-controller.ts b/packages/server/api/src/app/workers/engine-controller.ts
@@ -4,6 +4,7 @@ import {
 } from '@fastify/type-provider-typebox';
 import {
   GetRunForWorkerRequest,
+  logger,
   SharedSystemProp,
   system,
   UpdateFailureCountRequest,
@@ -125,6 +126,10 @@ export const flowEngineWorker: FastifyPluginAsyncTypebox = async (app) => {
       );
     }
 
+    logger.debug(
+      `Updating run ${runId} to ${getTerminalStatus(runDetails.status)}`,
+    );
+
     const populatedRun = await flowRunService.updateStatus({
       flowRunId: runId,
       status: getTerminalStatus(runDetails.status),
