Handle internal errors in trigger payload extraction (#2124)

Fixes OPS-3864

---------

Co-authored-by: Marcelo Gonçalves <marcelo@openops.com>

Author: Roman Snapko

packages/engine/src/lib/handler/context/engine-constants.ts

@@ -14,6 +14,7 @@ import {
   TestRunLimitSettings,
   TriggerHookType,
 } from '@openops/shared';
+import { parseJsonResponse } from '../../helper/response-helper';
 import {
   createPropsResolver,
   PropsResolver,
@@ -232,7 +233,7 @@ export class EngineConstants {
       },
     });
 
-    return (await response.json()) as Project;
+    return parseJsonResponse<Project>(response);
   }
 }
 

packages/engine/src/lib/helper/execution-errors.ts

@@ -100,6 +100,17 @@ export class FetchError extends ExecutionError {
   }
 }
 
+export class InfrastructureError extends ExecutionError {
+  constructor(message: string, cause?: unknown) {
+    super(
+      'InfrastructureError',
+      formatMessage(message),
+      ExecutionErrorType.ENGINE,
+      cause,
+    );
+  }
+}
+
 export class ExecutionLimitReachedError extends ExecutionError {
   formated: boolean;
 

packages/engine/src/lib/helper/response-helper.ts

@@ -0,0 +1,31 @@
+import { logger } from '@openops/server-shared';
+import { InfrastructureError } from './execution-errors';
+
+export const parseJsonResponse = async <T>(response: Response): Promise<T> => {
+  const contentType = response.headers.get('content-type');
+  const text = await response.text();
+
+  if (contentType && !contentType.includes('application/json')) {
+    logger.warn('Expected JSON response but received non-JSON content type', {
+      status: response.status,
+      contentType,
+      body: text,
+    });
+    throw new InfrastructureError(
+      `Expected JSON response, but received status ${response.status} and ${contentType}.`,
+    );
+  }
+
+  try {
+    return JSON.parse(text) as T;
+  } catch (e) {
+    logger.warn('Failed to parse JSON response', {
+      status: response.status,
+      body: text,
+    });
+    throw new InfrastructureError(
+      `Failed to parse JSON response with status ${response.status}.`,
+      e,
+    );
+  }
+};

packages/engine/src/lib/operations.ts

@@ -34,7 +34,10 @@ import {
 import { testExecutionContext } from './handler/context/test-execution-context';
 import { flowExecutor } from './handler/flow-executor';
 import { blockHelper } from './helper/block-helper';
-import { ExecutionLimitReachedError } from './helper/execution-errors';
+import {
+  ExecutionLimitReachedError,
+  InfrastructureError,
+} from './helper/execution-errors';
 import { triggerHelper } from './helper/trigger-helper';
 import { resolveVariable } from './resolve-variable';
 import { progressService } from './services/progress.service';
@@ -330,6 +333,10 @@ function evaluateError(error: Error): {
     message = error.getMessage();
   }
 
+  if (error instanceof InfrastructureError) {
+    status = FlowRunStatus.INFRASTRUCTURE_ERROR;
+  }
+
   return {
     status,
     message,

packages/engine/src/lib/services/connections.service.ts

@@ -11,9 +11,9 @@ import {
   ConnectionExpiredError,
   ConnectionLoadingError,
   ConnectionNotFoundError,
-  ExecutionError,
   FetchError,
 } from '../helper/execution-errors';
+import { parseJsonResponse } from '../helper/response-helper';
 
 export const createConnectionService = ({
   projectId,
@@ -40,16 +40,13 @@ export const createConnectionService = ({
             httpStatus: response.status,
           });
         }
-        const connection: AppConnection = await response.json();
+        const connection: AppConnection =
+          await parseJsonResponse<AppConnection>(response);
         if (connection.status === AppConnectionStatus.ERROR) {
           throw new ConnectionExpiredError(connectionName);
         }
         return getConnectionValue(connection);
       } catch (e) {
-        if (e instanceof ExecutionError) {
-          throw e;
-        }
-
         return handleFetchError({
           url,
           cause: e,

packages/engine/src/lib/services/files.service.ts

@@ -1,6 +1,7 @@
 import { FilesService, WorkflowFile } from '@openops/blocks-framework';
 import { isNil } from '@openops/shared';
 import fs from 'fs/promises';
+import { parseJsonResponse } from '../helper/response-helper';
 
 const FILE_PREFIX_URL = 'file://';
 const MEMORY_PREFIX_URL = 'memory://';
@@ -129,7 +130,7 @@ async function writeDbFile({
     throw new Error('Failed to store entry ' + response.body);
   }
 
-  const result = await response.json();
+  const result = await parseJsonResponse<{ url: string }>(response);
   return result.url;
 }
 

packages/engine/src/lib/services/storage.service.ts

@@ -19,6 +19,7 @@ import {
   StorageError,
   StorageLimitError,
 } from '../helper/execution-errors';
+import { parseJsonResponse } from '../helper/response-helper';
 
 export const createStorageService = ({
   engineToken,
@@ -42,7 +43,7 @@ export const createStorageService = ({
           });
         }
 
-        return await response.json();
+        return await parseJsonResponse<StoreEntry>(response);
       } catch (e) {
         return handleFetchError({
           url,
@@ -78,7 +79,7 @@ export const createStorageService = ({
           });
         }
 
-        return await response.json();
+        return await parseJsonResponse<StoreEntry>(response);
       } catch (e) {
         return handleFetchError({
           url,
@@ -136,7 +137,9 @@ export const createStorageService = ({
           throw new Error(`Failed to list keys: ${response.statusText}`);
         }
 
-        const result = await response.json();
+        const result = await parseJsonResponse<{
+          entries: Array<{ key: string; value: unknown }>;
+        }>(response);
         return result.entries || [];
       } catch (e) {
         return handleFetchError({

packages/engine/test/response-helper.test.ts

@@ -0,0 +1,80 @@
+import { parseJsonResponse } from '../src/lib/helper/response-helper';
+import { logger } from '@openops/server-shared';
+import { InfrastructureError } from '../src/lib/helper/execution-errors';
+
+jest.mock('@openops/server-shared', () => ({
+  logger: {
+    warn: jest.fn(),
+  },
+}));
+
+describe('parseJsonResponse', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('should return parsed JSON for a valid JSON response', async () => {
+    const payload = { id: 1, name: 'test' };
+    const response = {
+      status: 200,
+      headers: new Headers({ 'content-type': 'application/json' }),
+      text: jest.fn().mockResolvedValue(JSON.stringify(payload)),
+    } as unknown as Response;
+
+    const result = await parseJsonResponse(response);
+    expect(result).toEqual(payload);
+    expect(logger.warn).not.toHaveBeenCalled();
+  });
+
+  it('should attempt to parse JSON when content-type header is missing', async () => {
+    const payload = { id: 2 };
+    const response = {
+      status: 200,
+      headers: new Headers(),
+      text: jest.fn().mockResolvedValue(JSON.stringify(payload)),
+    } as unknown as Response;
+
+    const result = await parseJsonResponse(response);
+    expect(result).toEqual(payload);
+    expect(logger.warn).not.toHaveBeenCalled();
+  });
+
+  it('should include full status code and full body in InfrastructureError for non-json content-type', async () => {
+    const body = '<html><body><h1>502 Bad Gateway</h1><p>Nginx Error</p></body></html>';
+    const response = {
+      status: 502,
+      headers: new Headers({ 'content-type': 'text/html' }),
+      text: jest.fn().mockResolvedValue(body),
+    } as unknown as Response;
+
+    const promise = parseJsonResponse(response);
+    await expect(promise).rejects.toThrow(InfrastructureError);
+    await expect(promise).rejects.toThrow(
+      'Expected JSON response, but received status 502 and text/html.',
+    );
+    expect(logger.warn).toHaveBeenCalledWith(
+      'Expected JSON response but received non-JSON content type',
+      { status: 502, contentType: 'text/html', body },
+
+    );
+  });
+
+  it('should include status code in InfrastructureError for invalid json', async () => {
+    const body = 'invalid json';
+    const response = {
+      status: 200,
+      headers: new Headers({ 'content-type': 'application/json' }),
+      text: jest.fn().mockResolvedValue(body),
+    } as unknown as Response;
+
+    const promise = parseJsonResponse(response);
+    await expect(promise).rejects.toThrow(InfrastructureError);
+    await expect(promise).rejects.toThrow(
+      'Failed to parse JSON response with status 200.',
+    );
+    expect(logger.warn).toHaveBeenCalledWith(
+      'Failed to parse JSON response',
+      { status: 200, body },
+    );
+  });
+});

packages/react-ui/src/app/features/flow-runs/lib/flow-run-utils.ts

@@ -83,11 +83,6 @@ export const flowRunUtils = {
           variant: 'default',
           Icon: CircleStop,
         };
-      case FlowRunStatus.FAILED:
-        return {
-          variant: 'error',
-          Icon: X,
-        };
       case FlowRunStatus.IGNORED:
         return {
           variant: 'default',
@@ -103,12 +98,10 @@ export const flowRunUtils = {
           variant: 'default',
           Icon: PauseIcon,
         };
-      case FlowRunStatus.INTERNAL_ERROR:
-        return {
-          variant: 'error',
-          Icon: X,
-        };
       case FlowRunStatus.TIMEOUT:
+      case FlowRunStatus.INTERNAL_ERROR:
+      case FlowRunStatus.INFRASTRUCTURE_ERROR:
+      case FlowRunStatus.FAILED:
         return {
           variant: 'error',
           Icon: X,

packages/server/worker/src/lib/trigger/hooks/extract-trigger-payload-hooks.ts

@@ -6,6 +6,7 @@ import {
 import {
   ApplicationError,
   ErrorCode,
+  FlowRunStatus,
   FlowVersion,
   isNil,
   ProjectId,
@@ -38,6 +39,21 @@ export async function extractPayloads(
       handleFailureFlow(flowVersion, projectId, engineToken, true);
       return result.output as unknown[];
     } else {
+      if (
+        !isNil(result) &&
+        typeof result === 'object' &&
+        'status' in result &&
+        result.status === FlowRunStatus.INFRASTRUCTURE_ERROR
+      ) {
+        logger.warn('Failed to execute trigger due to infrastructure issue', {
+          result,
+          blockName,
+          blockVersion,
+          flowId: flowVersion.flowId,
+        });
+
+        return [];
+      }
       logger.error(
         {
           result,
@@ -47,6 +63,7 @@ export async function extractPayloads(
         },
         'Failed to execute trigger',
       );
+
       const errorMessage =
         result?.message ?? 'Trigger execution failed due to an unknown issue.';