From 66df0ac47c31d405229748205b119b191b6920ee Mon Sep 17 00:00:00 2001 From: Weiko Date: Thu, 23 Jul 2026 14:29:22 +0200 Subject: [PATCH] Switch application stop/start commands to Redis-backed global kill switch (#23202) ## Context [#23183](https://github.com/twentyhq/twenty/pull/23183) introduced the right enforcement point: every logic-function execution is rejected centrally before consuming the shared workspace throttle when its application is stopped. However, its server-wide path reads PostgreSQL for every execution attempt. A kill switch is most useful while an application is producing abnormal load, potentially while PostgreSQL is already under pressure. The enforcement mechanism should not add more database traffic in that situation. This state is also operational and temporary. It is used to troubleshoot an application, not as durable application configuration. ## What this PR changes - Uses one global Redis key per application universal identifier: ```text module:applications:kill-switch:{applicationUniversalIdentifier} ``` - Keeps the check in `LogicFunctionExecutorService`, before the workspace execution throttle. - Adds a 60-second process-local cache for both present and absent keys. - Deduplicates concurrent cache refreshes, so an execution burst causes at most one Redis read per application and process. - Fails open when Redis cannot be read and caches that result for the same minute, avoiding a Redis retry storm. - Removes the database columns, upgrade command, workspace-cache recomputation, registration lookup, and stop/start CLI commands introduced by #23183. - Keeps disabled queued executions non-retriable, without emitting one warning for every skipped payload. The switch is operated directly in Redis. For example: ```redis SET module:applications:kill-switch:{applicationUniversalIdentifier} 1 EX 3600 DEL module:applications:kill-switch:{applicationUniversalIdentifier} ``` Any value means stopped; deleting or expiring the key means enabled. ## Why this is a better fit | | #23183 | This PR | |---|---|---| | State | Durable PostgreSQL fields | Ephemeral Redis key | | Server-wide hot path | PostgreSQL lookup per execution | At most one Redis lookup per app/process/minute | | Scope | Workspace and application registration | Application universal identifier across all workspaces | | Operational cleanup | Explicit start command | `DEL`, eviction, restart, or operator-selected TTL | | Database dependency during an incident | Required | None | The trade-off is deliberate: a Redis change can take up to 60 seconds to reach every process, and the switch is lost when the cache key disappears. That is acceptable for a temporary troubleshooting control and keeps the normal execution path inexpensive. Existing in-flight functions are not interrupted. New direct or queued executions are rejected when they reach the executor. --- .../commands/database-command.module.ts | 8 - .../start-application-registration.command.ts | 46 --- .../commands/start-application.command.ts | 67 ---- .../stop-application-registration.command.ts | 43 --- .../commands/stop-application.command.ts | 55 ---- ...pplication-and-application-registration.ts | 27 -- .../instance-commands.constant.ts | 2 - .../application-stop.service.spec.ts | 306 ++++++------------ .../application-sync.service.ts | 1 - .../application-registration.entity.ts | 10 - .../application/application-stop.service.ts | 192 +++-------- .../application/application.entity.ts | 9 - .../types/cache-storage-namespace.enum.ts | 1 + .../logic-function-executor.service.ts | 19 +- .../jobs/logic-function-trigger.job.ts | 8 +- ...-to-flat-field-metadatas-to-create.spec.ts | 1 - ...-relation-flat-field-metadata-pair.spec.ts | 1 - 17 files changed, 135 insertions(+), 661 deletions(-) delete mode 100644 packages/twenty-server/src/database/commands/start-application-registration.command.ts delete mode 100644 packages/twenty-server/src/database/commands/start-application.command.ts delete mode 100644 packages/twenty-server/src/database/commands/stop-application-registration.command.ts delete mode 100644 packages/twenty-server/src/database/commands/stop-application.command.ts delete mode 100644 packages/twenty-server/src/database/commands/upgrade-version-command/2-24/2-24-instance-command-fast-1784734278506-add-stopped-at-to-application-and-application-registration.ts 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 928183b62f..86301c4666 100644 --- a/packages/twenty-server/src/database/commands/database-command.module.ts +++ b/packages/twenty-server/src/database/commands/database-command.module.ts @@ -12,10 +12,6 @@ import { ListOrphanedWorkspaceEntitiesCommand } from 'src/database/commands/list import { ConfirmationQuestion } from 'src/database/commands/questions/confirmation.question'; import { RebuildApplicationDefaultDepsCommand } from 'src/database/commands/rebuild-application-default-deps.command'; import { RunInstanceCommandsCommand } from 'src/database/commands/run-instance-commands.command'; -import { StartApplicationRegistrationCommand } from 'src/database/commands/start-application-registration.command'; -import { StartApplicationCommand } from 'src/database/commands/start-application.command'; -import { StopApplicationRegistrationCommand } from 'src/database/commands/stop-application-registration.command'; -import { StopApplicationCommand } from 'src/database/commands/stop-application.command'; import { UpgradeVersionCommandModule } from 'src/database/commands/upgrade-version-command/upgrade-version-command.module'; import { WorkspaceExportModule } from 'src/database/commands/workspace-export/workspace-export.module'; import { TypeORMModule } from 'src/database/typeorm/typeorm.module'; @@ -114,10 +110,6 @@ import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/au UpgradeStatusCommand, RebuildApplicationDefaultDepsCommand, InstallPreInstalledAppsCommand, - StopApplicationCommand, - StartApplicationCommand, - StopApplicationRegistrationCommand, - StartApplicationRegistrationCommand, provideWorkspaceScopedRepository(RoleEntity), ], }) diff --git a/packages/twenty-server/src/database/commands/start-application-registration.command.ts b/packages/twenty-server/src/database/commands/start-application-registration.command.ts deleted file mode 100644 index ba3f0a2f45..0000000000 --- a/packages/twenty-server/src/database/commands/start-application-registration.command.ts +++ /dev/null @@ -1,46 +0,0 @@ -import { Logger } from '@nestjs/common'; - -import { Command, CommandRunner, Option } from 'nest-commander'; - -import { ApplicationStopService } from 'src/engine/core-modules/application/application-stop.service'; - -@Command({ - name: 'application-registration:start', - description: - 'Lift the server-level kill switch set by application-registration:stop and resume logic function executions of all applications installed from the registration.', -}) -export class StartApplicationRegistrationCommand extends CommandRunner { - private readonly logger = new Logger( - StartApplicationRegistrationCommand.name, - ); - - constructor(private readonly applicationStopService: ApplicationStopService) { - super(); - } - - @Option({ - flags: - '-r, --application-registration-universal-identifier ', - description: - 'universal identifier of the application registration to start', - required: true, - }) - parseApplicationRegistrationUniversalIdentifier(value: string): string { - return value; - } - - override async run( - _passedParams: string[], - options: { applicationRegistrationUniversalIdentifier: string }, - ): Promise { - const { applicationRegistration, installedApplicationCount } = - await this.applicationStopService.startApplicationRegistration({ - applicationRegistrationUniversalIdentifier: - options.applicationRegistrationUniversalIdentifier, - }); - - this.logger.log( - `Started application registration "${applicationRegistration.name}" (universalIdentifier ${applicationRegistration.universalIdentifier}): the server-level stop is lifted for its ${installedApplicationCount} installed application(s). Workspace-level stops set with application:stop remain in effect.`, - ); - } -} diff --git a/packages/twenty-server/src/database/commands/start-application.command.ts b/packages/twenty-server/src/database/commands/start-application.command.ts deleted file mode 100644 index 4f82b943e8..0000000000 --- a/packages/twenty-server/src/database/commands/start-application.command.ts +++ /dev/null @@ -1,67 +0,0 @@ -import { Logger } from '@nestjs/common'; - -import { Command, CommandRunner, Option } from 'nest-commander'; -import { isDefined } from 'twenty-shared/utils'; - -import { ApplicationStopService } from 'src/engine/core-modules/application/application-stop.service'; - -type StartApplicationCommandOptions = { - workspaceId: string; - applicationUniversalIdentifier: string; -}; - -@Command({ - name: 'application:start', - description: - 'Lift the workspace-level kill switch set by application:stop and resume logic function executions of the application.', -}) -export class StartApplicationCommand extends CommandRunner { - private readonly logger = new Logger(StartApplicationCommand.name); - - constructor(private readonly applicationStopService: ApplicationStopService) { - super(); - } - - @Option({ - flags: '-w, --workspace-id ', - description: 'id of the workspace the application is installed in', - required: true, - }) - parseWorkspaceId(value: string): string { - return value; - } - - @Option({ - flags: '-a, --application-universal-identifier ', - description: 'universal identifier of the application to start', - required: true, - }) - parseApplicationUniversalIdentifier(value: string): string { - return value; - } - - override async run( - _passedParams: string[], - options: StartApplicationCommandOptions, - ): Promise { - const application = await this.applicationStopService.startApplication({ - workspaceId: options.workspaceId, - applicationUniversalIdentifier: options.applicationUniversalIdentifier, - }); - - this.logger.log( - `Started application "${application.name}" (universalIdentifier ${application.universalIdentifier}) in workspace ${application.workspaceId}: the workspace-level stop is lifted.`, - ); - - if ( - isDefined(application.applicationRegistrationId) && - (await this.applicationStopService.isApplicationRegistrationStopped( - application.applicationRegistrationId, - )) - ) { - this.logger.warn( - `Application registration ${application.applicationRegistrationId} is still stopped server-wide: executions of this application remain blocked until application-registration:start is run.`, - ); - } - } -} diff --git a/packages/twenty-server/src/database/commands/stop-application-registration.command.ts b/packages/twenty-server/src/database/commands/stop-application-registration.command.ts deleted file mode 100644 index c3fd5a3eda..0000000000 --- a/packages/twenty-server/src/database/commands/stop-application-registration.command.ts +++ /dev/null @@ -1,43 +0,0 @@ -import { Logger } from '@nestjs/common'; - -import { Command, CommandRunner, Option } from 'nest-commander'; - -import { ApplicationStopService } from 'src/engine/core-modules/application/application-stop.service'; - -@Command({ - name: 'application-registration:stop', - description: - 'Server-level kill switch: block all logic function executions of every application installed from a registration, across all workspaces. Reverse with application-registration:start.', -}) -export class StopApplicationRegistrationCommand extends CommandRunner { - private readonly logger = new Logger(StopApplicationRegistrationCommand.name); - - constructor(private readonly applicationStopService: ApplicationStopService) { - super(); - } - - @Option({ - flags: - '-r, --application-registration-universal-identifier ', - description: 'universal identifier of the application registration to stop', - required: true, - }) - parseApplicationRegistrationUniversalIdentifier(value: string): string { - return value; - } - - override async run( - _passedParams: string[], - options: { applicationRegistrationUniversalIdentifier: string }, - ): Promise { - const { applicationRegistration, installedApplicationCount } = - await this.applicationStopService.stopApplicationRegistration({ - applicationRegistrationUniversalIdentifier: - options.applicationRegistrationUniversalIdentifier, - }); - - this.logger.log( - `Stopped application registration "${applicationRegistration.name}" (universalIdentifier ${applicationRegistration.universalIdentifier}). All logic function executions of its ${installedApplicationCount} installed application(s) are now blocked, across all workspaces.`, - ); - } -} diff --git a/packages/twenty-server/src/database/commands/stop-application.command.ts b/packages/twenty-server/src/database/commands/stop-application.command.ts deleted file mode 100644 index 6d9413f191..0000000000 --- a/packages/twenty-server/src/database/commands/stop-application.command.ts +++ /dev/null @@ -1,55 +0,0 @@ -import { Logger } from '@nestjs/common'; - -import { Command, CommandRunner, Option } from 'nest-commander'; - -import { ApplicationStopService } from 'src/engine/core-modules/application/application-stop.service'; - -type StopApplicationCommandOptions = { - workspaceId: string; - applicationUniversalIdentifier: string; -}; - -@Command({ - name: 'application:stop', - description: - 'Workspace-level kill switch: block all logic function executions of one installed application. Reverse with application:start.', -}) -export class StopApplicationCommand extends CommandRunner { - private readonly logger = new Logger(StopApplicationCommand.name); - - constructor(private readonly applicationStopService: ApplicationStopService) { - super(); - } - - @Option({ - flags: '-w, --workspace-id ', - description: 'id of the workspace the application is installed in', - required: true, - }) - parseWorkspaceId(value: string): string { - return value; - } - - @Option({ - flags: '-a, --application-universal-identifier ', - description: 'universal identifier of the application to stop', - required: true, - }) - parseApplicationUniversalIdentifier(value: string): string { - return value; - } - - override async run( - _passedParams: string[], - options: StopApplicationCommandOptions, - ): Promise { - const application = await this.applicationStopService.stopApplication({ - workspaceId: options.workspaceId, - applicationUniversalIdentifier: options.applicationUniversalIdentifier, - }); - - this.logger.log( - `Stopped application "${application.name}" (universalIdentifier ${application.universalIdentifier}) in workspace ${application.workspaceId}. All its logic function executions are now blocked.`, - ); - } -} diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/2-24/2-24-instance-command-fast-1784734278506-add-stopped-at-to-application-and-application-registration.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/2-24/2-24-instance-command-fast-1784734278506-add-stopped-at-to-application-and-application-registration.ts deleted file mode 100644 index e4b1f38690..0000000000 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/2-24/2-24-instance-command-fast-1784734278506-add-stopped-at-to-application-and-application-registration.ts +++ /dev/null @@ -1,27 +0,0 @@ -import { QueryRunner } from 'typeorm'; - -import { RegisteredInstanceCommand } from 'src/engine/core-modules/upgrade/decorators/registered-instance-command.decorator'; -import { FastInstanceCommand } from 'src/engine/core-modules/upgrade/interfaces/fast-instance-command.interface'; - -@RegisteredInstanceCommand('2.24.0', 1784734278506) -export class AddStoppedAtToApplicationAndApplicationRegistrationFastInstanceCommand - implements FastInstanceCommand -{ - public async up(queryRunner: QueryRunner): Promise { - await queryRunner.query( - 'ALTER TABLE "core"."application" ADD COLUMN IF NOT EXISTS "stoppedAt" TIMESTAMP WITH TIME ZONE', - ); - await queryRunner.query( - 'ALTER TABLE "core"."applicationRegistration" ADD COLUMN IF NOT EXISTS "stoppedAt" TIMESTAMP WITH TIME ZONE', - ); - } - - public async down(queryRunner: QueryRunner): Promise { - await queryRunner.query( - 'ALTER TABLE "core"."applicationRegistration" DROP COLUMN IF EXISTS "stoppedAt"', - ); - await queryRunner.query( - 'ALTER TABLE "core"."application" DROP COLUMN IF EXISTS "stoppedAt"', - ); - } -} diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/instance-commands.constant.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/instance-commands.constant.ts index 47d5d3888b..5d18bc6e74 100644 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/instance-commands.constant.ts +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/instance-commands.constant.ts @@ -53,7 +53,6 @@ import { AllowServerScopedFileFastInstanceCommand } from 'src/database/commands/ import { AddCalendarEndFieldMetadataIdToViewFastInstanceCommand } from 'src/database/commands/upgrade-version-command/2-22/2-22-instance-command-fast-1783956795000-add-calendar-end-field-metadata-id-to-view'; import { AddSdkClientCoreChecksumToApplicationFastInstanceCommand } from 'src/database/commands/upgrade-version-command/2-23/2-23-instance-command-fast-1784625638000-add-sdk-client-core-checksum-to-application'; import { AddAutoUpgradeToApplicationFastInstanceCommand } from 'src/database/commands/upgrade-version-command/2-23/2-23-instance-command-fast-1784297307235-add-auto-upgrade-to-application'; -import { AddStoppedAtToApplicationAndApplicationRegistrationFastInstanceCommand } from 'src/database/commands/upgrade-version-command/2-24/2-24-instance-command-fast-1784734278506-add-stopped-at-to-application-and-application-registration'; import { AddSubFieldNameToViewSortEarlyFastInstanceCommand } from 'src/database/commands/upgrade-version-command/2-3/2-3-instance-command-fast-1747234200000-add-sub-field-name-to-view-sort'; import { AddRelationTargetFieldMetadataIdToViewFilterEarlyFastInstanceCommand } from 'src/database/commands/upgrade-version-command/2-3/2-3-instance-command-fast-1747234300000-add-relation-target-field-metadata-id-to-view-filter'; import { AddUpgradeMigrationWorkspaceIdIndexFastInstanceCommand } from 'src/database/commands/upgrade-version-command/2-3/2-3-instance-command-fast-1777308014234-add-upgrade-migration-workspace-id-index'; @@ -245,5 +244,4 @@ export const INSTANCE_COMMANDS = [ AddSdkClientCoreChecksumToApplicationFastInstanceCommand, AddStatusesToBillingSubscriptionIndexSlowInstanceCommand, AddOnConnectLogicFunctionToConnectionProviderFastInstanceCommand, - AddStoppedAtToApplicationAndApplicationRegistrationFastInstanceCommand, ]; diff --git a/packages/twenty-server/src/engine/core-modules/application/__tests__/application-stop.service.spec.ts b/packages/twenty-server/src/engine/core-modules/application/__tests__/application-stop.service.spec.ts index e03719a6f1..d7351d4fc5 100644 --- a/packages/twenty-server/src/engine/core-modules/application/__tests__/application-stop.service.spec.ts +++ b/packages/twenty-server/src/engine/core-modules/application/__tests__/application-stop.service.spec.ts @@ -1,243 +1,123 @@ import { Test, type TestingModule } from '@nestjs/testing'; -import { getRepositoryToken } from '@nestjs/typeorm'; -import { ApplicationRegistrationEntity } from 'src/engine/core-modules/application/application-registration/application-registration.entity'; -import { ApplicationRegistrationException } from 'src/engine/core-modules/application/application-registration/application-registration.exception'; -import { ApplicationStopService } from 'src/engine/core-modules/application/application-stop.service'; -import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity'; -import { ApplicationException } from 'src/engine/core-modules/application/application.exception'; -import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; +import { + APPLICATION_KILL_SWITCH_LOCAL_CACHE_TTL_MS, + ApplicationStopService, +} from 'src/engine/core-modules/application/application-stop.service'; +import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; -const APPLICATION_ID = 'application-1'; const APPLICATION_UNIVERSAL_IDENTIFIER = 'application-universal-identifier-1'; -const APPLICATION_REGISTRATION_ID = 'application-registration-1'; -const APPLICATION_REGISTRATION_UNIVERSAL_IDENTIFIER = - 'application-registration-universal-identifier-1'; -const WORKSPACE_ID = 'workspace-1'; +const KILL_SWITCH_KEY = `kill-switch:${APPLICATION_UNIVERSAL_IDENTIFIER}`; describe('ApplicationStopService', () => { let applicationStopService: ApplicationStopService; - const applicationRepository = { - findOne: jest.fn(), - update: jest.fn(), - count: jest.fn(), - }; - - const applicationRegistrationRepository = { - findOne: jest.fn(), - update: jest.fn(), - count: jest.fn(), - }; - - const workspaceCacheService = { - invalidateAndRecompute: jest.fn(), + const cacheStorageService = { + get: jest.fn(), }; beforeEach(async () => { - jest.clearAllMocks(); + jest.resetAllMocks(); const module: TestingModule = await Test.createTestingModule({ providers: [ ApplicationStopService, { - provide: getRepositoryToken(ApplicationEntity), - useValue: applicationRepository, - }, - { - provide: getRepositoryToken(ApplicationRegistrationEntity), - useValue: applicationRegistrationRepository, - }, - { - provide: WorkspaceCacheService, - useValue: workspaceCacheService, + provide: CacheStorageNamespace.ModuleApplications, + useValue: cacheStorageService, }, ], }).compile(); - applicationStopService = module.get( - ApplicationStopService, + applicationStopService = module.get(ApplicationStopService); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('returns true when the kill switch exists', async () => { + cacheStorageService.get.mockResolvedValue('stopped'); + + await expect( + applicationStopService.isApplicationStopped( + APPLICATION_UNIVERSAL_IDENTIFIER, + ), + ).resolves.toBe(true); + expect(cacheStorageService.get).toHaveBeenCalledWith(KILL_SWITCH_KEY); + }); + + it('returns false when the kill switch is absent', async () => { + cacheStorageService.get.mockResolvedValue(undefined); + + await expect( + applicationStopService.isApplicationStopped( + APPLICATION_UNIVERSAL_IDENTIFIER, + ), + ).resolves.toBe(false); + }); + + it('caches the Redis value for one minute', async () => { + const dateNow = jest.spyOn(Date, 'now').mockReturnValue(1_000); + + cacheStorageService.get + .mockResolvedValueOnce(undefined) + .mockResolvedValueOnce('stopped'); + + await expect( + applicationStopService.isApplicationStopped( + APPLICATION_UNIVERSAL_IDENTIFIER, + ), + ).resolves.toBe(false); + + dateNow.mockReturnValue( + 1_000 + APPLICATION_KILL_SWITCH_LOCAL_CACHE_TTL_MS - 1, ); + + await expect( + applicationStopService.isApplicationStopped( + APPLICATION_UNIVERSAL_IDENTIFIER, + ), + ).resolves.toBe(false); + expect(cacheStorageService.get).toHaveBeenCalledTimes(1); + + dateNow.mockReturnValue(1_000 + APPLICATION_KILL_SWITCH_LOCAL_CACHE_TTL_MS); + + await expect( + applicationStopService.isApplicationStopped( + APPLICATION_UNIVERSAL_IDENTIFIER, + ), + ).resolves.toBe(true); + expect(cacheStorageService.get).toHaveBeenCalledTimes(2); }); - describe('stopApplication', () => { - it('should set stoppedAt and invalidate the workspace application cache', async () => { - applicationRepository.findOne.mockResolvedValue({ - id: APPLICATION_ID, - universalIdentifier: APPLICATION_UNIVERSAL_IDENTIFIER, - workspaceId: WORKSPACE_ID, - stoppedAt: null, - }); + it('deduplicates concurrent Redis reads', async () => { + cacheStorageService.get.mockResolvedValue(undefined); - const application = await applicationStopService.stopApplication({ - workspaceId: WORKSPACE_ID, - applicationUniversalIdentifier: APPLICATION_UNIVERSAL_IDENTIFIER, - }); - - expect(applicationRepository.update).toHaveBeenCalledWith( - APPLICATION_ID, - { stoppedAt: expect.any(Date) }, - ); - expect(workspaceCacheService.invalidateAndRecompute).toHaveBeenCalledWith( - WORKSPACE_ID, - ['flatApplicationMaps'], - ); - expect(application.stoppedAt).toBeInstanceOf(Date); - }); - - it('should throw when the application does not exist', async () => { - applicationRepository.findOne.mockResolvedValue(null); - - await expect( - applicationStopService.stopApplication({ - workspaceId: WORKSPACE_ID, - applicationUniversalIdentifier: APPLICATION_UNIVERSAL_IDENTIFIER, - }), - ).rejects.toThrow(ApplicationException); - - expect(applicationRepository.update).not.toHaveBeenCalled(); - expect( - workspaceCacheService.invalidateAndRecompute, - ).not.toHaveBeenCalled(); - }); - }); - - describe('startApplication', () => { - it('should clear stoppedAt and invalidate the workspace application cache', async () => { - applicationRepository.findOne.mockResolvedValue({ - id: APPLICATION_ID, - universalIdentifier: APPLICATION_UNIVERSAL_IDENTIFIER, - workspaceId: WORKSPACE_ID, - stoppedAt: new Date(), - }); - - const application = await applicationStopService.startApplication({ - workspaceId: WORKSPACE_ID, - applicationUniversalIdentifier: APPLICATION_UNIVERSAL_IDENTIFIER, - }); - - expect(applicationRepository.update).toHaveBeenCalledWith( - APPLICATION_ID, - { stoppedAt: null }, - ); - expect(workspaceCacheService.invalidateAndRecompute).toHaveBeenCalledWith( - WORKSPACE_ID, - ['flatApplicationMaps'], - ); - expect(application.stoppedAt).toBeNull(); - }); - - it('should throw when the application does not exist', async () => { - applicationRepository.findOne.mockResolvedValue(null); - - await expect( - applicationStopService.startApplication({ - workspaceId: WORKSPACE_ID, - applicationUniversalIdentifier: APPLICATION_UNIVERSAL_IDENTIFIER, - }), - ).rejects.toThrow(ApplicationException); - - expect(applicationRepository.update).not.toHaveBeenCalled(); - expect( - workspaceCacheService.invalidateAndRecompute, - ).not.toHaveBeenCalled(); - }); - }); - - describe('stopApplicationRegistration', () => { - it('should set stoppedAt on the registration and report the installed application count', async () => { - applicationRegistrationRepository.findOne.mockResolvedValue({ - id: APPLICATION_REGISTRATION_ID, - universalIdentifier: APPLICATION_REGISTRATION_UNIVERSAL_IDENTIFIER, - name: 'My App', - stoppedAt: null, - }); - applicationRepository.count.mockResolvedValue(3); - - const { applicationRegistration, installedApplicationCount } = - await applicationStopService.stopApplicationRegistration({ - applicationRegistrationUniversalIdentifier: - APPLICATION_REGISTRATION_UNIVERSAL_IDENTIFIER, - }); - - expect(applicationRegistrationRepository.update).toHaveBeenCalledWith( - APPLICATION_REGISTRATION_ID, - { stoppedAt: expect.any(Date) }, - ); - expect(applicationRegistration.stoppedAt).toBeInstanceOf(Date); - expect(installedApplicationCount).toBe(3); - }); - - it('should throw when the registration does not exist', async () => { - applicationRegistrationRepository.findOne.mockResolvedValue(null); - - await expect( - applicationStopService.stopApplicationRegistration({ - applicationRegistrationUniversalIdentifier: - APPLICATION_REGISTRATION_UNIVERSAL_IDENTIFIER, - }), - ).rejects.toThrow(ApplicationRegistrationException); - - expect(applicationRegistrationRepository.update).not.toHaveBeenCalled(); - }); - }); - - describe('startApplicationRegistration', () => { - it('should clear stoppedAt on the registration', async () => { - applicationRegistrationRepository.findOne.mockResolvedValue({ - id: APPLICATION_REGISTRATION_ID, - universalIdentifier: APPLICATION_REGISTRATION_UNIVERSAL_IDENTIFIER, - name: 'My App', - stoppedAt: new Date(), - }); - applicationRepository.count.mockResolvedValue(1); - - const { applicationRegistration } = - await applicationStopService.startApplicationRegistration({ - applicationRegistrationUniversalIdentifier: - APPLICATION_REGISTRATION_UNIVERSAL_IDENTIFIER, - }); - - expect(applicationRegistrationRepository.update).toHaveBeenCalledWith( - APPLICATION_REGISTRATION_ID, - { stoppedAt: null }, - ); - expect(applicationRegistration.stoppedAt).toBeNull(); - }); - - it('should throw when the registration does not exist', async () => { - applicationRegistrationRepository.findOne.mockResolvedValue(null); - - await expect( - applicationStopService.startApplicationRegistration({ - applicationRegistrationUniversalIdentifier: - APPLICATION_REGISTRATION_UNIVERSAL_IDENTIFIER, - }), - ).rejects.toThrow(ApplicationRegistrationException); - - expect(applicationRegistrationRepository.update).not.toHaveBeenCalled(); - }); - }); - - describe('isApplicationRegistrationStopped', () => { - it('should return true when a stopped registration matches', async () => { - applicationRegistrationRepository.count.mockResolvedValue(1); - - await expect( - applicationStopService.isApplicationRegistrationStopped( - APPLICATION_REGISTRATION_ID, + await Promise.all( + Array.from({ length: 10 }, () => + applicationStopService.isApplicationStopped( + APPLICATION_UNIVERSAL_IDENTIFIER, ), - ).resolves.toBe(true); - }); + ), + ); - it('should return false when the registration is not stopped', async () => { - applicationRegistrationRepository.count.mockResolvedValue(0); + expect(cacheStorageService.get).toHaveBeenCalledTimes(1); + }); - await expect( - applicationStopService.isApplicationRegistrationStopped( - APPLICATION_REGISTRATION_ID, + it('fails open when the cache cannot be read', async () => { + cacheStorageService.get.mockRejectedValue(new Error('Redis unavailable')); + + await expect( + Promise.all([ + applicationStopService.isApplicationStopped( + APPLICATION_UNIVERSAL_IDENTIFIER, ), - ).resolves.toBe(false); - }); + applicationStopService.isApplicationStopped( + APPLICATION_UNIVERSAL_IDENTIFIER, + ), + ]), + ).resolves.toEqual([false, false]); + expect(cacheStorageService.get).toHaveBeenCalledTimes(1); }); }); diff --git a/packages/twenty-server/src/engine/core-modules/application/application-manifest/application-sync.service.ts b/packages/twenty-server/src/engine/core-modules/application/application-manifest/application-sync.service.ts index 4434ba5733..e1c31a2070 100644 --- a/packages/twenty-server/src/engine/core-modules/application/application-manifest/application-sync.service.ts +++ b/packages/twenty-server/src/engine/core-modules/application/application-manifest/application-sync.service.ts @@ -162,7 +162,6 @@ export class ApplicationSyncService { settingsCustomTabFrontComponentId: null, canBeUninstalled: true, autoUpgrade: false, - stoppedAt: null, isSdkLayerStale: false, sdkClientCoreChecksum: null, applicationRegistrationId: null, diff --git a/packages/twenty-server/src/engine/core-modules/application/application-registration/application-registration.entity.ts b/packages/twenty-server/src/engine/core-modules/application/application-registration/application-registration.entity.ts index f58f3d46a1..4c671d0456 100644 --- a/packages/twenty-server/src/engine/core-modules/application/application-registration/application-registration.entity.ts +++ b/packages/twenty-server/src/engine/core-modules/application/application-registration/application-registration.entity.ts @@ -138,16 +138,6 @@ export class ApplicationRegistrationEntity { @Column({ type: 'boolean', default: false }) isPreInstalled: boolean; - // Emergency kill switch: while set, every logic function execution of every - // application installed from this registration is blocked, across all - // workspaces (see `application-registration:stop` command). - @Column({ nullable: true, type: 'timestamptz' }) - @WasIntroducedInUpgrade({ - upgradeCommandName: - '2.24.0_AddStoppedAtToApplicationAndApplicationRegistrationFastInstanceCommand_1784734278506', - }) - stoppedAt: Date | null; - @Column({ type: 'jsonb', nullable: true }) manifest: Manifest | null; diff --git a/packages/twenty-server/src/engine/core-modules/application/application-stop.service.ts b/packages/twenty-server/src/engine/core-modules/application/application-stop.service.ts index 7c5a97edaf..0c7d45fe82 100644 --- a/packages/twenty-server/src/engine/core-modules/application/application-stop.service.ts +++ b/packages/twenty-server/src/engine/core-modules/application/application-stop.service.ts @@ -1,175 +1,55 @@ import { Injectable } from '@nestjs/common'; -import { InjectRepository } from '@nestjs/typeorm'; -import { isDefined } from 'twenty-shared/utils'; -import { IsNull, Not, Repository } 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 { PromiseMemoizer } from 'src/engine/twenty-orm/storage/promise-memoizer.storage'; -import { ApplicationRegistrationEntity } from 'src/engine/core-modules/application/application-registration/application-registration.entity'; -import { - ApplicationRegistrationException, - ApplicationRegistrationExceptionCode, -} from 'src/engine/core-modules/application/application-registration/application-registration.exception'; -import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity'; -import { - ApplicationException, - ApplicationExceptionCode, -} from 'src/engine/core-modules/application/application.exception'; -import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; +export const APPLICATION_KILL_SWITCH_LOCAL_CACHE_TTL_MS = 60_000; + +type ApplicationKillSwitchCacheEntry = { + isStopped: boolean; +}; -// Emergency kill switch for applications DDoSing the server or the database. -// Workspace level: stops one installed application. Server level: stops every -// application installed from an application registration, in all workspaces. @Injectable() export class ApplicationStopService { + private readonly memoizer = + new PromiseMemoizer( + APPLICATION_KILL_SWITCH_LOCAL_CACHE_TTL_MS, + ); + constructor( - @InjectRepository(ApplicationEntity) - private readonly applicationRepository: Repository, - @InjectRepository(ApplicationRegistrationEntity) - private readonly applicationRegistrationRepository: Repository, - private readonly workspaceCacheService: WorkspaceCacheService, + @InjectCacheStorage(CacheStorageNamespace.ModuleApplications) + private readonly cacheStorageService: CacheStorageService, ) {} - async stopApplication({ - workspaceId, - applicationUniversalIdentifier, - }: { - workspaceId: string; - applicationUniversalIdentifier: string; - }): Promise { - return this.setApplicationStoppedAt({ - workspaceId, - applicationUniversalIdentifier, - stoppedAt: new Date(), - }); - } - - async startApplication({ - workspaceId, - applicationUniversalIdentifier, - }: { - workspaceId: string; - applicationUniversalIdentifier: string; - }): Promise { - return this.setApplicationStoppedAt({ - workspaceId, - applicationUniversalIdentifier, - stoppedAt: null, - }); - } - - async stopApplicationRegistration({ - applicationRegistrationUniversalIdentifier, - }: { - applicationRegistrationUniversalIdentifier: string; - }): Promise<{ - applicationRegistration: ApplicationRegistrationEntity; - installedApplicationCount: number; - }> { - return this.setApplicationRegistrationStoppedAt({ - applicationRegistrationUniversalIdentifier, - stoppedAt: new Date(), - }); - } - - async startApplicationRegistration({ - applicationRegistrationUniversalIdentifier, - }: { - applicationRegistrationUniversalIdentifier: string; - }): Promise<{ - applicationRegistration: ApplicationRegistrationEntity; - installedApplicationCount: number; - }> { - return this.setApplicationRegistrationStoppedAt({ - applicationRegistrationUniversalIdentifier, - stoppedAt: null, - }); - } - - async isApplicationRegistrationStopped( - applicationRegistrationId: string, + async isApplicationStopped( + applicationUniversalIdentifier: string, ): Promise { - const stoppedApplicationRegistrationCount = - await this.applicationRegistrationRepository.count({ - where: { - id: applicationRegistrationId, - stoppedAt: Not(IsNull()), - }, - }); - - return stoppedApplicationRegistrationCount > 0; - } - - private async setApplicationStoppedAt({ - workspaceId, - applicationUniversalIdentifier, - stoppedAt, - }: { - workspaceId: string; - applicationUniversalIdentifier: string; - stoppedAt: Date | null; - }): Promise { - const application = await this.applicationRepository.findOne({ - where: { - workspaceId, - universalIdentifier: applicationUniversalIdentifier, - }, - }); - - if (!isDefined(application)) { - throw new ApplicationException( - `Application ${applicationUniversalIdentifier} not found in workspace ${workspaceId}`, - ApplicationExceptionCode.APPLICATION_NOT_FOUND, - ); - } - - await this.applicationRepository.update(application.id, { stoppedAt }); - - // The executor reads applications from the workspace cache, so the flag - // only takes effect once the cached flat application maps are rebuilt. - await this.workspaceCacheService.invalidateAndRecompute( - application.workspaceId, - ['flatApplicationMaps'], + const cacheEntry = await this.memoizer.memoizePromiseAndExecute( + `application-${applicationUniversalIdentifier}`, + () => this.readKillSwitch(applicationUniversalIdentifier), ); - return { ...application, stoppedAt }; + return cacheEntry?.isStopped ?? false; } - private async setApplicationRegistrationStoppedAt({ - applicationRegistrationUniversalIdentifier, - stoppedAt, - }: { - applicationRegistrationUniversalIdentifier: string; - stoppedAt: Date | null; - }): Promise<{ - applicationRegistration: ApplicationRegistrationEntity; - installedApplicationCount: number; - }> { - const applicationRegistration = - await this.applicationRegistrationRepository.findOne({ - where: { - universalIdentifier: applicationRegistrationUniversalIdentifier, - }, - }); + private async readKillSwitch( + applicationUniversalIdentifier: string, + ): Promise { + try { + const isStopped = + (await this.cacheStorageService.get( + this.getKillSwitchKey(applicationUniversalIdentifier), + )) !== undefined; - if (!isDefined(applicationRegistration)) { - throw new ApplicationRegistrationException( - `Application registration ${applicationRegistrationUniversalIdentifier} not found`, - ApplicationRegistrationExceptionCode.APPLICATION_REGISTRATION_NOT_FOUND, - ); + return { isStopped }; + } catch { + return { isStopped: false }; } + } - await this.applicationRegistrationRepository.update( - applicationRegistration.id, - { stoppedAt }, - ); - - const installedApplicationCount = await this.applicationRepository.count({ - where: { applicationRegistrationId: applicationRegistration.id }, - }); - - return { - applicationRegistration: { ...applicationRegistration, stoppedAt }, - installedApplicationCount, - }; + private getKillSwitchKey(applicationUniversalIdentifier: string): string { + return `kill-switch:${applicationUniversalIdentifier}`; } } diff --git a/packages/twenty-server/src/engine/core-modules/application/application.entity.ts b/packages/twenty-server/src/engine/core-modules/application/application.entity.ts index 91113389f8..9975c20b33 100644 --- a/packages/twenty-server/src/engine/core-modules/application/application.entity.ts +++ b/packages/twenty-server/src/engine/core-modules/application/application.entity.ts @@ -131,15 +131,6 @@ export class ApplicationEntity extends WorkspaceRelatedEntity { }) autoUpgrade: boolean; - // Emergency kill switch: while set, every logic function execution of this - // application is blocked in this workspace (see `application:stop` command). - @Column({ nullable: true, type: 'timestamptz' }) - @WasIntroducedInUpgrade({ - upgradeCommandName: - '2.24.0_AddStoppedAtToApplicationAndApplicationRegistrationFastInstanceCommand_1784734278506', - }) - stoppedAt: Date | null; - @Column({ nullable: false, type: 'boolean', default: false }) isSdkLayerStale: boolean; diff --git a/packages/twenty-server/src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum.ts b/packages/twenty-server/src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum.ts index ee9575f141..c0397c6725 100644 --- a/packages/twenty-server/src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum.ts +++ b/packages/twenty-server/src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum.ts @@ -1,4 +1,5 @@ export enum CacheStorageNamespace { + ModuleApplications = 'module:applications', ModuleMessaging = 'module:messaging', ModuleEmailing = 'module:emailing', ModuleCalendar = 'module:calendar', diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-executor/logic-function-executor.service.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-executor/logic-function-executor.service.ts index c579efdbff..d29e54b9ef 100644 --- a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-executor/logic-function-executor.service.ts +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-executor/logic-function-executor.service.ts @@ -239,27 +239,16 @@ export class LogicFunctionExecutorService { return driver.transpile(params); } - // Emergency kill switch checked on every execution: an application can be - // stopped for one workspace (application.stoppedAt) or server-wide for all - // workspaces (applicationRegistration.stoppedAt) when it degrades production. private async assertApplicationNotStopped( flatApplication: FlatApplication, ): Promise { - if (isDefined(flatApplication.stoppedAt)) { - throw new LogicFunctionException( - `Application ${flatApplication.id} is stopped in workspace ${flatApplication.workspaceId}`, - LogicFunctionExceptionCode.LOGIC_FUNCTION_DISABLED, - ); - } - if ( - isDefined(flatApplication.applicationRegistrationId) && - (await this.applicationStopService.isApplicationRegistrationStopped( - flatApplication.applicationRegistrationId, - )) + await this.applicationStopService.isApplicationStopped( + flatApplication.universalIdentifier, + ) ) { throw new LogicFunctionException( - `Application registration ${flatApplication.applicationRegistrationId} is stopped server-wide`, + `Application ${flatApplication.universalIdentifier} is temporarily stopped`, LogicFunctionExceptionCode.LOGIC_FUNCTION_DISABLED, ); } diff --git a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job.ts b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job.ts index cf23e5bdf5..61f5d855a3 100644 --- a/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job.ts +++ b/packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job.ts @@ -1,4 +1,4 @@ -import { Logger, Scope } from '@nestjs/common'; +import { Scope } from '@nestjs/common'; import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; @@ -22,8 +22,6 @@ export type LogicFunctionTriggerJobData = { scope: Scope.REQUEST, }) export class LogicFunctionTriggerJob { - private readonly logger = new Logger(LogicFunctionTriggerJob.name); - constructor( private readonly logicFunctionExecutorService: LogicFunctionExecutorService, ) {} @@ -47,10 +45,6 @@ export class LogicFunctionTriggerJob { error instanceof LogicFunctionException && error.code === LogicFunctionExceptionCode.LOGIC_FUNCTION_DISABLED ) { - this.logger.warn( - `Skipping execution of logic function ${logicFunctionPayload.logicFunctionId} in workspace ${logicFunctionPayload.workspaceId}: ${error.message}`, - ); - return; } diff --git a/packages/twenty-server/src/engine/metadata-modules/flat-field-metadata/__tests__/morph-relation-from-create-field-input-to-flat-field-metadatas-to-create.spec.ts b/packages/twenty-server/src/engine/metadata-modules/flat-field-metadata/__tests__/morph-relation-from-create-field-input-to-flat-field-metadatas-to-create.spec.ts index 3e8c8e9c49..35ee4695f1 100644 --- a/packages/twenty-server/src/engine/metadata-modules/flat-field-metadata/__tests__/morph-relation-from-create-field-input-to-flat-field-metadatas-to-create.spec.ts +++ b/packages/twenty-server/src/engine/metadata-modules/flat-field-metadata/__tests__/morph-relation-from-create-field-input-to-flat-field-metadatas-to-create.spec.ts @@ -42,7 +42,6 @@ const MOCK_FLAT_APPLICATION: FlatApplication = { settingsCustomTabFrontComponentId: null, canBeUninstalled: false, autoUpgrade: false, - stoppedAt: null, applicationRegistrationId: null, primaryPublicDomainId: null, createdAt: new Date(), diff --git a/packages/twenty-server/src/engine/metadata-modules/flat-field-metadata/utils/__tests__/generate-morph-or-relation-flat-field-metadata-pair.spec.ts b/packages/twenty-server/src/engine/metadata-modules/flat-field-metadata/utils/__tests__/generate-morph-or-relation-flat-field-metadata-pair.spec.ts index 1c0bee3e64..c31e2643e7 100644 --- a/packages/twenty-server/src/engine/metadata-modules/flat-field-metadata/utils/__tests__/generate-morph-or-relation-flat-field-metadata-pair.spec.ts +++ b/packages/twenty-server/src/engine/metadata-modules/flat-field-metadata/utils/__tests__/generate-morph-or-relation-flat-field-metadata-pair.spec.ts @@ -37,7 +37,6 @@ const MOCK_FLAT_APPLICATION: FlatApplication = { settingsCustomTabFrontComponentId: null, canBeUninstalled: false, autoUpgrade: false, - stoppedAt: null, applicationRegistrationId: null, primaryPublicDomainId: null, createdAt: new Date(),