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

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