feat(logic-function): add LIVE / PREBUILT execution modes (#20873)

## Summary ### Why 1. Sending the code to the lambda (~1Mb usually) is heavy on network and results to a constant traffic of ~30Mb/s on AWS which results into TB of network data every month 2. eval(1MB of code) is not that fast, it's heavy on memory and CPU on lambda side ### High level Adds two execution modes for logic functions, gated behind the new `IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED` workspace feature flag (off everywhere by default): - **LIVE** (current behavior, preserved bit-for-bit): the compiled bundle is read from object storage and shipped in every Lambda invoke payload. Used for fast iteration in the workflow editor / Settings test runs. - **PREBUILT** (new): the bundle is installed onto the per-function Lambda alongside the unified executor, and invocations carry only `{ params, env, handlerName }` — saving JSON payload egress and warm-start `import()` cost on every call. ### Key design choices - **Unified Lambda handler** (`constants/executor/index.mjs`) dispatches at runtime: `event.code` present ? LIVE (write to `/tmp`, dynamic import) : `import('./prebuilt-logic-function.mjs')`. Both code paths always coexist on the deployment package, so the same Lambda can serve either mode without redeploying. - **Install runs inside the `validateBuildAndRun` migration pipeline**, not at execute time. `Create/UpdateLogicFunctionActionHandlerService` calls `driver.installPrebuiltBundle` when `executionMode` flips LIVE?PREBUILT or `checksum` changes while PREBUILT, gated on `isBuildUpToDate=true` and a fresh checksum. - **Strict execute, no reconciliation**: `LogicFunctionExecutorService.execute` resolves `effectiveExecutionMode` (caller override > feature flag > entity column). For PREBUILT it asks the driver `getInstalledBundleChecksum` (Lambda `twenty:bundle-checksum` tag for AWS, sidecar file locally) and throws `LOGIC_FUNCTION_PREBUILT_BUNDLE_NOT_INSTALLED` on mismatch. - **Feature flag gates every side effect**: with the flag off the executor forces LIVE, the action-handler install hooks bail before AWS, and workflow activation does not flip the mode. Rollback is just turning the flag off. ### Lifecycle - New workflow CODE step ? `LIVE`, no install. - Workflow activated ? build + activation flips `executionMode=PREBUILT` ? action-handler installs the bundle + sets the Lambda tag. - Draft from active version ? duplicated logic function reset to `LIVE`. - App install ? manifest converter sets `PREBUILT`, create-action handler installs. - Test runs (`executeOneFromSource`, workflow editor) pass `executionMode=LIVE` explicitly. ### Observability `[lambda-timing]` log lines now include `effectiveExecutionMode` and `payloadBytes`; the action handler logs `install_duration_ms` for each install. ## Test plan - [x] `npx nx typecheck twenty-server` ? passes - [x] `npx oxlint --type-aware` on all changed files ? 0 warnings, 0 errors - [x] `npx nx test twenty-server` ? 588 suites / 5009 tests pass (no regressions vs main) - [x] New unit suite `flat-logic-function-validator.service.spec.ts` ? 9/9 - [x] Existing `workflow-version-step-operations.workspace-service.spec.ts` ? 8/8 (verified the new token-based DI avoids a circular-import regression) - [x] Snapshot for `ALL_UNIVERSAL_FLAT_ENTITY_PROPERTIES_TO_COMPARE_AND_STRINGIFY` updated to include `executionMode` - [x] Integration suite `logic-function-execution.integration-spec.ts` extended to assert `executionMode=LIVE` on newly-created functions and continues to exercise the LIVE happy path - [ ] Manual staging rollout: flip `IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED` per workspace, observe `[lambda-timing]` `payloadBytes` drop + `install_duration_ms`, then ramp in prod.
2026-06-02 13:14:39 +02:00
parent 1ae00d6753
commit 58907b733c
44 changed files with 2212 additions and 951 deletions
@@ -1,6 +1,7 @@
 import { Module } from '@nestjs/common';

 import { ApplicationModule } from 'src/engine/core-modules/application/application.module';
+import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { WorkspaceManyOrAllFlatEntityMapsCacheModule } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.module';
 import { LogicFunctionModule } from 'src/engine/metadata-modules/logic-function/logic-function.module';

@@ -11,6 +12,7 @@ import { CodeStepBuildService } from './services/code-step-build.service';
    ApplicationModule,
    WorkspaceManyOrAllFlatEntityMapsCacheModule,
    LogicFunctionModule,
+    FeatureFlagModule,
  ],
  providers: [CodeStepBuildService],
  exports: [CodeStepBuildService],
@@ -1,21 +1,31 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';

 import { SEED_WORKFLOW_ACTION_TRIGGER_SETTINGS } from 'twenty-shared/logic-function';
+import { FeatureFlagKey } from 'twenty-shared/types';
 import { isDefined } from 'twenty-shared/utils';

+import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
 import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
 import { findFlatEntityByIdInFlatEntityMaps } from 'src/engine/metadata-modules/flat-entity/utils/find-flat-entity-by-id-in-flat-entity-maps.util';
+import { LogicFunctionExecutionMode } from 'src/engine/metadata-modules/logic-function/logic-function.entity';
+import { LogicFunctionFromSourceHelperService } from 'src/engine/metadata-modules/logic-function/services/logic-function-from-source-helper.service';
 import { LogicFunctionFromSourceService } from 'src/engine/metadata-modules/logic-function/services/logic-function-from-source.service';
+import { extractCodeStepLogicFunctionIdsFromWorkflowSteps } from 'src/modules/workflow/workflow-builder/workflow-version-step/code-step/utils/extract-code-step-logic-function-ids-from-workflow-steps.util';
+import { type WorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
 import {
-  WorkflowActionType,
-  type WorkflowAction,
-} from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
+  WorkflowTriggerException,
+  WorkflowTriggerExceptionCode,
+} from 'src/modules/workflow/workflow-trigger/exceptions/workflow-trigger.exception';

@Injectable()
 export class CodeStepBuildService {
+  private readonly logger = new Logger(CodeStepBuildService.name);
+
  constructor(
    private readonly workspaceManyOrAllFlatEntityMapsCacheService: WorkspaceManyOrAllFlatEntityMapsCacheService,
    private readonly logicFunctionFromSourceService: LogicFunctionFromSourceService,
+    private readonly logicFunctionFromSourceHelperService: LogicFunctionFromSourceHelperService,
+    private readonly featureFlagService: FeatureFlagService,
  ) {}

  async createCodeStepLogicFunction({
@@ -56,21 +66,10 @@ export class CodeStepBuildService {
    workspaceId: string;
    steps: WorkflowAction[];
  }): Promise<void> {
-    const codeSteps = steps.filter(
-      (
-        step,
-      ): step is WorkflowAction & {
-        type: typeof WorkflowActionType.CODE;
-        settings: { input: { logicFunctionId: string } };
-      } =>
-        step.type === WorkflowActionType.CODE &&
-        isDefined(
-          (step.settings?.input as { logicFunctionId?: string })
-            ?.logicFunctionId,
-        ),
-    );
+    const logicFunctionIds =
+      extractCodeStepLogicFunctionIdsFromWorkflowSteps(steps);

-    if (codeSteps.length === 0) {
+    if (logicFunctionIds.length === 0) {
      return;
    }

@@ -82,8 +81,7 @@ export class CodeStepBuildService {
        },
      );

-    for (const step of codeSteps) {
-      const logicFunctionId = step.settings.input.logicFunctionId;
+    for (const logicFunctionId of logicFunctionIds) {
      const flatLogicFunction = findFlatEntityByIdInFlatEntityMaps({
        flatEntityId: logicFunctionId,
        flatEntityMaps: flatLogicFunctionMaps,
@@ -105,6 +103,11 @@ export class CodeStepBuildService {
        : undefined;

      if (!isDefined(applicationUniversalIdentifier)) {
+        this.logger.warn(
+          `Skipping build for logic function '${logicFunctionId}' (workspace=${workspaceId}): ` +
+            `applicationId=${flatLogicFunction.applicationId ?? 'null'} did not resolve to an application. ` +
+            `The function will not be rebuilt and may stay out of date.`,
+        );
        continue;
      }

@@ -114,4 +117,94 @@ export class CodeStepBuildService {
      });
    }
  }
+
+  async switchCodeStepLogicFunctionsToPrebuilt({
+    workspaceId,
+    steps,
+  }: {
+    workspaceId: string;
+    steps: WorkflowAction[];
+  }): Promise<void> {
+    const isPrebuiltModeEnabled =
+      await this.featureFlagService.isFeatureEnabled(
+        FeatureFlagKey.IS_LOGIC_FUNCTION_PREBUILT_MODE_ENABLED,
+        workspaceId,
+      );
+
+    if (!isPrebuiltModeEnabled) {
+      return;
+    }
+
+    const logicFunctionIds =
+      extractCodeStepLogicFunctionIdsFromWorkflowSteps(steps);
+
+    if (logicFunctionIds.length === 0) {
+      return;
+    }
+
+    const { flatLogicFunctionMaps, flatApplicationMaps } =
+      await this.workspaceManyOrAllFlatEntityMapsCacheService.getOrRecomputeManyOrAllFlatEntityMaps(
+        {
+          workspaceId,
+          flatMapsKeys: ['flatLogicFunctionMaps', 'flatApplicationMaps'],
+        },
+      );
+
+    for (const logicFunctionId of logicFunctionIds) {
+      const flatLogicFunction = findFlatEntityByIdInFlatEntityMaps({
+        flatEntityId: logicFunctionId,
+        flatEntityMaps: flatLogicFunctionMaps,
+      });
+
+      if (
+        !isDefined(flatLogicFunction) ||
+        isDefined(flatLogicFunction.deletedAt)
+      ) {
+        throw new WorkflowTriggerException(
+          `CODE step references logic function '${logicFunctionId}' that is missing or deleted`,
+          WorkflowTriggerExceptionCode.INVALID_WORKFLOW_VERSION,
+        );
+      }
+
+      if (
+        !flatLogicFunction.isBuildUpToDate ||
+        !isDefined(flatLogicFunction.checksum)
+      ) {
+        throw new WorkflowTriggerException(
+          `Logic function '${logicFunctionId}' has no fresh build or checksum after the build step`,
+          WorkflowTriggerExceptionCode.INTERNAL_ERROR,
+        );
+      }
+
+      if (
+        flatLogicFunction.executionMode === LogicFunctionExecutionMode.PREBUILT
+      ) {
+        continue;
+      }
+
+      const applicationUniversalIdentifier = isDefined(
+        flatLogicFunction.applicationId,
+      )
+        ? flatApplicationMaps.byId[flatLogicFunction.applicationId]
+            ?.universalIdentifier
+        : undefined;
+
+      if (!isDefined(applicationUniversalIdentifier)) {
+        throw new WorkflowTriggerException(
+          `Logic function '${logicFunctionId}' references applicationId='${flatLogicFunction.applicationId ?? 'null'}' that did not resolve to an application`,
+          WorkflowTriggerExceptionCode.INTERNAL_ERROR,
+        );
+      }
+
+      await this.logicFunctionFromSourceHelperService.updateOneFromMetadata({
+        flatLogicFunctionToUpdate: {
+          ...flatLogicFunction,
+          executionMode: LogicFunctionExecutionMode.PREBUILT,
+          updatedAt: new Date().toISOString(),
+        },
+        workspaceId,
+        applicationUniversalIdentifier,
+      });
+    }
+  }
 }
@@ -0,0 +1,69 @@
+import { WorkflowVersionStepException } from 'src/modules/workflow/common/exceptions/workflow-version-step.exception';
+import {
+  type CodeStepForLogicFunctionIdExtraction,
+  extractCodeStepLogicFunctionIdsFromWorkflowSteps,
+  type NonCodeStepForLogicFunctionIdExtraction,
+} from 'src/modules/workflow/workflow-builder/workflow-version-step/code-step/utils/extract-code-step-logic-function-ids-from-workflow-steps.util';
+import { WorkflowActionType } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
+
+const buildCodeStep = (
+  logicFunctionId: string,
+): CodeStepForLogicFunctionIdExtraction => ({
+  id: 'step-id',
+  type: WorkflowActionType.CODE,
+  settings: {
+    input: {
+      logicFunctionId,
+    },
+  },
+});
+
+const buildNonCodeStep = (): NonCodeStepForLogicFunctionIdExtraction => ({
+  id: 'send-email-id',
+  type: WorkflowActionType.SEND_EMAIL,
+});
+
+describe('extractCodeStepLogicFunctionIdsFromWorkflowSteps', () => {
+  it('should return logic function ids of CODE steps', () => {
+    const steps = [
+      buildCodeStep('logic-function-1'),
+      buildCodeStep('logic-function-2'),
+    ];
+
+    expect(extractCodeStepLogicFunctionIdsFromWorkflowSteps(steps)).toEqual([
+      'logic-function-1',
+      'logic-function-2',
+    ]);
+  });
+
+  it('should ignore non-CODE steps', () => {
+    const steps = [
+      buildCodeStep('logic-function-1'),
+      buildNonCodeStep(),
+      buildCodeStep('logic-function-2'),
+    ];
+
+    expect(extractCodeStepLogicFunctionIdsFromWorkflowSteps(steps)).toEqual([
+      'logic-function-1',
+      'logic-function-2',
+    ]);
+  });
+
+  it('should throw when a CODE step has an empty logic function id', () => {
+    const steps = [buildCodeStep(''), buildCodeStep('logic-function-1')];
+
+    expect(() =>
+      extractCodeStepLogicFunctionIdsFromWorkflowSteps(steps),
+    ).toThrow(WorkflowVersionStepException);
+  });
+
+  it('should return an empty array when there are no CODE steps', () => {
+    expect(
+      extractCodeStepLogicFunctionIdsFromWorkflowSteps([buildNonCodeStep()]),
+    ).toEqual([]);
+  });
+
+  it('should return an empty array for an empty list of steps', () => {
+    expect(extractCodeStepLogicFunctionIdsFromWorkflowSteps([])).toEqual([]);
+  });
+});
@@ -0,0 +1,54 @@
+import { isNonEmptyString } from '@sniptt/guards';
+
+import {
+  WorkflowVersionStepException,
+  WorkflowVersionStepExceptionCode,
+} from 'src/modules/workflow/common/exceptions/workflow-version-step.exception';
+import { type WorkflowCodeActionInput } from 'src/modules/workflow/workflow-executor/workflow-actions/code/types/workflow-code-action-input.type';
+import {
+  type WorkflowAction,
+  type WorkflowCodeAction,
+  WorkflowActionType,
+} from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
+
+export type CodeStepForLogicFunctionIdExtraction = Pick<
+  WorkflowCodeAction,
+  'id' | 'type'
+> & {
+  settings: {
+    input: Pick<WorkflowCodeActionInput, 'logicFunctionId'>;
+  };
+};
+
+export type NonCodeStepForLogicFunctionIdExtraction = Pick<
+  Exclude<WorkflowAction, WorkflowCodeAction>,
+  'id' | 'type'
+>;
+
+type WorkflowStepForLogicFunctionIdExtraction =
+  | CodeStepForLogicFunctionIdExtraction
+  | NonCodeStepForLogicFunctionIdExtraction;
+
+const isCodeStepForLogicFunction = (
+  step: WorkflowStepForLogicFunctionIdExtraction,
+): step is CodeStepForLogicFunctionIdExtraction =>
+  step.type === WorkflowActionType.CODE;
+
+export const extractCodeStepLogicFunctionIdsFromWorkflowSteps = (
+  steps: WorkflowStepForLogicFunctionIdExtraction[],
+): string[] => [
+  ...new Set(
+    steps.filter(isCodeStepForLogicFunction).map((step) => {
+      const logicFunctionId = step.settings.input.logicFunctionId;
+
+      if (!isNonEmptyString(logicFunctionId)) {
+        throw new WorkflowVersionStepException(
+          `CODE step '${step.id}' has no logic function id`,
+          WorkflowVersionStepExceptionCode.INVALID_REQUEST,
+        );
+      }
+
+      return logicFunctionId;
+    }),
+  ),
+];
@@ -136,6 +136,11 @@ export class WorkflowTriggerWorkspaceService {
          steps: workflowVersion.steps ?? [],
        });

+        await this.codeStepBuildService.switchCodeStepLogicFunctionsToPrebuilt({
+          workspaceId,
+          steps: workflowVersion.steps ?? [],
+        });
+
        await this.performActivationSteps(
          workflow,
          workflowVersion,