From c6c5f563e4ba53092aa467d12cc8e33950b4b4a6 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Mon, 10 Aug 2026 20:30:50 -0700 Subject: [PATCH] fix(deployments): stop superseded activations from dead-lettering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 29 workflow.deployment.prepare.v2 events dead-lettered with "Webhook registration operation is stale", every one at attempts = max_attempts. A full retry budget means the failure is deterministic, which rules out the preparation path: an attempt superseded while preparing is marked superseded, so its next attempt short-circuits at the top of the handler and completes. The branch a retry re-enters is the other one. isTerminalNonActiveOperation covers failed and superseded but not active, so an attempt that activated and was then superseded by the next deploy keeps its own active status, re-enters post-activation work on every retry, and re-fails the same generation fence until the event dies. The fence it fails is correct — it takes the same workflow row lock the generation bump takes, and compares generations exactly — so nothing about the detection is racy; only the reaction to it was wrong. Reaching it needs a handler timeout, which parks the row for the 10-minute reaper instead of the 2s/4s/8s backoff, opening a window wide enough for a redeploy to land. Gate the resume branch on the operation still owning the current generation, matching the sibling cleanup that already does this, and complete the event as a no-op when it does not. The newer generation adopts the leftover work anyway: it collects every retired registration below its own fence. Also reverse the post-activation order. The audit entry, analytics event, socket notification, and workspace event describe a cutover that is already durable, and each is separately checkpointed, but they ran behind retiring the previous generation's external subscriptions — one provider call per retired row, and by far the most failure-prone step there. A single flaky provider silently cost the deploy its audit trail and left clients on the old version until something else refreshed them. Both call sites now share one helper so the order cannot drift apart again. Co-Authored-By: Claude Opus 5 (1M context) --- .../lib/webhooks/registration-store.test.ts | 89 +++++++++++++++++-- .../sim/lib/workflows/deployment-lifecycle.ts | 7 ++ .../lib/workflows/deployment-outbox.test.ts | 60 +++++++++++++ apps/sim/lib/workflows/deployment-outbox.ts | 89 +++++++++++++------ 4 files changed, 212 insertions(+), 33 deletions(-) diff --git a/apps/sim/lib/webhooks/registration-store.test.ts b/apps/sim/lib/webhooks/registration-store.test.ts index 5a0641f3304..b54f67d131b 100644 --- a/apps/sim/lib/webhooks/registration-store.test.ts +++ b/apps/sim/lib/webhooks/registration-store.test.ts @@ -63,6 +63,14 @@ const FENCE: WebhookRegistrationOperationFence = { deploymentVersionId: 'version-3', } +/** The redeploy that lands seconds after {@link FENCE} and supersedes it. */ +const NEXT_FENCE: WebhookRegistrationOperationFence = { + workflowId: 'workflow-1', + operationId: 'operation-2', + generation: 4, + deploymentVersionId: 'version-4', +} + interface UpdateCall { payload: Record condition: Condition @@ -125,6 +133,15 @@ function createTx(selectResults: unknown[][]) { return { tx: tx as unknown as DbOrTx, updates, inserts, updateResults } } +/** Routes `db.transaction` at a queue-driven tx so store writes are observable. */ +function runInTx(selectResults: unknown[][]) { + const harness = createTx(selectResults) + dbChainMockFns.transaction.mockImplementation( + async (callback: (tx: DbOrTx) => Promise) => callback(harness.tx) + ) + return harness +} + function activeRow(overrides: Record = {}) { return { id: 'wh-active', @@ -224,14 +241,6 @@ describe('prepareWebhookRegistrationIntents', () => { }) }) - function runInTx(selectResults: unknown[][]) { - const harness = createTx(selectResults) - dbChainMockFns.transaction.mockImplementation( - async (callback: (tx: DbOrTx) => Promise) => callback(harness.tx) - ) - return harness - } - const desired = { blockId: 'block-1', provider: 'slack', @@ -341,3 +350,67 @@ describe('prepareWebhookRegistrationIntents', () => { expect(updates).toHaveLength(0) }) }) + +describe('redeploys racing within seconds', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + mockClaimWebhookPath.mockResolvedValue('hooks/a') + dbChainMockFns.transaction.mockImplementation(async () => { + throw new Error('db.transaction not configured for this test') + }) + }) + + const desired = { + blockId: 'block-1', + provider: 'slack', + path: 'hooks/a', + routingKey: null, + providerConfig: { url: 'https://example.test' }, + configFingerprint: 'fp-new', + } + + it('no-ops the superseded attempt and still lands the newer registration', async () => { + mockIsDeploymentOperationCurrent.mockResolvedValue(false) + const superseded = runInTx([[{ id: 'workflow-1' }]]) + + await expect( + prepareWebhookRegistrationIntents({ fence: FENCE, desired: [desired] }) + ).rejects.toBeInstanceOf(StaleWebhookRegistrationOperationError) + expect(superseded.inserts).toHaveLength(0) + expect(superseded.updates).toHaveLength(0) + expect(mockClaimWebhookPath).not.toHaveBeenCalled() + + mockIsDeploymentOperationCurrent.mockResolvedValue(true) + const winner = runInTx([[{ id: 'workflow-1' }], [], [activeRow()], [], []]) + + const work = await prepareWebhookRegistrationIntents({ fence: NEXT_FENCE, desired: [desired] }) + + expect(mockClaimWebhookPath).toHaveBeenCalledWith(expect.anything(), { + path: 'hooks/a', + workflowId: 'workflow-1', + generation: 4, + }) + expect(work.candidates).toHaveLength(1) + expect(winner.inserts).toHaveLength(1) + expect(winner.inserts[0].values).toEqual( + expect.objectContaining({ + registrationStatus: 'candidate', + registrationGeneration: 4, + deploymentVersionId: 'version-4', + }) + ) + + const activation = createTx([[{ id: 'workflow-1' }], [], []]) + await activateWebhookRegistrations(activation.tx, NEXT_FENCE) + + expect(activation.updates[1].payload).toEqual( + expect.objectContaining({ + registrationStatus: 'active', + deploymentVersionId: 'version-4', + isActive: true, + archivedAt: null, + }) + ) + }) +}) diff --git a/apps/sim/lib/workflows/deployment-lifecycle.ts b/apps/sim/lib/workflows/deployment-lifecycle.ts index 1ed98bf590b..d4ed7f51c7b 100644 --- a/apps/sim/lib/workflows/deployment-lifecycle.ts +++ b/apps/sim/lib/workflows/deployment-lifecycle.ts @@ -129,6 +129,13 @@ export function parseDeploymentReadiness(value: unknown): DeploymentReadiness | export const DEPLOYMENT_ERROR_CODES = { webhookPathConflict: 'webhook_path_conflict', invalidTriggerConfiguration: 'invalid_trigger_configuration', + /** + * A newer generation took over the workflow while this attempt was running. + * Never a failure — the newer attempt owns the outcome — so it is neither + * persisted on the operation nor counted as non-retryable; it exists to give + * the benign hand-off a greppable identity in logs. + */ + operationSuperseded: 'deployment_operation_superseded', } as const const NON_RETRYABLE_DEPLOYMENT_ERROR_CODES = new Set([ diff --git a/apps/sim/lib/workflows/deployment-outbox.test.ts b/apps/sim/lib/workflows/deployment-outbox.test.ts index b08db0ce3e9..5481e9eb8be 100644 --- a/apps/sim/lib/workflows/deployment-outbox.test.ts +++ b/apps/sim/lib/workflows/deployment-outbox.test.ts @@ -473,6 +473,66 @@ describe('versioned deployment preparation outbox', () => { expect(mockActivateDeploymentOperation).not.toHaveBeenCalled() }) + /** + * The production shape: an attempt activates, its post-activation phase is + * interrupted (handler timeout), and a redeploy lands before the reaper + * requeues it. Every resumed attempt then re-fails the same generation + * fence, so without the guard it exhausts the retry budget and dead-letters. + */ + it('skips post-activation work once a newer deploy supersedes an activated attempt', async () => { + mockGetDeploymentOperation.mockResolvedValue(operation({ status: 'active', completedAt: NOW })) + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + mockCleanupRetiredWebhookRegistrations.mockRejectedValue( + new Error('Webhook registration operation is stale') + ) + + await expect(handler()(payload(), context(new AbortController(), 3))).resolves.toBeUndefined() + + expect(mockCleanupRetiredWebhookRegistrations).not.toHaveBeenCalled() + expect(mockRecordAudit).not.toHaveBeenCalled() + expect(mockMarkDeploymentOperationFailed).not.toHaveBeenCalled() + expect(mockRecordDeploymentOperationRetry).not.toHaveBeenCalled() + }) + + it('resumes post-activation work while the activated attempt is still current', async () => { + mockIsDeploymentOperationCurrent.mockResolvedValue(true) + mockGetDeploymentOperation.mockResolvedValue(operation({ status: 'active', completedAt: NOW })) + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + + await handler()(payload(), context()) + + expect(mockCleanupRetiredWebhookRegistrations).toHaveBeenCalledTimes(1) + expect(mockRecordAudit).toHaveBeenCalledTimes(1) + expect(mockEmitWorkflowDeployedEvent).toHaveBeenCalledTimes(1) + }) + + /** + * Retiring the previous generation's provider subscriptions is the slowest + * step after cutover; a deploy that already went live must not lose its + * audit trail or its "deployment changed" notification when that step fails. + */ + it('records and notifies an activated deploy before retiring old subscriptions', async () => { + mockIsDeploymentOperationCurrent.mockResolvedValue(true) + mockGetDeploymentOperation.mockResolvedValue(operation({ status: 'active', completedAt: NOW })) + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + mockCleanupRetiredWebhookRegistrations.mockRejectedValue(new Error('provider unavailable')) + + await expect(handler()(payload(), context())).rejects.toThrow('provider unavailable') + + expect(mockRecordAudit).toHaveBeenCalledTimes(1) + expect(mockCaptureServerEvent).toHaveBeenCalledTimes(1) + expect(mockEmitWorkflowDeployedEvent).toHaveBeenCalledTimes(1) + expect(mockRecordAudit.mock.invocationCallOrder[0]).toBeLessThan( + mockCleanupRetiredWebhookRegistrations.mock.invocationCallOrder[0] + ) + }) + it('keeps v1 cleanup from deleting a candidate owned by the current v2 operation', async () => { queueTableRows(schemaMock.workflow, [ { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, diff --git a/apps/sim/lib/workflows/deployment-outbox.ts b/apps/sim/lib/workflows/deployment-outbox.ts index ed58b1d040a..14db4e5bddc 100644 --- a/apps/sim/lib/workflows/deployment-outbox.ts +++ b/apps/sim/lib/workflows/deployment-outbox.ts @@ -326,19 +326,33 @@ async function prepareDeploymentOperation( } if (operation.status === 'active') { - await cleanupRetiredWebhooksForOperation({ - payload, - workflow: workflowRecord as Record, - context, - }) - await cleanupInactiveDeploymentsForOperation({ - payload, - workflow: workflowRecord as Record, - checkpoints, - checkpoint, - context, + /** + * Resuming an attempt that already activated: every remaining step is + * fenced to this generation, so once a newer one exists they can only + * fail, identically, on every retry until the event dead-letters. The + * terminal short circuit above cannot catch this — a superseded-after- + * activation attempt keeps its own `active` status — and the newer + * generation adopts the leftover work anyway, retired registrations + * included (it collects every retired row below its own fence). + */ + context.signal.throwIfAborted() + const isCurrent = await isDeploymentOperationCurrent({ + workflowId: payload.workflowId, + operationId: payload.operationId, + generation: payload.generation, }) - await emitPostActivationSideEffects({ + context.signal.throwIfAborted() + if (!isCurrent) { + logger.info('Skipping post-activation work for a superseded generation', { + workflowId: payload.workflowId, + operationId: payload.operationId, + generation: payload.generation, + errorCode: DEPLOYMENT_ERROR_CODES.operationSuperseded, + }) + return + } + + await runPostActivationWork({ payload, operation, workflow: workflowRecord as Record, @@ -488,19 +502,7 @@ async function prepareDeploymentOperation( notifyMcpToolServers(affectedMcpServers) context.signal.throwIfAborted() - await cleanupRetiredWebhooksForOperation({ - payload, - workflow: workflowRecord as Record, - context, - }) - await cleanupInactiveDeploymentsForOperation({ - payload, - workflow: workflowRecord as Record, - checkpoints, - checkpoint, - context, - }) - await emitPostActivationSideEffects({ + await runPostActivationWork({ payload, operation, workflow: workflowRecord as Record, @@ -510,6 +512,43 @@ async function prepareDeploymentOperation( }) } +/** + * Runs everything that follows a committed cutover — notifications first. + * + * The ordering is load-bearing. The audit entry, analytics event, socket + * notification, and workspace event all describe an activation that is + * already durable, and each is individually checkpointed. Retiring the + * previous generation's external subscriptions is best-effort cleanup that + * makes one provider call per retired row and is by far the slowest, most + * failure-prone step here. Running cleanup first put every one of those + * notifications behind it, so a single flaky provider — or the handler + * timeout its latency burns through — silently cost the deploy its audit + * trail and left clients on the old version until something else refreshed + * them. Nothing below depends on the cleanup having run. + */ +async function runPostActivationWork(params: { + payload: PrepareDeploymentV2Payload + operation: WorkflowDeploymentOperation + workflow: Record + checkpoints: DeploymentPreparationCheckpoints + checkpoint: (patch: Partial) => Promise + context: OutboxEventContext +}): Promise { + await emitPostActivationSideEffects(params) + await cleanupRetiredWebhooksForOperation({ + payload: params.payload, + workflow: params.workflow, + context: params.context, + }) + await cleanupInactiveDeploymentsForOperation({ + payload: params.payload, + workflow: params.workflow, + checkpoints: params.checkpoints, + checkpoint: params.checkpoint, + context: params.context, + }) +} + async function prepareReadinessComponent(params: { payload: PrepareDeploymentV2Payload operation: WorkflowDeploymentOperation