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