22 branches 2 (#13051)

This PR is purely technical, it does produces any functional change to
the user

- add Lock mecanism to run steps concurrently
- update `workflow-executor.workspace-service.ts` to handle multi branch
workflow execution
  - stop passing `context` through steps, it causes race condition issue
  - refactor a little bit
- simplify `workflow-run.workspace-service.ts` to prepare `output` and
`context` removal
- move workflowRun status computing from `run-workflow.job.ts` to
`workflow-executor.workspace-service.ts`

## NOTA BENE
When a code step depends of 2 parents like in this config (see image
below)

If the form is submitted before the "Code - 2s" step succeed, the branch
merge "Form" step is launched twice.
- once because form is submission Succeed resumes the workflow in an
asynchronous job
- the second time is when the asynchronous job is launched when "Code -
2s" is succeeded
- the merge "Form" step makes the workflow waiting for response to
trigger the resume in another job
- during that time, the first resume job is launched, running the merge
"Form" step again

This issue only occurs with branch workflows. It will be solved by
checking if the currentStepToExecute is already in a SUCCESS state or
not

<img width="505" alt="image"
src="https://github.com/user-attachments/assets/b73839a1-16fe-45e1-a0d9-3efa26ab4f8b"
/>

packages/twenty-server/src/engine/core-modules/cache-lock/__tests__/cache-lock.service.spec.ts

@@ -0,0 +1,108 @@
+import { Test, TestingModule } from '@nestjs/testing';
+
+import { CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service';
+import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
+import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
+
+describe('CacheLockService', () => {
+  let service: CacheLockService;
+  let cacheStorageService: jest.Mocked<CacheStorageService>;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        CacheLockService,
+        {
+          provide: CacheStorageNamespace.EngineLock,
+          useValue: {
+            acquireLock: jest.fn(),
+            releaseLock: jest.fn(),
+          },
+        },
+        {
+          provide: CacheStorageService,
+          useValue: {
+            acquireLock: jest.fn(),
+            releaseLock: jest.fn(),
+          },
+        },
+      ],
+    }).compile();
+
+    service = module.get<CacheLockService>(CacheLockService);
+    cacheStorageService = module.get(CacheStorageNamespace.EngineLock);
+    jest.spyOn(service, 'delay').mockResolvedValue(undefined);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  it('should acquire the lock and execute the function', async () => {
+    cacheStorageService.acquireLock.mockResolvedValue(true);
+    cacheStorageService.releaseLock.mockResolvedValue(undefined);
+
+    const fn = jest.fn().mockResolvedValue('success');
+
+    const ttl = 100;
+
+    const result = await service.withLock(fn, 'key', {
+      ttl,
+    });
+
+    expect(result).toBe('success');
+    expect(fn).toHaveBeenCalled();
+    expect(cacheStorageService.acquireLock).toHaveBeenCalledTimes(1);
+    expect(cacheStorageService.acquireLock).toHaveBeenCalledWith('key', ttl);
+    expect(cacheStorageService.releaseLock).toHaveBeenCalledTimes(1);
+    expect(cacheStorageService.releaseLock).toHaveBeenCalledWith('key');
+  });
+
+  it('should throw an error if lock cannot be acquired after max retries', async () => {
+    cacheStorageService.acquireLock.mockResolvedValue(false);
+
+    const fn = jest.fn();
+    const ms = 1;
+    const maxRetries = 3;
+
+    await expect(
+      service.withLock(fn, 'key', { ms, maxRetries }),
+    ).rejects.toThrow('Failed to acquire lock for key: key');
+
+    expect(cacheStorageService.acquireLock).toHaveBeenCalledTimes(maxRetries);
+    expect(fn).not.toHaveBeenCalled();
+  });
+
+  it('should retry before acquiring the lock', async () => {
+    const mockAcquireLock = cacheStorageService.acquireLock;
+
+    mockAcquireLock
+      .mockResolvedValueOnce(false)
+      .mockResolvedValueOnce(false)
+      .mockResolvedValueOnce(true);
+
+    const fn = jest.fn().mockResolvedValue('retried success');
+
+    const result = await service.withLock(fn, 'key', {
+      maxRetries: 5,
+      ms: 1,
+    });
+
+    expect(result).toBe('retried success');
+    expect(fn).toHaveBeenCalledTimes(1);
+    expect(mockAcquireLock).toHaveBeenCalledTimes(3);
+    expect(cacheStorageService.releaseLock).toHaveBeenCalledWith('key');
+  });
+
+  it('should release the lock even if the function throws', async () => {
+    cacheStorageService.acquireLock.mockResolvedValue(true);
+    cacheStorageService.releaseLock.mockResolvedValue(undefined);
+
+    const fn = jest.fn().mockRejectedValue(new Error('fail'));
+
+    await expect(service.withLock(fn, 'key')).rejects.toThrow('fail');
+
+    expect(fn).toHaveBeenCalled();
+    expect(cacheStorageService.releaseLock).toHaveBeenCalledWith('key');
+  });
+});

packages/twenty-server/src/engine/core-modules/cache-lock/cache-lock.module.ts

@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+
+import { CacheLockService } from 'src/engine/core-modules/cache-lock/cache-lock.service';
+
+@Module({
+  imports: [],
+  providers: [CacheLockService],
+  exports: [CacheLockService],
+})
+export class CacheLockModule {}

packages/twenty-server/src/engine/core-modules/cache-lock/cache-lock.service.ts

@@ -0,0 +1,47 @@
+import { Injectable } from '@nestjs/common';
+
+import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator';
+import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
+import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
+
+export type CacheLockOptions = {
+  ms?: number;
+  maxRetries?: number;
+  ttl?: number;
+};
+
+@Injectable()
+export class CacheLockService {
+  constructor(
+    @InjectCacheStorage(CacheStorageNamespace.EngineLock)
+    private readonly cacheStorageService: CacheStorageService,
+  ) {}
+
+  async delay(ms: number) {
+    return new Promise((res) => setTimeout(res, ms));
+  }
+
+  async withLock<T>(
+    fn: () => Promise<T>,
+    key: string,
+    options?: CacheLockOptions,
+  ): Promise<T> {
+    const { ms = 50, maxRetries = 20, ttl = 500 } = options || {};
+
+    for (let attempt = 0; attempt < maxRetries; attempt++) {
+      const acquired = await this.cacheStorageService.acquireLock(key, ttl);
+
+      if (acquired) {
+        try {
+          return await fn();
+        } finally {
+          await this.cacheStorageService.releaseLock(key);
+        }
+      }
+
+      await this.delay(ms);
+    }
+
+    throw new Error(`Failed to acquire lock for key: ${key}`);
+  }
+}

packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts

@@ -122,6 +122,29 @@ export class CacheStorageService {
     } while (cursor !== 0);
   }
 
+  async acquireLock(key: string, ttl = 1000): Promise<boolean> {
+    if (!this.isRedisCache()) {
+      throw new Error('acquireLock is only supported with Redis cache');
+    }
+
+    const redisClient = (this.cache as RedisCache).store.client;
+
+    const result = await redisClient.set(this.getKey(key), 'lock', {
+      NX: true,
+      PX: ttl,
+    });
+
+    return result === 'OK';
+  }
+
+  async releaseLock(key: string): Promise<void> {
+    if (!this.isRedisCache()) {
+      throw new Error('releaseLock is only supported with Redis cache');
+    }
+
+    await this.del(key);
+  }
+
   private isRedisCache() {
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     return (this.cache.store as any)?.name === 'redis';

packages/twenty-server/src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum.ts

@@ -3,5 +3,6 @@ export enum CacheStorageNamespace {
   ModuleCalendar = 'module:calendar',
   ModuleWorkflow = 'module:workflow',
   EngineWorkspace = 'engine:workspace',
+  EngineLock = 'engine:lock',
   EngineHealth = 'engine:health',
 }