Save workflow run when trigger failed (#1577)

<!--
Ensure the title clearly reflects what was changed.
Provide a clear and concise description of the changes made. The PR
should only contain the changes related to the issue, and no other
unrelated changes.
-->

Fixes OPS-2926.

https://www.loom.com/share/ef9620a0a5404edebae9a019cff75385

---------

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

Author: Roman Snapko

packages/server/api/src/app/flows/flow-run/flow-run-service.ts

@@ -25,6 +25,8 @@ import {
   RunEnvironment,
   SeekPage,
   spreadIfDefined,
+  StepOutput,
+  StepOutputStatus,
   TERMINAL_STATUSES,
 } from '@openops/shared';
 import { nanoid } from 'nanoid';
@@ -131,6 +133,58 @@ function getEffectiveProgressUpdateType(
 }
 
 export const flowRunService = {
+  async recordTriggerFailure({
+    projectId,
+    flowVersionId,
+    errorMessage,
+    reason,
+    triggerInput,
+  }: {
+    projectId: ProjectId;
+    flowVersionId: FlowVersionId;
+    errorMessage: string;
+    reason: string;
+    triggerInput?: unknown;
+  }): Promise<void> {
+    const flowVersion = await flowVersionService.getOneOrThrow(flowVersionId);
+
+    const executionState: ExecutionState = {
+      steps: {
+        [flowVersion.trigger.name]: {
+          type: flowVersion.trigger.type,
+          status: StepOutputStatus.FAILED,
+          input: triggerInput,
+          errorMessage,
+        },
+      } as Record<string, StepOutput>,
+    };
+
+    const logsFileId = await updateLogs({
+      logsFileId: null,
+      projectId,
+      executionState,
+    });
+
+    const failedRun = {
+      id: openOpsId(),
+      projectId,
+      flowId: flowVersion.flowId,
+      flowVersionId: flowVersion.id,
+      environment: RunEnvironment.PRODUCTION,
+      flowDisplayName: flowVersion.displayName,
+      startTime: new Date().toISOString(),
+      finishTime: new Date().toISOString(),
+      status: FlowRunStatus.FAILED,
+      triggerSource: FlowRunTriggerSource.TRIGGERED,
+      terminationReason: reason,
+      tasks: 0,
+      duration: 0,
+      tags: [],
+      logsFileId,
+    };
+
+    await flowRunRepo().save(failedRun);
+  },
   async list({
     projectId,
     flowId,

packages/server/api/src/app/workers/engine-controller.ts

@@ -110,6 +110,19 @@ export const flowEngineWorker: FastifyPluginAsyncTypebox = async (app) => {
       projectId,
       success,
     });
+
+    if (success === false) {
+      const { flowVersionId, errorMessage, reason, triggerInput } =
+        request.body as Extract<UpdateFailureCountRequest, { success: false }>;
+
+      await flowRunService.recordTriggerFailure({
+        projectId,
+        flowVersionId,
+        errorMessage,
+        reason,
+        triggerInput,
+      });
+    }
   });
 
   app.post('/update-run', UpdateStepProgress, async (request) => {

packages/server/api/test/unit/flow-run/flow-run-service.test.ts

@@ -0,0 +1,161 @@
+export const mockedRepo = {
+  findOneBy: jest.fn(),
+  createQueryBuilder: jest.fn(),
+  save: jest.fn(),
+  update: jest.fn(),
+};
+
+jest.mock('../../../src/app/core/db/repo-factory', () => {
+  return {
+    repoFactory: () => jest.fn().mockReturnValue(mockedRepo),
+  };
+});
+
+const fileServiceMock = {
+  save: jest.fn(),
+};
+jest.mock('../../../src/app/file/file.service', () => {
+  return {
+    fileService: fileServiceMock,
+  };
+});
+
+const flowVersionServiceMock = {
+  getOneOrThrow: jest.fn(),
+};
+jest.mock('../../../src/app/flows/flow-version/flow-version.service', () => {
+  return {
+    flowVersionService: flowVersionServiceMock,
+  };
+});
+
+const logSerializerMock = {
+  serialize: jest.fn(),
+};
+jest.mock('../../../src/app/flows/flow-run/log-serializer', () => {
+  return {
+    logSerializer: logSerializerMock,
+  };
+});
+
+import {
+  FileCompression,
+  FileType,
+  FlowRunStatus,
+  FlowRunTriggerSource,
+  RunEnvironment,
+  StepOutputStatus,
+} from '@openops/shared';
+import { flowRunService } from '../../../src/app/flows/flow-run/flow-run-service';
+
+describe('flowRunService.recordTriggerFailure', () => {
+  const now = new Date('2024-01-02T03:04:05.000Z');
+
+  beforeAll(() => {
+    jest.useFakeTimers();
+  });
+
+  beforeEach(() => {
+    jest.setSystemTime(now);
+    jest.clearAllMocks();
+    logSerializerMock.serialize.mockResolvedValue(
+      Buffer.from('compressed-logs'),
+    );
+    fileServiceMock.save.mockResolvedValue({ id: 'file_123' });
+    flowVersionServiceMock.getOneOrThrow.mockResolvedValue({
+      id: 'fv_1',
+      flowId: 'flow_1',
+      displayName: 'My Flow V1',
+      trigger: {
+        name: 'triggerStep',
+        type: 'POLLING',
+      },
+    });
+  });
+
+  afterAll(() => {
+    jest.useRealTimers();
+  });
+
+  it('should create a failed flow run with logs when triggerInput is provided', async () => {
+    await flowRunService.recordTriggerFailure({
+      projectId: 'proj_1',
+      flowVersionId: 'fv_1',
+      errorMessage: 'Trigger failed to execute',
+      reason: 'TRIGGER_ERROR',
+      triggerInput: { a: 1 },
+    });
+
+    expect(logSerializerMock.serialize).toHaveBeenCalledTimes(1);
+    const serializeArg = (logSerializerMock.serialize as jest.Mock).mock
+      .calls[0][0];
+    expect(serializeArg).toEqual({
+      executionState: {
+        steps: {
+          triggerStep: {
+            type: 'POLLING',
+            status: StepOutputStatus.FAILED,
+            input: { a: 1 },
+            errorMessage: 'Trigger failed to execute',
+          },
+        },
+      },
+    });
+
+    expect(fileServiceMock.save).toHaveBeenCalledTimes(1);
+    const saveArg = (fileServiceMock.save as jest.Mock).mock.calls[0][0];
+    expect(saveArg).toEqual(
+      expect.objectContaining({
+        fileId: expect.any(String),
+        data: Buffer.from('compressed-logs'),
+        type: FileType.FLOW_RUN_LOG,
+        compression: FileCompression.GZIP,
+        projectId: 'proj_1',
+      }),
+    );
+
+    expect(mockedRepo.save).toHaveBeenCalledTimes(1);
+    const saved = mockedRepo.save.mock.calls[0][0];
+    expect(saved).toMatchObject({
+      projectId: 'proj_1',
+      flowId: 'flow_1',
+      flowVersionId: 'fv_1',
+      environment: RunEnvironment.PRODUCTION,
+      flowDisplayName: 'My Flow V1',
+      startTime: now.toISOString(),
+      finishTime: now.toISOString(),
+      status: FlowRunStatus.FAILED,
+      triggerSource: FlowRunTriggerSource.TRIGGERED,
+      terminationReason: 'TRIGGER_ERROR',
+      tasks: 0,
+      duration: 0,
+      tags: [],
+    });
+    expect(saved.logsFileId).toBe(saveArg.fileId);
+
+    expect(typeof saved.id).toBe('string');
+    expect(saved.id.length).toBeGreaterThan(0);
+  });
+
+  it('should create a failed flow run without triggerInput when not provided', async () => {
+    await flowRunService.recordTriggerFailure({
+      projectId: 'proj_2',
+      flowVersionId: 'fv_1',
+      errorMessage: 'Boom',
+      reason: 'TRIGGER_ERROR_NO_INPUT',
+    });
+
+    const serializeArg = (logSerializerMock.serialize as jest.Mock).mock
+      .calls[0][0];
+    expect(serializeArg.executionState.steps.triggerStep).toEqual({
+      type: 'POLLING',
+      status: StepOutputStatus.FAILED,
+      input: undefined,
+      errorMessage: 'Boom',
+    });
+
+    const saved = mockedRepo.save.mock.calls[0][0];
+    expect(saved.projectId).toBe('proj_2');
+    expect(saved.terminationReason).toBe('TRIGGER_ERROR_NO_INPUT');
+  });
+});

packages/server/shared/src/lib/job/index.ts

@@ -39,11 +39,22 @@ export const UpdateJobRequest = Type.Object({
 });
 export type UpdateJobRequest = Static<typeof UpdateJobRequest>;
 
-export const UpdateFailureCountRequest = Type.Object({
-  flowId: Type.String(),
-  projectId: Type.String(),
-  success: Type.Boolean(),
-});
+export const UpdateFailureCountRequest = Type.Union([
+  Type.Object({
+    flowId: Type.String(),
+    projectId: Type.String(),
+    success: Type.Literal(true),
+  }),
+  Type.Object({
+    flowId: Type.String(),
+    projectId: Type.String(),
+    success: Type.Literal(false),
+    flowVersionId: Type.String(),
+    reason: Type.String(),
+    errorMessage: Type.String(),
+    triggerInput: Type.Optional(Type.Unknown()),
+  }),
+]);
 
 export type UpdateFailureCountRequest = Static<
   typeof UpdateFailureCountRequest

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

@@ -1,4 +1,8 @@
-import { logger, rejectedPromiseHandler } from '@openops/server-shared';
+import {
+  logger,
+  rejectedPromiseHandler,
+  UpdateFailureCountRequest,
+} from '@openops/server-shared';
 import {
   ApplicationError,
   ErrorCode,
@@ -43,8 +47,15 @@ export async function extractPayloads(
         },
         'Failed to execute trigger',
       );
-      handleFailureFlow(flowVersion, projectId, engineToken, false);
+      const errorMessage =
+        result?.message ?? 'Trigger execution failed due to an unknown issue.';
 
+      handleFailureFlow(flowVersion, projectId, engineToken, false, {
+        reason: 'TRIGGER_HOOK_FAILED',
+        flowVersionId: flowVersion.id,
+        errorMessage,
+        triggerInput: result.input,
+      });
       return [];
     }
   } catch (e) {
@@ -61,7 +72,12 @@ export async function extractPayloads(
         },
         'Failed to execute trigger due to timeout',
       );
-      handleFailureFlow(flowVersion, projectId, engineToken, false);
+
+      handleFailureFlow(flowVersion, projectId, engineToken, false, {
+        reason: 'TRIGGER_TIMEOUT',
+        flowVersionId: flowVersion.id,
+        errorMessage: 'Trigger execution timed out',
+      });
       return [];
     }
     throw e;
@@ -73,16 +89,23 @@ function handleFailureFlow(
   projectId: ProjectId,
   engineToken: string,
   success: boolean,
+  failureDetails?: {
+    reason: string;
+    flowVersionId: string;
+    errorMessage: string;
+    triggerInput?: unknown;
+  },
 ): void {
   const engineController = engineApiService(engineToken);
 
-  rejectedPromiseHandler(
-    engineController.updateFailureCount({
-      flowId: flowVersion.flowId,
-      projectId,
-      success,
-    }),
-  );
+  const request = {
+    flowId: flowVersion.flowId,
+    projectId,
+    success,
+    ...(!success && failureDetails ? failureDetails : {}),
+  } as UpdateFailureCountRequest;
+
+  rejectedPromiseHandler(engineController.updateFailureCount(request));
 }
 
 type ExecuteTrigger = {