Prevent overlapping workspace cleanup executions (#23522)

## Context The suspended-workspace cleanup is a long-running scheduled job. Under database or cache pressure, BullMQ can consider an execution stalled and start a replacement on another worker while the original execution is still running. Both executions can then enumerate the same suspended workspaces and run destructive cleanup concurrently. This amplifies the initial slowdown: 1. Multiple cleanup transactions target the same workspace data. 2. Transactions wait on each other's locks. 3. Database connections remain occupied while waiting. 4. Other workers and API requests have fewer connections available. There is a second source of unnecessary lock duration in workspace deletion. The deletion transaction currently starts before field metadata is read from the workspace cache. If that lookup is slow, the transaction stays open during an unrelated cache wait. ## What changed ### Prevent overlapping scheduled cleanups - Acquire a non-blocking PostgreSQL advisory lock before listing suspended workspaces. - Skip the execution when another worker already holds the lock. - Keep the lock on one dedicated PostgreSQL session for the full callback. - Release the lock in all normal and error paths. - Discard the database connection if lock acquisition or release has an ambiguous failure, preventing a session that may still own the lock from returning to the pool. - Encapsulate this lifecycle in `PostgresAdvisoryLockService`, exported by `TypeORMModule`, so other coarse-grained jobs can reuse it without handling acquisition and release themselves. ### Shorten the workspace deletion transaction - Read field metadata and build deletion chunks before starting the transaction. - Pass the precomputed chunks into the transactional deletion loop. - Keep the existing deletion order and SQL behavior unchanged. ## Why a PostgreSQL advisory lock The lock needs to coordinate workers running in different pods. A PostgreSQL session advisory lock provides the required behavior: - It is shared across all workers using the same database. - Acquisition is non-blocking, a duplicate execution can exit immediately. - It has no TTL or renewal heartbeat that could expire during the same event-loop stall that caused BullMQ to recover the job. - PostgreSQL automatically releases it when the owning session or process disappears. This is deliberately scoped to `CleanSuspendedWorkspacesJob`. It prevents overlapping scheduled executions, but it is not an exactly-once mechanism or a global mutex around every workspace-deletion entry point. ## Expected impact - Prevent one slow cleanup execution from becoming several concurrent cleanup executions. - Reduce database lock contention and connection-pool pressure during cleanup. - Avoid holding deletion transaction locks while waiting for workspace-cache data. - Reduce cleanup-related API latency bursts without changing normal cleanup semantics. The advisory lock holds one core database connection for the duration of the scheduled cleanup. This is intentional and bounded to the single lock owner. ## Validation - Focused advisory-lock tests cover successful execution, contention, callback failure, and unsafe connection disposal when unlock fails. - Cleanup-job tests cover both the lock-owner and skipped-execution paths. - Workspace-service coverage verifies that field metadata is loaded before the deletion transaction starts. - `yarn nx typecheck twenty-server` - Oxlint, Prettier, and Oxfmt checks on the changed files
2026-07-30 14:34:27 +02:00
parent dbd2eac69c
commit e4e1d24731
7 changed files with 566 additions and 30 deletions
@@ -0,0 +1,74 @@
+import { type Repository } from 'typeorm';
+
+import { type PostgresAdvisoryLockService } from 'src/database/typeorm/postgres-advisory-lock.service';
+import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
+import { CleanSuspendedWorkspacesJob } from 'src/engine/workspace-manager/workspace-cleaner/crons/clean-suspended-workspaces.job';
+import { type CleanerWorkspaceService } from 'src/engine/workspace-manager/workspace-cleaner/services/cleaner.workspace-service';
+
+jest.mock(
+  'src/engine/workspace-manager/workspace-cleaner/services/cleaner.workspace-service',
+  () => ({
+    CleanerWorkspaceService: class {},
+  }),
+);
+
+describe('CleanSuspendedWorkspacesJob', () => {
+  const workspaceRepository = {
+    find: jest.fn(),
+  };
+  const cleanerWorkspaceService = {
+    batchWarnOrCleanSuspendedWorkspaces: jest.fn(),
+  };
+  const postgresAdvisoryLockService = {
+    tryWithLock: jest.fn(),
+  };
+
+  const createJob = () =>
+    new CleanSuspendedWorkspacesJob(
+      cleanerWorkspaceService as unknown as CleanerWorkspaceService,
+      workspaceRepository as unknown as Repository<WorkspaceEntity>,
+      postgresAdvisoryLockService as unknown as PostgresAdvisoryLockService,
+    );
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    workspaceRepository.find.mockResolvedValue([{ id: 'workspace-id' }]);
+  });
+
+  it('skips cleanup when another execution holds the lock', async () => {
+    postgresAdvisoryLockService.tryWithLock.mockResolvedValue({
+      acquired: false,
+    });
+
+    await createJob().handle();
+
+    expect(workspaceRepository.find).not.toHaveBeenCalled();
+    expect(
+      cleanerWorkspaceService.batchWarnOrCleanSuspendedWorkspaces,
+    ).not.toHaveBeenCalled();
+  });
+
+  it('cleans suspended workspaces while holding the lock', async () => {
+    postgresAdvisoryLockService.tryWithLock.mockImplementation(
+      async (_lockName, callback) => ({
+        acquired: true,
+        value: await callback(),
+      }),
+    );
+
+    await createJob().handle();
+
+    expect(workspaceRepository.find).toHaveBeenCalledWith({
+      select: ['id'],
+      where: {
+        activationStatus: 'SUSPENDED',
+      },
+      withDeleted: true,
+    });
+    expect(
+      cleanerWorkspaceService.batchWarnOrCleanSuspendedWorkspaces,
+    ).toHaveBeenCalledWith({
+      workspaceIds: ['workspace-id'],
+    });
+  });
+});
@@ -1,8 +1,10 @@
+import { Logger } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';

 import { Repository } from 'typeorm';
 import { WorkspaceActivationStatus } from 'twenty-shared/workspace';

+import { PostgresAdvisoryLockService } from 'src/database/typeorm/postgres-advisory-lock.service';
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
 import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
@@ -11,12 +13,17 @@ import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.ent
 import { cleanSuspendedWorkspaceCronPattern } from 'src/engine/workspace-manager/workspace-cleaner/crons/clean-suspended-workspaces.cron.pattern';
 import { CleanerWorkspaceService } from 'src/engine/workspace-manager/workspace-cleaner/services/cleaner.workspace-service';

+const CLEAN_SUSPENDED_WORKSPACES_LOCK_NAME = 'clean-suspended-workspaces-job';
+
@Processor(MessageQueue.cronQueue)
 export class CleanSuspendedWorkspacesJob {
+  private readonly logger = new Logger(CleanSuspendedWorkspacesJob.name);
+
  constructor(
    private readonly cleanerWorkspaceService: CleanerWorkspaceService,
    @InjectRepository(WorkspaceEntity)
    private readonly workspaceRepository: Repository<WorkspaceEntity>,
+    private readonly postgresAdvisoryLockService: PostgresAdvisoryLockService,
  ) {}

  @Process(CleanSuspendedWorkspacesJob.name)
@@ -25,16 +32,27 @@ export class CleanSuspendedWorkspacesJob {
    cleanSuspendedWorkspaceCronPattern,
  )
  async handle(): Promise<void> {
-    const suspendedWorkspaceIds = await this.workspaceRepository.find({
-      select: ['id'],
-      where: {
-        activationStatus: WorkspaceActivationStatus.SUSPENDED,
-      },
-      withDeleted: true,
-    });
+    const result = await this.postgresAdvisoryLockService.tryWithLock(
+      CLEAN_SUSPENDED_WORKSPACES_LOCK_NAME,
+      async () => {
+        const suspendedWorkspaceIds = await this.workspaceRepository.find({
+          select: ['id'],
+          where: {
+            activationStatus: WorkspaceActivationStatus.SUSPENDED,
+          },
+          withDeleted: true,
+        });

-    await this.cleanerWorkspaceService.batchWarnOrCleanSuspendedWorkspaces({
-      workspaceIds: suspendedWorkspaceIds.map((workspace) => workspace.id),
-    });
+        await this.cleanerWorkspaceService.batchWarnOrCleanSuspendedWorkspaces({
+          workspaceIds: suspendedWorkspaceIds.map((workspace) => workspace.id),
+        });
+      },
+    );
+
+    if (!result.acquired) {
+      this.logger.log(
+        'Skipping suspended workspace cleanup because another execution is running',
+      );
+    }
  }
 }