fix(deployments): stop superseded activations from dead-lettering

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) <noreply@anthropic.com>

Author: icecrasher321

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<string, unknown>
   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<unknown>) => callback(harness.tx)
+  )
+  return harness
+}
+
 function activeRow(overrides: Record<string, unknown> = {}) {
   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<unknown>) => 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,
+      })
+    )
+  })
+})

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<string>([

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' },

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<string, unknown>,
-      context,
-    })
-    await cleanupInactiveDeploymentsForOperation({
-      payload,
-      workflow: workflowRecord as Record<string, unknown>,
-      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<string, unknown>,
@@ -488,19 +502,7 @@ async function prepareDeploymentOperation(
   notifyMcpToolServers(affectedMcpServers)
   context.signal.throwIfAborted()
 
-  await cleanupRetiredWebhooksForOperation({
-    payload,
-    workflow: workflowRecord as Record<string, unknown>,
-    context,
-  })
-  await cleanupInactiveDeploymentsForOperation({
-    payload,
-    workflow: workflowRecord as Record<string, unknown>,
-    checkpoints,
-    checkpoint,
-    context,
-  })
-  await emitPostActivationSideEffects({
+  await runPostActivationWork({
     payload,
     operation,
     workflow: workflowRecord as Record<string, unknown>,
@@ -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<string, unknown>
+  checkpoints: DeploymentPreparationCheckpoints
+  checkpoint: (patch: Partial<DeploymentPreparationCheckpoints>) => Promise<void>
+  context: OutboxEventContext
+}): Promise<void> {
+  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