mirror of
https://github.com/n8n-io/n8n.git
synced 2026-09-24 23:22:38 +08:00
feat(core): Enforce policy on workflow publish (no-changelog) (#36955)
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
04edd2b911
commit
1f19b11c40
@@ -202,6 +202,13 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
|
||||
* skew visible with no in-flight record is a real divergence (e.g. a stalled
|
||||
* processor writing the mapping after losing its lease), never a normal
|
||||
* mid-flight state.
|
||||
*
|
||||
* Workflows whose most recent record is a terminal `failed` for the version
|
||||
* that is currently active are excluded, as in
|
||||
* {@link findTriggerStatusDriftedWorkflowIds}: a publication failing before the
|
||||
* mapping advances leaves the skew forever, so this would loop every pass.
|
||||
* Matching on the version keeps the unpublish direction healing — there the
|
||||
* active version is null, so a failed teardown never matches and is retried.
|
||||
*/
|
||||
async findVersionSkewedWorkflowIds(): Promise<string[]> {
|
||||
const outboxTableName = this.getTableName('workflow_publication_outbox');
|
||||
@@ -211,6 +218,9 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
|
||||
// `(x IS NULL) <> (y IS NULL) OR x <> y` is the portable spelling of
|
||||
// `activeVersionId IS DISTINCT FROM publishedVersionId`, which sqlite
|
||||
// lacks; both-null (never published, no mapping) compares as equal.
|
||||
//
|
||||
// The failed-record match relies on plain `=`: an unpublished workflow has a
|
||||
// null `activeVersionId`, so nothing matches and its skew stays detectable.
|
||||
const rows: Array<{ workflowId: string }> = await this.query(
|
||||
`SELECT w."id" AS "workflowId"
|
||||
FROM ${workflowTableName} w
|
||||
@@ -223,6 +233,16 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
|
||||
SELECT 1 FROM ${outboxTableName} o
|
||||
WHERE o."workflowId" = w."id"
|
||||
AND o."status" IN ('${Status.Pending}', '${Status.InProgress}')
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM ${outboxTableName} o
|
||||
WHERE o."workflowId" = w."id"
|
||||
AND o."status" = '${Status.Failed}'
|
||||
AND o."publishedVersionId" = w."activeVersionId"
|
||||
AND o."id" = (
|
||||
SELECT MAX(latest."id") FROM ${outboxTableName} latest
|
||||
WHERE latest."workflowId" = w."id"
|
||||
)
|
||||
)`,
|
||||
);
|
||||
|
||||
|
||||
@@ -43,6 +43,8 @@ import type {
|
||||
ScheduleTriggerCollectionSession,
|
||||
ScheduleTriggerJobRegistrar,
|
||||
} from '@/scheduling/schedule-trigger-node/schedule-trigger-job-registrar';
|
||||
import type { PolicyEnforcementService } from '@/policy/policy-enforcement.service';
|
||||
import { PolicyViolationError } from '@/policy/policy-violation.error';
|
||||
import type { OwnershipService } from '@/services/ownership.service';
|
||||
import type { PollCursorService } from '@/workflows/triggers/poll-cursor.service';
|
||||
import { TriggerExecutionContextFactory } from '@/workflows/triggers/trigger-execution-context.factory';
|
||||
@@ -58,8 +60,17 @@ describe('ActiveWorkflowManager', () => {
|
||||
const workflowRepository = mock<WorkflowRepository>();
|
||||
const workflowsConfig = mock<WorkflowsConfig>({ useWorkflowPublicationService: false });
|
||||
|
||||
// Shared by every construction below; clears by default, like an instance with
|
||||
// no policy backend.
|
||||
const policyEnforcementService = mock<PolicyEnforcementService>();
|
||||
const ownershipService = mock<OwnershipService>();
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
// `clearAllMocks` keeps implementations, so restore the clearing defaults.
|
||||
ownershipService.getWorkflowProjectCached.mockResolvedValue(mock<Project>({ id: 'project-1' }));
|
||||
policyEnforcementService.hasChecksFor.mockReturnValue(true);
|
||||
policyEnforcementService.enforceWorkflowPublish.mockResolvedValue(mock());
|
||||
activeWorkflowManager = new ActiveWorkflowManager(
|
||||
mockLogger(),
|
||||
mock(),
|
||||
@@ -79,6 +90,8 @@ describe('ActiveWorkflowManager', () => {
|
||||
mock(),
|
||||
mock(), // scheduleTriggerJobRegistrar
|
||||
mock(), // pollTriggerJobRegistrar
|
||||
policyEnforcementService,
|
||||
ownershipService,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -181,6 +194,223 @@ describe('ActiveWorkflowManager', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('policy enforcement', () => {
|
||||
// `chunk` coerces an unset `activationBatchSize` to 0 and yields no batches, so
|
||||
// the startup loop would silently never run.
|
||||
const policyWorkflowsConfig = mock<WorkflowsConfig>({
|
||||
useWorkflowPublicationService: false,
|
||||
activationBatchSize: 1,
|
||||
});
|
||||
const activationErrorsService = mock<ActivationErrorsService>();
|
||||
const activeWorkflowTriggers = mock<ActiveWorkflowTriggers>();
|
||||
const errorReporter = mock<ErrorReporter>();
|
||||
|
||||
const VERSION_NODES = [{ id: 'node-1', name: 'Active trigger' } as INode];
|
||||
const DRAFT_NODES = [{ id: 'node-2', name: 'Draft node' } as INode];
|
||||
|
||||
const makeManager = () =>
|
||||
new ActiveWorkflowManager(
|
||||
mockLogger(),
|
||||
errorReporter,
|
||||
activeWorkflowTriggers,
|
||||
mock(),
|
||||
nodeTypes,
|
||||
mock(),
|
||||
workflowRepository,
|
||||
activationErrorsService,
|
||||
mock(),
|
||||
mock(),
|
||||
instanceSettings,
|
||||
mock(),
|
||||
policyWorkflowsConfig,
|
||||
mock(),
|
||||
mock<TriggerExecutionContextFactory>(),
|
||||
mock(),
|
||||
mock(), // scheduleTriggerJobRegistrar
|
||||
mock(), // pollTriggerJobRegistrar
|
||||
policyEnforcementService,
|
||||
ownershipService,
|
||||
);
|
||||
|
||||
const makeWorkflow = (overrides: Partial<WorkflowEntity> = {}) =>
|
||||
mock<WorkflowEntity>({
|
||||
id: 'wf-1',
|
||||
name: 'My workflow',
|
||||
active: true,
|
||||
isArchived: false,
|
||||
activeVersionId: 'v1',
|
||||
// Differs from the published version on purpose: only the latter runs.
|
||||
nodes: DRAFT_NODES,
|
||||
activeVersion: mock<WorkflowHistory>({
|
||||
versionId: 'v1',
|
||||
nodes: VERSION_NODES,
|
||||
connections: {},
|
||||
}),
|
||||
...overrides,
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
Object.assign(instanceSettings, { isLeader: true, isFollower: false });
|
||||
activeWorkflowManager = makeManager();
|
||||
});
|
||||
|
||||
test('enforces with the published version nodes, not the draft', async () => {
|
||||
workflowRepository.findById.mockResolvedValue(makeWorkflow());
|
||||
|
||||
// Registration fails here (no real node types); the check runs before it.
|
||||
await activeWorkflowManager.add('wf-1', 'activate').catch(() => {});
|
||||
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).toHaveBeenCalledExactlyOnceWith({
|
||||
workflow: { id: 'wf-1', name: 'My workflow', nodes: VERSION_NODES },
|
||||
projectId: 'project-1',
|
||||
});
|
||||
});
|
||||
|
||||
test('registers nothing and records an activation error when policy blocks', async () => {
|
||||
workflowRepository.findById.mockResolvedValue(makeWorkflow());
|
||||
const addWebhooksSpy = vi.spyOn(activeWorkflowManager, 'addWebhooks');
|
||||
const addNonWebhookTriggersSpy = vi.spyOn(activeWorkflowManager, 'addNonWebhookTriggers');
|
||||
policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(
|
||||
new PolicyViolationError([
|
||||
{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
|
||||
]),
|
||||
);
|
||||
|
||||
await expect(activeWorkflowManager.add('wf-1', 'activate')).rejects.toBeInstanceOf(
|
||||
PolicyViolationError,
|
||||
);
|
||||
|
||||
expect(addWebhooksSpy).not.toHaveBeenCalled();
|
||||
expect(addNonWebhookTriggersSpy).not.toHaveBeenCalled();
|
||||
expect(activationErrorsService.register).toHaveBeenCalledWith('wf-1', 'Blocked by policy');
|
||||
});
|
||||
|
||||
test('does not enforce for a workflow that is no longer active', async () => {
|
||||
workflowRepository.findById.mockResolvedValue(
|
||||
makeWorkflow({ active: false, activeVersionId: null, activeVersion: null }),
|
||||
);
|
||||
|
||||
await activeWorkflowManager.add('wf-1', 'init');
|
||||
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// An unevaluated project rule is not a passed one, so the lookup is unguarded.
|
||||
test('propagates a failed ownership lookup instead of policing a null scope', async () => {
|
||||
workflowRepository.findById.mockResolvedValue(makeWorkflow());
|
||||
ownershipService.getWorkflowProjectCached.mockRejectedValue(new Error('no owner row'));
|
||||
|
||||
await expect(activeWorkflowManager.add('wf-1', 'activate')).rejects.toThrow('no owner row');
|
||||
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// A feature that is merely absent must not cost a lookup on every activation.
|
||||
test('does not resolve ownership when no check is registered', async () => {
|
||||
policyEnforcementService.hasChecksFor.mockReturnValue(false);
|
||||
workflowRepository.findById.mockResolvedValue(makeWorkflow());
|
||||
|
||||
await activeWorkflowManager.add('wf-1', 'activate').catch(() => {});
|
||||
|
||||
expect(ownershipService.getWorkflowProjectCached).not.toHaveBeenCalled();
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('does not enforce for an archived workflow', async () => {
|
||||
workflowRepository.findById.mockResolvedValue(makeWorkflow({ isArchived: true }));
|
||||
|
||||
await activeWorkflowManager.add('wf-1', 'activate');
|
||||
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// The forward is unconditional on multi-main, not leader-specific; the leader
|
||||
// enforces when it handles the pubsub command with `shouldPublish: false`.
|
||||
test('does not enforce when a multi-main instance forwards the activation', async () => {
|
||||
Object.assign(instanceSettings, { isMultiMain: true });
|
||||
workflowRepository.findById.mockResolvedValue(makeWorkflow());
|
||||
|
||||
try {
|
||||
await activeWorkflowManager.add('wf-1', 'activate');
|
||||
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
Object.assign(instanceSettings, { isMultiMain: false });
|
||||
}
|
||||
});
|
||||
|
||||
// Already queued for an unrelated transient failure: if policy then blocks it,
|
||||
// the retry must drop it rather than reschedule forever.
|
||||
test('drops a queued retry when policy blocks on the retry attempt', async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const dbWorkflow = makeWorkflow();
|
||||
workflowRepository.findById.mockResolvedValue(dbWorkflow);
|
||||
const manager = activeWorkflowManager as unknown as {
|
||||
addQueuedWorkflowActivation: (
|
||||
mode: WorkflowActivateMode,
|
||||
workflow: WorkflowEntity,
|
||||
) => void;
|
||||
queuedActivations: Record<string, unknown>;
|
||||
};
|
||||
|
||||
manager.addQueuedWorkflowActivation('update', dbWorkflow);
|
||||
expect(manager.queuedActivations['wf-1']).toBeDefined();
|
||||
|
||||
policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(
|
||||
new PolicyViolationError([
|
||||
{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
|
||||
]),
|
||||
);
|
||||
|
||||
await vi.runOnlyPendingTimersAsync();
|
||||
|
||||
expect(manager.queuedActivations['wf-1']).toBeUndefined();
|
||||
// Nor a fault report: the refusal is expected and permanent.
|
||||
expect(errorReporter.error).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
// A refusal, not a failure: the owner's error automation must not fire.
|
||||
test('does not run the error workflow when policy blocks at startup', async () => {
|
||||
workflowRepository.getAllActiveIds.mockResolvedValue(['wf-1']);
|
||||
workflowRepository.findById.mockResolvedValue(makeWorkflow());
|
||||
policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(
|
||||
new PolicyViolationError([
|
||||
{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
|
||||
]),
|
||||
);
|
||||
const errorWorkflowSpy = vi.spyOn(activeWorkflowManager, 'executeErrorWorkflow');
|
||||
|
||||
await activeWorkflowManager.addActiveWorkflows('init');
|
||||
|
||||
expect(errorWorkflowSpy).not.toHaveBeenCalled();
|
||||
// Nor a fault report, or every restart alerts on an expected refusal.
|
||||
expect(errorReporter.error).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// A block is permanent, so the indefinite retry must not pick it up.
|
||||
test('does not queue a startup retry when policy blocks', async () => {
|
||||
workflowRepository.getAllActiveIds.mockResolvedValue(['wf-1']);
|
||||
workflowRepository.findById.mockResolvedValue(makeWorkflow());
|
||||
policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(
|
||||
new PolicyViolationError([
|
||||
{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
|
||||
]),
|
||||
);
|
||||
const queueSpy = vi.spyOn(
|
||||
activeWorkflowManager as unknown as { addQueuedWorkflowActivation: () => void },
|
||||
'addQueuedWorkflowActivation',
|
||||
);
|
||||
|
||||
await activeWorkflowManager.addActiveWorkflows('init');
|
||||
|
||||
expect(queueSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('handleAddWebhooksAndNonWebhookTriggers', () => {
|
||||
const push = mock<Push>();
|
||||
const publisher = mock<Publisher>();
|
||||
@@ -205,6 +435,33 @@ describe('ActiveWorkflowManager', () => {
|
||||
mock(),
|
||||
mock(), // scheduleTriggerJobRegistrar
|
||||
mock(), // pollTriggerJobRegistrar
|
||||
policyEnforcementService,
|
||||
ownershipService,
|
||||
);
|
||||
});
|
||||
|
||||
// The leader's generic failure path clears `activeVersionId`. A refusal must not
|
||||
// unpublish a workflow — that decision is not this path's to make.
|
||||
test('does not unpublish the workflow when policy blocks the forwarded activation', async () => {
|
||||
const violation = new PolicyViolationError([
|
||||
{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
|
||||
]);
|
||||
vi.spyOn(activeWorkflowManager, 'add').mockRejectedValue(violation);
|
||||
const clearWebhooksSpy = vi.spyOn(activeWorkflowManager, 'clearWebhooks');
|
||||
const removeTriggersSpy = vi.spyOn(activeWorkflowManager, 'removeNonWebhookTriggers');
|
||||
|
||||
await activeWorkflowManager.handleAddWebhooksAndNonWebhookTriggers({
|
||||
workflowId: 'wf-1',
|
||||
activeVersionId: 'v1',
|
||||
activationMode: 'init',
|
||||
});
|
||||
|
||||
expect(workflowRepository.update).not.toHaveBeenCalled();
|
||||
expect(clearWebhooksSpy).not.toHaveBeenCalled();
|
||||
expect(removeTriggersSpy).not.toHaveBeenCalled();
|
||||
// The user is still told why it did not start.
|
||||
expect(push.broadcast).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ type: 'workflowFailedToActivate' }),
|
||||
);
|
||||
});
|
||||
|
||||
@@ -463,6 +720,8 @@ describe('ActiveWorkflowManager', () => {
|
||||
mock(), // eventBus
|
||||
mock(), // scheduleTriggerJobRegistrar
|
||||
mock(), // pollTriggerJobRegistrar
|
||||
policyEnforcementService,
|
||||
ownershipService,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -914,6 +1173,8 @@ describe('ActiveWorkflowManager', () => {
|
||||
mock(),
|
||||
scheduleTriggerJobRegistrar,
|
||||
pollTriggerJobRegistrar,
|
||||
policyEnforcementService,
|
||||
ownershipService,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -1100,6 +1361,8 @@ describe('ActiveWorkflowManager', () => {
|
||||
mock(),
|
||||
scheduleTriggerJobRegistrar,
|
||||
pollTriggerJobRegistrar,
|
||||
policyEnforcementService,
|
||||
ownershipService,
|
||||
);
|
||||
|
||||
beforeEach(() => vi.clearAllMocks());
|
||||
@@ -1177,6 +1440,8 @@ describe('ActiveWorkflowManager', () => {
|
||||
mock(),
|
||||
scheduleTriggerJobRegistrar,
|
||||
mock(), // pollTriggerJobRegistrar
|
||||
policyEnforcementService,
|
||||
ownershipService,
|
||||
);
|
||||
|
||||
const makeWorkflow = () => {
|
||||
|
||||
@@ -44,6 +44,8 @@ import { ActivationErrorsService } from '@/activation-errors.service';
|
||||
import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
|
||||
import { ExternalHooks } from '@/external-hooks';
|
||||
import { NodeTypes } from '@/node-types';
|
||||
import { PolicyEnforcementService } from '@/policy/policy-enforcement.service';
|
||||
import { isPolicyRefusal } from '@/policy/policy-violation.error';
|
||||
import { Push } from '@/push';
|
||||
import { Publisher } from '@/scaling/pubsub/publisher.service';
|
||||
import { PubSubCommandMap } from '@/scaling/pubsub/pubsub.event-map';
|
||||
@@ -51,6 +53,7 @@ import type { ScheduleTriggerCollectionSession } from '@/scheduling/schedule-tri
|
||||
import { PollTriggerJobRegistrar } from '@/scheduling/poll-trigger-node/poll-trigger-job-registrar';
|
||||
import { ScheduleTriggerJobRegistrar } from '@/scheduling/schedule-trigger-node/schedule-trigger-job-registrar';
|
||||
import { ActiveWorkflowsService } from '@/services/active-workflows.service';
|
||||
import { OwnershipService } from '@/services/ownership.service';
|
||||
import * as WebhookHelpers from '@/webhooks/webhook-helpers';
|
||||
import { WebhookService } from '@/webhooks/webhook.service';
|
||||
import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';
|
||||
@@ -89,6 +92,8 @@ export class ActiveWorkflowManager {
|
||||
private readonly eventBus: MessageEventBus,
|
||||
private readonly scheduleTriggerJobRegistrar: ScheduleTriggerJobRegistrar,
|
||||
private readonly pollTriggerJobRegistrar: PollTriggerJobRegistrar,
|
||||
private readonly policyEnforcementService: PolicyEnforcementService,
|
||||
private readonly ownershipService: OwnershipService,
|
||||
) {
|
||||
this.logger = this.logger.scoped(['workflow-activation']);
|
||||
}
|
||||
@@ -446,6 +451,16 @@ export class ActiveWorkflowManager {
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
// An expected refusal, permanent until policy or workflow changes: no fault
|
||||
// report, no error workflow and no retry, or every restart alerts.
|
||||
if (isPolicyRefusal(error)) {
|
||||
this.logger.warn(`Publication of ${formatWorkflow(dbWorkflow)} blocked by policy`, {
|
||||
workflowId: dbWorkflow.id,
|
||||
});
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
this.errorReporter.error(error);
|
||||
this.logger.error(
|
||||
`Issue on initial workflow activation try of ${formatWorkflow(dbWorkflow)} (startup)`,
|
||||
@@ -583,6 +598,17 @@ export class ActiveWorkflowManager {
|
||||
dbWorkflow.nodes = nodes;
|
||||
dbWorkflow.connections = connections;
|
||||
|
||||
// Trigger and poller nodes run code at registration, so this gates startup
|
||||
// and leadership change too, not just the activate button.
|
||||
if (this.policyEnforcementService.hasChecksFor('workflowPublish')) {
|
||||
const project = await this.ownershipService.getWorkflowProjectCached(dbWorkflow.id);
|
||||
|
||||
await this.policyEnforcementService.enforceWorkflowPublish({
|
||||
workflow: { id: dbWorkflow.id, name: dbWorkflow.name, nodes },
|
||||
projectId: project.id,
|
||||
});
|
||||
}
|
||||
|
||||
workflow = new Workflow({
|
||||
id: dbWorkflow.id,
|
||||
name: dbWorkflow.name,
|
||||
@@ -741,44 +767,48 @@ export class ActiveWorkflowManager {
|
||||
return;
|
||||
}
|
||||
|
||||
const dbWorkflow = await this.workflowRepository.findById(workflowId);
|
||||
// A policy refusal happens before anything is registered, so there is
|
||||
// nothing to tear down — and unpublishing is not this path's call to make.
|
||||
if (!isPolicyRefusal(error)) {
|
||||
const dbWorkflow = await this.workflowRepository.findById(workflowId);
|
||||
|
||||
// Activation may have failed partway with triggers already registered,
|
||||
// in memory and as durable jobs. Tear them down before the
|
||||
// deactivation below so the active version is still resolvable, or
|
||||
// they keep firing a workflow marked inactive. Each teardown is caught
|
||||
// on its own so a webhook failure never skips the durable-job cleanup.
|
||||
try {
|
||||
await this.clearWebhooks(workflowId);
|
||||
} catch (cleanupError) {
|
||||
this.logger.error(`Failed to remove webhooks of workflow "${workflowId}"`, {
|
||||
workflowId,
|
||||
error: ensureError(cleanupError),
|
||||
});
|
||||
}
|
||||
|
||||
try {
|
||||
await this.removeNonWebhookTriggers(workflowId);
|
||||
} catch (cleanupError) {
|
||||
this.logger.error(`Failed to remove triggers of workflow "${workflowId}"`, {
|
||||
workflowId,
|
||||
error: ensureError(cleanupError),
|
||||
});
|
||||
}
|
||||
|
||||
await this.workflowRepository.update(workflowId, { active: false, activeVersionId: null });
|
||||
|
||||
if (dbWorkflow && (activationMode === 'init' || activationMode === 'leadershipChange')) {
|
||||
void this.eventBus.sendAuditEvent({
|
||||
eventName: 'n8n.audit.workflow.deactivated',
|
||||
payload: {
|
||||
// Activation may have failed partway with triggers already registered,
|
||||
// in memory and as durable jobs. Tear them down before the
|
||||
// deactivation below so the active version is still resolvable, or
|
||||
// they keep firing a workflow marked inactive. Each teardown is caught
|
||||
// on its own so a webhook failure never skips the durable-job cleanup.
|
||||
try {
|
||||
await this.clearWebhooks(workflowId);
|
||||
} catch (cleanupError) {
|
||||
this.logger.error(`Failed to remove webhooks of workflow "${workflowId}"`, {
|
||||
workflowId,
|
||||
workflowName: dbWorkflow.name,
|
||||
deactivatedVersionId: dbWorkflow.activeVersionId ?? null,
|
||||
activationMode,
|
||||
reason: error.name,
|
||||
},
|
||||
});
|
||||
error: ensureError(cleanupError),
|
||||
});
|
||||
}
|
||||
|
||||
try {
|
||||
await this.removeNonWebhookTriggers(workflowId);
|
||||
} catch (cleanupError) {
|
||||
this.logger.error(`Failed to remove triggers of workflow "${workflowId}"`, {
|
||||
workflowId,
|
||||
error: ensureError(cleanupError),
|
||||
});
|
||||
}
|
||||
|
||||
await this.workflowRepository.update(workflowId, { active: false, activeVersionId: null });
|
||||
|
||||
if (dbWorkflow && (activationMode === 'init' || activationMode === 'leadershipChange')) {
|
||||
void this.eventBus.sendAuditEvent({
|
||||
eventName: 'n8n.audit.workflow.deactivated',
|
||||
payload: {
|
||||
workflowId,
|
||||
workflowName: dbWorkflow.name,
|
||||
deactivatedVersionId: dbWorkflow.activeVersionId ?? null,
|
||||
activationMode,
|
||||
reason: error.name,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
this.push.broadcast({
|
||||
@@ -833,6 +863,7 @@ export class ActiveWorkflowManager {
|
||||
) {
|
||||
const workflowId = workflowData.id;
|
||||
const workflowName = workflowData.name;
|
||||
const own: { activation?: QueuedActivation } = {};
|
||||
|
||||
const retryFunction = async () => {
|
||||
this.logger.info(`Try to activate workflow "${workflowName}" (${workflowId})`, {
|
||||
@@ -842,7 +873,23 @@ export class ActiveWorkflowManager {
|
||||
try {
|
||||
await this.add(workflowId, activationMode, workflowData, { shouldPublish: false });
|
||||
} catch (error) {
|
||||
// An expected refusal, permanent until policy or workflow changes: no fault
|
||||
// report, and leave the queue rather than retrying forever.
|
||||
if (isPolicyRefusal(error)) {
|
||||
this.logger.warn(`Publication of workflow "${workflowId}" blocked by policy`, {
|
||||
workflowId,
|
||||
});
|
||||
|
||||
// Only our own entry: a newer failure may have replaced it since.
|
||||
if (this.queuedActivations[workflowId] === own.activation) {
|
||||
this.removeQueuedWorkflowActivation(workflowId);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
this.errorReporter.error(error);
|
||||
|
||||
const queuedActivation = this.queuedActivations[workflowId];
|
||||
if (!queuedActivation) {
|
||||
return;
|
||||
@@ -878,12 +925,13 @@ export class ActiveWorkflowManager {
|
||||
// multiple run in parallel
|
||||
this.removeQueuedWorkflowActivation(workflowId);
|
||||
|
||||
this.queuedActivations[workflowId] = {
|
||||
own.activation = {
|
||||
activationMode,
|
||||
lastTimeout: WORKFLOW_REACTIVATE_INITIAL_TIMEOUT,
|
||||
timeout: setTimeout(retryFunction, WORKFLOW_REACTIVATE_INITIAL_TIMEOUT),
|
||||
workflowData,
|
||||
};
|
||||
this.queuedActivations[workflowId] = own.activation;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -7,6 +7,7 @@ import type { INode, IRun, IWorkflowBase } from 'n8n-workflow';
|
||||
import { createRunExecutionData, NodeOperationError } from 'n8n-workflow';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import { PolicyViolationError } from '@/policy/policy-violation.error';
|
||||
import { OwnershipService } from '@/services/ownership.service';
|
||||
import { UrlService } from '@/services/url.service';
|
||||
import { WorkflowExecutionService } from '@/workflows/workflow-execution.service';
|
||||
@@ -126,4 +127,85 @@ describe('executeErrorWorkflow', () => {
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('policy violations', () => {
|
||||
const makeRunData = (error: Error): IRun => ({
|
||||
data: createRunExecutionData({
|
||||
resultData: { error: error as IRun['data']['resultData']['error'], runData: {} },
|
||||
}),
|
||||
mode: 'trigger',
|
||||
startedAt: new Date(),
|
||||
storedAt: 'db',
|
||||
status: 'error',
|
||||
});
|
||||
|
||||
const violation = () =>
|
||||
new PolicyViolationError([
|
||||
{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
|
||||
]);
|
||||
|
||||
// A refusal repeats on every attempt, so it must not reach user automation.
|
||||
it('does not run a configured error workflow', () => {
|
||||
const workflowData = mock<IWorkflowBase>({
|
||||
id: 'workflow-123',
|
||||
settings: { errorWorkflow: 'error-workflow-456' },
|
||||
nodes: [],
|
||||
});
|
||||
|
||||
executeErrorWorkflow(workflowData, makeRunData(violation()), 'trigger');
|
||||
|
||||
expect(workflowExecutionService.executeErrorWorkflow).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// The worse case: this would execute the graph policy just refused.
|
||||
it('does not run the workflow through its own Error Trigger', () => {
|
||||
const workflowData = mock<IWorkflowBase>({
|
||||
id: 'workflow-123',
|
||||
settings: {},
|
||||
nodes: [mock<INode>({ type: 'n8n-nodes-base.errorTrigger' })],
|
||||
});
|
||||
|
||||
executeErrorWorkflow(workflowData, makeRunData(violation()), 'internal');
|
||||
|
||||
expect(workflowExecutionService.executeErrorWorkflow).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// What `WorkflowRunner.processError` hands to the lifecycle hook: the error
|
||||
// spread into a plain object, so `instanceof` no longer holds.
|
||||
it('does not run the error workflow for a flattened refusal', () => {
|
||||
const error = violation();
|
||||
const workflowData = mock<IWorkflowBase>({
|
||||
id: 'workflow-123',
|
||||
settings: { errorWorkflow: 'error-workflow-456' },
|
||||
nodes: [],
|
||||
});
|
||||
|
||||
executeErrorWorkflow(
|
||||
workflowData,
|
||||
makeRunData({ ...error, message: error.message, stack: error.stack } as Error),
|
||||
'trigger',
|
||||
'execution-1',
|
||||
);
|
||||
|
||||
expect(workflowExecutionService.executeErrorWorkflow).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('still runs the error workflow for an ordinary failure', async () => {
|
||||
const workflowData = mock<IWorkflowBase>({
|
||||
id: 'workflow-123',
|
||||
settings: { errorWorkflow: 'error-workflow-456' },
|
||||
nodes: [],
|
||||
});
|
||||
ownershipService.getWorkflowProjectCached.mockResolvedValue({ id: 'project-1' } as never);
|
||||
|
||||
executeErrorWorkflow(
|
||||
workflowData,
|
||||
makeRunData(new NodeOperationError(mockNode, 'Test error')),
|
||||
'trigger',
|
||||
);
|
||||
await new Promise(process.nextTick);
|
||||
|
||||
expect(workflowExecutionService.executeErrorWorkflow).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -5,6 +5,7 @@ import { ErrorReporter } from 'n8n-core';
|
||||
import type { IRun, IWorkflowBase, WorkflowExecuteMode } from 'n8n-workflow';
|
||||
|
||||
import type { IWorkflowErrorData } from '@/interfaces';
|
||||
import { isPolicyRefusal } from '@/policy/policy-violation.error';
|
||||
import { OwnershipService } from '@/services/ownership.service';
|
||||
import { UrlService } from '@/services/url.service';
|
||||
|
||||
@@ -55,6 +56,11 @@ export function executeErrorWorkflow(
|
||||
}
|
||||
|
||||
if (fullRunData.data.resultData.error !== undefined) {
|
||||
// An administrative refusal, not a workflow failure — and where a workflow is
|
||||
// its own error handler, this would run the graph policy just refused.
|
||||
// Not `instanceof`: a failed execution arrives here with its error flattened.
|
||||
if (isPolicyRefusal(fullRunData.data.resultData.error)) return;
|
||||
|
||||
let workflowErrorData: IWorkflowErrorData;
|
||||
const workflowId = workflowData.id;
|
||||
|
||||
|
||||
@@ -4,7 +4,11 @@ import { UserError } from 'n8n-workflow';
|
||||
import { classifyHttpError, HttpErrorKind } from '@/errors/http-error-classifier';
|
||||
import { serializeInternalRestError } from '@/errors/http-error-serializers';
|
||||
|
||||
import { PolicyViolationError, type NonEmptyViolations } from '../policy-violation.error';
|
||||
import {
|
||||
isPolicyRefusal,
|
||||
PolicyViolationError,
|
||||
type NonEmptyViolations,
|
||||
} from '../policy-violation.error';
|
||||
|
||||
const violation = (overrides: Partial<PolicyViolation> = {}): PolicyViolation => ({
|
||||
kind: 'node-type-unavailable',
|
||||
@@ -88,4 +92,34 @@ describe('PolicyViolationError', () => {
|
||||
expect(body.meta).toEqual({ violations: [violation()] });
|
||||
});
|
||||
});
|
||||
|
||||
describe('isPolicyRefusal', () => {
|
||||
const violation = () =>
|
||||
new PolicyViolationError([
|
||||
{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked' },
|
||||
]);
|
||||
|
||||
it('recognises a live error', () => {
|
||||
expect(isPolicyRefusal(violation())).toBe(true);
|
||||
});
|
||||
|
||||
// `WorkflowRunner.processError` spreads a failed execution's error into a plain
|
||||
// object, dropping the prototype, so `instanceof` no longer holds.
|
||||
it('recognises one flattened by execution failure serialization', () => {
|
||||
const error = violation();
|
||||
const flattened = { ...error, message: error.message, stack: error.stack };
|
||||
|
||||
expect(flattened instanceof PolicyViolationError).toBe(false);
|
||||
expect(isPolicyRefusal(flattened)).toBe(true);
|
||||
});
|
||||
|
||||
it.each([
|
||||
['an ordinary error', new Error('boom')],
|
||||
['an unrelated object', { violations: [] }],
|
||||
['null', null],
|
||||
['undefined', undefined],
|
||||
])('does not recognise %s', (_label, value) => {
|
||||
expect(isPolicyRefusal(value)).toBe(false);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -31,6 +31,13 @@ export class PolicyViolationError extends UserError {
|
||||
|
||||
readonly meta: { violations: PolicyViolation[] };
|
||||
|
||||
/**
|
||||
* An own property, so it survives the flattening that execution failure
|
||||
* serialization does to errors — where `instanceof` no longer holds. Read it
|
||||
* through {@link isPolicyRefusal}, never directly.
|
||||
*/
|
||||
readonly isPolicyRefusal = true;
|
||||
|
||||
/**
|
||||
* @param violations All of them, not just the first — a user fixing a workflow deserves the
|
||||
* whole list.
|
||||
@@ -45,3 +52,23 @@ export class PolicyViolationError extends UserError {
|
||||
this.meta = { violations: this.violations };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether an error is a policy refusal, including one flattened into a plain
|
||||
* object by execution failure serialization, where `instanceof` no longer holds.
|
||||
*
|
||||
* The default way to ask. Use `instanceof` only where the typed instance itself
|
||||
* is needed, e.g. to read {@link PolicyViolationError.violations}.
|
||||
*/
|
||||
export function isPolicyRefusal(error: unknown): boolean {
|
||||
if (error instanceof PolicyViolationError) return true;
|
||||
|
||||
// `hasOwn`, not `in`: the marker is an own property by contract, and an inherited
|
||||
// one would let an unrelated error suppress cleanup and retries.
|
||||
return (
|
||||
typeof error === 'object' &&
|
||||
error !== null &&
|
||||
Object.hasOwn(error, 'isPolicyRefusal') &&
|
||||
(error as { isPolicyRefusal?: unknown }).isPolicyRefusal === true
|
||||
);
|
||||
}
|
||||
|
||||
@@ -35,6 +35,7 @@ import type { RedactionEnforcementService } from '@/modules/redaction/redaction-
|
||||
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 { PollTriggerJobRegistrar } from '@/scheduling/poll-trigger-node/poll-trigger-job-registrar';
|
||||
import type { ScheduleTriggerJobRegistrar } from '@/scheduling/schedule-trigger-node/schedule-trigger-job-registrar';
|
||||
import type { OwnershipService } from '@/services/ownership.service';
|
||||
@@ -1062,6 +1063,8 @@ describe('WorkflowService', () => {
|
||||
let pollTriggerJobRegistrarMock: MockProxy<PollTriggerJobRegistrar>;
|
||||
let workflowPublishGuardMock: MockProxy<WorkflowPublishGuardProxy>;
|
||||
let workflowMutationHooksMock: MockProxy<WorkflowMutationHooksProxy>;
|
||||
let policyEnforcementServiceMock: MockProxy<PolicyEnforcementService>;
|
||||
let ownershipServiceMock: MockProxy<OwnershipService>;
|
||||
|
||||
const WORKFLOW_ID = 'workflow-1';
|
||||
const PREVIOUS_VERSION_ID = 'v1';
|
||||
@@ -1117,6 +1120,14 @@ describe('WorkflowService', () => {
|
||||
pollTriggerJobRegistrarMock = mock();
|
||||
workflowPublishGuardMock = mock<WorkflowPublishGuardProxy>();
|
||||
workflowMutationHooksMock = mock<WorkflowMutationHooksProxy>();
|
||||
// Stands in for the dummy always-allow check: with no backend registered the
|
||||
// real service clears every publish.
|
||||
policyEnforcementServiceMock = mock<PolicyEnforcementService>();
|
||||
policyEnforcementServiceMock.hasChecksFor.mockReturnValue(true);
|
||||
ownershipServiceMock = mock<OwnershipService>();
|
||||
ownershipServiceMock.getWorkflowProjectCached.mockResolvedValue(
|
||||
mock<Project>({ id: 'project-1' }),
|
||||
);
|
||||
|
||||
workflowRepositoryMock.create.mockImplementation(
|
||||
(data) => Object.assign(new WorkflowEntity(), data) as WorkflowEntity,
|
||||
@@ -1127,7 +1138,7 @@ describe('WorkflowService', () => {
|
||||
mock(), // sharedWorkflowRepository
|
||||
workflowRepositoryMock, // workflowRepository
|
||||
mock(), // workflowTagMappingRepository
|
||||
mock(), // ownershipService
|
||||
ownershipServiceMock, // ownershipService
|
||||
mock(), // tagService
|
||||
workflowHistoryServiceMock, // workflowHistoryService
|
||||
externalHooksMock, // externalHooks
|
||||
@@ -1156,7 +1167,7 @@ describe('WorkflowService', () => {
|
||||
workflowHookContextServiceMock, // workflowHookContextService
|
||||
workflowPublishGuardMock, // workflowPublishGuard
|
||||
workflowMutationHooksMock, // workflowMutationHooks
|
||||
mock(), // policyEnforcementService
|
||||
policyEnforcementServiceMock, // policyEnforcementService
|
||||
);
|
||||
|
||||
// Bypass validation internals
|
||||
@@ -1622,6 +1633,175 @@ describe('WorkflowService', () => {
|
||||
|
||||
expect(externalHooksMock.run).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
describe('policy enforcement', () => {
|
||||
const arrangeSuccessfulActivation = (workflow: WorkflowEntity) => {
|
||||
workflowFinderServiceMock.findWorkflowForUser.mockResolvedValue(workflow);
|
||||
workflowRepositoryMock.findOne.mockResolvedValue(workflow);
|
||||
externalHooksMock.run.mockResolvedValue(undefined);
|
||||
vi.spyOn(
|
||||
workflowService as unknown as { _addToActiveWorkflowManager: () => Promise<void> },
|
||||
'_addToActiveWorkflowManager',
|
||||
).mockResolvedValue(undefined);
|
||||
};
|
||||
|
||||
test('enforces the publish with the version being activated and the owning project', async () => {
|
||||
const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID });
|
||||
const versionToActivate = makeVersionToActivate();
|
||||
arrangeSuccessfulActivation(workflow);
|
||||
workflowHistoryServiceMock.getVersion.mockResolvedValue(versionToActivate);
|
||||
|
||||
await workflowService.activateWorkflow(mock<User>(), WORKFLOW_ID, {
|
||||
versionId: TARGET_VERSION_ID,
|
||||
});
|
||||
|
||||
expect(policyEnforcementServiceMock.enforceWorkflowPublish).toHaveBeenCalledExactlyOnceWith(
|
||||
{
|
||||
workflow: {
|
||||
id: WORKFLOW_ID,
|
||||
name: workflow.name,
|
||||
nodes: versionToActivate.nodes,
|
||||
},
|
||||
projectId: 'project-1',
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
// The candidate shares its node array with the version row, so an in-place
|
||||
// mutation would otherwise be policed even though it is never persisted.
|
||||
test('polices the version even when the hook mutates the nodes in place', async () => {
|
||||
const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID });
|
||||
const versionToActivate = makeVersionToActivate();
|
||||
const originalNodes = [...versionToActivate.nodes];
|
||||
arrangeSuccessfulActivation(workflow);
|
||||
workflowHistoryServiceMock.getVersion.mockResolvedValue(versionToActivate);
|
||||
|
||||
externalHooksMock.run.mockImplementation(async (_name, args) => {
|
||||
const [candidate] = args as unknown as [WorkflowEntity];
|
||||
candidate.nodes.push({ name: 'Injected by hook' } as INode);
|
||||
});
|
||||
|
||||
await workflowService.activateWorkflow(mock<User>(), WORKFLOW_ID, {
|
||||
versionId: TARGET_VERSION_ID,
|
||||
});
|
||||
|
||||
expect(policyEnforcementServiceMock.enforceWorkflowPublish).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
workflow: expect.objectContaining({ nodes: originalNodes }),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
// Only the version row gets registered, so a hook that rewrites the candidate
|
||||
// changes a graph that never runs.
|
||||
test('polices the version being published, not a hook-mutated candidate', async () => {
|
||||
const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID });
|
||||
const versionToActivate = makeVersionToActivate();
|
||||
arrangeSuccessfulActivation(workflow);
|
||||
workflowHistoryServiceMock.getVersion.mockResolvedValue(versionToActivate);
|
||||
|
||||
externalHooksMock.run.mockImplementation(async (_name, args) => {
|
||||
const [candidate] = args as unknown as [WorkflowEntity];
|
||||
candidate.nodes = [{ name: 'Rewritten by hook' } as INode];
|
||||
});
|
||||
|
||||
await workflowService.activateWorkflow(mock<User>(), WORKFLOW_ID, {
|
||||
versionId: TARGET_VERSION_ID,
|
||||
});
|
||||
|
||||
expect(policyEnforcementServiceMock.enforceWorkflowPublish).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
workflow: expect.objectContaining({ nodes: versionToActivate.nodes }),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
// Unlike the review gate: re-applying the current version still re-registers.
|
||||
test('enforces when re-applying the already-published version', async () => {
|
||||
const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID });
|
||||
arrangeSuccessfulActivation(workflow);
|
||||
workflowHistoryServiceMock.getVersion.mockResolvedValue(makeActiveVersion());
|
||||
|
||||
await workflowService.activateWorkflow(mock<User>(), WORKFLOW_ID, {
|
||||
versionId: PREVIOUS_VERSION_ID,
|
||||
});
|
||||
|
||||
expect(workflowPublishGuardMock.assertCanPublish).not.toHaveBeenCalled();
|
||||
expect(policyEnforcementServiceMock.enforceWorkflowPublish).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test.each([
|
||||
['current publication path', false],
|
||||
['outbox publication path', true],
|
||||
] as const)(
|
||||
'publishes nothing on the %s when policy blocks the version',
|
||||
async (_path, useWorkflowPublicationService) => {
|
||||
globalConfigMock.workflows.useWorkflowPublicationService = useWorkflowPublicationService;
|
||||
const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID });
|
||||
arrangeSuccessfulActivation(workflow);
|
||||
workflowHistoryServiceMock.getVersion.mockResolvedValue(makeVersionToActivate());
|
||||
policyEnforcementServiceMock.enforceWorkflowPublish.mockRejectedValue(
|
||||
new PolicyViolationError([
|
||||
{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked' },
|
||||
]),
|
||||
);
|
||||
|
||||
await expect(
|
||||
workflowService.activateWorkflow(mock<User>(), WORKFLOW_ID, {
|
||||
versionId: TARGET_VERSION_ID,
|
||||
}),
|
||||
).rejects.toBeInstanceOf(PolicyViolationError);
|
||||
|
||||
expect(workflow.activeVersionId).toBe(PREVIOUS_VERSION_ID);
|
||||
expect(workflowRepositoryMock.update).not.toHaveBeenCalled();
|
||||
expect(activeWorkflowManagerMock.add).not.toHaveBeenCalled();
|
||||
expect(activeWorkflowManagerMock.remove).not.toHaveBeenCalled();
|
||||
expect(outboxRepositoryMock.enqueue).not.toHaveBeenCalled();
|
||||
expect(workflowPublishHistoryRepositoryMock.addRecord).not.toHaveBeenCalled();
|
||||
expect(workflowMutationHooksMock.afterWorkflowPublished).not.toHaveBeenCalled();
|
||||
},
|
||||
);
|
||||
|
||||
test('does not enforce while unpublishing', async () => {
|
||||
const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID });
|
||||
workflowFinderServiceMock.findWorkflowForUser.mockResolvedValue(workflow);
|
||||
|
||||
await workflowService.deactivateWorkflow(mock<User>(), WORKFLOW_ID);
|
||||
|
||||
expect(policyEnforcementServiceMock.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// An unevaluated project rule is not a passed one, so the lookup is unguarded.
|
||||
test('propagates a failed ownership lookup instead of policing a null scope', async () => {
|
||||
const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID });
|
||||
arrangeSuccessfulActivation(workflow);
|
||||
workflowHistoryServiceMock.getVersion.mockResolvedValue(makeVersionToActivate());
|
||||
ownershipServiceMock.getWorkflowProjectCached.mockRejectedValue(new Error('no owner row'));
|
||||
|
||||
await expect(
|
||||
workflowService.activateWorkflow(mock<User>(), WORKFLOW_ID, {
|
||||
versionId: TARGET_VERSION_ID,
|
||||
}),
|
||||
).rejects.toThrow('no owner row');
|
||||
|
||||
expect(policyEnforcementServiceMock.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// A feature that is merely absent must not cost a lookup on every publish.
|
||||
test('does not resolve ownership when no check is registered', async () => {
|
||||
const workflow = makeWorkflowEntity({ activeVersionId: PREVIOUS_VERSION_ID });
|
||||
arrangeSuccessfulActivation(workflow);
|
||||
workflowHistoryServiceMock.getVersion.mockResolvedValue(makeVersionToActivate());
|
||||
policyEnforcementServiceMock.hasChecksFor.mockReturnValue(false);
|
||||
|
||||
await workflowService.activateWorkflow(mock<User>(), WORKFLOW_ID, {
|
||||
versionId: TARGET_VERSION_ID,
|
||||
});
|
||||
|
||||
expect(ownershipServiceMock.getWorkflowProjectCached).not.toHaveBeenCalled();
|
||||
expect(policyEnforcementServiceMock.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('deactivateWorkflowAsSystem()', () => {
|
||||
|
||||
@@ -10,6 +10,7 @@ import { mock } from 'vitest-mock-extended';
|
||||
import type { ErrorReporter } from 'n8n-core';
|
||||
|
||||
import type { ActivationErrorsService } from '@/activation-errors.service';
|
||||
import { PolicyViolationError } from '@/policy/policy-violation.error';
|
||||
import type { Push } from '@/push';
|
||||
import type { Publisher } from '@/scaling/pubsub/publisher.service';
|
||||
import { PublicationStatusReporter } from '@/workflows/publication/publication-status-reporter';
|
||||
@@ -305,6 +306,23 @@ describe('PublicationStatusReporter', () => {
|
||||
});
|
||||
});
|
||||
|
||||
// An expected denial: the record must still fail and the UI must still be told,
|
||||
// but it is not a fault to report.
|
||||
test('failed by policy marks the record failed without reporting a fault', async () => {
|
||||
const error = new PolicyViolationError([
|
||||
{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
|
||||
]);
|
||||
|
||||
await reporter.report(makeRecord(), { type: 'failed', error });
|
||||
|
||||
expect(errorReporter.error).not.toHaveBeenCalled();
|
||||
expect(outboxRepository.markFailed).toHaveBeenCalledWith(1, error.message, entityManager);
|
||||
expect(push.broadcast).toHaveBeenCalledWith({
|
||||
type: 'workflowFailedToActivate',
|
||||
data: { workflowId: 'wf-1', errorMessage: error.message },
|
||||
});
|
||||
});
|
||||
|
||||
test('failed with triggerStatuses writes rows before marking failed', async () => {
|
||||
const error = new Error('partial registration failed');
|
||||
|
||||
|
||||
+133
@@ -15,6 +15,9 @@ import { WebhookPathTakenError } from 'n8n-workflow';
|
||||
import { TELEMETRY_EVENT } from '@n8n/telemetry';
|
||||
|
||||
import type { NodeTypes } from '@/node-types';
|
||||
import type { PolicyEnforcementService } from '@/policy/policy-enforcement.service';
|
||||
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 { WorkflowTriggerActivator } from '@/workflows/triggers/workflow-trigger-activator';
|
||||
@@ -32,6 +35,9 @@ describe('WorkflowPublicationApplier', () => {
|
||||
const nodeTypes = mock<NodeTypes>();
|
||||
const workflowService = mock<WorkflowService>();
|
||||
const telemetry = mock<Telemetry>();
|
||||
// Clears by default, which is what the real service does with no policy backend.
|
||||
const policyEnforcementService = mock<PolicyEnforcementService>();
|
||||
const ownershipService = mock<OwnershipService>();
|
||||
|
||||
const applier = new WorkflowPublicationApplier(
|
||||
logger,
|
||||
@@ -43,6 +49,8 @@ describe('WorkflowPublicationApplier', () => {
|
||||
nodeTypes,
|
||||
workflowService,
|
||||
telemetry,
|
||||
policyEnforcementService,
|
||||
ownershipService,
|
||||
);
|
||||
|
||||
function makeRecord(
|
||||
@@ -129,6 +137,10 @@ describe('WorkflowPublicationApplier', () => {
|
||||
workflowTriggerActivator.getTriggerKinds.mockImplementation(
|
||||
(nodes) => new Map(nodes.map((node) => [node.id, 'in-memory'])),
|
||||
);
|
||||
ownershipService.getWorkflowProjectCached.mockResolvedValue(mock({ id: 'project-1' }));
|
||||
// `clearAllMocks` keeps implementations, so restore the clearing default.
|
||||
policyEnforcementService.hasChecksFor.mockReturnValue(true);
|
||||
policyEnforcementService.enforceWorkflowPublish.mockResolvedValue(mock());
|
||||
});
|
||||
|
||||
test('skips with workflow-not-found when the workflow is gone', async () => {
|
||||
@@ -242,6 +254,127 @@ describe('WorkflowPublicationApplier', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('policy enforcement', () => {
|
||||
const violation = () =>
|
||||
new PolicyViolationError([
|
||||
{ kind: 'node-type-unavailable', checkId: 'check-1', message: 'Blocked by policy' },
|
||||
]);
|
||||
|
||||
test('enforces with the version being published and the owning project', async () => {
|
||||
workflowRepository.findOneBy.mockResolvedValue(
|
||||
makeWorkflow({ activeVersionId: 'v-1', name: 'My workflow' }),
|
||||
);
|
||||
const versionWithNodes = {
|
||||
...makeVersion('v-2'),
|
||||
nodes: [triggerNode('a')],
|
||||
} as WorkflowHistory;
|
||||
workflowHistoryRepository.findOneBy.mockResolvedValue(versionWithNodes);
|
||||
|
||||
await applier.apply(makeRecord(), abort);
|
||||
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).toHaveBeenCalledExactlyOnceWith({
|
||||
workflow: {
|
||||
id: 'wf-1',
|
||||
name: 'My workflow',
|
||||
nodes: versionWithNodes.nodes,
|
||||
},
|
||||
projectId: 'project-1',
|
||||
});
|
||||
});
|
||||
|
||||
// The no-change branch still advances the published version, which running
|
||||
// triggers re-read on their next fire.
|
||||
test('enforces even when the trigger diff is empty', async () => {
|
||||
const trigger = triggerNode('a');
|
||||
setTriggerSets([trigger], [{ ...trigger }]);
|
||||
|
||||
const result = await applier.apply(makeRecord(), abort);
|
||||
|
||||
expect(result.type).toBe('completed');
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test('fails the record without advancing or touching triggers when policy blocks', async () => {
|
||||
policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(violation());
|
||||
|
||||
const result = await applier.apply(makeRecord(), abort);
|
||||
|
||||
expect(result).toMatchObject({ type: 'failed' });
|
||||
expect((result as { error: Error }).error).toBeInstanceOf(PolicyViolationError);
|
||||
expect(workflowPublishedVersionRepository.setPublishedVersion).not.toHaveBeenCalled();
|
||||
expect(workflowTriggerActivator.activate).not.toHaveBeenCalled();
|
||||
expect(workflowTriggerActivator.deactivate).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// Left as `activated`, these rows read as drift and get re-enqueued forever.
|
||||
test('reports every desired trigger as failed so no activated rows survive', async () => {
|
||||
workflowTriggerActivator.getEnabledTriggerNodes.mockReturnValue([
|
||||
triggerNode('a'),
|
||||
triggerNode('b'),
|
||||
]);
|
||||
policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(violation());
|
||||
|
||||
const result = await applier.apply(makeRecord(), abort);
|
||||
|
||||
expect(result).toMatchObject({
|
||||
type: 'failed',
|
||||
triggerStatuses: [
|
||||
{
|
||||
nodeId: 'a',
|
||||
nodeName: 'a',
|
||||
status: 'failed',
|
||||
triggerKind: 'in-memory',
|
||||
errorMessage: 'Blocked by policy',
|
||||
},
|
||||
{
|
||||
nodeId: 'b',
|
||||
nodeName: 'b',
|
||||
status: 'failed',
|
||||
triggerKind: 'in-memory',
|
||||
errorMessage: 'Blocked by policy',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
// An unevaluated project rule is not a passed one, so the lookup is unguarded.
|
||||
test('propagates a failed ownership lookup instead of policing a null scope', async () => {
|
||||
ownershipService.getWorkflowProjectCached.mockRejectedValue(new Error('no owner row'));
|
||||
|
||||
await expect(applier.apply(makeRecord(), abort)).rejects.toThrow('no owner row');
|
||||
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// A feature that is merely absent must not cost a lookup on every publication.
|
||||
test('does not resolve ownership when no check is registered', async () => {
|
||||
policyEnforcementService.hasChecksFor.mockReturnValue(false);
|
||||
|
||||
const result = await applier.apply(makeRecord(), abort);
|
||||
|
||||
expect(result.type).toBe('completed');
|
||||
expect(ownershipService.getWorkflowProjectCached).not.toHaveBeenCalled();
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('does not enforce while unpublishing', async () => {
|
||||
workflowRepository.findOneBy.mockResolvedValue(
|
||||
makeWorkflow({ active: true, activeVersionId: null }),
|
||||
);
|
||||
|
||||
await applier.apply(makeRecord(), abort);
|
||||
|
||||
expect(policyEnforcementService.enforceWorkflowPublish).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// Only a violation is a verdict; a broken check must not read as "blocked".
|
||||
test('propagates a non-violation error instead of failing the record', async () => {
|
||||
policyEnforcementService.enforceWorkflowPublish.mockRejectedValue(new Error('boom'));
|
||||
|
||||
await expect(applier.apply(makeRecord(), abort)).rejects.toThrow('boom');
|
||||
});
|
||||
});
|
||||
|
||||
test('returns version-missing when the published version history row is gone', async () => {
|
||||
workflowHistoryRepository.findOneBy.mockResolvedValue(null);
|
||||
|
||||
|
||||
@@ -12,6 +12,7 @@ import { ErrorReporter } from 'n8n-core';
|
||||
import { OperationalError } from 'n8n-workflow';
|
||||
|
||||
import { ActivationErrorsService } from '@/activation-errors.service';
|
||||
import { isPolicyRefusal } from '@/policy/policy-violation.error';
|
||||
import { Push } from '@/push';
|
||||
import { Publisher } from '@/scaling/pubsub/publisher.service';
|
||||
import type {
|
||||
@@ -103,7 +104,11 @@ export class PublicationStatusReporter {
|
||||
}
|
||||
await this.outboxRepository.markFailed(record.id, result.error.message, trx);
|
||||
});
|
||||
this.errorReporter.error(result.error, { shouldBeLogged: true });
|
||||
// An expected denial, already logged as a warning by the applier — the
|
||||
// terminal state and the UI push stand, the fault report does not.
|
||||
if (!isPolicyRefusal(result.error)) {
|
||||
this.errorReporter.error(result.error, { shouldBeLogged: true });
|
||||
}
|
||||
this.pushFailedToActivate(record.workflowId, result.error.message);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -15,6 +15,9 @@ import { ensureError } from '@n8n/utils/errors/ensure-error';
|
||||
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 { OwnershipService } from '@/services/ownership.service';
|
||||
import { Telemetry } from '@/telemetry';
|
||||
import { healNodeIds } from '@/workflows/publication/heal-node-ids';
|
||||
import type {
|
||||
@@ -69,6 +72,8 @@ export class WorkflowPublicationApplier {
|
||||
private readonly nodeTypes: NodeTypes,
|
||||
private readonly workflowService: WorkflowService,
|
||||
private readonly telemetry: Telemetry,
|
||||
private readonly policyEnforcementService: PolicyEnforcementService,
|
||||
private readonly ownershipService: OwnershipService,
|
||||
) {
|
||||
this.logger = this.logger.scoped('workflow-publication');
|
||||
}
|
||||
@@ -132,6 +137,9 @@ export class WorkflowPublicationApplier {
|
||||
const healSkip = await this.healBrokenNodeIds(workflow, newVersion);
|
||||
if (healSkip !== null) return healSkip;
|
||||
|
||||
const blocked = await this.enforcePublishPolicy(workflow, newVersion);
|
||||
if (blocked !== null) return blocked;
|
||||
|
||||
const oldTriggerNodes = this.workflowTriggerActivator.getEnabledTriggerNodes(oldVersion);
|
||||
const desiredTriggerNodes = this.workflowTriggerActivator.getEnabledTriggerNodes(newVersion);
|
||||
const triggerKinds = this.workflowTriggerActivator.getTriggerKinds(desiredTriggerNodes);
|
||||
@@ -238,6 +246,60 @@ export class WorkflowPublicationApplier {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Blocks a version that policy objects to, or `null` to carry on.
|
||||
*
|
||||
* Runs before the trigger diff: the no-change branch still advances the
|
||||
* published version, which running triggers re-read on their next fire.
|
||||
*/
|
||||
private async enforcePublishPolicy(
|
||||
workflow: WorkflowEntity,
|
||||
newVersion: WorkflowHistory,
|
||||
): Promise<PublicationResult | null> {
|
||||
if (!this.policyEnforcementService.hasChecksFor('workflowPublish')) return null;
|
||||
|
||||
// Unguarded, as in `PolicyLifecycleHandler`: an unevaluated project rule is not
|
||||
// a passed one, so a failed lookup fails the publication.
|
||||
const project = await this.ownershipService.getWorkflowProjectCached(workflow.id);
|
||||
|
||||
try {
|
||||
await this.policyEnforcementService.enforceWorkflowPublish({
|
||||
workflow: { id: workflow.id, name: workflow.name, nodes: newVersion.nodes },
|
||||
projectId: project.id,
|
||||
});
|
||||
|
||||
return null;
|
||||
} catch (e) {
|
||||
// `instanceof`, not `isPolicyRefusal`: the violations and the `Error` shape
|
||||
// below both need the typed instance.
|
||||
if (!(e instanceof PolicyViolationError)) throw e;
|
||||
|
||||
this.logger.warn('Workflow publication blocked by policy', {
|
||||
workflowId: workflow.id,
|
||||
versionId: newVersion.versionId,
|
||||
violations: e.violations.map((violation) => violation.kind),
|
||||
});
|
||||
|
||||
// Report every trigger as failed: `activated` rows left behind read as drift
|
||||
// to the reconciler, which would then re-enqueue this forever.
|
||||
const desiredTriggerNodes = this.workflowTriggerActivator.getEnabledTriggerNodes(newVersion);
|
||||
const triggerKinds = this.workflowTriggerActivator.getTriggerKinds(desiredTriggerNodes);
|
||||
|
||||
return {
|
||||
type: 'failed',
|
||||
error: e,
|
||||
triggerStatuses: this.buildTriggerStatuses(desiredTriggerNodes, triggerKinds, {
|
||||
activated: [],
|
||||
failures: desiredTriggerNodes.map((node) => ({
|
||||
nodeId: node.id,
|
||||
nodeName: node.name,
|
||||
error: e,
|
||||
})),
|
||||
}),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A first publication (no old version) reports `activate`, so the n8n Trigger's
|
||||
* "Workflow Published" event fires; otherwise the mode follows the record's reason.
|
||||
|
||||
@@ -862,6 +862,10 @@ export class WorkflowService {
|
||||
this._validateTriggerNodeIds(workflowId, versionToActivate);
|
||||
}
|
||||
|
||||
// The candidate below shares this array with the version row, and the hook may
|
||||
// mutate it in place, so snapshot what will actually be registered.
|
||||
const nodesToPublish = structuredClone(versionToActivate.nodes);
|
||||
|
||||
// Run hook before destructive state changes so a rejection leaves
|
||||
// the previous active version running instead of deactivating it.
|
||||
const candidateWorkflow = this.workflowRepository.create({
|
||||
@@ -886,6 +890,23 @@ export class WorkflowService {
|
||||
});
|
||||
}
|
||||
|
||||
// Polices what gets registered — the version row, not the hook's candidate.
|
||||
// Enforced on a same-version republish too.
|
||||
if (this.policyEnforcementService.hasChecksFor('workflowPublish')) {
|
||||
// Unguarded, as in `PolicyLifecycleHandler`: an unevaluated project rule is
|
||||
// not a passed one, so a failed lookup fails the publish.
|
||||
const project = await this.ownershipService.getWorkflowProjectCached(workflowId);
|
||||
|
||||
await this.policyEnforcementService.enforceWorkflowPublish({
|
||||
workflow: {
|
||||
id: workflowId,
|
||||
name: workflow.name,
|
||||
nodes: nodesToPublish,
|
||||
},
|
||||
projectId: project.id,
|
||||
});
|
||||
}
|
||||
|
||||
// re-applying the already-published version (e.g. a settings-only update)
|
||||
// publishes no new version, so the review gate must not block it.
|
||||
//
|
||||
|
||||
@@ -25,6 +25,7 @@ import { ActiveWorkflowManager } from '@/active-workflow-manager';
|
||||
import { ExecutionService } from '@/executions/execution.service';
|
||||
import { ExternalHooks } from '@/external-hooks';
|
||||
import { NodeTypes } from '@/node-types';
|
||||
import { PolicyEnforcementService } from '@/policy/policy-enforcement.service';
|
||||
import { Push } from '@/push';
|
||||
import { OwnershipService } from '@/services/ownership.service';
|
||||
import * as WebhookHelpers from '@/webhooks/webhook-helpers';
|
||||
@@ -136,6 +137,21 @@ describe('init()', () => {
|
||||
|
||||
expect(validateWorkflowHasTriggerLikeNodeSpy).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
// Startup reactivation is a publish in its own right. Spied rather than registered
|
||||
// via `setImplementation`, which is single-shot and would leak; `restoreMocks` undoes
|
||||
// a spy between tests.
|
||||
it('should enforce the publish policy for each workflow it reactivates', async () => {
|
||||
const policyEnforcementService = Container.get(PolicyEnforcementService);
|
||||
vi.spyOn(policyEnforcementService, 'hasChecksFor').mockReturnValue(true);
|
||||
const enforceSpy = vi.spyOn(policyEnforcementService, 'enforceWorkflowPublish');
|
||||
await Promise.all([createActiveWorkflow(), createActiveWorkflow()]);
|
||||
|
||||
await activeWorkflowManager.init();
|
||||
|
||||
expect(enforceSpy).toHaveBeenCalledTimes(2);
|
||||
expect(activeWorkflowManager.allActiveInMemory()).toHaveLength(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe('add()', () => {
|
||||
|
||||
@@ -20,6 +20,7 @@ import type { INode, INodeTypeData } from 'n8n-workflow';
|
||||
|
||||
import { ActiveExecutions } from '@/active-executions';
|
||||
import { ActiveWorkflowManager } from '@/active-workflow-manager';
|
||||
import { EventService } from '@/events/event.service';
|
||||
import { ExecutionService } from '@/executions/execution.service';
|
||||
import { ExternalHooks } from '@/external-hooks';
|
||||
import { Push } from '@/push';
|
||||
@@ -330,6 +331,81 @@ describe('WorkflowPublicationReconciler (integration)', () => {
|
||||
expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
|
||||
});
|
||||
|
||||
// A publication failing before the mapping advances (e.g. a policy block) leaves
|
||||
// the skew permanently, so re-enqueueing would loop on every pass.
|
||||
test('leaves a workflow skewed by a terminally failed publication alone', async () => {
|
||||
const owner = await createOwner();
|
||||
|
||||
const trigger = scheduleNode('skew-failed');
|
||||
const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner);
|
||||
await setActiveVersion(workflow.id, workflow.versionId);
|
||||
|
||||
await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish');
|
||||
await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal);
|
||||
|
||||
// A newer version is made active, but its publication fails before the
|
||||
// mapping advances — so activeVersionId and publishedVersionId diverge.
|
||||
const newVersionId = 'version-2-failed-publish';
|
||||
await createWorkflowHistory(workflow, owner, undefined, {
|
||||
versionId: newVersionId,
|
||||
nodes: [trigger],
|
||||
});
|
||||
await setActiveVersion(workflow.id, newVersionId);
|
||||
await outboxRepository.enqueue(workflow.id, newVersionId, 'publish');
|
||||
const failing = (await outboxRepository.claimNextPendingRecord())!;
|
||||
await outboxRepository.markFailed(failing.id, 'Blocked by policy');
|
||||
|
||||
expect(await publishedVersionRepository.getPublishedVersionId(workflow.id)).toBe(
|
||||
workflow.versionId,
|
||||
);
|
||||
|
||||
await reconciler.reconcile('reconcile');
|
||||
|
||||
// No third record: the skew is real but permanent, so the reconciler must
|
||||
// not keep retrying it.
|
||||
expect(await outboxRepository.countBy({ workflowId: workflow.id })).toBe(2);
|
||||
expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
|
||||
});
|
||||
|
||||
// The failed-record exclusion above is scoped to the active version so it never
|
||||
// reaches an unpublish, where the mapping is all that is left to heal.
|
||||
test('removes a mapping left behind by a failed unpublish, despite the failed record', async () => {
|
||||
const owner = await createOwner();
|
||||
|
||||
const trigger = scheduleNode('unpublish-failed');
|
||||
const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner);
|
||||
await setActiveVersion(workflow.id, workflow.versionId);
|
||||
|
||||
await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish');
|
||||
await consumer.processRecord((await outboxRepository.claimNextPendingRecord())!, abortSignal);
|
||||
|
||||
// An unpublish another main tore down and reported (triggers gone, status
|
||||
// rows cleared) whose mapping removal never landed, so its record is
|
||||
// terminal `failed`. With no in-memory trigger and no status rows, the
|
||||
// registry detections are blind — only the version comparison can see it.
|
||||
await Container.get(WorkflowRepository).update(workflow.id, { activeVersionId: null });
|
||||
await activeWorkflowTriggers.remove(workflow.id);
|
||||
await triggerStatusRepository.delete({ workflowId: workflow.id });
|
||||
await outboxRepository.enqueue(workflow.id, workflow.versionId, 'publish');
|
||||
const failing = (await outboxRepository.claimNextPendingRecord())!;
|
||||
await outboxRepository.markFailed(failing.id, 'Removing the published version failed');
|
||||
|
||||
expect(await outboxRepository.findVersionSkewedWorkflowIds()).toContain(workflow.id);
|
||||
|
||||
const emitSpy = vi.spyOn(Container.get(EventService), 'emit');
|
||||
await reconciler.reconcile('reconcile');
|
||||
|
||||
// Attribution: the skew detector drove the repair, not another pass.
|
||||
expect(emitSpy).toHaveBeenCalledWith(
|
||||
'workflow-publication-reconciliation',
|
||||
expect.objectContaining({ versionSkewCount: 1, deficientCount: 0, surplusCount: 0 }),
|
||||
);
|
||||
expect(await publishedVersionRepository.getPublishedVersionId(workflow.id)).toBeNull();
|
||||
expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
|
||||
|
||||
emitSpy.mockRestore();
|
||||
});
|
||||
|
||||
test('removes a published-version mapping left behind by a missed unpublish', async () => {
|
||||
const owner = await createOwner();
|
||||
|
||||
|
||||
@@ -114,8 +114,8 @@ beforeAll(async () => {
|
||||
Container.get(WorkflowHookContextService), // workflowHookContextService
|
||||
workflowPublishGuard,
|
||||
mock(), // workflowMutationHooks
|
||||
// Real service on purpose: with no policy backend registered it clears every save,
|
||||
// so these tests also prove save behavior is unchanged when the module is off.
|
||||
// Real service on purpose: with no backend registered it clears every save and
|
||||
// publish, so these tests also prove behavior is unchanged with the module off.
|
||||
Container.get(PolicyEnforcementService), // policyEnforcementService
|
||||
);
|
||||
});
|
||||
@@ -342,6 +342,29 @@ describe('update()', () => {
|
||||
});
|
||||
|
||||
describe('activateWorkflow()', () => {
|
||||
// The rest of this suite runs with no checks registered, proving activation is
|
||||
// unchanged when the module is off. Spied rather than registered via
|
||||
// `setImplementation`, which is single-shot and would leak into those tests.
|
||||
test('should enforce the publish policy with the version being activated', async () => {
|
||||
const owner = await createOwner();
|
||||
const workflow = await createWorkflowWithHistory({}, owner);
|
||||
const policyEnforcementService = Container.get(PolicyEnforcementService);
|
||||
vi.spyOn(policyEnforcementService, 'hasChecksFor').mockReturnValue(true);
|
||||
const enforceSpy = vi.spyOn(policyEnforcementService, 'enforceWorkflowPublish');
|
||||
|
||||
const updatedWorkflow = await workflowService.activateWorkflow(owner, workflow.id);
|
||||
|
||||
expect(enforceSpy).toHaveBeenCalledExactlyOnceWith({
|
||||
workflow: {
|
||||
id: workflow.id,
|
||||
name: workflow.name,
|
||||
nodes: expect.any(Array),
|
||||
},
|
||||
projectId: expect.any(String),
|
||||
});
|
||||
expect(updatedWorkflow.activeVersionId).toBe(workflow.versionId);
|
||||
});
|
||||
|
||||
test('should activate current workflow version if no version provided', async () => {
|
||||
const owner = await createOwner();
|
||||
const workflow = await createWorkflowWithHistory({}, owner);
|
||||
|
||||
Reference in New Issue
Block a user