Add a lock on function creation (#19428)

Fixes
https://twenty-v7.sentry.io/issues/7368127776/?environment=prod&project=4507072499810304&query=is%3Aunresolved&referrer=issue-stream

- Add a distributed lock (via CacheLockService) around the Lambda
executor build in LambdaDriver to prevent concurrent workflow runs from
racing on CreateFunctionCommand for the same logic function
- Use double-checked locking: isAlreadyBuilt is checked lock-free first
(fast path), then re-checked inside the lock to avoid redundant rebuilds

packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda.driver.ts

@@ -1,6 +1,6 @@
 import { createHash } from 'crypto';
 import * as fs from 'fs/promises';
-import { resolve, join } from 'path';
+import { join, resolve } from 'path';
 
 import {
   CreateFunctionCommand,
@@ -20,8 +20,8 @@ import {
   waitUntilFunctionActiveV2,
 } from '@aws-sdk/client-lambda';
 import { PutObjectCommand, S3Client } from '@aws-sdk/client-s3';
-import { getSignedUrl } from '@aws-sdk/s3-request-presigner';
 import { AssumeRoleCommand, STSClient } from '@aws-sdk/client-sts';
+import { getSignedUrl } from '@aws-sdk/s3-request-presigner';
 import { isDefined } from 'twenty-shared/utils';
 
 import {
@@ -34,7 +34,9 @@ import {
 
 import { ASSET_PATH } from 'src/constants/assets-path';
 import { type FlatApplication } from 'src/engine/core-modules/application/types/flat-application.type';
+import { type CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service';
 import { COMMON_LAYER_DEPENDENCIES_DIRNAME } from 'src/engine/core-modules/logic-function/logic-function-drivers/constants/common-layer-dependencies-dirname';
+import { callWithTimeout } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/call-with-timeout';
 import { copyBuilder } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-builder';
 import { copyCommonLayerDependencies } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-common-layer-dependencies';
 import { copyExecutor } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/copy-executor';
@@ -43,7 +45,6 @@ import { createZipFile } from 'src/engine/core-modules/logic-function/logic-func
 import { TemporaryDirManager } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/temporary-dir-manager';
 import { type LogicFunctionResourceService } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.service';
 import { type SdkClientArchiveService } from 'src/engine/core-modules/sdk-client/sdk-client-archive.service';
-import { callWithTimeout } from 'src/engine/core-modules/logic-function/logic-function-drivers/utils/call-with-timeout';
 import { LogicFunctionExecutionStatus } from 'src/engine/metadata-modules/logic-function/dtos/logic-function-execution-result.dto';
 import { LogicFunctionRuntime } from 'src/engine/metadata-modules/logic-function/logic-function.entity';
 import {
@@ -60,7 +61,6 @@ const COMMON_LAYER_NAME_PREFIX = 'twenty-common-layer';
 const BUILDER_LAMBDA_TIMEOUT_SECONDS = 60;
 const BUILDER_LAMBDA_MEMORY_MB = 512;
 const LAMBDA_EPHEMERAL_STORAGE_MB = 2048;
-
 const YARN_INSTALL_HANDLER_PATH = resolve(
   __dirname,
   join(
@@ -109,6 +109,7 @@ export type BuilderLambdaResult = {
 export interface LambdaDriverOptions extends LambdaClientConfig {
   logicFunctionResourceService: LogicFunctionResourceService;
   sdkClientArchiveService: SdkClientArchiveService;
+  cacheLockService: CacheLockService;
   region: string;
   lambdaRole: string;
   subhostingRole?: string;
@@ -125,12 +126,14 @@ export class LambdaDriver implements LogicFunctionDriver {
   private readonly options: LambdaDriverOptions;
   private readonly logicFunctionResourceService: LogicFunctionResourceService;
   private readonly sdkClientArchiveService: SdkClientArchiveService;
+  private readonly cacheLockService: CacheLockService;
 
   constructor(options: LambdaDriverOptions) {
     this.options = options;
     this.lambdaClient = undefined;
     this.logicFunctionResourceService = options.logicFunctionResourceService;
     this.sdkClientArchiveService = options.sdkClientArchiveService;
+    this.cacheLockService = options.cacheLockService;
   }
 
   private areAssumeRoleCredentialsExpired(): boolean {
@@ -892,17 +895,66 @@ export class LambdaDriver implements LogicFunctionDriver {
     flatApplication: FlatApplication;
     applicationUniversalIdentifier: string;
   }) {
-    if (
+    const buildArgs = {
+      flatLogicFunction,
+      flatApplication,
+      applicationUniversalIdentifier,
+    };
+
+    if (await this.canSkipBuild(buildArgs)) {
+      return;
+    }
+
+    const buildLockTtlMs = 120_000;
+    const buildLockRetryMs = 500;
+    const buildLockMaxRetries = 240;
+
+    await this.cacheLockService.withLock(
+      async () => {
+        // Need to check again inside the lock in case lock was not acquired immediately.
+        if (await this.canSkipBuild(buildArgs)) {
+          return;
+        }
+
+        await this.createLambdaExecutor(buildArgs);
+      },
+      `lambda-build:${flatLogicFunction.id}`,
+      {
+        ttl: buildLockTtlMs,
+        ms: buildLockRetryMs,
+        maxRetries: buildLockMaxRetries,
+      },
+    );
+  }
+
+  private async canSkipBuild({
+    flatLogicFunction,
+    flatApplication,
+    applicationUniversalIdentifier,
+  }: {
+    flatLogicFunction: FlatLogicFunction;
+    flatApplication: FlatApplication;
+    applicationUniversalIdentifier: string;
+  }) {
+    return (
       !flatApplication.isSdkLayerStale &&
       (await this.isAlreadyBuilt({
         flatLogicFunction,
         flatApplication,
         applicationUniversalIdentifier,
       }))
-    ) {
-      return;
-    }
+    );
+  }
 
+  private async createLambdaExecutor({
+    flatLogicFunction,
+    flatApplication,
+    applicationUniversalIdentifier,
+  }: {
+    flatLogicFunction: FlatLogicFunction;
+    flatApplication: FlatApplication;
+    applicationUniversalIdentifier: string;
+  }) {
     await this.delete(flatLogicFunction);
 
     const depsLayerArn = await this.getLayerArn({

packages/twenty-server/src/engine/core-modules/logic-function/logic-function-drivers/logic-function-driver.factory.ts

@@ -7,6 +7,7 @@ import {
   LogicFunctionDriverType,
 } from 'src/engine/core-modules/logic-function/logic-function-drivers/interfaces/logic-function-driver.interface';
 
+import { CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service';
 import { DisabledDriver } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/disabled.driver';
 import { LambdaDriver } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/lambda.driver';
 import { LocalDriver } from 'src/engine/core-modules/logic-function/logic-function-drivers/drivers/local.driver';
@@ -22,6 +23,7 @@ export class LogicFunctionDriverFactory extends DriverFactoryBase<LogicFunctionD
     twentyConfigService: TwentyConfigService,
     private readonly logicFunctionResourceService: LogicFunctionResourceService,
     private readonly sdkClientArchiveService: SdkClientArchiveService,
+    private readonly cacheLockService: CacheLockService,
   ) {
     super(twentyConfigService);
   }
@@ -77,6 +79,7 @@ export class LogicFunctionDriverFactory extends DriverFactoryBase<LogicFunctionD
 
         return new LambdaDriver({
           logicFunctionResourceService: this.logicFunctionResourceService,
+          cacheLockService: this.cacheLockService,
           credentials: accessKeyId
             ? { accessKeyId, secretAccessKey }
             : fromNodeProviderChain({ clientConfig: { region } }),

packages/twenty-server/src/engine/core-modules/logic-function/logic-function.module.ts

@@ -1,5 +1,6 @@
 import { type DynamicModule, Global, Module } from '@nestjs/common';
 
+import { CacheLockModule } from 'src/engine/core-modules/cache-lock/cache-lock.module';
 import { LogicFunctionDriverFactory } from 'src/engine/core-modules/logic-function/logic-function-drivers/logic-function-driver.factory';
 import { LogicFunctionResourceModule } from 'src/engine/core-modules/logic-function/logic-function-resource/logic-function-resource.module';
 import { LogicFunctionTriggerModule } from 'src/engine/core-modules/logic-function/logic-function-trigger/logic-function-trigger.module';
@@ -15,6 +16,7 @@ export class LogicFunctionModule {
       module: LogicFunctionModule,
       imports: [
         TwentyConfigModule,
+        CacheLockModule,
         LogicFunctionResourceModule,
         LogicFunctionTriggerModule,
         LogicFunctionExecutorModule,