diff --git a/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts b/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts index 43f261dc424..a632dd74652 100644 --- a/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-publication-outbox.repository.ts @@ -456,13 +456,18 @@ export class WorkflowPublicationOutboxRepository extends Repository { + /** + * 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 { 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); } diff --git a/packages/cli/src/constants.ts b/packages/cli/src/constants.ts index 44f8b8909e9..156a132fa89 100644 --- a/packages/cli/src/constants.ts +++ b/packages/cli/src/constants.ts @@ -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 = diff --git a/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts b/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts index d00cd93c47d..ebb23611e4c 100644 --- a/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts +++ b/packages/cli/src/workflows/publication/__tests__/publication-status-reporter.test.ts @@ -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 }); diff --git a/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts b/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts index fe219707cb3..a42076b480e 100644 --- a/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts +++ b/packages/cli/src/workflows/publication/__tests__/workflow-publication-applier.test.ts @@ -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(), { diff --git a/packages/cli/src/workflows/publication/publication-result.ts b/packages/cli/src/workflows/publication/publication-result.ts index 5780067c2fb..580b9a6b061 100644 --- a/packages/cli/src/workflows/publication/publication-result.ts +++ b/packages/cli/src/workflows/publication/publication-result.ts @@ -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[]; + }; diff --git a/packages/cli/src/workflows/publication/publication-status-reporter.ts b/packages/cli/src/workflows/publication/publication-status-reporter.ts index 79658782751..9c18f5c6122 100644 --- a/packages/cli/src/workflows/publication/publication-status-reporter.ts +++ b/packages/cli/src/workflows/publication/publication-status-reporter.ts @@ -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 { 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 { 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); } diff --git a/packages/cli/src/workflows/publication/workflow-publication-applier.ts b/packages/cli/src/workflows/publication/workflow-publication-applier.ts index 76d78114f80..8359a9455f2 100644 --- a/packages/cli/src/workflows/publication/workflow-publication-applier.ts +++ b/packages/cli/src/workflows/publication/workflow-publication-applier.ts @@ -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' }; } /** diff --git a/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts b/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts index 95af9d5ab9a..a9580caa988 100644 --- a/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/workflow-trigger-activator.test.ts @@ -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(); const webhookData = mock({ 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.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(); @@ -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(), + ); + + const callOrder: string[] = []; + const webhookTriggerRegistrar = mock(); + const webhookData = mock({ 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.getTriggerNodeIds.mockReturnValue([]); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + + await activator.deactivate( + mock({ 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(), + ); + + const webhookTriggerRegistrar = mock(); + webhookTriggerRegistrar.getNodeWebhookTriggers.mockImplementation(() => { + throw new Error('discovery failed'); + }); + const nonWebhookTriggerRegistrar = mock(); + nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + + const { externalTeardownFailures } = await activator.deactivate( + mock({ 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(), + ); + + const webhookTriggerRegistrar = mock(); + const webhookOk = mock({ node: 'Webhook OK' }); + webhookTriggerRegistrar.getNodeWebhookTriggers.mockImplementation((_workflow, node) => { + if (node.name === 'Webhook Broken') throw new Error('discovery failed'); + return [webhookOk]; + }); + const nonWebhookTriggerRegistrar = mock(); + nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + + const { externalTeardownFailures } = await activator.deactivate( + mock({ 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(), + ); + + const webhookTriggerRegistrar = mock(); + webhookTriggerRegistrar.getNodeWebhookTriggers.mockImplementation((_workflow, node) => { + if (node.name === 'Webhook') throw new Error('discovery failed'); + return []; + }); + const nonWebhookTriggerRegistrar = mock(); + nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue(['trigger-node']); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + + const { externalTeardownFailures } = await activator.deactivate( + mock({ 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(); const webhookOk = mock({ node: 'Webhook OK' }); const webhookBroken = mock({ 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) { + vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue( + mock(), + ); + + const webhookTriggerRegistrar = mock(); + const webhookData = mock({ node: 'Webhook' }); + webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([webhookData]); + let attempt = 0; + webhookTriggerRegistrar.deregister.mockImplementation( + async () => await deregisterImpl(attempt++), + ); + const nonWebhookTriggerRegistrar = mock(); + nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + const deactivate = async () => + await activator.deactivate( + mock({ 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(), + ); + + const webhookTriggerRegistrar = mock(); + webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([ + mock({ node: 'Webhook', path: 'one' }), + mock({ node: 'Webhook', path: 'two' }), + ]); + webhookTriggerRegistrar.deregister.mockRejectedValue(new Error('remote unreachable')); + const nonWebhookTriggerRegistrar = mock(); + nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue([]); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + + const { externalTeardownFailures } = await activator.deactivate( + mock({ 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(), ); const webhookTriggerRegistrar = mock(); - webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([]); + webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([]); const nonWebhookTriggerRegistrar = mock(); nonWebhookTriggerRegistrar.getTriggerNodeIds.mockReturnValue(['t']); nonWebhookTriggerRegistrar.deregister.mockRejectedValue( @@ -1139,7 +1422,7 @@ describe('WorkflowTriggerActivator', () => { mock(), ); const webhookTriggerRegistrar = mock(); - webhookTriggerRegistrar.getWebhookTriggers.mockReturnValue([]); + webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([]); const nonWebhookTriggerRegistrar = mock(); 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.getWebhookTriggers.mockReturnValue([]); + webhookTriggerRegistrar.getNodeWebhookTriggers.mockReturnValue([]); const nonWebhookTriggerRegistrar = mock(); 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(); + const webhookBroken = mock({ node: 'Webhook Broken', path: 'broken' }); + const webhookStuck = mock({ 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.getTriggerNodeIds.mockReturnValue([]); + + const activator = buildActivator({ webhookTriggerRegistrar, nonWebhookTriggerRegistrar }); + const controller = new AbortController(); + + const deactivation = activator.deactivate( + mock({ 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(); const webhookOk = mock({ 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({ node: 'Webhook Stuck', path: 'hang' }); - const webhookStuckSibling = mock({ 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', ]); }); }); diff --git a/packages/cli/src/workflows/triggers/webhook-trigger-registrar.ts b/packages/cli/src/workflows/triggers/webhook-trigger-registrar.ts index 3d4b1453820..7273dc045ba 100644 --- a/packages/cli/src/workflows/triggers/webhook-trigger-registrar.ts +++ b/packages/cli/src/workflows/triggers/webhook-trigger-registrar.ts @@ -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. */ diff --git a/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts b/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts index 8dab0225cfe..adf13dd6094 100644 --- a/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts +++ b/packages/cli/src/workflows/triggers/workflow-trigger-activator.ts @@ -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, abort: TriggerOperationAbort, - ) { - if (nodeIds.size === 0) return; + ): Promise { + 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, abort: TriggerOperationAbort, - ) { - await this.tracing.startSpan( + ): Promise { + 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, abort: TriggerOperationAbort, - ) { - const removedNodeNames: string[] = []; - let firstFailure: Error | undefined; + ): Promise { + // 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(); + + // 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(); - 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, + additionalData: IWorkflowExecuteAdditionalData, + failuresByNode: Map, + ): 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 { + 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( diff --git a/packages/cli/test/integration/database/repositories/workflow-publication-outbox.repository.test.ts b/packages/cli/test/integration/database/repositories/workflow-publication-outbox.repository.test.ts index c5516327aaa..bfe40de3d37 100644 --- a/packages/cli/test/integration/database/repositories/workflow-publication-outbox.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/workflow-publication-outbox.repository.test.ts @@ -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();