github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 13:39:44 +02:00
Code Issues Actions 13 Packages Projects Releases Wiki Activity Graph

feat(core): Deprovision a workflow's schedules when it stops owning them (#37135)

Browse Source
This commit is contained in:
Lorent Lempereur 2026-09-02 20:10:44 +00:00 committed by GitHub
parent 7675c614bf
commit 2a24001fac
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 601 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
packages/@n8n/db/src/repositories/__tests__/scheduled-job.repository.test.ts
View File

@ -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);
});

37
packages/@n8n/db/src/repositories/scheduled-job.repository.ts
View File

@ -167,6 +167,11 @@ export class ScheduledJobRepository extends Repository<ScheduledJob> {
* 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<ScheduledJob> {
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<ScheduledJob> {
}
}
/** A read-back row, carrying the owner the id is only valid under. */
type InsertedRow = Pick<ScheduledJob, 'id' | 'name' | 'ownerType' | 'ownerId' | 'ownerMemberId'>;
/**
* 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;
});
}

77
packages/cli/src/workflows/__tests__/workflow.service.test.ts
View File

@ -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<WorkflowPublishedVersionRepository>;
let workflowMutationHooksMock: MockProxy<WorkflowMutationHooksProxy>;
let ownershipServiceMock: MockProxy<OwnershipService>;
let workflowScheduledJobOwnerMock: MockProxy<WorkflowScheduledJobOwner>;
let durableJobProvisionerMock: MockProxy<DurableJobProvisioner>;
let trxMock: MockProxy<EntityManager>;
const WORKFLOW_ID = 'workflow-1';
const WORKFLOW_OWNER_REF = { ownerType: 'workflow', ownerId: WORKFLOW_ID };
function makeWorkflowEntity(overrides: Partial<WorkflowEntity> = {}): WorkflowEntity {
const workflow = new WorkflowEntity();
@ -2065,10 +2079,18 @@ describe('WorkflowService', () => {
workflowMutationHooksMock = mock<WorkflowMutationHooksProxy>();
workflowPublishedVersionRepositoryMock = mock<WorkflowPublishedVersionRepository>();
workflowPublishedVersionRepositoryMock.getPublishedVersionId.mockResolvedValue(null);
workflowScheduledJobOwnerMock = mock<WorkflowScheduledJobOwner>();
workflowScheduledJobOwnerMock.ref.mockReturnValue(WORKFLOW_OWNER_REF);
durableJobProvisionerMock = mock<DurableJobProvisioner>();
globalConfigMock = mock<GlobalConfig>({
workflows: mock<WorkflowsConfig>({ useWorkflowPublicationService: true }),
});
trxMock = mock<EntityManager>();
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<User>(), 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<User>(), 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<User>(), 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<User>(), 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

28
packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts
View File

@ -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<PolicyEnforcementService>();
const ownershipService = mock<OwnershipService>();
const workflowScheduledJobOwner = mock<WorkflowScheduledJobOwner>();
workflowScheduledJobOwner.ref.mockImplementation((workflowId) => ({
ownerType: 'workflow',
ownerId: workflowId,
}));
const durableJobProvisioner = mock<DurableJobProvisioner>();
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',
});
});
});

18
packages/cli/src/workflows/publication/workflow-publication-applier.ts
View File

@ -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

98
packages/cli/src/workflows/triggers/__tests__/non-webhook-trigger-registrar.test.ts
View File

@ -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>();
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<void>(() => {});
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);
});
});
});

18
packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts
View File

@ -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),
);
});
});

58
packages/cli/src/workflows/triggers/non-webhook-trigger-registrar.ts
View File

@ -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<unknown>) => 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);
}
}
}

5
packages/cli/src/workflows/triggers/workflow-trigger-activator.ts
View File

@ -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', {

20
packages/cli/src/workflows/workflow.service.ts
View File

@ -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]);

233
packages/cli/test/integration/scheduling/scheduled-job-owner-teardown.test.ts Normal file
View File

@ -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<typeof createDueJobFactory>;
const publishWorkflow = async (): Promise<WorkflowEntity> => {
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,
});
});
});

2
packages/cli/test/integration/workflows/workflow.service.test.ts
View File

@ -111,6 +111,8 @@ beforeAll(async () => {
workflowPublicationNotifier,
mock(), // scheduleTriggerJobRegistrar
mock(), // pollTriggerJobRegistrar
mock(), // workflowScheduledJobOwner
mock(), // durableJobProvisioner
workflowPublishedVersionRepository,
Container.get(WorkflowHookContextService), // workflowHookContextService
workflowPublishGuard,