fix(core): Stop webhook routing locally before best-effort external teardown (#37241)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
mfsiega 2026-08-31 11:55:28 +00:00 committed by GitHub
parent 6e9c3e2ecf
commit cae0c8be00
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 886 additions and 117 deletions

View File

@ -456,13 +456,18 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
}
}
/** Mark a claimed record as successfully processed. Pass `trx` to enroll in an existing transaction. */
async markCompleted(id: number, trx?: EntityManager): Promise<void> {
/**
* Mark a claimed record as successfully processed. Pass `trx` to enroll in an
* existing transaction. An optional `warningMessage` is stored in
* `errorMessage` for a record that completed with a non-fatal side effect
* (e.g. an abandoned external webhook deregistration).
*/
async markCompleted(id: number, trx?: EntityManager, warningMessage?: string): Promise<void> {
const manager = trx ?? this.manager;
const result = await manager.update(
WorkflowPublicationOutbox,
{ id, status: Status.InProgress },
{ status: Status.Completed, errorMessage: null },
{ status: Status.Completed, errorMessage: warningMessage ?? null },
);
this.assertSingleRowAffected(result.affected, id, Status.Completed);
}

View File

@ -105,6 +105,12 @@ export const WORKFLOW_REACTIVATE_MAX_TIMEOUT = 24 * 60 * 60 * 1000; // 1 day
/** Max in-process attempts to activate a single trigger node before recording it as failed. */
export const TRIGGER_ACTIVATION_MAX_ATTEMPTS = 5;
/** Max in-process attempts to externally deregister a single webhook before abandoning it. */
export const TRIGGER_TEARDOWN_MAX_ATTEMPTS = 3;
/** Base delay between external webhook deregistration attempts; doubles per attempt. */
export const TRIGGER_TEARDOWN_RETRY_INITIAL_DELAY_MS = 1000;
export const SETTINGS_LICENSE_CERT_KEY = 'license.cert';
export const UM_FIX_INSTRUCTION =

View File

@ -97,7 +97,7 @@ describe('PublicationStatusReporter', () => {
],
entityManager,
);
expect(outboxRepository.markCompleted).toHaveBeenCalledWith(1, entityManager);
expect(outboxRepository.markCompleted).toHaveBeenCalledWith(1, entityManager, undefined);
expect(activationErrorsService.deregister).toHaveBeenCalledWith('wf-1');
expect(outboxRepository.markFailed).not.toHaveBeenCalled();
expect(push.broadcast).toHaveBeenCalledWith({
@ -113,6 +113,111 @@ describe('PublicationStatusReporter', () => {
});
});
describe('external teardown failures', () => {
const failure = { nodeName: 'Trello Trigger', error: new Error('remote unreachable') };
test('unpublished still completes the record, recording the failures as a warning', async () => {
await reporter.report(makeRecord(), { type: 'unpublished', teardownFailures: [failure] });
expect(outboxRepository.markCompleted).toHaveBeenCalledWith(
1,
entityManager,
'External webhook deregistration failed for: "Trello Trigger": remote unreachable',
);
expect(outboxRepository.markFailed).not.toHaveBeenCalled();
expect(push.broadcast).toHaveBeenCalledWith({
type: 'workflowDeactivated',
data: { workflowId: 'wf-1' },
});
});
test('unpublished reports the abandoned deregistrations once, at error level', async () => {
await reporter.report(makeRecord(), { type: 'unpublished', teardownFailures: [failure] });
expect(errorReporter.error).toHaveBeenCalledTimes(1);
expect(errorReporter.error).toHaveBeenCalledWith(
expect.objectContaining({
message:
'External webhook deregistration failed for: "Trello Trigger": remote unreachable',
level: 'error',
cause: failure.error,
}),
{ shouldBeLogged: true },
);
});
test('partial reports the abandoned deregistrations alongside the activation failures', async () => {
await reporter.report(makeRecord(), {
type: 'partial',
triggerStatuses: [
{ nodeId: 'a', nodeName: 'Schedule', status: 'activated', triggerKind: 'in-memory' },
{
nodeId: 'b',
nodeName: 'Broken',
status: 'failed',
triggerKind: 'in-memory',
errorMessage: 'boom',
},
],
teardownFailures: [failure],
});
expect(errorReporter.error).toHaveBeenCalledWith(
expect.objectContaining({
message:
'External webhook deregistration failed for: "Trello Trigger": remote unreachable',
}),
{ shouldBeLogged: true },
);
expect(outboxRepository.markPartialSuccess).toHaveBeenCalled();
});
test('failed reports the abandoned deregistrations alongside the activation error', async () => {
const activationError = new Error('registration failed');
await reporter.report(makeRecord(), {
type: 'failed',
error: activationError,
teardownFailures: [failure],
});
expect(errorReporter.error).toHaveBeenCalledWith(
expect.objectContaining({
message:
'External webhook deregistration failed for: "Trello Trigger": remote unreachable',
}),
{ shouldBeLogged: true },
);
// The activation error stays the record's failure and its own report.
expect(errorReporter.error).toHaveBeenCalledWith(activationError, { shouldBeLogged: true });
expect(outboxRepository.markFailed).toHaveBeenCalledWith(
1,
'registration failed',
entityManager,
);
});
test('completed still completes the record and reports, keeping the activation push', async () => {
await reporter.report(makeRecord(), {
type: 'completed',
triggerStatuses: [
{ nodeId: 'a', nodeName: 'Schedule', status: 'activated', triggerKind: 'in-memory' },
],
teardownFailures: [failure],
});
expect(outboxRepository.markCompleted).toHaveBeenCalledWith(
1,
entityManager,
'External webhook deregistration failed for: "Trello Trigger": remote unreachable',
);
expect(errorReporter.error).toHaveBeenCalledTimes(1);
expect(push.broadcast).toHaveBeenCalledWith({
type: 'workflowActivated',
data: { workflowId: 'wf-1', activeVersionId: 'v-2' },
});
});
});
test('unpublished clears trigger rows, marks the record completed, clears errors, and pushes deactivation', async () => {
await reporter.report(makeRecord(), { type: 'unpublished' });
@ -121,7 +226,7 @@ describe('PublicationStatusReporter', () => {
[],
entityManager,
);
expect(outboxRepository.markCompleted).toHaveBeenCalledWith(1, entityManager);
expect(outboxRepository.markCompleted).toHaveBeenCalledWith(1, entityManager, undefined);
expect(activationErrorsService.deregister).toHaveBeenCalledWith('wf-1');
expect(outboxRepository.markFailed).not.toHaveBeenCalled();
expect(push.broadcast).toHaveBeenCalledWith({
@ -143,7 +248,7 @@ describe('PublicationStatusReporter', () => {
async ({ reason, message }) => {
await reporter.report(makeRecord(), { type: 'skipped', reason });
expect(outboxRepository.markCompleted).toHaveBeenCalledWith(1, entityManager);
expect(outboxRepository.markCompleted).toHaveBeenCalledWith(1, entityManager, undefined);
expect(activationErrorsService.deregister).toHaveBeenCalledWith('wf-1');
expect(outboxRepository.markFailed).not.toHaveBeenCalled();
expect(push.broadcast).not.toHaveBeenCalled();
@ -335,7 +440,7 @@ describe('PublicationStatusReporter', () => {
type: 'workflowDeactivated',
data: { workflowId: 'wf-1' },
});
expect(outboxRepository.markCompleted).toHaveBeenCalledWith(1, entityManager);
expect(outboxRepository.markCompleted).toHaveBeenCalledWith(1, entityManager, undefined);
// The rejection is handled asynchronously; flush the microtask queue.
await new Promise(process.nextTick);
expect(errorReporter.error).toHaveBeenCalledWith(publishError, { shouldBeLogged: true });

View File

@ -123,7 +123,7 @@ describe('WorkflowPublicationApplier', () => {
workflowTriggerActivator.getUnregisteredNonWebhookTriggerNodeIds.mockReturnValue(new Set());
workflowTriggerActivator.getNodesWithUnregisteredWebhooks.mockResolvedValue(new Set());
workflowTriggerActivator.activate.mockResolvedValue({ activated: [], failures: [] });
workflowTriggerActivator.deactivate.mockResolvedValue(undefined);
workflowTriggerActivator.deactivate.mockResolvedValue({ externalTeardownFailures: [] });
workflowTriggerActivator.updateTriggerCount.mockResolvedValue(undefined);
// The test nodes are scheduleTrigger nodes, so default every node to 'in-memory'.
workflowTriggerActivator.getTriggerKinds.mockImplementation(
@ -208,6 +208,28 @@ describe('WorkflowPublicationApplier', () => {
expect(workflowPublishedVersionRepository.setPublishedVersion).not.toHaveBeenCalled();
});
test('completes as unpublished carrying external teardown failures, still removing the mapping', async () => {
// Local routing already stopped inside `deactivate` (rows deleted before
// any external call), so an abandoned external deregistration must not
// keep the mapping alive — that would make the reconciler re-enqueue the
// unpublish forever and block deleting the workflow.
workflowPublishedVersionRepository.findOne.mockResolvedValue(
makePublishedVersion(oldVersion),
);
workflowTriggerActivator.getEnabledTriggerNodes.mockReturnValue([triggerNode('a')]);
const failure = { nodeName: 'a', error: new Error('remote unreachable') };
workflowTriggerActivator.deactivate.mockResolvedValue({
externalTeardownFailures: [failure],
});
const result = await applier.apply(makeRecord(), abort);
expect(result).toEqual({ type: 'unpublished', teardownFailures: [failure] });
expect(workflowPublishedVersionRepository.removePublishedVersion).toHaveBeenCalledWith(
'wf-1',
);
});
test('propagates a teardown failure and leaves the mapping in place', async () => {
workflowPublishedVersionRepository.findOne.mockResolvedValue(
makePublishedVersion(oldVersion),
@ -531,6 +553,7 @@ describe('WorkflowPublicationApplier', () => {
const callOrder: string[] = [];
workflowTriggerActivator.deactivate.mockImplementation(async () => {
callOrder.push('remove');
return { externalTeardownFailures: [] };
});
workflowPublishedVersionRepository.setPublishedVersion.mockImplementation(async () => {
callOrder.push('advance');
@ -573,6 +596,30 @@ describe('WorkflowPublicationApplier', () => {
expect(callOrder).toEqual(['remove', 'invalidate', 'advance', 'refresh', 'add']);
});
test('completes carrying external teardown failures from removed triggers, still advancing', async () => {
setTriggerSets([triggerNode('a'), triggerNode('b')], [triggerNode('a')]);
const failure = { nodeName: 'b', error: new Error('remote unreachable') };
workflowTriggerActivator.deactivate.mockResolvedValue({
externalTeardownFailures: [failure],
});
const result = await applier.apply(makeRecord(), abort);
// The new version must not be blocked by a third party refusing to release
// an old trigger: the publication completes and the failures ride along.
expect(result).toEqual({
type: 'completed',
triggerStatuses: [
{ nodeId: 'a', nodeName: 'a', status: 'activated', triggerKind: 'in-memory' },
],
teardownFailures: [failure],
});
expect(workflowPublishedVersionRepository.setPublishedVersion).toHaveBeenCalledWith(
'wf-1',
'v-2',
);
});
test('propagates without advancing when removing triggers throws', async () => {
setTriggerSets([triggerNode('a'), triggerNode('b')], [triggerNode('a')]);
workflowTriggerActivator.deactivate.mockRejectedValue(new Error('teardown failed'));
@ -633,6 +680,63 @@ describe('WorkflowPublicationApplier', () => {
expect(workflowTriggerActivator.deactivate).not.toHaveBeenCalled();
});
test('carries external teardown failures on a partial activation outcome', async () => {
// The teardown ran (and the version advanced) before activation classified
// the result: the abandoned deregistration must survive the classification
// so the reporter can still surface it.
setTriggerSets(
[triggerNode('a'), triggerNode('removed')],
[triggerNode('a'), triggerNode('b')],
);
const teardownFailure = { nodeName: 'removed', error: new Error('remote unreachable') };
workflowTriggerActivator.deactivate.mockResolvedValue({
externalTeardownFailures: [teardownFailure],
});
workflowTriggerActivator.activate.mockResolvedValue({
activated: ['a'],
failures: [{ nodeId: 'b', nodeName: 'b', error: new Error('third-party unavailable') }],
});
const result = await applier.apply(makeRecord(), abort);
expect(result).toMatchObject({ type: 'partial', teardownFailures: [teardownFailure] });
});
test('carries external teardown failures when adding triggers throws', async () => {
setTriggerSets(
[triggerNode('a'), triggerNode('removed')],
[triggerNode('a'), triggerNode('b')],
);
const teardownFailure = { nodeName: 'removed', error: new Error('remote unreachable') };
workflowTriggerActivator.deactivate.mockResolvedValue({
externalTeardownFailures: [teardownFailure],
});
workflowTriggerActivator.activate.mockRejectedValue(new Error('registration failed'));
const result = await applier.apply(makeRecord(), abort);
expect(result).toMatchObject({ type: 'failed', teardownFailures: [teardownFailure] });
});
test('carries external teardown failures when advancing the version throws', async () => {
// The teardown (and its abandoned deregistrations) already happened;
// an advance failure must not throw past them and lose the report.
setTriggerSets([triggerNode('a'), triggerNode('removed')], [triggerNode('a')]);
const teardownFailure = { nodeName: 'removed', error: new Error('remote unreachable') };
workflowTriggerActivator.deactivate.mockResolvedValue({
externalTeardownFailures: [teardownFailure],
});
workflowPublishedVersionRepository.setPublishedVersion.mockRejectedValue(new Error('db down'));
const result = await applier.apply(makeRecord(), abort);
expect(result).toMatchObject({
type: 'failed',
error: expect.objectContaining({ message: 'db down' }),
teardownFailures: [teardownFailure],
});
});
test('returns partial when a deterministic failure coexists with an activated trigger', async () => {
setTriggerSets([triggerNode('a')], [triggerNode('a'), triggerNode('b')]);
const error = new WebhookPathTakenError('b');
@ -803,6 +907,7 @@ describe('WorkflowPublicationApplier', () => {
const controller = new AbortController();
workflowTriggerActivator.deactivate.mockImplementation(async () => {
controller.abort(new Error('deadline'));
return { externalTeardownFailures: [] };
});
const result = await applier.apply(makeRecord(), {

View File

@ -21,6 +21,8 @@ export type PublicationSkipReason =
import type { WorkflowPublicationTriggerKind } from '@n8n/db';
import type { TriggerTeardownFailure } from '@/workflows/triggers/workflow-trigger-activator';
/** A trigger that activated successfully; carries no error. */
type ActivatedTriggerPublicationStatus = {
nodeId: string;
@ -51,14 +53,26 @@ export type TriggerPublicationStatus =
* the single place mapping outcomes to terminal statuses and side effects.
*/
export type PublicationResult =
/** Triggers reconciled (or no change needed); the published version advanced. */
| { type: 'completed'; triggerStatuses: TriggerPublicationStatus[] }
/**
* Triggers reconciled (or no change needed); the published version advanced.
* `teardownFailures`, when present, lists removed webhook nodes whose
* external deregistration was abandoned after retries — local routing has
* stopped, but a third-party subscription may remain. The record still
* completes; the reporter surfaces the failures.
*/
| {
type: 'completed';
triggerStatuses: TriggerPublicationStatus[];
teardownFailures?: TriggerTeardownFailure[];
}
/**
* The workflow was unpublished: the triggers of the previously published
* version were torn down and the `workflow_published_version` mapping removed.
* The record is completed and a deactivation status is pushed to the UI.
* `teardownFailures` carries abandoned external webhook deregistrations,
* as on `completed`.
*/
| { type: 'unpublished' }
| { type: 'unpublished'; teardownFailures?: TriggerTeardownFailure[] }
/** No trigger work was required; the record is completed without changes. */
| { type: 'skipped'; reason: PublicationSkipReason }
/** The history row for the published version is gone; the record is failed. */
@ -67,7 +81,22 @@ export type PublicationResult =
* The published version advanced and some triggers are running, but others
* failed to register. The record is marked `partial_success` and the workflow
* stays published (no auto-unpublish); per-trigger detail is in `triggerStatuses`.
* `teardownFailures` carries abandoned external webhook deregistrations from
* the remove phase (which ran before the version advanced), as on `completed`.
*/
| { type: 'partial'; triggerStatuses: TriggerPublicationStatus[] }
/** The publication failed; the record is failed and the error is reported. */
| { type: 'failed'; error: Error; triggerStatuses?: TriggerPublicationStatus[] };
| {
type: 'partial';
triggerStatuses: TriggerPublicationStatus[];
teardownFailures?: TriggerTeardownFailure[];
}
/**
* The publication failed; the record is failed and the error is reported.
* `teardownFailures` as on `partial`: the remove phase already ran, so its
* abandoned deregistrations must still be surfaced.
*/
| {
type: 'failed';
error: Error;
triggerStatuses?: TriggerPublicationStatus[];
teardownFailures?: TriggerTeardownFailure[];
};

View File

@ -9,6 +9,7 @@ import {
import { OnPubSubEvent } from '@n8n/decorators';
import { Service } from '@n8n/di';
import { ErrorReporter } from 'n8n-core';
import { OperationalError } from 'n8n-workflow';
import { ActivationErrorsService } from '@/activation-errors.service';
import { Push } from '@/push';
@ -19,6 +20,7 @@ import type {
PublicationSkipReason,
TriggerPublicationStatus,
} from '@/workflows/publication/publication-result';
import type { TriggerTeardownFailure } from '@/workflows/triggers/workflow-trigger-activator';
/**
* Turns a {@link PublicationResult} into terminal state. This is the only place
@ -49,7 +51,8 @@ export class PublicationStatusReporter {
async report(record: WorkflowPublicationOutbox, result: PublicationResult): Promise<void> {
switch (result.type) {
case 'completed': {
await this.complete(record, this.toRows(record, result.triggerStatuses));
const warningMessage = this.surfaceTeardownFailures(result.teardownFailures);
await this.complete(record, this.toRows(record, result.triggerStatuses), warningMessage);
this.pushStatus({
type: 'workflowActivated',
data: { workflowId: record.workflowId, activeVersionId: record.publishedVersionId },
@ -58,7 +61,8 @@ export class PublicationStatusReporter {
}
case 'unpublished': {
await this.complete(record, /*triggerStatuses=*/ []);
const warningMessage = this.surfaceTeardownFailures(result.teardownFailures);
await this.complete(record, /*triggerStatuses=*/ [], warningMessage);
this.pushStatus({
type: 'workflowDeactivated',
data: { workflowId: record.workflowId },
@ -85,6 +89,9 @@ export class PublicationStatusReporter {
}
case 'failed': {
// The record's failure stays the activation error; the abandoned
// deregistrations (whose teardown already ran) get their own report.
this.surfaceTeardownFailures(result.teardownFailures);
const { triggerStatuses } = result;
await this.outboxRepository.manager.transaction(async (trx) => {
if (triggerStatuses) {
@ -102,6 +109,8 @@ export class PublicationStatusReporter {
}
case 'partial': {
// As on 'failed': the stored/pushed message stays about activation.
this.surfaceTeardownFailures(result.teardownFailures);
await this.reportPartial(record, result.triggerStatuses);
return;
}
@ -204,6 +213,32 @@ export class PublicationStatusReporter {
this.push.broadcast(pushMsg);
}
/**
* Surfaces abandoned external webhook deregistrations carried on a
* successful result: the publication itself succeeded (local routing has
* stopped), but a third-party subscription may remain. Reports once at
* error level — explicit, since `OperationalError` defaults to `warning`,
* which the error reporter filters from Sentry — and returns the message
* to store on the completed record for diagnostics.
*/
private surfaceTeardownFailures(
teardownFailures: TriggerTeardownFailure[] | undefined,
): string | undefined {
if (!teardownFailures?.length) return undefined;
const detail = teardownFailures
.map((failure) => `"${failure.nodeName}": ${failure.error.message}`)
.join('; ');
const message = `External webhook deregistration failed for: ${detail}`;
this.errorReporter.error(
new OperationalError(message, { level: 'error', cause: teardownFailures[0].error }),
{ shouldBeLogged: true },
);
return message;
}
/**
* Marks the record completed and clears any activation errors for the workflow.
* If there are any per-trigger statuses passed in, they are persisted in the same transaction.
@ -211,6 +246,7 @@ export class PublicationStatusReporter {
private async complete(
record: WorkflowPublicationOutbox,
triggerStatuses?: TriggerStatusRow[],
warningMessage?: string,
): Promise<void> {
await this.outboxRepository.manager.transaction(async (trx) => {
if (triggerStatuses !== undefined) {
@ -220,7 +256,7 @@ export class PublicationStatusReporter {
trx,
);
}
await this.outboxRepository.markCompleted(record.id, trx);
await this.outboxRepository.markCompleted(record.id, trx, warningMessage);
});
await this.activationErrorsService.deregister(record.workflowId);
}

View File

@ -29,6 +29,7 @@ import {
type TriggerActivationFailure,
type TriggerActivationOutcome,
type TriggerOperationAbort,
type TriggerTeardownFailure,
} from '@/workflows/triggers/workflow-trigger-activator';
import { WorkflowPublishedDataService } from '@/workflows/workflow-published-data.service';
@ -180,17 +181,32 @@ export class WorkflowPublicationApplier {
abort.signal.throwIfAborted();
// Must happen BEFORE advancing the version, using the currently published
// version so the right webhooks are deregistered. A retryable teardown
// failure bubbles up so the version is not advanced; one that can never
// succeed (a UserError) is abandoned inside `deactivate`, since retrying
// it would block this publication forever.
// version so the right webhooks are deregistered. Teardown is best-effort
// for external webhook state (local routing stops first inside
// `deactivate`): abandoned external deregistrations ride along on the
// result rather than blocking the new version behind a third party. Only
// a failure that leaves local trigger state possibly live (a non-webhook
// close failure, or an abort) bubbles up so the version is not advanced.
let teardownFailures: TriggerTeardownFailure[] = [];
if (toRemove.size > 0 && oldVersion) {
await this.workflowTriggerActivator.deactivate(workflow, oldVersion, toRemove, abort);
({ externalTeardownFailures: teardownFailures } =
await this.workflowTriggerActivator.deactivate(workflow, oldVersion, toRemove, abort));
}
// Whatever the activation outcome, the remove phase already ran (and the
// version advances below), so its abandoned external deregistrations must
// ride along for the reporter to surface.
const withTeardownFailures = (result: PublicationResult): PublicationResult =>
(result.type === 'completed' || result.type === 'partial' || result.type === 'failed') &&
teardownFailures.length > 0
? { ...result, teardownFailures }
: result;
await this.advancePublishedVersion(record);
// Inside the try: once teardown has run, any failure — advancing the
// version included — must return a `failed` result (not throw) so the
// teardown failures collected above still reach the reporter.
try {
await this.advancePublishedVersion(record);
abort.signal.throwIfAborted();
if (toAdd.size > 0) {
const activationMode = this.resolveActivationMode(record, oldVersion);
@ -201,23 +217,25 @@ export class WorkflowPublicationApplier {
activationMode,
abort,
);
return this.classifyActivationOutcome(outcome, desiredTriggerNodes, triggerKinds);
return withTeardownFailures(
this.classifyActivationOutcome(outcome, desiredTriggerNodes, triggerKinds),
);
}
if (toRemove.size > 0) {
await this.workflowTriggerActivator.updateTriggerCount(workflow, newVersion);
}
} catch (e) {
return { type: 'failed', error: ensureError(e) };
return withTeardownFailures({ type: 'failed', error: ensureError(e) });
}
return {
return withTeardownFailures({
type: 'completed',
triggerStatuses: this.buildTriggerStatuses(desiredTriggerNodes, triggerKinds, {
activated: [],
failures: [],
}),
};
});
}
/**
@ -245,12 +263,14 @@ export class WorkflowPublicationApplier {
* successful `unpublished` to support idempotent retries — the reporter then
* clears any trigger-status rows left behind by an interrupted unpublish.
*
* A teardown failure bubbles up (the consumer turns it into a `failed` result)
* so the mapping is only removed once teardown has succeeded. A teardown
* failure that can never succeed on retry (e.g. a webhook's delete hook
* needs a credential that was deleted) is abandoned inside `deactivate`
* rather than surfaced — the mapping is what the reconciler treats as drift,
* so such a failure would otherwise re-enqueue this unpublish forever.
* Only a teardown failure that leaves local trigger state possibly live (a
* non-webhook close failure, or an abort) bubbles up — the consumer turns it
* into a `failed` result and the mapping stays for the retry. A webhook's
* external deregistration failure does not: local routing already stopped
* inside `deactivate` (rows deleted before any external call), so the
* unpublish completes with the abandoned nodes in `teardownFailures`.
* Keeping the mapping for external garbage would make the reconciler
* re-enqueue this unpublish forever and block deleting the workflow.
*/
private async unpublish(
workflow: WorkflowEntity,
@ -265,9 +285,11 @@ export class WorkflowPublicationApplier {
this.workflowTriggerActivator.getEnabledTriggerNodes(oldVersion).map((node) => node.id),
);
let teardownFailures: TriggerTeardownFailure[] = [];
if (oldVersion && toRemove.size > 0) {
abort.signal.throwIfAborted();
await this.workflowTriggerActivator.deactivate(workflow, oldVersion, toRemove, abort);
({ externalTeardownFailures: teardownFailures } =
await this.workflowTriggerActivator.deactivate(workflow, oldVersion, toRemove, abort));
}
// Invalidate before the mapping is removed, so reads fall through to the
@ -276,7 +298,9 @@ export class WorkflowPublicationApplier {
await this.workflowPublishedDataService.invalidateCache(record.workflowId);
await this.workflowPublishedVersionRepository.removePublishedVersion(record.workflowId);
return { type: 'unpublished' };
return teardownFailures.length > 0
? { type: 'unpublished', teardownFailures }
: { type: 'unpublished' };
}
/**

View File

@ -14,7 +14,7 @@ import {
import { mock, type MockProxy } from 'vitest-mock-extended';
import type { ActivationErrorsService } from '@/activation-errors.service';
import { TRIGGER_ACTIVATION_MAX_ATTEMPTS } from '@/constants';
import { TRIGGER_ACTIVATION_MAX_ATTEMPTS, TRIGGER_TEARDOWN_MAX_ATTEMPTS } from '@/constants';
import type { EventService } from '@/events/event.service';
import * as WorkflowExecuteAdditionalData from '@/workflow-execute-additional-data';
import type {
@ -419,7 +419,9 @@ describe('WorkflowTriggerActivator', () => {
const deregisterB = createDeferredPromise();
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
const webhookData = mock<IWebhookData>({ node: 'Webhook' });
webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([webhookData]);
webhookTriggerRegistrar.getNodeWebhookTriggers.mockImplementation((_workflow, node) =>
node.name === 'Webhook' ? [webhookData] : [],
);
webhookTriggerRegistrar.deregister.mockImplementation(async () => {
callOrder.push('deregister-webhooks');
return 'Webhook';
@ -459,13 +461,13 @@ describe('WorkflowTriggerActivator', () => {
expect(callOrder).toEqual(
expect.arrayContaining([
'deregister-webhooks',
'clear-webhook-rows',
'deregister-webhooks',
'deregister-non-webhook:trigger-a',
]),
);
expect(callOrder.indexOf('deregister-webhooks')).toBeLessThan(
callOrder.indexOf('clear-webhook-rows'),
expect(callOrder.indexOf('clear-webhook-rows')).toBeLessThan(
callOrder.indexOf('deregister-webhooks'),
);
expect(deactivateSettled).toBe(false);
@ -491,9 +493,11 @@ describe('WorkflowTriggerActivator', () => {
const callOrder: string[] = [];
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
webhookTriggerRegistrar.getWebhookTriggers.mockImplementation(() => {
callOrder.push('webhook-discovery-fail');
throw new Error('webhook discovery failed');
// A failed local row deletion is (still) fatal to the webhook phase:
// routing was not stopped, so the operation must fail for retry.
webhookTriggerRegistrar.clearWorkflowWebhooksForNodes.mockImplementation(async () => {
callOrder.push('webhook-row-cleanup-fail');
throw new Error('webhook row cleanup failed');
});
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
@ -519,11 +523,163 @@ describe('WorkflowTriggerActivator', () => {
new Set(['webhook-node', 'trigger-node']),
abort,
),
).rejects.toThrow('webhook discovery failed');
).rejects.toThrow('webhook row cleanup failed');
expect(callOrder).toContain('deregister-non-webhook-finish');
});
describe('local-first webhook teardown', () => {
test('deletes local webhook rows before attempting external deregistration', async () => {
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
mock<IWorkflowExecuteAdditionalData>(),
);
const callOrder: string[] = [];
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
const webhookData = mock<IWebhookData>({ node: 'Webhook' });
webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([webhookData]);
webhookTriggerRegistrar.clearWorkflowWebhooksForNodes.mockImplementation(async () => {
callOrder.push('clear-webhook-rows');
});
webhookTriggerRegistrar.deregister.mockImplementation(async () => {
callOrder.push('deregister-external');
return 'Webhook';
});
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]);
const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar });
await activator.deactivate(
mock<WorkflowEntity>({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }),
{ nodes: [node('webhook-node', 'webhook', { name: 'Webhook' })], connections: {} },
new Set(['webhook-node']),
abort,
);
expect(callOrder).toEqual(['clear-webhook-rows', 'deregister-external']);
expect(webhookTriggerRegistrar.clearWorkflowWebhooksForNodes).toHaveBeenCalledWith('wf-1', [
'Webhook',
]);
});
test('deletes local webhook rows and proceeds when webhook discovery throws', async () => {
// Discovery re-evaluates webhook expressions and can throw when the
// evaluation context drifted since publish. Rows are already deleted at
// that point, so the failure is collected — not thrown — or a publish
// would fail after killing the old version's routing.
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
mock<IWorkflowExecuteAdditionalData>(),
);
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
webhookTriggerRegistrar.getNodeWebhookTriggers.mockImplementation(() => {
throw new Error('discovery failed');
});
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]);
const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar });
const { externalTeardownFailures } = await activator.deactivate(
mock<WorkflowEntity>({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }),
{ nodes: [node('webhook-node', 'webhook', { name: 'Webhook' })], connections: {} },
new Set(['webhook-node']),
abort,
);
expect(externalTeardownFailures).toEqual([
{
nodeName: 'Webhook',
error: expect.objectContaining({ message: 'discovery failed' }),
},
]);
expect(webhookTriggerRegistrar.clearWorkflowWebhooksForNodes).toHaveBeenCalledWith('wf-1', [
'Webhook',
]);
});
test('a discovery failure on one node still deregisters the other nodes externally', async () => {
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
mock<IWorkflowExecuteAdditionalData>(),
);
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
const webhookOk = mock<IWebhookData>({ node: 'Webhook OK' });
webhookTriggerRegistrar.getNodeWebhookTriggers.mockImplementation((_workflow, node) => {
if (node.name === 'Webhook Broken') throw new Error('discovery failed');
return [webhookOk];
});
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]);
const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar });
const { externalTeardownFailures } = await activator.deactivate(
mock<WorkflowEntity>({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }),
{
nodes: [
node('ok-node', 'webhook', { name: 'Webhook OK' }),
node('broken-node', 'webhook', { name: 'Webhook Broken' }),
],
connections: {},
},
new Set(['ok-node', 'broken-node']),
abort,
);
// The healthy node's external subscription is still cleaned up; only
// the node whose discovery failed is abandoned.
expect(webhookTriggerRegistrar.deregister).toHaveBeenCalledWith(
expect.objectContaining({ webhookData: webhookOk }),
);
expect(externalTeardownFailures).toEqual([
{
nodeName: 'Webhook Broken',
error: expect.objectContaining({ message: 'discovery failed' }),
},
]);
});
test('a discovery failure names only the failing node', async () => {
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
mock<IWorkflowExecuteAdditionalData>(),
);
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
webhookTriggerRegistrar.getNodeWebhookTriggers.mockImplementation((_workflow, node) => {
if (node.name === 'Webhook') throw new Error('discovery failed');
return [];
});
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue(['trigger-node']);
const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar });
const { externalTeardownFailures } = await activator.deactivate(
mock<WorkflowEntity>({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }),
{
nodes: [
node('webhook-node', 'webhook', { name: 'Webhook' }),
node('trigger-node', 'trigger'),
],
connections: {},
},
new Set(['webhook-node', 'trigger-node']),
abort,
);
// The schedule-style trigger has no external webhook to leak; naming it
// in an "external webhook deregistration failed" report would be wrong.
expect(externalTeardownFailures).toEqual([
{
nodeName: 'Webhook',
error: expect.objectContaining({ message: 'discovery failed' }),
},
]);
});
});
describe('deactivate teardown failures', () => {
function buildDeactivationSetup(deregisterError: Error) {
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
@ -533,7 +689,9 @@ describe('WorkflowTriggerActivator', () => {
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
const webhookOk = mock<IWebhookData>({ node: 'Webhook OK' });
const webhookBroken = mock<IWebhookData>({ node: 'Webhook Broken' });
webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([webhookOk, webhookBroken]);
webhookTriggerRegistrar.getNodeWebhookTriggers.mockImplementation((_workflow, node) =>
node.name === 'Webhook OK' ? [webhookOk] : [webhookBroken],
);
webhookTriggerRegistrar.deregister.mockImplementation(async ({ webhookData }) => {
if (webhookData.node === 'Webhook Broken') throw deregisterError;
return webhookData.node;
@ -559,10 +717,10 @@ describe('WorkflowTriggerActivator', () => {
return { webhookTriggerRegistrar, deactivate };
}
test('counts a webhook whose deregistration fails with a UserError as removed', async () => {
test('abandons a webhook whose deregistration fails with a UserError', async () => {
// The failure can never succeed on retry (e.g. the delete hook's
// credential was deleted): the remote registration is abandoned and the
// node's rows are cleared instead of retained.
// credential was deleted): the remote registration is abandoned; the
// node's rows were already cleared up front.
const { webhookTriggerRegistrar, deactivate } = buildDeactivationSetup(
new UserError('Credential with ID "c-1" does not exist for type "trelloApi".'),
);
@ -575,24 +733,149 @@ describe('WorkflowTriggerActivator', () => {
]);
});
test('a retryable failure still fails and retains the node rows', async () => {
test('collects a transient external failure instead of throwing, with rows already cleared', async () => {
// Local routing already stopped (rows deleted up front), so an external
// deregistration failure must not fail the whole deactivation — it is
// returned for the caller to surface, leaving only external garbage.
const { webhookTriggerRegistrar, deactivate } = buildDeactivationSetup(
new Error('remote unreachable'),
);
await expect(deactivate()).rejects.toThrow('remote unreachable');
const { externalTeardownFailures } = await deactivate();
expect(externalTeardownFailures).toEqual([
{
nodeName: 'Webhook Broken',
error: expect.objectContaining({ message: 'remote unreachable' }),
},
]);
expect(webhookTriggerRegistrar.clearWorkflowWebhooksForNodes).toHaveBeenCalledWith('wf-1', [
'Webhook OK',
'Webhook Broken',
]);
});
test('an abandoned UserError failure is not returned as an external teardown failure', async () => {
const { deactivate } = buildDeactivationSetup(
new UserError('Credential with ID "c-1" does not exist for type "trelloApi".'),
);
const { externalTeardownFailures } = await deactivate();
expect(externalTeardownFailures).toEqual([]);
});
test('does not burn retries on a failure that can never succeed', async () => {
const { webhookTriggerRegistrar, deactivate } = buildDeactivationSetup(
new UserError('Credential with ID "c-1" does not exist for type "trelloApi".'),
);
await deactivate();
// One call per webhook: 'Webhook OK' + a single, un-retried 'Webhook Broken'.
expect(webhookTriggerRegistrar.deregister).toHaveBeenCalledTimes(2);
});
});
describe('external deregistration retries', () => {
function buildRetrySetup(deregisterImpl: (attempt: number) => Promise<string>) {
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
mock<IWorkflowExecuteAdditionalData>(),
);
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
const webhookData = mock<IWebhookData>({ node: 'Webhook' });
webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([webhookData]);
let attempt = 0;
webhookTriggerRegistrar.deregister.mockImplementation(
async () => await deregisterImpl(attempt++),
);
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]);
const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar });
const deactivate = async () =>
await activator.deactivate(
mock<WorkflowEntity>({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }),
{ nodes: [node('webhook-node', 'webhook', { name: 'Webhook' })], connections: {} },
new Set(['webhook-node']),
abort,
);
return { webhookTriggerRegistrar, deactivate };
}
test('retries a transient external failure and reports nothing when a retry succeeds', async () => {
const { webhookTriggerRegistrar, deactivate } = buildRetrySetup(async (attempt) => {
if (attempt === 0) throw new Error('remote unreachable');
return 'Webhook';
});
const { externalTeardownFailures } = await deactivate();
expect(externalTeardownFailures).toEqual([]);
expect(webhookTriggerRegistrar.deregister).toHaveBeenCalledTimes(2);
});
test('reports a node with multiple failing webhooks as a single teardown failure', async () => {
// Failures are per node, not per webhook: metrics subtract the failure
// count from the node count, and the reporter names failed nodes — a
// multi-webhook node must not be counted (or named) twice.
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
mock<IWorkflowExecuteAdditionalData>(),
);
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([
mock<IWebhookData>({ node: 'Webhook', path: 'one' }),
mock<IWebhookData>({ node: 'Webhook', path: 'two' }),
]);
webhookTriggerRegistrar.deregister.mockRejectedValue(new Error('remote unreachable'));
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]);
const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar });
const { externalTeardownFailures } = await activator.deactivate(
mock<WorkflowEntity>({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }),
{ nodes: [node('webhook-node', 'webhook', { name: 'Webhook' })], connections: {} },
new Set(['webhook-node']),
abort,
);
expect(externalTeardownFailures).toEqual([
{
nodeName: 'Webhook',
error: expect.objectContaining({ message: 'remote unreachable' }),
},
]);
});
test('gives up on a transient external failure after exhausting the retry budget', async () => {
const { webhookTriggerRegistrar, deactivate } = buildRetrySetup(async () => {
throw new Error('remote unreachable');
});
const { externalTeardownFailures } = await deactivate();
expect(externalTeardownFailures).toEqual([
{
nodeName: 'Webhook',
error: expect.objectContaining({ message: 'remote unreachable' }),
},
]);
expect(webhookTriggerRegistrar.deregister).toHaveBeenCalledTimes(
TRIGGER_TEARDOWN_MAX_ATTEMPTS,
);
});
test('skips a non-webhook trigger whose deregistration fails with a UserError, even when wrapped', async () => {
vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
mock<IWorkflowExecuteAdditionalData>(),
);
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([]);
webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([]);
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue(['t']);
nonWebhookTriggerRegistrar.deregister.mockRejectedValue(
@ -1139,7 +1422,7 @@ describe('WorkflowTriggerActivator', () => {
mock<IWorkflowExecuteAdditionalData>(),
);
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([]);
webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([]);
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue(['trigger-a']);
@ -1211,7 +1494,7 @@ describe('WorkflowTriggerActivator', () => {
test('a deactivation whose teardown hangs rejects with the abort reason once the signal fires', async () => {
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([]);
webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([]);
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue(['stuck']);
nonWebhookTriggerRegistrar.deregister.mockImplementation(
@ -1235,18 +1518,52 @@ describe('WorkflowTriggerActivator', () => {
expect(onDetached).toHaveBeenCalledWith(expect.any(Promise));
});
test('clears webhook rows only for nodes whose every webhook deregistered before the abort', async () => {
test('an aborted deactivation rejects with the abort reason, not an earlier webhook failure', async () => {
// One webhook fails with a UserError (policy: abandon silently) before a
// sibling's hang triggers the deadline abort. The operation must fail
// with the abort reason — blaming the UserError would mark the record
// failed with an error the code explicitly abandons.
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
const webhookBroken = mock<IWebhookData>({ node: 'Webhook Broken', path: 'broken' });
const webhookStuck = mock<IWebhookData>({ node: 'Webhook Stuck', path: 'hang' });
webhookTriggerRegistrar.getNodeWebhookTriggers.mockImplementation((_workflow, node) =>
node.name === 'Webhook Broken' ? [webhookBroken] : [webhookStuck],
);
webhookTriggerRegistrar.deregister.mockImplementation(async ({ webhookData }) => {
if (webhookData.path === 'hang') await new Promise(() => {});
throw new UserError('Credential with ID "c-1" does not exist for type "trelloApi".');
});
const nonWebhookTriggerRegistrar = mock<NonWebhookTriggerRegistrar>();
nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]);
const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar });
const controller = new AbortController();
const deactivation = activator.deactivate(
mock<WorkflowEntity>({ id: 'wf-1', name: 'Test workflow', staticData: {}, settings: {} }),
{
nodes: [
node('broken-node', 'webhook', { name: 'Webhook Broken' }),
node('stuck-node', 'webhook', { name: 'Webhook Stuck' }),
],
connections: {},
},
new Set(['broken-node', 'stuck-node']),
{ signal: controller.signal, onDetached: vi.fn() },
);
await flushPromises();
controller.abort(new Error('deadline'));
await expect(deactivation).rejects.toThrow('deadline');
});
test('clears all target webhook rows up front even when a deregistration hangs into the abort', async () => {
const webhookTriggerRegistrar = mock<WebhookTriggerRegistrar>();
const webhookOk = mock<IWebhookData>({ node: 'Webhook OK', path: 'ok' });
// The stuck node has two webhooks; only one of them hangs. Its fulfilled
// sibling must not cause the node's rows to be cleared.
const webhookStuck = mock<IWebhookData>({ node: 'Webhook Stuck', path: 'hang' });
const webhookStuckSibling = mock<IWebhookData>({ node: 'Webhook Stuck', path: 'fine' });
webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([
webhookOk,
webhookStuck,
webhookStuckSibling,
]);
webhookTriggerRegistrar.getNodeWebhookTriggers.mockImplementation((_workflow, node) =>
node.name === 'Webhook OK' ? [webhookOk] : [webhookStuck],
);
webhookTriggerRegistrar.deregister.mockImplementation(async ({ webhookData }) => {
if (webhookData.path === 'hang') await new Promise(() => {});
return webhookData.node;
@ -1272,9 +1589,13 @@ describe('WorkflowTriggerActivator', () => {
await flushPromises();
controller.abort(new Error('deadline'));
// Executions stop regardless of external teardown: rows for every target
// node were deleted before the external calls, and the abort still
// surfaces so the record is retried for external cleanup.
await expect(deactivation).rejects.toThrow('deadline');
expect(webhookTriggerRegistrar.clearWorkflowWebhooksForNodes).toHaveBeenCalledWith('wf-1', [
'Webhook OK',
'Webhook Stuck',
]);
});
});

View File

@ -68,6 +68,24 @@ export class WebhookTriggerRegistrar {
return WebhookHelpers.getWorkflowWebhooks(workflow, additionalData, undefined, true);
}
/**
* The webhook triggers of a single node. Lets a caller isolate one node's
* expression-evaluation failure instead of losing every node's webhooks to
* the wholesale {@link getWebhookTriggers}. Same isolate note as there.
*/
getNodeWebhookTriggers(
workflow: Workflow,
node: INode,
additionalData: IWorkflowExecuteAdditionalData,
): IWebhookData[] {
return this.webhookService.getNodeWebhooks(
workflow,
node,
additionalData,
/* ignoreRestartWebhooks */ true,
);
}
/**
* Register one workflow-defined webhook in storage and with third-party services.
*/

View File

@ -8,6 +8,7 @@ import { Service } from '@n8n/di';
import { ErrorReporter, SpanStatus, Tracing } from 'n8n-core';
import { ensureError } from '@n8n/utils/errors/ensure-error';
import { createResultError, createResultOk, type Result } from '@n8n/utils/result';
import { sleep } from '@n8n/utils/sleep';
import type {
INode,
IWebhookData,
@ -27,7 +28,11 @@ import {
} from 'n8n-workflow';
import { ActivationErrorsService } from '@/activation-errors.service';
import { TRIGGER_ACTIVATION_MAX_ATTEMPTS } from '@/constants';
import {
TRIGGER_ACTIVATION_MAX_ATTEMPTS,
TRIGGER_TEARDOWN_MAX_ATTEMPTS,
TRIGGER_TEARDOWN_RETRY_INITIAL_DELAY_MS,
} from '@/constants';
import { EventService } from '@/events/event.service';
import type {
PublicationOperationResult,
@ -101,6 +106,23 @@ export type TriggerActivationFailure = {
error: Error;
};
/**
* A webhook node whose external deregistration kept failing during teardown.
* Local routing has already stopped (the node's `webhook_entity` rows are
* deleted before any external call), so this only means an orphaned
* third-party subscription may remain — the caller surfaces it instead of
* failing the operation.
*/
export type TriggerTeardownFailure = {
nodeName: string;
error: Error;
};
/** The result of deactivating a set of trigger nodes. */
export type TriggerDeactivationOutcome = {
externalTeardownFailures: TriggerTeardownFailure[];
};
/**
* The per-node result of activating a set of trigger nodes. Activation is
* resilient: a failing node is recorded in `failures` and the remaining nodes
@ -378,20 +400,34 @@ export class WorkflowTriggerActivator {
* Deregisters only the given trigger nodes (webhook and non-webhook) of the
* given workflow version, leaving the rest active. The caller passes the
* currently published version so the right webhooks are deregistered.
*
* The two teardown surfaces fail differently, because they differ in what
* a failure leaves live:
*
* - A non-webhook trigger whose close fails stays tracked and possibly live
* in memory, so its failure rejects — the record must be retried until
* the local state is truly down.
* - A webhook's external deregistration failure leaves only third-party
* garbage: local routing already stopped when the node's rows were
* deleted (before any external call). Such failures resolve into
* `externalTeardownFailures` for the caller to surface — retrying the
* operation forever would block the publication on a third party.
*/
async deactivate(
dbWorkflow: WorkflowEntity,
version: WorkflowTriggerVersion,
nodeIds: Set<INode['id']>,
abort: TriggerOperationAbort,
) {
if (nodeIds.size === 0) return;
): Promise<TriggerDeactivationOutcome> {
if (nodeIds.size === 0) return { externalTeardownFailures: [] };
const startedAt = Date.now();
try {
await this.deactivateInternal(dbWorkflow, version, nodeIds, abort);
const outcome = await this.deactivateInternal(dbWorkflow, version, nodeIds, abort);
const failed = outcome.externalTeardownFailures.length;
this.emitTriggerOperation('deactivate', 'success', startedAt);
this.emitTriggerNodeOperations('deactivate', nodeIds.size, 0);
this.emitTriggerNodeOperations('deactivate', nodeIds.size - failed, failed);
return outcome;
} catch (error) {
this.emitTriggerOperation('deactivate', 'failure', startedAt);
this.emitTriggerNodeOperations('deactivate', 0, nodeIds.size);
@ -404,8 +440,8 @@ export class WorkflowTriggerActivator {
version: WorkflowTriggerVersion,
nodeIds: Set<INode['id']>,
abort: TriggerOperationAbort,
) {
await this.tracing.startSpan(
): Promise<TriggerDeactivationOutcome> {
return await this.tracing.startSpan(
{
name: 'Trigger deactivation',
op: 'publication.trigger.deactivate',
@ -425,13 +461,17 @@ export class WorkflowTriggerActivator {
// The non-webhook phase doesn't touch the expression isolate that the
// webhook deregister acquires, so the two phases can overlap.
const phaseResults = await Promise.allSettled([
const [webhookPhase, nonWebhookPhase] = await Promise.allSettled([
this.deregisterWebhookTriggers(workflow, additionalData, nodeIds, abort),
this.deregisterNonWebhookTriggers(dbWorkflow.id, workflow, nodeIds, abort),
]);
this.throwRejectedPhaseError(phaseResults);
this.throwRejectedPhaseError([webhookPhase, nonWebhookPhase]);
span.setStatus({ code: SpanStatus.ok });
return {
externalTeardownFailures: webhookPhase.status === 'fulfilled' ? webhookPhase.value : [],
};
},
);
}
@ -629,68 +669,133 @@ export class WorkflowTriggerActivator {
additionalData: IWorkflowExecuteAdditionalData,
nodeIds: Set<INode['id']>,
abort: TriggerOperationAbort,
) {
const removedNodeNames: string[] = [];
let firstFailure: Error | undefined;
): Promise<TriggerTeardownFailure[]> {
// Keyed by node: a node can expose several webhooks, but failures are
// reported (and counted against the node metrics) per node, keeping the
// first error.
const failuresByNode = new Map<string, TriggerTeardownFailure>();
// We remove the local routing so executions stop firing, and THEN attempt to deregister any
// external state.
const targetNodeNames = Object.values(workflow.nodes)
.filter((node) => nodeIds.has(node.id))
.map((node) => node.name);
await this.webhookTriggerRegistrar.clearWorkflowWebhooksForNodes(workflow.id, targetNodeNames);
await workflow.expression.acquireIsolate();
try {
const webhooks = this.getWebhookTriggersForNodeIds(workflow, additionalData, nodeIds);
const webhooks = this.getNodeWebhooks(workflow, nodeIds, additionalData, failuresByNode);
const deregistrationResults = await Promise.allSettled(
webhooks.map(
async (webhookData) =>
await raceAbort(
this.webhookTriggerRegistrar.deregister({ workflow, webhookData }),
abort,
),
await this.deregisterExternalWebhookWithRetry(workflow, webhookData, abort),
),
);
// Row cleanup is per node, so only a node whose EVERY webhook
// deregistered may be cleared: a failed or abandoned webhook's row is
// the only record left for deregistering it externally later.
const pendingWebhooksByNode = new Map<string, number>();
for (const webhookData of webhooks) {
pendingWebhooksByNode.set(
webhookData.node,
(pendingWebhooksByNode.get(webhookData.node) ?? 0) + 1,
);
}
deregistrationResults.forEach((result, index) => {
if (result.status === 'rejected') {
const error = ensureError(result.reason);
if (this.shouldAbandonFailedTeardown(error)) {
// The webhook counts as removed, so its row is cleared below
// instead of retained.
this.logger.warn('Abandoned webhook whose deregistration can never succeed', {
workflowId: workflow.id,
nodeName: webhooks[index].node,
error: error.message,
});
} else {
firstFailure ??= error;
return;
}
}
for (const [index, result] of deregistrationResults.entries()) {
if (result.status !== 'rejected') continue;
const error = ensureError(result.reason);
// An abort is not a teardown outcome: it must fail the operation so
// the record is retried for the external cleanup it interrupted. Throw
// the abort reason itself — this iteration's error may be an earlier,
// unrelated failure (e.g. an abandoned UserError), and blaming it
// would mark the record failed with an error the policy ignores.
if (abort.signal.aborted) throw ensureError(abort.signal.reason);
const nodeName = webhooks[index].node;
const pending = (pendingWebhooksByNode.get(nodeName) ?? 0) - 1;
pendingWebhooksByNode.set(nodeName, pending);
if (pending === 0) removedNodeNames.push(nodeName);
});
// We don't add these to failuresByNode because we don't intend to retry them.
if (this.shouldAbandonFailedTeardown(error)) {
this.logger.warn('Abandoned webhook whose deregistration can never succeed', {
workflowId: workflow.id,
nodeName,
error: error.message,
});
} else {
this.logger.warn('Abandoned webhook whose external deregistration kept failing', {
workflowId: workflow.id,
nodeName,
error: error.message,
});
if (!failuresByNode.has(nodeName)) failuresByNode.set(nodeName, { nodeName, error });
}
}
} finally {
await workflow.expression.releaseIsolate();
}
// Clear rows for the nodes that fully deregistered even when another
// node's teardown failed or was abandoned, so their `webhook_entity` rows
// don't outlive the deregistration.
await this.webhookTriggerRegistrar.clearWorkflowWebhooksForNodes(workflow.id, removedNodeNames);
if (firstFailure) throw firstFailure;
await this.workflowStaticDataService.saveStaticData(workflow);
return removedNodeNames;
return [...failuresByNode.values()];
}
// NOTE: acquire the expression evaluation isolate before calling this.
private getNodeWebhooks(
workflow: Workflow,
nodeIds: Set<INode['id']>,
additionalData: IWorkflowExecuteAdditionalData,
failuresByNode: Map<string, TriggerTeardownFailure>,
): IWebhookData[] {
const webhooks: IWebhookData[] = [];
for (const targetNode of Object.values(workflow.nodes)) {
if (!nodeIds.has(targetNode.id)) continue;
try {
webhooks.push(
...this.webhookTriggerRegistrar.getNodeWebhookTriggers(
workflow,
targetNode,
additionalData,
),
);
} catch (discoveryError) {
// NOTE: hitting an error can happen if the node's webhook expressions are invalid,
// but this is pretty rare. We still log but tolerate it and continue with teardown.
const error = ensureError(discoveryError);
this.logger.warn('Abandoned external webhook cleanup after failed discovery', {
workflowId: workflow.id,
nodeName: targetNode.name,
error: error.message,
});
failuresByNode.set(targetNode.name, { nodeName: targetNode.name, error });
}
}
return webhooks;
}
/**
* Externally deregisters one webhook, retrying a plausibly transient
* failure (e.g. an unreachable third party) with short exponential backoff
* before giving up. A failure that can never succeed (see
* {@link shouldAbandonFailedTeardown}) and an abort are not retried. The
* `sleep` rejects with the abort reason as soon as the signal fires, so an
* abandoned retry never sleeps out its backoff.
*
* Everything else is deliberately treated as retryable, including failures
* that may turn out permanent: node `delete()` errors are heterogeneous
* across integrations and carry no reliable transient/permanent signal
* (e.g. a 429 is retryable while a 403 is not, and many errors carry no
* status at all). Misclassifying a transient failure as permanent orphans
* an external subscription that one retry would have cleaned up, while the
* cost of retrying a permanent failure is a few bounded seconds inside the
* record's abort deadline.
*/
private async deregisterExternalWebhookWithRetry(
workflow: Workflow,
webhookData: IWebhookData,
abort: TriggerOperationAbort,
): Promise<void> {
for (let attempt = 0; ; attempt++) {
try {
await raceAbort(this.webhookTriggerRegistrar.deregister({ workflow, webhookData }), abort);
return;
} catch (error) {
const isLastAttempt = attempt >= TRIGGER_TEARDOWN_MAX_ATTEMPTS - 1;
if (this.shouldAbandonFailedTeardown(error) || isLastAttempt || abort.signal.aborted) {
throw error;
}
await sleep(TRIGGER_TEARDOWN_RETRY_INITIAL_DELAY_MS * 2 ** attempt, abort.signal);
}
}
}
private getWebhookTriggersForNodeIds(

View File

@ -174,6 +174,21 @@ describe('WorkflowPublicationOutboxRepository', () => {
expect(record?.errorMessage).toBeNull();
});
it('marks a claimed record as completed with a warning message', async () => {
// A record can complete with a non-fatal side effect (e.g. an abandoned
// external webhook deregistration); the warning lands in `errorMessage`
// for diagnostics while the status stays `completed`.
await repository.enqueue('wf-1', 'v-1', 'publish');
const claimed = await repository.claimNextPendingRecord();
assert(claimed);
await repository.markCompleted(claimed.id, undefined, 'external deregistration abandoned');
const record = await repository.findOneBy({ id: claimed.id });
expect(record?.status).toBe('completed');
expect(record?.errorMessage).toBe('external deregistration abandoned');
});
it('marks a claimed record as failed and records the error', async () => {
await repository.enqueue('wf-1', 'v-1', 'publish');
const claimed = await repository.claimNextPendingRecord();