diff --git a/packages/twenty-server/src/database/commands/cron-register-all.command.ts b/packages/twenty-server/src/database/commands/cron-register-all.command.ts index 4f13794235..09ce49886d 100644 --- a/packages/twenty-server/src/database/commands/cron-register-all.command.ts +++ b/packages/twenty-server/src/database/commands/cron-register-all.command.ts @@ -2,6 +2,7 @@ import { Logger } from '@nestjs/common'; import { Command, CommandRunner } from 'nest-commander'; +import { CheckCustomDomainValidRecordsCronCommand } from 'src/engine/core-modules/domain-manager/crons/commands/check-custom-domain-valid-records.cron.command'; import { CleanupOrphanedFilesCronCommand } from 'src/engine/core-modules/file/crons/commands/cleanup-orphaned-files.cron.command'; import { CalendarEventListFetchCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-event-list-fetch.cron.command'; import { CalendarEventsImportCronCommand } from 'src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-import.cron.command'; @@ -9,8 +10,9 @@ import { CalendarOngoingStaleCronCommand } from 'src/modules/calendar/calendar-e import { MessagingMessageListFetchCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-message-list-fetch.cron.command'; import { MessagingMessagesImportCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-messages-import.cron.command'; import { MessagingOngoingStaleCronCommand } from 'src/modules/messaging/message-import-manager/crons/commands/messaging-ongoing-stale.cron.command'; -import { CronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command'; -import { CheckCustomDomainValidRecordsCronCommand } from 'src/engine/core-modules/domain-manager/crons/commands/check-custom-domain-valid-records.cron.command'; +import { WorkflowHandleStaledRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command'; +import { WorkflowRunEnqueueCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command'; +import { WorkflowCronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command'; @Command({ name: 'cron:register:all', @@ -26,9 +28,11 @@ export class CronRegisterAllCommand extends CommandRunner { private readonly calendarEventListFetchCronCommand: CalendarEventListFetchCronCommand, private readonly calendarEventsImportCronCommand: CalendarEventsImportCronCommand, private readonly calendarOngoingStaleCronCommand: CalendarOngoingStaleCronCommand, - private readonly cronTriggerCronCommand: CronTriggerCronCommand, + private readonly workflowCronTriggerCronCommand: WorkflowCronTriggerCronCommand, private readonly cleanupOrphanedFilesCronCommand: CleanupOrphanedFilesCronCommand, private readonly checkCustomDomainValidRecordsCronCommand: CheckCustomDomainValidRecordsCronCommand, + private readonly workflowRunEnqueueCronCommand: WorkflowRunEnqueueCronCommand, + private readonly workflowHandleStaledRunsCronCommand: WorkflowHandleStaledRunsCronCommand, ) { super(); } @@ -61,7 +65,6 @@ export class CronRegisterAllCommand extends CommandRunner { name: 'CalendarOngoingStale', command: this.calendarOngoingStaleCronCommand, }, - { name: 'CronTrigger', command: this.cronTriggerCronCommand }, { name: 'CleanupOrphanedFiles', command: this.cleanupOrphanedFilesCronCommand, @@ -70,6 +73,18 @@ export class CronRegisterAllCommand extends CommandRunner { name: 'CheckCustomDomainValidRecords', command: this.checkCustomDomainValidRecordsCronCommand, }, + { + name: 'WorkflowCronTrigger', + command: this.workflowCronTriggerCronCommand, + }, + { + name: 'WorkflowRunEnqueue', + command: this.workflowRunEnqueueCronCommand, + }, + { + name: 'WorkflowHandleStaledRuns', + command: this.workflowHandleStaledRunsCronCommand, + }, ]; let successCount = 0; diff --git a/packages/twenty-server/src/database/commands/database-command.module.ts b/packages/twenty-server/src/database/commands/database-command.module.ts index 16e43f875c..464be1473b 100644 --- a/packages/twenty-server/src/database/commands/database-command.module.ts +++ b/packages/twenty-server/src/database/commands/database-command.module.ts @@ -7,6 +7,7 @@ import { UpgradeVersionCommandModule } from 'src/database/commands/upgrade-versi import { MigrateViewsToCoreCommand } from 'src/database/commands/views-migration/migrate-views-to-core.command'; import { TypeORMModule } from 'src/database/typeorm/typeorm.module'; import { ApiKeyModule } from 'src/engine/core-modules/api-key/api-key.module'; +import { DomainManagerModule } from 'src/engine/core-modules/domain-manager/domain-manager.module'; import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module'; import { FileModule } from 'src/engine/core-modules/file/file.module'; import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; @@ -18,24 +19,22 @@ import { DevSeederModule } from 'src/engine/workspace-manager/dev-seeder/dev-see import { WorkspaceManagerModule } from 'src/engine/workspace-manager/workspace-manager.module'; import { CalendarEventImportManagerModule } from 'src/modules/calendar/calendar-event-import-manager/calendar-event-import-manager.module'; import { MessagingImportManagerModule } from 'src/modules/messaging/message-import-manager/messaging-import-manager.module'; +import { WorkflowRunQueueModule } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module'; import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module'; -import { DomainManagerModule } from 'src/engine/core-modules/domain-manager/domain-manager.module'; import { DataSeedWorkspaceCommand } from './data-seed-dev-workspace.command'; @Module({ imports: [ UpgradeVersionCommandModule, - TypeOrmModule.forFeature([Workspace], 'core'), - // Cron command dependencies MessagingImportManagerModule, CalendarEventImportManagerModule, AutomatedTriggerModule, FileModule, DomainManagerModule, - + WorkflowRunQueueModule, // Data seeding dependencies TypeORMModule, FieldMetadataModule, diff --git a/packages/twenty-server/src/engine/workspace-manager/workspace-sync-metadata/constants/standard-field-ids.ts b/packages/twenty-server/src/engine/workspace-manager/workspace-sync-metadata/constants/standard-field-ids.ts index e0b887ec39..f2f37d4d16 100644 --- a/packages/twenty-server/src/engine/workspace-manager/workspace-sync-metadata/constants/standard-field-ids.ts +++ b/packages/twenty-server/src/engine/workspace-manager/workspace-sync-metadata/constants/standard-field-ids.ts @@ -466,6 +466,7 @@ export const WORKFLOW_RUN_STANDARD_FIELD_IDS = { name: '20202020-b840-4253-aef9-4e5013694587', workflowVersion: '20202020-2f52-4ba8-8dc4-d0d6adb9578d', workflow: '20202020-8c57-4e7f-84f5-f373f68e1b82', + enqueuedAt: '20202020-f1e3-4de1-a461-b5c4fdbc861d', startedAt: '20202020-a234-4e2d-bd15-85bcea6bb183', endedAt: '20202020-e1c1-4b6b-bbbd-b2beaf2e159e', status: '20202020-6b3e-4f9c-8c2b-2e5b8e6d6f3b', diff --git a/packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-run.workspace-entity.ts b/packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-run.workspace-entity.ts index 2a73f4555b..50433a3e3d 100644 --- a/packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-run.workspace-entity.ts +++ b/packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-run.workspace-entity.ts @@ -90,6 +90,16 @@ export class WorkflowRunWorkspaceEntity extends BaseWorkspaceEntity { }) name: string; + @WorkspaceField({ + standardId: WORKFLOW_RUN_STANDARD_FIELD_IDS.enqueuedAt, + type: FieldMetadataType.DATE_TIME, + label: msg`Workflow run enqueued at`, + description: msg`Workflow run enqueued at`, + icon: 'IconHistory', + }) + @WorkspaceIsNullable() + enqueuedAt: Date | null; + @WorkspaceField({ standardId: WORKFLOW_RUN_STANDARD_FIELD_IDS.startedAt, type: FieldMetadataType.DATE_TIME, diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts index f6b4fde97a..782e82ace8 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/jobs/run-workflow.job.ts @@ -2,6 +2,8 @@ import { Scope } from '@nestjs/common'; import { isDefined } from 'twenty-shared/utils'; +import { FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum'; +import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service'; import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; @@ -20,8 +22,6 @@ import { getRootSteps } from 'src/modules/workflow/workflow-runner/utils/get-roo import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; import { WorkflowTriggerType } from 'src/modules/workflow/workflow-trigger/types/workflow-trigger.type'; -import { FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum'; -import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service'; export type RunWorkflowJobData = { workspaceId: string; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts new file mode 100644 index 0000000000..9b8ac7e683 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit.ts @@ -0,0 +1 @@ +export const WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100; diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command.ts new file mode 100644 index 0000000000..e950e66d6c --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command.ts @@ -0,0 +1,34 @@ +import { Command, CommandRunner } from 'nest-commander'; + +import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; +import { + WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN, + WorkflowHandleStaledRunsJob, +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job'; + +@Command({ + name: 'cron:workflow:handle-staled-runs', + description: 'Handles staled workflow runs', +}) +export class WorkflowHandleStaledRunsCronCommand extends CommandRunner { + constructor( + @InjectMessageQueue(MessageQueue.cronQueue) + private readonly messageQueueService: MessageQueueService, + ) { + super(); + } + + async run(): Promise { + await this.messageQueueService.addCron({ + jobName: WorkflowHandleStaledRunsJob.name, + data: undefined, + options: { + repeat: { + pattern: WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN, + }, + }, + }); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-workflow-run-enqueue.cron.command.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts similarity index 90% rename from packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-workflow-run-enqueue.cron.command.ts rename to packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts index 0a3fcd1448..7542fc76da 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-workflow-run-enqueue.cron.command.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command.ts @@ -6,13 +6,13 @@ import { MessageQueueService } from 'src/engine/core-modules/message-queue/servi import { WORKFLOW_RUN_ENQUEUE_CRON_PATTERN, WorkflowRunEnqueueJob, -} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job'; +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job'; @Command({ name: 'cron:workflow:enqueue-awaiting-workflow-run', description: 'Enqueues awaiting workflow runs', }) -export class CronWorkflowRunEnqueueCommand extends CommandRunner { +export class WorkflowRunEnqueueCronCommand extends CommandRunner { constructor( @InjectMessageQueue(MessageQueue.cronQueue) private readonly messageQueueService: MessageQueueService, diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job.ts new file mode 100644 index 0000000000..983cfe313b --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job.ts @@ -0,0 +1,60 @@ +import { LessThan } from 'typeorm'; + +import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; +import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; +import { + WorkflowRunStatus, + WorkflowRunWorkspaceEntity, +} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; + +export type WorkflowHandleStaledRunsPerWorkspaceJobData = { + workspaceId: string; +}; + +@Processor(MessageQueue.workflowQueue) +export class WorkflowHandleStaledRunsPerWorkspaceJob { + constructor( + private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService, + private readonly twentyORMGlobalManager: TwentyORMGlobalManager, + ) {} + + @Process(WorkflowHandleStaledRunsPerWorkspaceJob.name) + async handle(data: WorkflowHandleStaledRunsPerWorkspaceJobData) { + const { workspaceId } = data; + + const workflowRunRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + const oneHourAgo = new Date(Date.now() - 60 * 60 * 1000); + + const staledWorkflowRuns = await workflowRunRepository.find({ + where: { + status: WorkflowRunStatus.ENQUEUED, + enqueuedAt: LessThan(oneHourAgo), + }, + }); + + if (staledWorkflowRuns.length <= 0) { + return; + } + + await workflowRunRepository.update( + staledWorkflowRuns.map((workflowRun) => workflowRun.id), + { + enqueuedAt: null, + status: WorkflowRunStatus.NOT_STARTED, + }, + ); + + await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount( + workspaceId, + ); + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job.ts new file mode 100644 index 0000000000..6820aa4bea --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job.ts @@ -0,0 +1,50 @@ +import { InjectRepository } from '@nestjs/typeorm'; + +import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; +import { Repository } from 'typeorm'; + +import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; +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'; +import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; +import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; +import { + WorkflowHandleStaledRunsPerWorkspaceJob, + type WorkflowHandleStaledRunsPerWorkspaceJobData, +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job'; + +export const WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN = '0 * * * *'; + +@Processor(MessageQueue.cronQueue) +export class WorkflowHandleStaledRunsJob { + constructor( + @InjectRepository(Workspace, 'core') + private readonly workspaceRepository: Repository, + @InjectMessageQueue(MessageQueue.workflowQueue) + private readonly messageQueueService: MessageQueueService, + ) {} + + @Process(WorkflowHandleStaledRunsJob.name) + @SentryCronMonitor( + WorkflowHandleStaledRunsJob.name, + WORKFLOW_HANDLE_STALED_RUNS_CRON_PATTERN, + ) + async handle() { + const activeWorkspaces = await this.workspaceRepository.find({ + where: { + activationStatus: WorkspaceActivationStatus.ACTIVE, + }, + }); + + for (const activeWorkspace of activeWorkspaces) { + await this.messageQueueService.add( + WorkflowHandleStaledRunsPerWorkspaceJob.name, + { + workspaceId: activeWorkspace.id, + }, + ); + } + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job.ts new file mode 100644 index 0000000000..e10f1848f4 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job.ts @@ -0,0 +1,107 @@ +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'; +import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; +import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; +import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type'; +import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; +import { + WorkflowRunStatus, + WorkflowRunWorkspaceEntity, +} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { + RunWorkflowJob, + type RunWorkflowJobData, +} from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job'; +import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; + +export type WorkflowRunEnqueuePerWorkspaceJobData = { + workspaceId: string; +}; + +@Processor(MessageQueue.workflowQueue) +export class WorkflowRunEnqueuePerWorkspaceJob { + constructor( + private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService, + @InjectMessageQueue(MessageQueue.workflowQueue) + private readonly messageQueueService: MessageQueueService, + private readonly metricsService: MetricsService, + private readonly twentyORMGlobalManager: TwentyORMGlobalManager, + ) {} + + @Process(WorkflowRunEnqueuePerWorkspaceJob.name) + async handle(data: WorkflowRunEnqueuePerWorkspaceJobData) { + const { workspaceId } = data; + + try { + const workflowRunRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + const remainingWorkflowRunToEnqueueCount = + await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCountFromDatabase( + workspaceId, + ); + + if (remainingWorkflowRunToEnqueueCount <= 0) { + await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount( + workspaceId, + ); + + return; + } + + const workflowRunsToEnqueue = await workflowRunRepository.find({ + where: { + status: WorkflowRunStatus.NOT_STARTED, + }, + order: { + createdAt: 'ASC', + }, + take: remainingWorkflowRunToEnqueueCount, + }); + + if (workflowRunsToEnqueue.length <= 0) { + await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount( + workspaceId, + ); + + return; + } + + const workflowRunIds = workflowRunsToEnqueue.map( + (workflowRun: WorkflowRunWorkspaceEntity) => workflowRun.id, + ); + + await workflowRunRepository.update(workflowRunIds, { + enqueuedAt: new Date().toISOString(), + status: WorkflowRunStatus.ENQUEUED, + }); + + for (const workflowRunId of workflowRunIds) { + await this.messageQueueService.add( + RunWorkflowJob.name, + { + workflowRunId, + workspaceId, + }, + ); + } + + await this.workflowRunQueueWorkspaceService.recomputeWorkflowRunQueuedCount( + workspaceId, + ); + } catch (error) { + this.metricsService.incrementCounter({ + key: MetricsKeys.WorkflowRunFailedToEnqueue, + eventId: workspaceId, + }); + + throw error; + } + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts deleted file mode 100644 index d4607ece53..0000000000 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job.ts +++ /dev/null @@ -1,138 +0,0 @@ -import { Logger } from '@nestjs/common'; -import { InjectRepository } from '@nestjs/typeorm'; - -import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; -import { Repository } from 'typeorm'; - -import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; -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'; -import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; -import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; -import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; -import { MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; -import { MetricsKeys } from 'src/engine/core-modules/metrics/types/metrics-keys.type'; -import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; -import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; -import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; -import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service'; -import { - WorkflowRunStatus, - WorkflowRunWorkspaceEntity, -} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; -import { - RunWorkflowJob, - type RunWorkflowJobData, -} from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; - -export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '* * * * *'; - -@Processor(MessageQueue.cronQueue) -export class WorkflowRunEnqueueJob { - private readonly logger = new Logger(WorkflowRunEnqueueJob.name); - - constructor( - @InjectRepository(Workspace, 'core') - private readonly workspaceRepository: Repository, - private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService, - private readonly workspaceDataSourceService: WorkspaceDataSourceService, - @InjectMessageQueue(MessageQueue.workflowQueue) - private readonly messageQueueService: MessageQueueService, - private readonly metricsService: MetricsService, - private readonly twentyORMGlobalManager: TwentyORMGlobalManager, - ) {} - - @Process(WorkflowRunEnqueueJob.name) - @SentryCronMonitor( - WorkflowRunEnqueueJob.name, - WORKFLOW_RUN_ENQUEUE_CRON_PATTERN, - ) - async handle() { - const activeWorkspaces = await this.workspaceRepository.find({ - where: { - activationStatus: WorkspaceActivationStatus.ACTIVE, - }, - }); - - const mainDataSource = - await this.workspaceDataSourceService.connectToMainDataSource(); - - for (const activeWorkspace of activeWorkspaces) { - let enqueuedWorkflowRunCount = 0; - - try { - const remainingWorkflowRunCount = - await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCount( - activeWorkspace.id, - ); - - if (remainingWorkflowRunCount <= 0) { - continue; - } - - const schemaName = getWorkspaceSchemaName(activeWorkspace.id); - - // Using raw query to avoid storing repository in cache - const workflowRuns = await mainDataSource.query( - `SELECT id FROM ${schemaName}."workflowRun" WHERE status = '${WorkflowRunStatus.NOT_STARTED}' ORDER BY "createdAt" ASC`, - ); - - const workflowRunsToEnqueueCount = Math.min( - remainingWorkflowRunCount, - workflowRuns.length, - ); - - if (workflowRunsToEnqueueCount <= 0) { - continue; - } - - const workflowRunRepository = - await this.twentyORMGlobalManager.getRepositoryForWorkspace( - activeWorkspace.id, - WorkflowRunWorkspaceEntity, - { shouldBypassPermissionChecks: true }, - ); - - for ( - let runIndex = 0; - runIndex < workflowRunsToEnqueueCount; - runIndex++ - ) { - const workflowRunId = workflowRuns[runIndex].id; - - await this.messageQueueService.add( - RunWorkflowJob.name, - { - workflowRunId, - workspaceId: activeWorkspace.id, - }, - ); - - await workflowRunRepository.update(workflowRunId, { - status: WorkflowRunStatus.ENQUEUED, - }); - - enqueuedWorkflowRunCount++; - } - } catch (error) { - this.logger.error( - `Error enqueuing workflow runs for workspace ${activeWorkspace.id}`, - error, - ); - - this.metricsService.incrementCounter({ - key: MetricsKeys.WorkflowRunFailedToEnqueue, - eventId: activeWorkspace.id, - }); - } finally { - if (enqueuedWorkflowRunCount > 0) { - await this.workflowRunQueueWorkspaceService.increaseWorkflowRunQueuedCount( - activeWorkspace.id, - enqueuedWorkflowRunCount, - ); - } - } - } - } -} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job.ts new file mode 100644 index 0000000000..605943d370 --- /dev/null +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job.ts @@ -0,0 +1,50 @@ +import { InjectRepository } from '@nestjs/typeorm'; + +import { WorkspaceActivationStatus } from 'twenty-shared/workspace'; +import { Repository } from 'typeorm'; + +import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; +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'; +import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; +import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; +import { + WorkflowRunEnqueuePerWorkspaceJob, + type WorkflowRunEnqueuePerWorkspaceJobData, +} from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job'; + +export const WORKFLOW_RUN_ENQUEUE_CRON_PATTERN = '* * * * *'; + +@Processor(MessageQueue.cronQueue) +export class WorkflowRunEnqueueJob { + constructor( + @InjectRepository(Workspace, 'core') + private readonly workspaceRepository: Repository, + @InjectMessageQueue(MessageQueue.workflowQueue) + private readonly messageQueueService: MessageQueueService, + ) {} + + @Process(WorkflowRunEnqueueJob.name) + @SentryCronMonitor( + WorkflowRunEnqueueJob.name, + WORKFLOW_RUN_ENQUEUE_CRON_PATTERN, + ) + async handle() { + const activeWorkspaces = await this.workspaceRepository.find({ + where: { + activationStatus: WorkspaceActivationStatus.ACTIVE, + }, + }); + + for (const activeWorkspace of activeWorkspaces) { + await this.messageQueueService.add( + WorkflowRunEnqueuePerWorkspaceJob.name, + { + workspaceId: activeWorkspace.id, + }, + ); + } + } +} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts index 298f0fa4e6..78e76494e6 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module.ts @@ -6,11 +6,15 @@ import { MessageQueueModule } from 'src/engine/core-modules/message-queue/messag import { MetricsModule } from 'src/engine/core-modules/metrics/metrics.module'; import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module'; -import { CronWorkflowRunEnqueueCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-workflow-run-enqueue.cron.command'; -import { WorkflowRunEnqueueJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.cron.job'; -import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; -import { CleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job'; import { CronCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/cron-clean-workflow-runs.cron.command'; +import { WorkflowHandleStaledRunsCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-handle-staled-runs.cron.command'; +import { WorkflowRunEnqueueCronCommand } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/command/workflow-run-enqueue.cron.command'; +import { CleanWorkflowRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/clean-workflow-runs.cron.job'; +import { WorkflowHandleStaledRunsPerWorkspaceJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs-per-workspace.job'; +import { WorkflowHandleStaledRunsJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-handle-staled-runs.job'; +import { WorkflowRunEnqueuePerWorkspaceJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue-per-workspace.job'; +import { WorkflowRunEnqueueJob } from 'src/modules/workflow/workflow-runner/workflow-run-queue/cron/jobs/workflow-run-enqueue.job'; +import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; @Module({ imports: [ @@ -22,11 +26,19 @@ import { CronCleanWorkflowRunsCommand } from 'src/modules/workflow/workflow-runn ], providers: [ WorkflowRunQueueWorkspaceService, + WorkflowRunEnqueueCronCommand, WorkflowRunEnqueueJob, + WorkflowRunEnqueuePerWorkspaceJob, + WorkflowHandleStaledRunsCronCommand, + WorkflowHandleStaledRunsJob, + WorkflowHandleStaledRunsPerWorkspaceJob, CleanWorkflowRunsJob, - CronWorkflowRunEnqueueCommand, CronCleanWorkflowRunsCommand, ], - exports: [WorkflowRunQueueWorkspaceService], + exports: [ + WorkflowRunQueueWorkspaceService, + WorkflowRunEnqueueCronCommand, + WorkflowHandleStaledRunsCronCommand, + ], }) export class WorkflowRunQueueModule {} diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts index d3e358bd3f..758721bade 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service.ts @@ -1,17 +1,24 @@ import { Injectable } from '@nestjs/common'; +import { In } from 'typeorm'; + import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; 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 { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; +import { + WorkflowRunStatus, + WorkflowRunWorkspaceEntity, +} from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity'; +import { WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT } from 'src/modules/workflow/workflow-runner/workflow-run-queue/constants/workflow-run-queue-throttle-limit'; import { getWorkflowRunQueuedCountCacheKey } from 'src/modules/workflow/workflow-runner/workflow-run-queue/utils/get-cache-workflow-run-count-key.util'; @Injectable() export class WorkflowRunQueueWorkspaceService { - private readonly WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT = 100; - constructor( @InjectCacheStorage(CacheStorageNamespace.ModuleWorkflow) private readonly cacheStorage: CacheStorageService, + private readonly twentyORMGlobalManager: TwentyORMGlobalManager, ) {} async increaseWorkflowRunQueuedCount( @@ -40,11 +47,64 @@ export class WorkflowRunQueueWorkspaceService { ); } - async getRemainingRunsToEnqueueCount(workspaceId: string): Promise { + async recomputeWorkflowRunQueuedCount(workspaceId: string): Promise { + const workflowRunRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + const currentlyEnqueuedWorkflowRunCount = await workflowRunRepository.count( + { + where: { + status: In([WorkflowRunStatus.ENQUEUED, WorkflowRunStatus.RUNNING]), + }, + }, + ); + + await this.setWorkflowRunQueuedCount( + workspaceId, + currentlyEnqueuedWorkflowRunCount, + ); + } + + async getRemainingRunsToEnqueueCountFromCache( + workspaceId: string, + ): Promise { const currentCount = await this.getCurrentWorkflowRunQueuedCount(workspaceId); - return this.WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - currentCount; + return WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - currentCount; + } + + async getRemainingRunsToEnqueueCountFromDatabase( + workspaceId: string, + ): Promise { + const workflowRunRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + WorkflowRunWorkspaceEntity, + { shouldBypassPermissionChecks: true }, + ); + + const currentCount = await workflowRunRepository.count({ + where: { + status: In([WorkflowRunStatus.ENQUEUED, WorkflowRunStatus.RUNNING]), + }, + }); + + return WORKFLOW_RUN_QUEUE_THROTTLE_LIMIT - currentCount; + } + + private async setWorkflowRunQueuedCount( + workspaceId: string, + count: number, + ): Promise { + await this.cacheStorage.set( + getWorkflowRunQueuedCountCacheKey(workspaceId), + count, + ); } private async getCurrentWorkflowRunQueuedCount( diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts index 4abf2ad030..9a41143cd3 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service.ts @@ -1,7 +1,7 @@ import { Injectable } from '@nestjs/common'; import { isDefined } from 'twenty-shared/utils'; -import { StepStatus, WorkflowRunStepInfo } from 'twenty-shared/workflow'; +import { StepStatus, type WorkflowRunStepInfo } from 'twenty-shared/workflow'; import { type QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity'; import { v4 } from 'uuid'; @@ -44,9 +44,9 @@ export class WorkflowRunWorkspaceService { }: { workflowVersionId: string; createdBy: ActorMetadata; - workflowRunId?: string; status: WorkflowRunStatus.NOT_STARTED | WorkflowRunStatus.ENQUEUED; triggerPayload: object; + workflowRunId?: string; }) { const workspaceId = this.scopedWorkspaceContextFactory.create()?.workspaceId; @@ -124,6 +124,8 @@ export class WorkflowRunWorkspaceService { status, position, state: initState, + enqueuedAt: + status === WorkflowRunStatus.ENQUEUED ? new Date().toISOString() : null, }); await workflowRunRepository.insert(workflowRun); @@ -144,14 +146,21 @@ export class WorkflowRunWorkspaceService { workspaceId, }); + if ( + workflowRunToUpdate.status === WorkflowRunStatus.COMPLETED || + workflowRunToUpdate.status === WorkflowRunStatus.FAILED + ) { + throw new WorkflowRunException( + 'Cannot start a workflow run already ended', + WorkflowRunExceptionCode.INVALID_OPERATION, + ); + } + if ( workflowRunToUpdate.status !== WorkflowRunStatus.ENQUEUED && workflowRunToUpdate.status !== WorkflowRunStatus.NOT_STARTED ) { - throw new WorkflowRunException( - 'Workflow run already started', - WorkflowRunExceptionCode.INVALID_OPERATION, - ); + return; } const partialUpdate = { diff --git a/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts b/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts index c3ffc34c2f..5a372290ca 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service.ts @@ -9,7 +9,7 @@ import { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/ import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service'; import { RunWorkflowJob, - type RunWorkflowJobData, + RunWorkflowJobData, } from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job'; import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service'; import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service'; @@ -56,7 +56,7 @@ export class WorkflowRunnerWorkspaceService { }); const remainingRunsToEnqueueCount = - await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCount( + await this.workflowRunQueueWorkspaceService.getRemainingRunsToEnqueueCountFromCache( workspaceId, ); diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts index dbff9374c0..60b53f0498 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module.ts @@ -2,12 +2,12 @@ import { Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; import { Workspace } 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'; import { AutomatedTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.workspace-service'; -import { CronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command'; +import { WorkflowCronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command'; import { CronTriggerCronJob } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/cron-trigger.cron.job'; import { DatabaseEventTriggerListener } from 'src/modules/workflow/workflow-trigger/automated-trigger/listeners/database-event-trigger.listener'; -import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module'; @Module({ imports: [ @@ -19,8 +19,8 @@ import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/works AutomatedTriggerWorkspaceService, DatabaseEventTriggerListener, CronTriggerCronJob, - CronTriggerCronCommand, + WorkflowCronTriggerCronCommand, ], - exports: [AutomatedTriggerWorkspaceService, CronTriggerCronCommand], + exports: [AutomatedTriggerWorkspaceService, WorkflowCronTriggerCronCommand], }) export class AutomatedTriggerModule {} diff --git a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command.ts b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command.ts similarity index 94% rename from packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command.ts rename to packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command.ts index aa00ffd538..de07c9361b 100644 --- a/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command.ts +++ b/packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command.ts @@ -12,7 +12,7 @@ import { name: 'cron:workflow:automated-cron-trigger', description: 'Starts a cron job to trigger cron triggered workflows', }) -export class CronTriggerCronCommand extends CommandRunner { +export class WorkflowCronTriggerCronCommand extends CommandRunner { constructor( @InjectMessageQueue(MessageQueue.cronQueue) private readonly messageQueueService: MessageQueueService,