diff --git a/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts b/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts index 06f8ee8863a..81d971fdca1 100644 --- a/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts +++ b/packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts @@ -32,6 +32,10 @@ const newJob = (name: string): NewScheduledJob => ({ misfireGraceSeconds: 60, }); +/** A row as the post-insert read-back returns it: id, name and the owner it belongs to. */ +const insertedRow = (id: number, name: string, owner = OWNER): ScheduledJob => + ({ id, name, ...owner }) as ScheduledJob; + /** A chainable insert query-builder mock; `execute` is set per test. */ const insertQb = () => ({ insert: vi.fn().mockReturnThis(), @@ -169,8 +173,8 @@ describe('ScheduledJobRepository', () => { entityManager.createQueryBuilder.mockReturnValue(qb as never); // Read-back deliberately out of input order to prove the result is reordered. entityManager.find.mockResolvedValueOnce([ - { id: 20, name: 'wf:node:1' } as ScheduledJob, - { id: 10, name: 'wf:node:0' } as ScheduledJob, + insertedRow(20, 'wf:node:1'), + insertedRow(10, 'wf:node:0'), ]); const ids = await repository.insertMany(entityManager, jobs); @@ -180,7 +184,7 @@ describe('ScheduledJobRepository', () => { expect(qb.returning).not.toHaveBeenCalled(); expect(entityManager.find).toHaveBeenCalledWith(ScheduledJob, { where: { name: In(['wf:node:0', 'wf:node:1']) }, - select: { id: true, name: true }, + select: { id: true, name: true, ownerType: true, ownerId: true, ownerMemberId: true }, }); expect(ids).toEqual([10, 20]); }); @@ -192,11 +196,8 @@ describe('ScheduledJobRepository', () => { entityManager.createQueryBuilder.mockReturnValue(qb as never); // Read-back is chunked too; each chunk resolves only its own rows. entityManager.find - .mockResolvedValueOnce([ - { id: 10, name: 'wf:node:0' } as ScheduledJob, - { id: 20, name: 'wf:node:1' } as ScheduledJob, - ]) - .mockResolvedValueOnce([{ id: 30, name: 'wf:node:2' } as ScheduledJob]); + .mockResolvedValueOnce([insertedRow(10, 'wf:node:0'), insertedRow(20, 'wf:node:1')]) + .mockResolvedValueOnce([insertedRow(30, 'wf:node:2')]); // 3 jobs at 2/chunk -> two insert statements and two read-back queries. const ids = await repository.insertMany(entityManager, jobs, 2); @@ -206,11 +207,11 @@ describe('ScheduledJobRepository', () => { expect(qb.values).toHaveBeenNthCalledWith(2, [jobs[2]]); expect(entityManager.find).toHaveBeenNthCalledWith(1, ScheduledJob, { where: { name: In(['wf:node:0', 'wf:node:1']) }, - select: { id: true, name: true }, + select: { id: true, name: true, ownerType: true, ownerId: true, ownerMemberId: true }, }); expect(entityManager.find).toHaveBeenNthCalledWith(2, ScheduledJob, { where: { name: In(['wf:node:2']) }, - select: { id: true, name: true }, + select: { id: true, name: true, ownerType: true, ownerId: true, ownerMemberId: true }, }); expect(ids).toEqual([10, 20, 30]); }); @@ -223,12 +224,8 @@ describe('ScheduledJobRepository', () => { qb.execute.mockResolvedValue(undefined); entityManager.createQueryBuilder.mockReturnValue(qb as never); entityManager.find - .mockResolvedValueOnce( - jobs.slice(0, 500).map((job, i) => ({ id: i + 1, name: job.name }) as ScheduledJob), - ) - .mockResolvedValueOnce( - jobs.slice(500).map((job, i) => ({ id: 501 + i, name: job.name }) as ScheduledJob), - ); + .mockResolvedValueOnce(jobs.slice(0, 500).map((job, i) => insertedRow(i + 1, job.name))) + .mockResolvedValueOnce(jobs.slice(500).map((job, i) => insertedRow(501 + i, job.name))); const ids = await repository.insertMany(entityManager, jobs, 10_000_000); @@ -243,8 +240,8 @@ describe('ScheduledJobRepository', () => { qb.execute.mockResolvedValue(undefined); entityManager.createQueryBuilder.mockReturnValue(qb as never); entityManager.find.mockResolvedValueOnce([ - { id: 10, name: 'wf:node:0' } as ScheduledJob, - { id: 20, name: 'wf:node:1' } as ScheduledJob, + insertedRow(10, 'wf:node:0'), + insertedRow(20, 'wf:node:1'), ]); const ids = await postgresRepository.insertMany(entityManager, jobs); @@ -252,7 +249,7 @@ describe('ScheduledJobRepository', () => { expect(qb.returning).not.toHaveBeenCalled(); expect(entityManager.find).toHaveBeenCalledWith(ScheduledJob, { where: { name: In(['wf:node:0', 'wf:node:1']) }, - select: { id: true, name: true }, + select: { id: true, name: true, ownerType: true, ownerId: true, ownerMemberId: true }, }); expect(ids).toEqual([10, 20]); }); @@ -265,8 +262,8 @@ describe('ScheduledJobRepository', () => { // `wf:node:1` was already held by another writer; orIgnore skipped our row, // but the read-back still finds every input name (id 99 is the other writer's). entityManager.find.mockResolvedValueOnce([ - { id: 10, name: 'wf:node:0' } as ScheduledJob, - { id: 99, name: 'wf:node:1' } as ScheduledJob, + insertedRow(10, 'wf:node:0'), + insertedRow(99, 'wf:node:1'), ]); const ids = await repository.insertMany(entityManager, jobs); @@ -274,6 +271,22 @@ describe('ScheduledJobRepository', () => { expect(ids).toEqual([10, 99]); }); + it('throws when the row holding a name belongs to another owner', async () => { + // Job names are scoped by convention, not by the schema. Returning the other + // owner's id would have this call reseed and reschedule their job. + const jobs = [newJob('wf:node:0')]; + const qb = insertQb(); + qb.execute.mockResolvedValue(undefined); + entityManager.createQueryBuilder.mockReturnValue(qb as never); + entityManager.find.mockResolvedValueOnce([ + insertedRow(10, 'wf:node:0', { ...OWNER, ownerMemberId: 'other-node' }), + ]); + + await expect(repository.insertMany(entityManager, jobs)).rejects.toThrow( + 'Scheduled job name is already taken by another owner', + ); + }); + it('throws when a name has no row after insert', async () => { const jobs = [newJob('wf:node:0'), newJob('wf:node:1')]; const qb = insertQb(); @@ -281,7 +294,7 @@ describe('ScheduledJobRepository', () => { entityManager.createQueryBuilder.mockReturnValue(qb as never); // A name unexpectedly missing from the read-back must fail loud rather than // return a short array that would misalign the caller's index-based zip. - entityManager.find.mockResolvedValueOnce([{ id: 10, name: 'wf:node:0' } as ScheduledJob]); + entityManager.find.mockResolvedValueOnce([insertedRow(10, 'wf:node:0')]); await expect(repository.insertMany(entityManager, jobs)).rejects.toThrow(UnexpectedError); }); diff --git a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts index 4618fda141e..fbfd2418381 100644 --- a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts +++ b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts @@ -167,6 +167,11 @@ export class ScheduledJobRepository extends Repository { * would come back without an id. `name` is unique and every input job has a row * once the insert returns (ours, or the concurrent writer's), so the read-back * yields exactly one id per job. + * + * @throws {UnexpectedError} when a name's row belongs to a different owner. Job + * names are scoped by convention, not by the schema, so a collision would + * otherwise have `orIgnore` hand back another owner's id and let this call + * reschedule their job. */ async insertMany( manager: EntityManager, @@ -198,7 +203,7 @@ export class ScheduledJobRepository extends Repository { for (let start = 0; start < names.length; start += size) { const found = await manager.find(ScheduledJob, { where: { name: In(names.slice(start, start + size)) }, - select: { id: true, name: true }, + select: { id: true, name: true, ownerType: true, ownerId: true, ownerMemberId: true }, }); rows.push(...found); } @@ -500,22 +505,32 @@ export class ScheduledJobRepository extends Repository { } } +/** A read-back row, carrying the owner the id is only valid under. */ +type InsertedRow = Pick; + /** - * Ids for `jobs`, in input order, from rows carrying their `{ id, name }`. + * Ids for `jobs`, in input order, from the rows read back after the insert. * Throws on a name with no row: the caller zips the result back to `jobs` by - * index, so a gap would misalign every id after it. + * index, so a gap would misalign every id after it. Throws too on a row owned by + * someone else, rather than handing back an id this caller has no claim to. */ -function orderIdsByName( - rows: Array<{ id: number; name: string }>, - jobs: NewScheduledJob[], -): number[] { - const idByName = new Map(rows.map((row) => [row.name, row.id])); +function orderIdsByName(rows: InsertedRow[], jobs: NewScheduledJob[]): number[] { + const rowByName = new Map(rows.map((row) => [row.name, row])); return jobs.map((job) => { - const id = idByName.get(job.name); - if (id === undefined) { + const row = rowByName.get(job.name); + if (row === undefined) { throw new UnexpectedError(`No row found for scheduled job "${job.name}" after insert`); } - return id; + if ( + row.ownerType !== job.ownerType || + row.ownerId !== job.ownerId || + row.ownerMemberId !== job.ownerMemberId + ) { + throw new UnexpectedError('Scheduled job name is already taken by another owner', { + extra: { name: job.name, ownerType: job.ownerType, ownerId: job.ownerId }, + }); + } + return row.id; }); } diff --git a/packages/cli/src/workflows/__tests__/workflow.service.test.ts b/packages/cli/src/workflows/__tests__/workflow.service.test.ts index 326d2c477ae..ee1c88923c0 100644 --- a/packages/cli/src/workflows/__tests__/workflow.service.test.ts +++ b/packages/cli/src/workflows/__tests__/workflow.service.test.ts @@ -36,8 +36,10 @@ import type { PolicyCleared } from '@n8n/decorators'; import { userHasScopes } from '@/permissions.ee/check-access'; import type { PolicyEnforcementService } from '@/policy/policy-enforcement.service'; import { PolicyViolationError } from '@/policy/policy-violation.error'; +import type { DurableJobProvisioner } from '@/scheduling/durable-job-provisioner'; import type { PollTriggerJobRegistrar } from '@/scheduling/poll-trigger-node/poll-trigger-job-registrar'; import type { ScheduleTriggerJobRegistrar } from '@/scheduling/schedule-trigger-node/schedule-trigger-job-registrar'; +import type { WorkflowScheduledJobOwner } from '@/scheduling/workflow-scheduled-job-owner'; import type { OwnershipService } from '@/services/ownership.service'; import type { RoleService } from '@/services/role.service'; import type { TagService } from '@/services/tag.service'; @@ -120,6 +122,8 @@ describe('WorkflowService', () => { mock(), // workflowPublicationNotifier mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + mock(), // workflowScheduledJobOwner + mock(), // durableJobProvisioner mock(), // workflowPublishedVersionRepository mock(), // workflowHookContextService mock(), // workflowPublishGuard @@ -486,6 +490,8 @@ describe('WorkflowService', () => { mock(), // workflowPublicationNotifier mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + mock(), // workflowScheduledJobOwner + mock(), // durableJobProvisioner mock(), // workflowPublishedVersionRepository workflowHookContextServiceMock, // workflowHookContextService mock(), // workflowPublishGuard @@ -1267,6 +1273,8 @@ describe('WorkflowService', () => { mock(), // workflowPublicationNotifier scheduleTriggerJobRegistrarMock, // scheduleTriggerJobRegistrar pollTriggerJobRegistrarMock, // pollTriggerJobRegistrar + mock(), // workflowScheduledJobOwner + mock(), // durableJobProvisioner mock(), // workflowPublishedVersionRepository workflowHookContextServiceMock, // workflowHookContextService workflowPublishGuardMock, // workflowPublishGuard @@ -1975,6 +1983,8 @@ describe('WorkflowService', () => { mock(), // workflowPublicationNotifier mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + mock(), // workflowScheduledJobOwner + mock(), // durableJobProvisioner mock(), // workflowPublishedVersionRepository mock(), // workflowHookContextService mock(), // workflowPublishGuard @@ -2041,8 +2051,12 @@ describe('WorkflowService', () => { let workflowPublishedVersionRepositoryMock: MockProxy; let workflowMutationHooksMock: MockProxy; let ownershipServiceMock: MockProxy; + let workflowScheduledJobOwnerMock: MockProxy; + let durableJobProvisionerMock: MockProxy; + let trxMock: MockProxy; const WORKFLOW_ID = 'workflow-1'; + const WORKFLOW_OWNER_REF = { ownerType: 'workflow', ownerId: WORKFLOW_ID }; function makeWorkflowEntity(overrides: Partial = {}): WorkflowEntity { const workflow = new WorkflowEntity(); @@ -2065,10 +2079,18 @@ describe('WorkflowService', () => { workflowMutationHooksMock = mock(); workflowPublishedVersionRepositoryMock = mock(); workflowPublishedVersionRepositoryMock.getPublishedVersionId.mockResolvedValue(null); + workflowScheduledJobOwnerMock = mock(); + workflowScheduledJobOwnerMock.ref.mockReturnValue(WORKFLOW_OWNER_REF); + durableJobProvisionerMock = mock(); globalConfigMock = mock({ workflows: mock({ useWorkflowPublicationService: true }), }); + trxMock = mock(); + workflowRepositoryMock.runInTransaction.mockImplementation( + async (ctx, runInTransaction) => await runInTransaction(trxMock, ctx), + ); + workflowService = new WorkflowService( mock(), // logger mock(), // sharedWorkflowRepository @@ -2097,6 +2119,8 @@ describe('WorkflowService', () => { mock(), // workflowPublicationNotifier mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + workflowScheduledJobOwnerMock, // workflowScheduledJobOwner + durableJobProvisionerMock, // durableJobProvisioner workflowPublishedVersionRepositoryMock, // workflowPublishedVersionRepository mock(), // workflowHookContextService mock(), // workflowPublishGuard @@ -2114,7 +2138,7 @@ describe('WorkflowService', () => { ConflictError, ); - expect(workflowRepositoryMock.delete).not.toHaveBeenCalled(); + expect(trxMock.delete).not.toHaveBeenCalled(); }); test('throws ConflictError while the published-version mapping still exists', async () => { @@ -2127,7 +2151,7 @@ describe('WorkflowService', () => { 'Workflow is still being unpublished. Please try again in a few moments.', ); - expect(workflowRepositoryMock.delete).not.toHaveBeenCalled(); + expect(trxMock.delete).not.toHaveBeenCalled(); }); test('deletes a workflow whose active version was set while publication service was off', async () => { @@ -2138,7 +2162,7 @@ describe('WorkflowService', () => { await workflowService.delete(mock(), WORKFLOW_ID, true); expect(activeWorkflowManagerMock.remove).toHaveBeenCalledWith(WORKFLOW_ID); - expect(workflowRepositoryMock.delete).toHaveBeenCalledWith(WORKFLOW_ID); + expect(trxMock.delete).toHaveBeenCalledWith(WorkflowEntity, { id: WORKFLOW_ID }); }); test('deletes an unpublished workflow when publication service is on', async () => { @@ -2147,7 +2171,38 @@ describe('WorkflowService', () => { await workflowService.delete(mock(), WORKFLOW_ID, true); - expect(workflowRepositoryMock.delete).toHaveBeenCalledWith(WORKFLOW_ID); + expect(trxMock.delete).toHaveBeenCalledWith(WorkflowEntity, { id: WORKFLOW_ID }); + }); + + // Nothing in the database removes a workflow's scheduled jobs with it, so the + // delete has to do it in the same transaction as the row delete. A deprovision + // of its own would strip the schedules of a workflow that survives a failed + // delete. + test('deprovisions the scheduled jobs the workflow owned in the row delete transaction', async () => { + const workflow = makeWorkflowEntity({ isArchived: true, activeVersionId: null }); + workflowFinderServiceMock.findWorkflowForUser.mockResolvedValue(workflow); + + await workflowService.delete(mock(), WORKFLOW_ID, true); + + expect(workflowScheduledJobOwnerMock.ref).toHaveBeenCalledWith(WORKFLOW_ID); + expect( + durableJobProvisionerMock.deprovisionOwnerInTransaction, + ).toHaveBeenCalledExactlyOnceWith(trxMock, WORKFLOW_OWNER_REF); + expect(durableJobProvisionerMock.deprovisionOwner).not.toHaveBeenCalled(); + expect( + durableJobProvisionerMock.deprovisionOwnerInTransaction.mock.invocationCallOrder[0], + ).toBeLessThan(trxMock.delete.mock.invocationCallOrder[0]); + }); + + test('deprovisions no scheduled jobs when deletion is rejected', async () => { + const workflow = makeWorkflowEntity({ activeVersionId: 'v1' }); + workflowFinderServiceMock.findWorkflowForUser.mockResolvedValue(workflow); + + await expect(workflowService.delete(mock(), WORKFLOW_ID, true)).rejects.toBeInstanceOf( + ConflictError, + ); + + expect(durableJobProvisionerMock.deprovisionOwnerInTransaction).not.toHaveBeenCalled(); }); test('runs the beforeWorkflowDeleted lifecycle hook before the row delete', async () => { @@ -2162,7 +2217,7 @@ describe('WorkflowService', () => { ); expect( workflowMutationHooksMock.beforeWorkflowDeleted.mock.invocationCallOrder[0], - ).toBeLessThan(workflowRepositoryMock.delete.mock.invocationCallOrder[0]); + ).toBeLessThan(trxMock.delete.mock.invocationCallOrder[0]); }); test('does not run the beforeWorkflowDeleted lifecycle hook when deletion is rejected', async () => { @@ -2201,7 +2256,7 @@ describe('WorkflowService', () => { ]); expect( workflowMutationHooksMock.afterWorkflowsDeleted.mock.invocationCallOrder[0], - ).toBeGreaterThan(workflowRepositoryMock.delete.mock.invocationCallOrder[0]); + ).toBeGreaterThan(trxMock.delete.mock.invocationCallOrder[0]); }); test('deletes the workflow executions before the workflow itself', async () => { @@ -2213,7 +2268,7 @@ describe('WorkflowService', () => { expect(executionPersistenceMock.hardDeleteByWorkflowId).toHaveBeenCalledWith(WORKFLOW_ID); expect( executionPersistenceMock.hardDeleteByWorkflowId.mock.invocationCallOrder[0], - ).toBeLessThan(workflowRepositoryMock.delete.mock.invocationCallOrder[0]); + ).toBeLessThan(trxMock.delete.mock.invocationCallOrder[0]); }); test('invalidates the cached project for the deleted workflow', async () => { @@ -2318,6 +2373,8 @@ describe('WorkflowService', () => { mock(), // workflowPublicationNotifier mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + mock(), // workflowScheduledJobOwner + mock(), // durableJobProvisioner mock(), // workflowPublishedVersionRepository mock(), // workflowHookContextService mock(), // workflowPublishGuard @@ -2485,6 +2542,8 @@ describe('WorkflowService', () => { mock(), // workflowPublicationNotifier mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + mock(), // workflowScheduledJobOwner + mock(), // durableJobProvisioner mock(), // workflowPublishedVersionRepository mock(), // workflowHookContextService mock(), // workflowPublishGuard @@ -2663,6 +2722,8 @@ describe('WorkflowService', () => { mock(), // workflowPublicationNotifier mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + mock(), // workflowScheduledJobOwner + mock(), // durableJobProvisioner mock(), // workflowPublishedVersionRepository mock(), // workflowHookContextService mock(), // workflowPublishGuard @@ -2763,6 +2824,8 @@ describe('WorkflowService', () => { mock(), // workflowPublicationNotifier mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + mock(), // workflowScheduledJobOwner + mock(), // durableJobProvisioner mock(), // workflowPublishedVersionRepository mock(), // workflowHookContextService mock(), // workflowPublishGuard diff --git a/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts b/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts index 1eef8c00612..c1e2ff681d6 100644 --- a/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts +++ b/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts @@ -20,6 +20,8 @@ import { PolicyViolationError } from '@/policy/policy-violation.error'; import type { OwnershipService } from '@/services/ownership.service'; import type { Telemetry } from '@/telemetry'; import { WorkflowPublicationApplier } from '@/workflows/publication/workflow-publication-applier'; +import type { DurableJobProvisioner } from '@/scheduling/durable-job-provisioner'; +import type { WorkflowScheduledJobOwner } from '@/scheduling/workflow-scheduled-job-owner'; import type { WorkflowTriggerActivator } from '@/workflows/triggers/workflow-trigger-activator'; import type { WorkflowPublishedDataService } from '@/workflows/workflow-published-data.service'; import type { WorkflowService } from '@/workflows/workflow.service'; @@ -38,6 +40,12 @@ describe('WorkflowPublicationApplier', () => { // Clears by default, which is what the real service does with no policy backend. const policyEnforcementService = mock(); const ownershipService = mock(); + const workflowScheduledJobOwner = mock(); + workflowScheduledJobOwner.ref.mockImplementation((workflowId) => ({ + ownerType: 'workflow', + ownerId: workflowId, + })); + const durableJobProvisioner = mock(); const applier = new WorkflowPublicationApplier( logger, @@ -51,6 +59,8 @@ describe('WorkflowPublicationApplier', () => { telemetry, policyEnforcementService, ownershipService, + workflowScheduledJobOwner, + durableJobProvisioner, ); function makeRecord( @@ -173,6 +183,10 @@ describe('WorkflowPublicationApplier', () => { const result = await applier.apply(makeRecord(), abort); expect(result).toEqual({ type: 'unpublished' }); + expect(durableJobProvisioner.deprovisionOwner).toHaveBeenCalledWith({ + ownerType: 'workflow', + ownerId: 'wf-1', + }); expect(workflowTriggerActivator.getEnabledTriggerNodes).toHaveBeenCalledWith(oldVersion); expect(workflowTriggerActivator.deactivate).toHaveBeenCalledWith( expect.objectContaining({ id: 'wf-1' }), @@ -251,6 +265,20 @@ describe('WorkflowPublicationApplier', () => { await expect(applier.apply(makeRecord(), abort)).rejects.toThrow('teardown boom'); expect(workflowPublishedVersionRepository.removePublishedVersion).not.toHaveBeenCalled(); + expect(durableJobProvisioner.deprovisionOwner).not.toHaveBeenCalled(); + }); + + test('deprovisions the scheduled jobs the workflow owned even when its mapping is already gone', async () => { + // The retried leg of an interrupted unpublish. Nothing left to deactivate, but + // jobs the first attempt never reached would otherwise keep firing. + workflowPublishedVersionRepository.findOne.mockResolvedValue(makePublishedVersion(null)); + + await applier.apply(makeRecord(), abort); + + expect(durableJobProvisioner.deprovisionOwner).toHaveBeenCalledWith({ + ownerType: 'workflow', + ownerId: 'wf-1', + }); }); }); diff --git a/packages/cli/src/workflows/publication/workflow-publication-applier.ts b/packages/cli/src/workflows/publication/workflow-publication-applier.ts index 1887b22cd84..f9dd5770102 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-applier.ts +++ b/packages/cli/src/workflows/publication/workflow-publication-applier.ts @@ -17,6 +17,8 @@ import type { INode, WorkflowActivateMode } from 'n8n-workflow'; import { NodeTypes } from '@/node-types'; import { PolicyEnforcementService } from '@/policy/policy-enforcement.service'; import { PolicyViolationError } from '@/policy/policy-violation.error'; +import { DurableJobProvisioner } from '@/scheduling/durable-job-provisioner'; +import { WorkflowScheduledJobOwner } from '@/scheduling/workflow-scheduled-job-owner'; import { OwnershipService } from '@/services/ownership.service'; import { Telemetry } from '@/telemetry'; import { healNodeIds } from '@/workflows/publication/heal-node-ids'; @@ -74,6 +76,8 @@ export class WorkflowPublicationApplier { private readonly telemetry: Telemetry, private readonly policyEnforcementService: PolicyEnforcementService, private readonly ownershipService: OwnershipService, + private readonly workflowScheduledJobOwner: WorkflowScheduledJobOwner, + private readonly durableJobProvisioner: DurableJobProvisioner, ) { this.logger = this.logger.scoped('workflow-publication'); } @@ -358,6 +362,20 @@ export class WorkflowPublicationApplier { // database instead of the cache ever serving a version for an unpublished // workflow. No repopulation follows: the end state has no published version. await this.workflowPublishedDataService.invalidateCache(record.workflowId); + + // Before the mapping goes, not after. The mapping is what makes the workflow an + // owner of scheduled jobs, so removing it first would turn any job this misses + // into an orphan only the reconciliation sweep could find. The two are not one + // transaction: a crash between them, or a failing mapping removal, leaves no + // jobs and an intact mapping. The reconciler re-enqueues that unpublish, and + // the retry deprovisions nothing before it removes the mapping. + // + // Keyed by the workflow alone, not by the nodes deactivated above, so a job + // whose node is gone from the version goes too. + await this.durableJobProvisioner.deprovisionOwner( + this.workflowScheduledJobOwner.ref(record.workflowId), + ); + await this.workflowPublishedVersionRepository.removePublishedVersion(record.workflowId); return teardownFailures.length > 0 diff --git a/packages/cli/src/workflows/triggers/__tests__/non-webhook-trigger-registrar.test.ts b/packages/cli/src/workflows/triggers/__tests__/non-webhook-trigger-registrar.test.ts index d9671ca7739..fd2df35507c 100644 --- a/packages/cli/src/workflows/triggers/__tests__/non-webhook-trigger-registrar.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/non-webhook-trigger-registrar.test.ts @@ -3,6 +3,7 @@ import type { WorkflowEntity } from '@n8n/db'; import { mock } from 'vitest-mock-extended'; import type { ActiveWorkflowTriggers, Span, Tracing } from 'n8n-core'; import type { IWorkflowBase, IWorkflowExecuteAdditionalData } from 'n8n-workflow'; +import { UserError, WorkflowDeactivationError } from 'n8n-workflow'; import type { PollTriggerJobRegistrar } from '@/scheduling/poll-trigger-node/poll-trigger-job-registrar'; import type { @@ -88,6 +89,7 @@ describe('NonWebhookTriggerRegistrar', () => { test('deregisters one trigger or poll node id', async () => { const activeWorkflowTriggers = mock(); + activeWorkflowTriggers.removeTriggers.mockResolvedValue(); const registrar = new NonWebhookTriggerRegistrar( logger, activeWorkflowTriggers, @@ -196,6 +198,7 @@ describe('NonWebhookTriggerRegistrar', () => { test('deregister removes durable jobs alongside in-memory triggers', async () => { const { registrar, activeWorkflowTriggers } = makeRegistrar(); + activeWorkflowTriggers.removeTriggers.mockResolvedValue(); await registrar.deregister('wf-1', 'trigger-a'); @@ -206,5 +209,100 @@ describe('NonWebhookTriggerRegistrar', () => { expect(scheduleTriggerJobRegistrar.remove).toHaveBeenCalledWith('wf-1', 'trigger-a'); expect(pollTriggerJobRegistrar.remove).toHaveBeenCalledWith('wf-1', 'trigger-a'); }); + + test('deregister removes durable jobs even when the in-memory teardown fails', async () => { + // A close function that can never succeed is abandoned by the caller, and + // the publication advances anyway, so the workflow stays an owner. Nothing + // would ever retire these rows if this teardown skipped them. + const { registrar, activeWorkflowTriggers } = makeRegistrar(); + activeWorkflowTriggers.removeTriggers.mockRejectedValue( + new WorkflowDeactivationError('Failed to deactivate trigger', { + cause: new UserError('Credential not found'), + workflowId: 'wf-1', + }), + ); + + await expect(registrar.deregister('wf-1', 'trigger-a')).rejects.toThrow( + 'Failed to deactivate trigger', + ); + + expect(scheduleTriggerJobRegistrar.remove).toHaveBeenCalledWith('wf-1', 'trigger-a'); + expect(pollTriggerJobRegistrar.remove).toHaveBeenCalledWith('wf-1', 'trigger-a'); + }); + + test('deregister removes durable jobs even when the in-memory teardown never settles', async () => { + // A close function that never settles is abandoned by the caller, and the + // abandoned promise is parked under the publication lock, so the durable + // removals must not wait behind it. + const { registrar, activeWorkflowTriggers } = makeRegistrar(); + activeWorkflowTriggers.removeTriggers.mockReturnValue(new Promise(() => {})); + + void registrar.deregister('wf-1', 'trigger-a'); + + await vi.waitFor(() => { + expect(scheduleTriggerJobRegistrar.remove).toHaveBeenCalledWith('wf-1', 'trigger-a'); + expect(pollTriggerJobRegistrar.remove).toHaveBeenCalledWith('wf-1', 'trigger-a'); + }); + }); + + test('deregister removes the poll jobs even when the schedule job removal fails', async () => { + const { registrar, activeWorkflowTriggers } = makeRegistrar(); + activeWorkflowTriggers.removeTriggers.mockResolvedValue(); + scheduleTriggerJobRegistrar.remove.mockRejectedValue(new Error('db down')); + + await expect(registrar.deregister('wf-1', 'trigger-a')).rejects.toThrow('db down'); + + expect(pollTriggerJobRegistrar.remove).toHaveBeenCalledWith('wf-1', 'trigger-a'); + }); + + test('deregister propagates a durable removal failure over an abandonable in-memory failure', async () => { + // The caller abandons an in-memory UserError as permanent, so surfacing it + // while a durable removal failed would leave the rows behind with no retry. + const { registrar, activeWorkflowTriggers } = makeRegistrar(); + activeWorkflowTriggers.removeTriggers.mockRejectedValue( + new WorkflowDeactivationError('Failed to deactivate trigger', { + cause: new UserError('Credential not found'), + workflowId: 'wf-1', + }), + ); + scheduleTriggerJobRegistrar.remove.mockRejectedValue(new Error('db down')); + + await expect(registrar.deregister('wf-1', 'trigger-a')).rejects.toThrow('db down'); + + expect(logger.error).toHaveBeenCalledWith( + 'Failed to deregister a trigger node from memory', + expect.objectContaining({ + workflowId: 'wf-1', + nodeId: 'trigger-a', + error: expect.any(WorkflowDeactivationError), + }), + ); + }); + + test('deregister propagates a durable removal failure even when the in-memory teardown never settles', async () => { + // The caller abandons a deregistration that never settles, so a durable + // failure held back behind the hung in-memory teardown would never retry. + const { registrar, activeWorkflowTriggers } = makeRegistrar(); + activeWorkflowTriggers.removeTriggers.mockReturnValue(new Promise(() => {})); + scheduleTriggerJobRegistrar.remove.mockRejectedValue(new Error('db down')); + + await expect(registrar.deregister('wf-1', 'trigger-a')).rejects.toThrow('db down'); + }); + + test('deregister hands a pending in-memory teardown to onDetached when a durable removal fails', async () => { + // The rejection escapes deregister while the in-memory teardown still + // runs, so the caller's lifecycle lock must be told to outlive it. + const { registrar, activeWorkflowTriggers } = makeRegistrar(); + const pendingInMemory = new Promise(() => {}); + activeWorkflowTriggers.removeTriggers.mockReturnValue(pendingInMemory); + scheduleTriggerJobRegistrar.remove.mockRejectedValue(new Error('db down')); + const onDetached = vi.fn(); + + await expect(registrar.deregister('wf-1', 'trigger-a', onDetached)).rejects.toThrow( + 'db down', + ); + + expect(onDetached).toHaveBeenCalledWith(pendingInMemory); + }); }); }); diff --git a/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts b/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts index a9580caa988..4208b9c4889 100644 --- a/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts @@ -482,8 +482,16 @@ describe('WorkflowTriggerActivator', () => { expect(deactivateSettled).toBe(true); expect(nonWebhookTriggerRegistrar.deregister).toHaveBeenCalledTimes(2); - expect(nonWebhookTriggerRegistrar.deregister).toHaveBeenCalledWith('wf-1', 'trigger-a'); - expect(nonWebhookTriggerRegistrar.deregister).toHaveBeenCalledWith('wf-1', 'trigger-b'); + expect(nonWebhookTriggerRegistrar.deregister).toHaveBeenCalledWith( + 'wf-1', + 'trigger-a', + expect.any(Function), + ); + expect(nonWebhookTriggerRegistrar.deregister).toHaveBeenCalledWith( + 'wf-1', + 'trigger-b', + expect.any(Function), + ); }); test('waits for both concurrent deactivation phases before surfacing a phase error', async () => { @@ -893,7 +901,11 @@ describe('WorkflowTriggerActivator', () => { abort, ); - expect(nonWebhookTriggerRegistrar.deregister).toHaveBeenCalledWith('wf-1', 't'); + expect(nonWebhookTriggerRegistrar.deregister).toHaveBeenCalledWith( + 'wf-1', + 't', + expect.any(Function), + ); }); }); diff --git a/packages/cli/src/workflows/triggers/non-webhook-trigger-registrar.ts b/packages/cli/src/workflows/triggers/non-webhook-trigger-registrar.ts index bffe44844b4..2a4e3959f84 100644 --- a/packages/cli/src/workflows/triggers/non-webhook-trigger-registrar.ts +++ b/packages/cli/src/workflows/triggers/non-webhook-trigger-registrar.ts @@ -1,6 +1,7 @@ import { Logger } from '@n8n/backend-common'; import type { WorkflowEntity } from '@n8n/db'; import { Service } from '@n8n/di'; +import { ensureError } from '@n8n/utils/errors/ensure-error'; import { ActiveWorkflowTriggers, SpanStatus, @@ -165,9 +166,17 @@ export class NonWebhookTriggerRegistrar { } /** - * Deregister one active, poll, or schedule trigger node from memory. + * Deregister one active, poll, or schedule trigger node from memory, and drop + * the durable jobs it provisioned. When a durable removal failure propagates + * while the in-memory teardown is still pending, that teardown is handed to + * `onDetached`: it may still mutate the registry when it settles, so the + * caller must not release the workflow's lifecycle lock until it does. */ - async deregister(workflowId: WorkflowId, nodeId: INode['id']) { + async deregister( + workflowId: WorkflowId, + nodeId: INode['id'], + onDetached?: (work: Promise) => void, + ) { await this.tracing.startSpan( { name: 'Non-webhook trigger deregister', @@ -178,12 +187,51 @@ export class NonWebhookTriggerRegistrar { }, }, async (span) => { - await this.activeWorkflowTriggers.removeTriggers(workflowId, new Set([nodeId])); - await this.scheduleTriggerJobRegistrar.remove(workflowId, nodeId); - await this.pollTriggerJobRegistrar.remove(workflowId, nodeId); + // Start the in-memory teardown now, but await it only after the durable + // removal below. The durable rows are database state the node no longer + // owns, so a durable failure must reach the caller for a retry even when + // the in-memory teardown never settles. + const inMemory = this.activeWorkflowTriggers.removeTriggers(workflowId, new Set([nodeId])); + // Logs an in-memory failure a durable failure would otherwise swallow, and + // keeps its rejection handled while the durable removal is awaited first. + void inMemory.catch((error: unknown) => { + this.logger.error('Failed to deregister a trigger node from memory', { + workflowId, + nodeId, + error: ensureError(error), + }); + }); + + try { + await this.removeDurableJobs(workflowId, nodeId); + } catch (error) { + // The in-memory teardown may still be pending, so hand it back before + // this failure lets the caller release the lifecycle lock. + onDetached?.(inMemory); + throw ensureError(error); + } + + try { + await inMemory; + } catch (error) { + throw ensureError(error); + } span.setStatus({ code: SpanStatus.ok }); }, ); } + + private async removeDurableJobs(workflowId: WorkflowId, nodeId: INode['id']) { + const results = await Promise.allSettled([ + this.scheduleTriggerJobRegistrar.remove(workflowId, nodeId), + this.pollTriggerJobRegistrar.remove(workflowId, nodeId), + ]); + const failure = results.find( + (result): result is PromiseRejectedResult => result.status === 'rejected', + ); + if (failure) { + throw ensureError(failure.reason); + } + } } diff --git a/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts b/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts index adf13dd6094..a7fb5cd5834 100644 --- a/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts +++ b/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts @@ -900,7 +900,10 @@ export class WorkflowTriggerActivator { await Promise.all( triggerNodeIds.map(async (nodeId) => { try { - await raceAbort(this.nonWebhookTriggerRegistrar.deregister(workflowId, nodeId), abort); + await raceAbort( + this.nonWebhookTriggerRegistrar.deregister(workflowId, nodeId, abort.onDetached), + abort, + ); } catch (error) { if (!this.shouldAbandonFailedTeardown(ensureError(error))) throw error; this.logger.warn('Abandoned trigger whose deregistration can never succeed', { diff --git a/packages/cli/src/workflows/workflow.service.ts b/packages/cli/src/workflows/workflow.service.ts index 16a1889c35b..f678bd7ae46 100644 --- a/packages/cli/src/workflows/workflow.service.ts +++ b/packages/cli/src/workflows/workflow.service.ts @@ -61,8 +61,10 @@ import { userHasScopes } from '@/permissions.ee/check-access'; import { PolicyEnforcementService } from '@/policy/policy-enforcement.service'; import type { ListQuery } from '@/requests'; import { hasSharing } from '@/requests'; +import { DurableJobProvisioner } from '@/scheduling/durable-job-provisioner'; import { PollTriggerJobRegistrar } from '@/scheduling/poll-trigger-node/poll-trigger-job-registrar'; import { ScheduleTriggerJobRegistrar } from '@/scheduling/schedule-trigger-node/schedule-trigger-job-registrar'; +import { WorkflowScheduledJobOwner } from '@/scheduling/workflow-scheduled-job-owner'; import { OwnershipService } from '@/services/ownership.service'; import { ProjectService } from '@/services/project.service.ee'; import { RoleService } from '@/services/role.service'; @@ -119,6 +121,8 @@ export class WorkflowService { private readonly workflowPublicationNotifier: WorkflowPublicationNotifier, private readonly scheduleTriggerJobRegistrar: ScheduleTriggerJobRegistrar, private readonly pollTriggerJobRegistrar: PollTriggerJobRegistrar, + private readonly workflowScheduledJobOwner: WorkflowScheduledJobOwner, + private readonly durableJobProvisioner: DurableJobProvisioner, private readonly workflowPublishedVersionRepository: WorkflowPublishedVersionRepository, private readonly workflowHookContextService: WorkflowHookContextService, private readonly workflowPublishGuard: WorkflowPublishGuardProxy, @@ -1484,7 +1488,21 @@ export class WorkflowService { // to hit a DB statement timeout, however large the execution history. await this.executionPersistence.hardDeleteByWorkflowId(workflowId); - await this.workflowRepository.delete(workflowId); + // The workflow stops owning scheduled jobs here, and nothing in the database + // removes them with it. Normally there are none left, since a published + // workflow cannot be deleted and unpublishing deprovisions them. This covers + // the paths that never published and anything an interrupted unpublish left. + // + // Both writes commit together. A deprovision of its own would leave a workflow + // whose delete then failed alive with its schedules stripped. + await this.workflowRepository.runInTransaction({}, async (trx) => { + await this.durableJobProvisioner.deprovisionOwnerInTransaction( + trx, + this.workflowScheduledJobOwner.ref(workflowId), + ); + + await trx.delete(WorkflowEntity, { id: workflowId }); + }); await this.ownershipService.invalidateWorkflowProjectCacheByIds([workflowId]); diff --git a/packages/cli/test/integration/scheduling/scheduled-job-owner-teardown.test.ts b/packages/cli/test/integration/scheduling/scheduled-job-owner-teardown.test.ts new file mode 100644 index 00000000000..9a5064833d8 --- /dev/null +++ b/packages/cli/test/integration/scheduling/scheduled-job-owner-teardown.test.ts @@ -0,0 +1,233 @@ +import { createWorkflowWithHistory, testDb } from '@n8n/backend-test-utils'; +import type { WorkflowEntity } from '@n8n/db'; +import { + DataSource, + ScheduledJobRepository, + ScheduledTaskRepository, + WorkflowPublishedVersionRepository, +} from '@n8n/db'; +import { Container } from '@n8n/di'; +import { ScheduledJobMisfirePolicy } from '@n8n/constants'; +import { v4 as uuid } from 'uuid'; + +import { DurableJobProvisioner } from '@/scheduling/durable-job-provisioner'; +import { POLL_TRIGGER_TASK_TYPE } from '@/scheduling/poll-trigger-node/poll-trigger-task'; +import { SCHEDULE_TRIGGER_TASK_TYPE } from '@/scheduling/schedule-trigger-node/schedule-trigger-task'; +import { WorkflowScheduledJobOwner } from '@/scheduling/workflow-scheduled-job-owner'; + +import { createDueJobFactory, seedDueTask, workflowOwned } from './shared/job-factory'; + +/** + * Owner teardown against real rows. + * + * `scheduled_job.workflowId` used to carry a foreign key to + * `workflow_published_version` with `ON DELETE CASCADE`, so unpublishing a workflow + * removed its jobs in the database. That key is gone and teardown is now an + * explicit call every owner module owes. This suite pins both halves: that the + * cascade is gone, and that the explicit call covers what it used to. + */ +describe('scheduled job owner teardown', () => { + const TASK_TYPE = 'integration-owner-teardown'; + + let jobRepo: ScheduledJobRepository; + let taskRepo: ScheduledTaskRepository; + let publishedVersions: WorkflowPublishedVersionRepository; + let owner: WorkflowScheduledJobOwner; + let provisioner: DurableJobProvisioner; + let dataSource: DataSource; + let createJob: ReturnType; + + const publishWorkflow = async (): Promise => { + const workflow = await createWorkflowWithHistory({ active: true }); + await publishedVersions.setPublishedVersion(workflow.id, workflow.versionId); + return workflow; + }; + + beforeAll(async () => { + await testDb.init(); + jobRepo = Container.get(ScheduledJobRepository); + taskRepo = Container.get(ScheduledTaskRepository); + publishedVersions = Container.get(WorkflowPublishedVersionRepository); + owner = Container.get(WorkflowScheduledJobOwner); + provisioner = Container.get(DurableJobProvisioner); + dataSource = Container.get(DataSource); + createJob = createDueJobFactory(jobRepo, TASK_TYPE, 'owner-teardown'); + }); + + beforeEach(async () => { + await testDb.truncate([ + 'ScheduledTask', + 'ScheduledJob', + 'WorkflowPublishedVersion', + 'WorkflowEntity', + 'WorkflowHistory', + ]); + }); + + afterAll(async () => { + await testDb.terminate(); + }); + + it('no longer removes a workflow’s jobs when its published version goes away', async () => { + // The pin on the dropped cascade, and why every teardown path has to + // deprovision for itself. + const workflow = await publishWorkflow(); + const job = await createJob(workflowOwned(workflow.id, uuid())); + + await publishedVersions.removePublishedVersion(workflow.id); + + expect(await jobRepo.findOneBy({ id: job.id })).not.toBeNull(); + }); + + it('deletes every job the workflow owns, across nodes and task types, with their queued runs', async () => { + const workflow = await publishWorkflow(); + const other = await publishWorkflow(); + const nodeA = uuid(); + const nodeB = uuid(); + const scheduleJob = await createJob({ + ...workflowOwned(workflow.id, nodeA), + taskType: SCHEDULE_TRIGGER_TASK_TYPE, + }); + const pollJob = await createJob({ + ...workflowOwned(workflow.id, nodeB), + taskType: POLL_TRIGGER_TASK_TYPE, + }); + const foreignJob = await createJob(workflowOwned(other.id, nodeA)); + await seedDueTask(taskRepo, TASK_TYPE, scheduleJob.id); + await seedDueTask(taskRepo, TASK_TYPE, pollJob.id); + await seedDueTask(taskRepo, TASK_TYPE, foreignJob.id); + + const { removed } = await provisioner.deprovisionOwner(owner.ref(workflow.id)); + + expect(removed).toBe(2); + expect(await jobRepo.findOneBy({ id: scheduleJob.id })).toBeNull(); + expect(await jobRepo.findOneBy({ id: pollJob.id })).toBeNull(); + expect(await taskRepo.countBy({ jobId: scheduleJob.id })).toBe(0); + expect(await taskRepo.countBy({ jobId: pollJob.id })).toBe(0); + // Another workflow's jobs are untouched. + expect(await jobRepo.findOneBy({ id: foreignJob.id })).not.toBeNull(); + expect(await taskRepo.countBy({ jobId: foreignJob.id })).toBe(1); + }); + + it('is a no-op for a workflow that owns nothing, so a retried teardown is safe', async () => { + const workflow = await publishWorkflow(); + + const deprovision = async () => await provisioner.deprovisionOwner(owner.ref(workflow.id)); + + await expect(deprovision()).resolves.toEqual({ removed: 0 }); + await expect(deprovision()).resolves.toEqual({ removed: 0 }); + }); + + it('commits the teardown with the caller’s transaction, and rolls back with it', async () => { + const workflow = await publishWorkflow(); + const job = await createJob(workflowOwned(workflow.id, uuid())); + + await expect( + dataSource.transaction(async (manager) => { + await provisioner.deprovisionOwnerInTransaction(manager, owner.ref(workflow.id)); + throw new Error('caller failed after deprovisioning'); + }), + ).rejects.toThrow('caller failed after deprovisioning'); + + // The rollback took the delete with it, so the job's removal is not independent + // of the write that made the workflow stop owning it. + expect(await jobRepo.findOneBy({ id: job.id })).not.toBeNull(); + + await dataSource.transaction(async (manager) => { + await provisioner.deprovisionOwnerInTransaction(manager, owner.ref(workflow.id)); + }); + + expect(await jobRepo.findOneBy({ id: job.id })).toBeNull(); + }); + + it('refuses to provision an owner type with no liveness resolver, writing nothing', async () => { + await expect( + provisioner.provision({ + owner: { ownerType: 'unregistered-kind', ownerId: 'subject-1', ownerMemberId: null }, + taskType: TASK_TYPE, + payload: {}, + desired: [ + { + name: 'unregistered-kind:0', + schedule: { kind: 'interval', intervalSeconds: 60 }, + firstRunAt: new Date(), + }, + ], + misfirePolicy: ScheduledJobMisfirePolicy.Skip, + }), + ).rejects.toThrow('no registered liveness resolver'); + + expect(await jobRepo.countBy({ ownerType: 'unregistered-kind' })).toBe(0); + }); + + it('lifts a quarantine the sweep left behind when the owner provisions again', async () => { + // The recovery path for a teardown that never ran. The sweep disabled the jobs + // of a workflow it believed gone, then the workflow publishes again. Without + // this the trigger stays silently dead until the next sweep. + const workflow = await publishWorkflow(); + const nodeId = uuid(); + const desired = [ + { + name: `${workflow.id}:${nodeId}:0`, + schedule: { kind: 'interval', intervalSeconds: 3600 } as const, + firstRunAt: new Date(Date.now() + 3600_000), + }, + ]; + const provisionOnce = async () => + await provisioner.provision({ + owner: owner.member(workflow.id, nodeId), + taskType: SCHEDULE_TRIGGER_TASK_TYPE, + payload: { workflowId: workflow.id, nodeId }, + desired, + misfirePolicy: ScheduledJobMisfirePolicy.Skip, + }); + + const { inserted } = await provisionOnce(); + const jobId = inserted[0].id; + await jobRepo.quarantineByOwnerIds( + 'workflow', + [workflow.id], + new Date(), + new Date(Date.now() + 60_000), + ); + const quarantined = await jobRepo.findOneByOrFail({ id: jobId }); + expect(quarantined).toMatchObject({ enabled: true, nextRunAt: null }); + expect(quarantined.orphanedAt).not.toBeNull(); + + // Same schedule as before. Only the quarantine changed, so an unchanged plan + // has to be enough to bring it back. + await provisionOnce(); + + const revived = await jobRepo.findOneByOrFail({ id: jobId }); + expect(revived).toMatchObject({ enabled: true, orphanedAt: null }); + expect(revived.nextRunAt).not.toBeNull(); + }); + + it('provisions a workflow owner, whose type the workflow owner module claims', async () => { + const workflow = await publishWorkflow(); + const nodeId = uuid(); + + const summary = await provisioner.provision({ + owner: owner.member(workflow.id, nodeId), + taskType: SCHEDULE_TRIGGER_TASK_TYPE, + payload: { workflowId: workflow.id, nodeId }, + desired: [ + { + name: `${workflow.id}:${nodeId}:0`, + schedule: { kind: 'interval', intervalSeconds: 3600 }, + firstRunAt: new Date(Date.now() + 3600_000), + }, + ], + misfirePolicy: ScheduledJobMisfirePolicy.Skip, + }); + + expect(summary.inserted).toHaveLength(1); + const stored = await jobRepo.findOneByOrFail({ id: summary.inserted[0].id }); + expect(stored).toMatchObject({ + ownerType: 'workflow', + ownerId: workflow.id, + ownerMemberId: nodeId, + orphanedAt: null, + }); + }); +}); diff --git a/packages/cli/test/integration/workflows/workflow.service.test.ts b/packages/cli/test/integration/workflows/workflow.service.test.ts index daa547b1b9e..dc65c97f044 100644 --- a/packages/cli/test/integration/workflows/workflow.service.test.ts +++ b/packages/cli/test/integration/workflows/workflow.service.test.ts @@ -111,6 +111,8 @@ beforeAll(async () => { workflowPublicationNotifier, mock(), // scheduleTriggerJobRegistrar mock(), // pollTriggerJobRegistrar + mock(), // workflowScheduledJobOwner + mock(), // durableJobProvisioner workflowPublishedVersionRepository, Container.get(WorkflowHookContextService), // workflowHookContextService workflowPublishGuard,