fix(deployments): retire inactive-version side effects by row with bounded outbox continuation

Author: icecrasher321

apps/sim/lib/admin/member-operation.test.ts

@@ -51,6 +51,11 @@ vi.mock('@/lib/workspaces/organization-workspaces', () => ({
   ownedAttachableWorkspacesWhere: vi.fn(() => undefined),
 }))
 vi.mock('@/lib/core/outbox/service', () => ({
+  continueOutboxHandler: (reason: string) => ({
+    outcome: 'deferred',
+    reason,
+    consumeAttempt: false,
+  }),
   deferOutboxHandler: (reason: string, _minimum?: number, consumeAttempt = true) => ({
     outcome: 'deferred',
     reason,

apps/sim/lib/admin/member-operation.ts

@@ -16,7 +16,7 @@ import {
 import { reconcileOrganizationSeats } from '@/lib/billing/organizations/seats'
 import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/utils'
 import {
-  deferOutboxHandler,
+  continueOutboxHandler,
   enqueueOutboxEvent,
   type OutboxHandler,
   outboxEventHasSourceOperationId,
@@ -688,7 +688,7 @@ export const processAdminMemberOperation: OutboxHandler<unknown> = async (rawPay
   }
 
   if (nextWorkspaceIndex < payload.request.workspaceIds.length) {
-    return deferOutboxHandler('Continuing bounded member workspace moves', undefined, false)
+    return continueOutboxHandler('Continuing bounded member workspace moves')
   }
 }
 

apps/sim/lib/billing/enterprise-provisioning.test.ts

@@ -61,6 +61,11 @@ vi.mock('@/lib/billing/webhooks/enterprise-reconciliation-lease', () => ({
   ),
 }))
 vi.mock('@/lib/core/outbox/service', () => ({
+  continueOutboxHandler: (reason: string) => ({
+    outcome: 'deferred',
+    reason,
+    consumeAttempt: false,
+  }),
   deferOutboxHandler: (reason: string, minimumBackoffMs?: number, consumeAttempt = true) => ({
     outcome: 'deferred',
     reason,

apps/sim/lib/billing/enterprise-provisioning.ts

@@ -79,6 +79,7 @@ import { withEnterpriseReconciliationLease } from '@/lib/billing/webhooks/enterp
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-handlers'
 import { env } from '@/lib/core/config/env'
 import {
+  continueOutboxHandler,
   deferOutboxHandler,
   enqueueOutboxEvent,
   type OutboxEventContext,
@@ -3108,7 +3109,7 @@ export const reconcileEnterpriseMembers: OutboxHandler<unknown> = async (rawPayl
 
   if (!nextCursor) return
   await context.checkpointPayload({ afterUserId: nextCursor })
-  return deferOutboxHandler('Continuing bounded Enterprise member reconciliation', undefined, false)
+  return continueOutboxHandler('Continuing bounded Enterprise member reconciliation')
 }
 
 export const enterpriseIssuanceOutboxHandlers = {

apps/sim/lib/core/outbox/service.test.ts

@@ -25,6 +25,7 @@ vi.mock('@sim/utils/id', () => ({
 }))
 
 import {
+  continueOutboxHandler,
   deferOutboxHandler,
   enqueueOrReschedulePendingOutboxEvent,
   enqueueOutboxEvent,
@@ -387,6 +388,20 @@ describe('processOutboxEvents — handler success and retry', () => {
     expect(deferredUpdate).toMatchObject({ attempts: 4, lastError: null, lockedAt: null })
   })
 
+  it('re-runs a continued handler without consuming its attempt budget', async () => {
+    const handler = vi.fn(async () => continueOutboxHandler('continuing bounded cleanup'))
+    queueTableRows(outboxEvent, [makePendingRow({ attempts: 4, maxAttempts: 5 })])
+    holdLease()
+
+    const result = await processOutboxEvents({ 'test.event': handler })
+
+    expect(result.retried).toBe(1)
+    const continuedUpdate = updateSets().find(
+      (set) => set.status === 'pending' && 'attempts' in set
+    )
+    expect(continuedUpdate).toMatchObject({ attempts: 4, lastError: null, lockedAt: null })
+  })
+
   it('dead-letters on failure when attempts reaches maxAttempts', async () => {
     const handler = vi.fn(async () => {
       throw new Error('permanent failure')

apps/sim/lib/core/outbox/service.ts

@@ -81,8 +81,9 @@ export interface DeferredOutboxHandlerResult {
   minimumBackoffMs?: number
   /**
    * Defaults to true for an external acknowledgement with a finite retry
-   * budget. Set false only for an internal dependency whose own outbox row
-   * independently reaches completed or dead-letter.
+   * budget. False is reserved for waits on an internal dependency whose own
+   * outbox row independently reaches completed or dead-letter, and for
+   * bounded continuation after durable progress (`continueOutboxHandler`).
    */
   consumeAttempt?: boolean
 }
@@ -100,6 +101,19 @@ export function deferOutboxHandler(
   }
 }
 
+/**
+ * Yields after durable progress so the worker re-runs the event without
+ * spending an attempt. For bounded batches whose remaining work shrinks on
+ * every run; a run that made no progress must throw or `deferOutboxHandler`
+ * instead, or the event never reaches a terminal state.
+ */
+export function continueOutboxHandler(
+  reason: string,
+  minimumBackoffMs?: number
+): DeferredOutboxHandlerResult {
+  return deferOutboxHandler(reason, minimumBackoffMs, false)
+}
+
 export type OutboxHandler<T = unknown> = (
   payload: T,
   context: OutboxEventContext

apps/sim/lib/webhooks/deploy.test.ts

@@ -1,9 +1,15 @@
 /**
  * @vitest-environment node
  */
-import { account, credential } from '@sim/db/schema'
-import { queueTableRows, resetDbChainMock, resetEnvFlagsMock, setEnvFlags } from '@sim/testing'
-import { eq } from 'drizzle-orm'
+import { account, credential, webhook, workflowDeploymentVersion } from '@sim/db/schema'
+import {
+  dbChainMockFns,
+  queueTableRows,
+  resetDbChainMock,
+  resetEnvFlagsMock,
+  setEnvFlags,
+} from '@sim/testing'
+import { eq, ne } from 'drizzle-orm'
 import { afterAll, beforeEach, describe, expect, it, type Mock, vi } from 'vitest'
 import type { SubBlockConfig } from '@/blocks/types'
 import type { BlockState } from '@/stores/workflows/workflow/types'
@@ -29,6 +35,12 @@ vi.mock('@/lib/webhooks/utils.server', () => ({
 vi.mock('@/lib/webhooks/pending-verification', () => ({
   PendingWebhookVerificationTracker: vi.fn(),
 }))
+const { mockIsDeploymentVersionProtected } = vi.hoisted(() => ({
+  mockIsDeploymentVersionProtected: vi.fn(),
+}))
+vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({
+  isDeploymentVersionProtectedByCurrentOperation: mockIsDeploymentVersionProtected,
+}))
 
 const {
   mockGetSlackBotCredential,
@@ -52,9 +64,11 @@ vi.mock('@/lib/webhooks/providers/slack', () => ({
 
 import {
   buildProviderConfig,
+  cleanupInactiveDeploymentWebhooks,
   resolveTriggerCredentialId,
   resolveWebhookConfigForBlock,
 } from '@/lib/webhooks/deploy'
+import { cleanupExternalWebhook } from '@/lib/webhooks/provider-subscriptions'
 import { getBlock } from '@/blocks'
 import { getTrigger } from '@/triggers'
 
@@ -639,3 +653,95 @@ describe('resolveWebhookConfigForBlock — TikTok routing', () => {
     expect(result?.error.message).toContain('Reconnect')
   })
 })
+
+describe('cleanupInactiveDeploymentWebhooks', () => {
+  const workflow = { id: 'workflow-1', userId: 'user-1', workspaceId: 'workspace-1' }
+  const input = {
+    workflowId: 'workflow-1',
+    workflow,
+    requestId: 'request-1',
+    protectedDeploymentVersionId: null,
+    limit: 5,
+  }
+
+  function staleWebhookRow(id: string) {
+    return {
+      id,
+      workflowId: 'workflow-1',
+      deploymentVersionId: 'version-1',
+      provider: 'github',
+      providerConfig: {},
+      archivedAt: null,
+      createdAt: new Date('2026-07-14T08:00:00.000Z'),
+    }
+  }
+
+  beforeEach(() => {
+    mockIsDeploymentVersionProtected.mockResolvedValue(false)
+  })
+
+  it('retires one bounded batch of stale rows and reports the remainder', async () => {
+    queueTableRows(webhook, [
+      staleWebhookRow('wh-1'),
+      staleWebhookRow('wh-2'),
+      staleWebhookRow('wh-3'),
+    ])
+    queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }])
+    queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }])
+
+    await expect(cleanupInactiveDeploymentWebhooks({ ...input, limit: 2 })).resolves.toEqual({
+      hasMore: true,
+    })
+
+    expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledTimes(2)
+    expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledWith(
+      expect.objectContaining({ id: 'wh-1' }),
+      workflow,
+      'request-1',
+      { throwOnError: true }
+    )
+    expect(dbChainMockFns.delete).toHaveBeenCalledTimes(2)
+  })
+
+  it('reports completion once the batch drains every stale row', async () => {
+    queueTableRows(webhook, [staleWebhookRow('wh-1')])
+    queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }])
+
+    await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: false })
+
+    expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledTimes(1)
+    expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1)
+  })
+
+  it('excludes the version the current operation is preparing from the batch', async () => {
+    queueTableRows(webhook, [])
+
+    await expect(
+      cleanupInactiveDeploymentWebhooks({ ...input, protectedDeploymentVersionId: 'version-3' })
+    ).resolves.toEqual({ hasMore: false })
+
+    expect(ne).toHaveBeenCalledWith(webhook.deploymentVersionId, 'version-3')
+  })
+
+  it('stops before any provider call once the fence reports a change', async () => {
+    queueTableRows(webhook, [staleWebhookRow('wh-1')])
+
+    await expect(
+      cleanupInactiveDeploymentWebhooks({ ...input, shouldContinue: async () => false })
+    ).resolves.toEqual({ hasMore: true })
+
+    expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled()
+    expect(dbChainMockFns.delete).not.toHaveBeenCalled()
+  })
+
+  it('leaves a row alone when its version became the current candidate mid-batch', async () => {
+    queueTableRows(webhook, [staleWebhookRow('wh-1')])
+    mockIsDeploymentVersionProtected.mockResolvedValue(true)
+
+    await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: true })
+
+    expect(mockIsDeploymentVersionProtected).toHaveBeenCalledWith('workflow-1', 'version-1')
+    expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled()
+    expect(dbChainMockFns.delete).not.toHaveBeenCalled()
+  })
+})