fix(server): dispatch each cron trigger exactly once (#22113)

## Problem App/logic-function crons occasionally fire **twice, ~1 minute apart**. The most visible symptom is a notification cron sending the same Discord DM (or channel post) at e.g. `17:00` and again at `17:01`. ## Root cause `CronTriggerCronJob` runs every minute (`* * * * *`) and re-dispatches any logic function whose pattern is "due" according to `shouldRunNow`: ```ts const diff = Math.abs(prevTriggerDate.getTime() - now.getTime()); return diff < rootCronIntervalMs; // 60_000 ``` The detection window (`60_000ms`) is **equal to** the 60s tick interval. So when a root tick drifts across a minute boundary (runs slightly early/late, or BullMQ fires a catch-up), two adjacent ticks can both see the *same* trigger as "within the last 60s" and each enqueue a `LogicFunctionTriggerJob`. The dispatch isn't idempotent, so the function runs twice. ## Fix Make dispatch idempotent, keyed on the trigger itself: - New `getMatchingTriggerTimestamp(pattern, now)` returns the epoch-ms of the matched trigger (stable regardless of *when* within the window the root job runs), or `null`. `shouldRunNow` now delegates to it — behaviour unchanged. - Before enqueuing, `CronTriggerCronJob` claims a `logic-function-cron:{workspace}:{function}:{triggerTs}` key in the `EngineLock` cache. A second tick that resolves to the same trigger finds the key and skips. Distinct triggers always have distinct timestamps (hence distinct keys), so a later legitimate run is never suppressed. The TTL (2 min) only needs to outlive the detection window. ## Notes - `WorkflowCronTriggerCronJob` uses the same `shouldRunNow` pattern and has the same latent double-dispatch; left out of this PR to keep it focused, but the new helper makes the same guard a small follow-up. - The cache `get`-then-`set` isn't atomic; for the observed failure mode (ticks ~1 min apart, sequential) it's reliable. A Redis `SET NX` would also close the rare concurrent-multi-instance race. ## Test plan - [x] `should-run-now.utils.spec.ts` extended: two ticks within one window resolve to the same timestamp; out-of-window and invalid patterns return `null`. All 8 pass. - [x] `oxlint --type-aware` + `oxfmt` clean on changed files.  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22113?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-06-24 19:14:19 +02:00
parent 02120aac42
commit ad61d6d8a3
9 changed files with 105 additions and 73 deletions
@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';

 import { CacheStorageModule } from 'src/engine/core-modules/cache-storage/cache-storage.module';
+import { CronModule } from 'src/engine/core-modules/cron/cron.module';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module';
 import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
@@ -14,6 +15,7 @@ import { WorkflowDatabaseEventTriggerListener } from 'src/modules/workflow/workf
  imports: [
    TypeOrmModule.forFeature([WorkspaceEntity]),
    CacheStorageModule,
+    CronModule,
    WorkflowCommonModule,
    WorkspaceDataSourceModule,
  ],
@@ -2,6 +2,7 @@ import { Test, type TestingModule } from '@nestjs/testing';
 import { getDataSourceToken, getRepositoryToken } from '@nestjs/typeorm';

 import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
+import { CronTriggerDeduplicationService } from 'src/engine/core-modules/cron/services/cron-trigger-deduplication.service';
 import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
 import { WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { WORKFLOW_CRON_TRIGGER_CACHE_KEY } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/constants/workflow-cron-trigger-cache-key.constant';
@@ -35,6 +36,10 @@ const mockCacheStorageService = {
  hashSetWithExpire: jest.fn(),
 };

+const mockCronTriggerDeduplicationService = {
+  shouldDispatch: jest.fn(),
+};
+
 describe('WorkflowCronTriggerCronJob', () => {
  let job: WorkflowCronTriggerCronJob;

@@ -42,6 +47,7 @@ describe('WorkflowCronTriggerCronJob', () => {
    jest.clearAllMocks();
    jest.useFakeTimers();
    jest.setSystemTime(new Date('2026-04-02T15:00:30.000Z'));
+    mockCronTriggerDeduplicationService.shouldDispatch.mockResolvedValue(true);

    const module: TestingModule = await Test.createTestingModule({
      providers: [
@@ -66,6 +72,10 @@ describe('WorkflowCronTriggerCronJob', () => {
          provide: CacheStorageNamespace.ModuleWorkflow,
          useValue: mockCacheStorageService,
        },
+        {
+          provide: CronTriggerDeduplicationService,
+          useValue: mockCronTriggerDeduplicationService,
+        },
      ],
    }).compile();

@@ -117,7 +127,10 @@ describe('WorkflowCronTriggerCronJob', () => {
      );
    });

-    it('should not enqueue jobs when cron pattern does not match', async () => {
+    it('should not enqueue jobs when the trigger is not due', async () => {
+      mockCronTriggerDeduplicationService.shouldDispatch.mockResolvedValue(
+        false,
+      );
      mockCacheStorageService.hashGetValues.mockResolvedValue([
        JSON.stringify({
          workspaceId: WORKSPACE_1,
@@ -9,6 +9,7 @@ import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decora
 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';
 import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { CronTriggerDeduplicationService } from 'src/engine/core-modules/cron/services/cron-trigger-deduplication.service';
 import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
@@ -26,7 +27,6 @@ import {
  WorkflowTriggerJob,
  type WorkflowTriggerJobData,
 } from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';
-import { shouldRunNow } from 'src/utils/should-run-now.utils';

 export const WORKFLOW_CRON_TRIGGER_CRON_PATTERN = '* * * * *';

@@ -44,6 +44,7 @@ export class WorkflowCronTriggerCronJob {
    private readonly exceptionHandlerService: ExceptionHandlerService,
    @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow)
    private readonly cacheStorageService: CacheStorageService,
+    private readonly cronTriggerDeduplicationService: CronTriggerDeduplicationService,
  ) {}

  @Process(WorkflowCronTriggerCronJob.name)
@@ -82,7 +83,14 @@ export class WorkflowCronTriggerCronJob {
          continue;
        }

-        if (!shouldRunNow(trigger.pattern, now)) {
+        const shouldDispatch =
+          await this.cronTriggerDeduplicationService.shouldDispatch(
+            `workflow-cron:${trigger.workspaceId}:${trigger.workflowId}`,
+            trigger.pattern,
+            now,
+          );
+
+        if (!shouldDispatch) {
          continue;
        }

@@ -178,7 +186,14 @@ export class WorkflowCronTriggerCronJob {

        triggersToCache.push(cachedTrigger);

-        if (shouldRunNow(settings.pattern, now)) {
+        const shouldDispatch =
+          await this.cronTriggerDeduplicationService.shouldDispatch(
+            `workflow-cron:${workspaceId}:${trigger.workflowId}`,
+            settings.pattern,
+            now,
+          );
+
+        if (shouldDispatch) {
          this.logger.log(
            `Trigger ${trigger.id}: enqueuing WorkflowTriggerJob for workflow ${trigger.workflowId}`,
          );