From cd81e90589947fe7eddb10b111114aa3881c5bb9 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Wed, 2 Sep 2026 11:53:47 -0700 Subject: [PATCH 1/2] fix(deployments): retire inactive-version side effects by row with bounded outbox continuation --- apps/sim/lib/admin/member-operation.test.ts | 5 + apps/sim/lib/admin/member-operation.ts | 4 +- .../billing/enterprise-provisioning.test.ts | 5 + .../lib/billing/enterprise-provisioning.ts | 3 +- apps/sim/lib/core/outbox/service.test.ts | 15 ++ apps/sim/lib/core/outbox/service.ts | 18 +- apps/sim/lib/webhooks/deploy.test.ts | 112 ++++++++- apps/sim/lib/webhooks/deploy.ts | 171 +++++++++++--- .../lib/workflows/deployment-outbox.test.ts | 153 +++++++++++- apps/sim/lib/workflows/deployment-outbox.ts | 222 +++++++++++------- .../persistence/deployment-operations.test.ts | 28 +++ .../persistence/deployment-operations.ts | 42 +++- .../lib/workflows/schedules/deploy.test.ts | 92 +++++++- apps/sim/lib/workflows/schedules/deploy.ts | 75 +++++- apps/sim/lib/workflows/schedules/index.ts | 2 + 15 files changed, 790 insertions(+), 157 deletions(-) diff --git a/apps/sim/lib/admin/member-operation.test.ts b/apps/sim/lib/admin/member-operation.test.ts index 445c82dd07f..e915c4b24b2 100644 --- a/apps/sim/lib/admin/member-operation.test.ts +++ b/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, diff --git a/apps/sim/lib/admin/member-operation.ts b/apps/sim/lib/admin/member-operation.ts index 71eb2981bc8..e68090a08e4 100644 --- a/apps/sim/lib/admin/member-operation.ts +++ b/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 = async (rawPay } if (nextWorkspaceIndex < payload.request.workspaceIds.length) { - return deferOutboxHandler('Continuing bounded member workspace moves', undefined, false) + return continueOutboxHandler('Continuing bounded member workspace moves') } } diff --git a/apps/sim/lib/billing/enterprise-provisioning.test.ts b/apps/sim/lib/billing/enterprise-provisioning.test.ts index a71b36f8bb5..58afc58ab15 100644 --- a/apps/sim/lib/billing/enterprise-provisioning.test.ts +++ b/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, diff --git a/apps/sim/lib/billing/enterprise-provisioning.ts b/apps/sim/lib/billing/enterprise-provisioning.ts index 739ca721211..6a5f1152037 100644 --- a/apps/sim/lib/billing/enterprise-provisioning.ts +++ b/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 = 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 = { diff --git a/apps/sim/lib/core/outbox/service.test.ts b/apps/sim/lib/core/outbox/service.test.ts index dc4fd2f2504..9817363630c 100644 --- a/apps/sim/lib/core/outbox/service.test.ts +++ b/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') diff --git a/apps/sim/lib/core/outbox/service.ts b/apps/sim/lib/core/outbox/service.ts index a36600aaa43..a4ede767836 100644 --- a/apps/sim/lib/core/outbox/service.ts +++ b/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 = ( payload: T, context: OutboxEventContext diff --git a/apps/sim/lib/webhooks/deploy.test.ts b/apps/sim/lib/webhooks/deploy.test.ts index 9732e9d0308..1423f20ddce 100644 --- a/apps/sim/lib/webhooks/deploy.test.ts +++ b/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() + }) +}) diff --git a/apps/sim/lib/webhooks/deploy.ts b/apps/sim/lib/webhooks/deploy.ts index 136720edcc1..141f03b6e8c 100644 --- a/apps/sim/lib/webhooks/deploy.ts +++ b/apps/sim/lib/webhooks/deploy.ts @@ -3,7 +3,7 @@ import { account, credential, webhook, workflowDeploymentVersion } from '@sim/db import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { generateShortId } from '@sim/utils/id' -import { and, eq, inArray, isNull, or } from 'drizzle-orm' +import { and, asc, eq, inArray, isNull, ne, or } from 'drizzle-orm' import type { NextRequest } from 'next/server' import { isSlackExtendedScopesEnabled } from '@/lib/core/config/env-flags' import { getProviderIdFromServiceId } from '@/lib/oauth' @@ -33,6 +33,7 @@ import { replaceSlackStreamAuthoringConfig, } from '@/lib/webhooks/slack-stream-config' import { findConflictingWebhookPathOwner } from '@/lib/webhooks/utils.server' +import { isDeploymentVersionProtectedByCurrentOperation } from '@/lib/workflows/persistence/deployment-operations' import { buildCanonicalIndex, buildSubBlockValues, @@ -1276,39 +1277,15 @@ export async function cleanupWebhooksForWorkflow( if (!skipExternalCleanup) { for (const wh of existingWebhooks) { - if (shouldDeleteWebhook && !(await shouldDeleteWebhook())) { - logger.info(`[${requestId}] Stopping webhook cleanup because deployment became active`, { - workflowId, - deploymentVersionId, - webhookId: wh.id, - }) - return - } - - try { - await cleanupExternalWebhook(wh, workflow, requestId, { - throwOnError: strictExternalCleanup, - }) - } catch (cleanupError) { - logger.warn(`[${requestId}] Failed to cleanup external webhook ${wh.id}`, cleanupError) - if (strictExternalCleanup) throw cleanupError - // Continue with other webhooks even if one fails - } - - const deleted = await deleteWebhookRecordAfterCleanup({ - workflowId, + const deleted = await cleanupWebhookRow({ + webhook: wh, + workflow, + requestId, deploymentVersionId, - webhookId: wh.id, + strictExternalCleanup, shouldDeleteWebhook, }) - if (!deleted) { - logger.info(`[${requestId}] Stopping webhook DB cleanup because deployment became active`, { - workflowId, - deploymentVersionId, - webhookId: wh.id, - }) - return - } + if (!deleted) return } } else { for (const wh of existingWebhooks) { @@ -1336,6 +1313,138 @@ export async function cleanupWebhooksForWorkflow( ) } +type WebhookRow = typeof webhook.$inferSelect + +/** + * Tears down one webhook's provider subscription and then deletes its row. + * Returns false when `shouldDeleteWebhook` reports the deployment became + * active again, in which case the caller must stop touching its rows. + */ +async function cleanupWebhookRow(params: { + webhook: WebhookRow + workflow: Record + requestId: string + deploymentVersionId?: string | null + strictExternalCleanup: boolean + shouldDeleteWebhook?: () => Promise +}): Promise { + const { webhook: wh, workflow, requestId, deploymentVersionId, strictExternalCleanup } = params + const workflowId = wh.workflowId + if (params.shouldDeleteWebhook && !(await params.shouldDeleteWebhook())) { + logger.info(`[${requestId}] Stopping webhook cleanup because deployment became active`, { + workflowId, + deploymentVersionId, + webhookId: wh.id, + }) + return false + } + + try { + await cleanupExternalWebhook(wh, workflow, requestId, { throwOnError: strictExternalCleanup }) + } catch (cleanupError) { + logger.warn(`[${requestId}] Failed to cleanup external webhook ${wh.id}`, cleanupError) + if (strictExternalCleanup) throw cleanupError + } + + const deleted = await deleteWebhookRecordAfterCleanup({ + workflowId, + deploymentVersionId, + webhookId: wh.id, + shouldDeleteWebhook: params.shouldDeleteWebhook, + }) + if (!deleted) { + logger.info(`[${requestId}] Stopping webhook DB cleanup because deployment became active`, { + workflowId, + deploymentVersionId, + webhookId: wh.id, + }) + } + return deleted +} + +export interface InactiveDeploymentWebhookCleanupResult { + /** True when rows remain beyond this batch and the caller should run again. */ + hasMore: boolean +} + +/** + * Tears down webhooks still owned by inactive deployment versions of a + * workflow, at most `limit` rows per call. Provider teardown costs one call + * per row, so the work is bounded here and `hasMore` asks the caller to come + * back; every finished row leaves the remaining set smaller, so repeated calls + * converge. `protectedDeploymentVersionId` is the version an in-flight + * operation is preparing, inactive until cutover but live preparation state; + * each row is re-checked against the current operation right before its + * delete because that can change while the batch runs. + */ +export async function cleanupInactiveDeploymentWebhooks(params: { + workflowId: string + workflow: Record + requestId: string + protectedDeploymentVersionId: string | null + limit: number + shouldContinue?: () => Promise +}): Promise { + const { workflowId, workflow, requestId, shouldContinue } = params + const inactiveVersionIds = db + .select({ id: workflowDeploymentVersion.id }) + .from(workflowDeploymentVersion) + .where( + and( + eq(workflowDeploymentVersion.workflowId, workflowId), + eq(workflowDeploymentVersion.isActive, false) + ) + ) + const staleWebhooks = await db + .select() + .from(webhook) + .where( + and( + eq(webhook.workflowId, workflowId), + isNull(webhook.archivedAt), + inArray(webhook.deploymentVersionId, inactiveVersionIds), + params.protectedDeploymentVersionId + ? ne(webhook.deploymentVersionId, params.protectedDeploymentVersionId) + : undefined + ) + ) + .orderBy(asc(webhook.createdAt)) + .limit(params.limit + 1) + + const batch = staleWebhooks.slice(0, params.limit) + if (batch.length === 0) return { hasMore: false } + + logger.info( + `[${requestId}] Cleaning up ${batch.length} webhook(s) owned by inactive deployments`, + { + workflowId, + webhookIds: batch.map((wh) => wh.id), + } + ) + + for (const wh of batch) { + const deploymentVersionId = wh.deploymentVersionId + const deleted = await cleanupWebhookRow({ + webhook: wh, + workflow, + requestId, + deploymentVersionId, + strictExternalCleanup: true, + shouldDeleteWebhook: async () => { + if (shouldContinue && !(await shouldContinue())) return false + if (!deploymentVersionId) return true + return !(await isDeploymentVersionProtectedByCurrentOperation( + workflowId, + deploymentVersionId + )) + }, + }) + if (!deleted) return { hasMore: true } + } + + return { hasMore: staleWebhooks.length > params.limit } +} + /** * Deletes a webhook record unless the deployment became active again. * diff --git a/apps/sim/lib/workflows/deployment-outbox.test.ts b/apps/sim/lib/workflows/deployment-outbox.test.ts index 4483575af30..92d4089ecb5 100644 --- a/apps/sim/lib/workflows/deployment-outbox.test.ts +++ b/apps/sim/lib/workflows/deployment-outbox.test.ts @@ -30,6 +30,9 @@ const { mockRecordAudit, mockEmitWorkflowDeployedEvent, mockCaptureServerEvent, + mockCleanupInactiveDeploymentWebhooks, + mockDeleteInactiveDeploymentSchedules, + mockGetProtectedDeploymentVersionId, mockTx, } = vi.hoisted(() => ({ mockPrepareWebhooks: vi.fn(), @@ -51,6 +54,9 @@ const { mockRecordAudit: vi.fn(), mockEmitWorkflowDeployedEvent: vi.fn(), mockCaptureServerEvent: vi.fn(), + mockCleanupInactiveDeploymentWebhooks: vi.fn(), + mockDeleteInactiveDeploymentSchedules: vi.fn(), + mockGetProtectedDeploymentVersionId: vi.fn(), mockTx: { select: vi.fn(), update: vi.fn(), execute: vi.fn() }, })) @@ -66,6 +72,11 @@ vi.mock('@sim/audit', () => ({ vi.mock('@sim/db', () => ({ ...dbChainMock, ...schemaMock })) vi.mock('@/lib/core/outbox/service', () => ({ + continueOutboxHandler: (reason: string) => ({ + outcome: 'deferred', + reason, + consumeAttempt: false, + }), enqueueOutboxEvent: vi.fn(), processOutboxEventById: vi.fn(), })) @@ -85,6 +96,7 @@ vi.mock('@/lib/mcp/workflow-mcp-sync', () => ({ })) vi.mock('@/lib/webhooks/deploy', () => ({ + cleanupInactiveDeploymentWebhooks: mockCleanupInactiveDeploymentWebhooks, cleanupWebhooksForWorkflow: mockCleanupWebhooksForWorkflow, prepareStableTriggerWebhooksForDeploy: vi.fn(), saveTriggerWebhooksForDeploy: vi.fn(), @@ -102,6 +114,7 @@ vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ activateDeploymentOperation: mockActivateDeploymentOperation, beginDeploymentOperationActivation: mockBeginDeploymentOperationActivation, getDeploymentOperation: mockGetDeploymentOperation, + getProtectedDeploymentVersionId: mockGetProtectedDeploymentVersionId, isDeploymentOperationCurrent: mockIsDeploymentOperationCurrent, isDeploymentVersionProtectedByCurrentOperation: mockIsDeploymentVersionProtectedByCurrentOperation, @@ -113,6 +126,7 @@ vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ vi.mock('@/lib/workflows/schedules', () => ({ createSchedulesForDeploy: mockCreateSchedulesForDeploy, + deleteInactiveDeploymentSchedules: mockDeleteInactiveDeploymentSchedules, deleteSchedulesForWorkflow: vi.fn(), })) @@ -219,6 +233,9 @@ describe('versioned deployment preparation outbox', () => { }) mockIsDeploymentOperationCurrent.mockResolvedValue(false) mockIsDeploymentVersionProtectedByCurrentOperation.mockResolvedValue(false) + mockGetProtectedDeploymentVersionId.mockResolvedValue(null) + mockDeleteInactiveDeploymentSchedules.mockResolvedValue({ status: 'deleted', count: 0 }) + mockCleanupInactiveDeploymentWebhooks.mockResolvedValue({ hasMore: false }) }) it('activates only after every preparation component is ready', async () => { @@ -558,6 +575,8 @@ describe('versioned deployment preparation outbox', () => { await expect(handler()(payload(), context(new AbortController(), 3))).resolves.toBeUndefined() expect(mockCleanupRetiredWebhookRegistrations).not.toHaveBeenCalled() + expect(mockDeleteInactiveDeploymentSchedules).not.toHaveBeenCalled() + expect(mockCleanupInactiveDeploymentWebhooks).not.toHaveBeenCalled() expect(mockMarkDeploymentOperationFailed).not.toHaveBeenCalled() expect(mockRecordDeploymentOperationRetry).not.toHaveBeenCalled() }) @@ -589,11 +608,35 @@ describe('versioned deployment preparation outbox', () => { { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, ]) - await handler()(payload(), context()) + const outboxContext = context() + + await expect(handler()(payload(), outboxContext)).resolves.toBeUndefined() expect(mockCleanupRetiredWebhookRegistrations).toHaveBeenCalledTimes(1) expect(mockRecordAudit).toHaveBeenCalledTimes(1) expect(mockEmitWorkflowDeployedEvent).toHaveBeenCalledTimes(1) + expect(mockDeleteInactiveDeploymentSchedules).toHaveBeenCalledWith({ + workflowId: 'workflow-1', + operationFence: { + workflowId: 'workflow-1', + operationId: 'operation-1', + generation: 2, + deploymentVersionId: 'version-2', + statuses: ['active'], + }, + }) + expect(mockCleanupInactiveDeploymentWebhooks).toHaveBeenCalledWith( + expect.objectContaining({ + workflowId: 'workflow-1', + protectedDeploymentVersionId: null, + limit: 20, + }) + ) + expect(outboxContext.checkpointPayload).toHaveBeenCalledWith( + expect.objectContaining({ + checkpoints: expect.objectContaining({ inactiveCleanupCompleted: true }), + }) + ) }) /** @@ -619,29 +662,115 @@ describe('versioned deployment preparation outbox', () => { ) }) - it('keeps v1 cleanup from deleting a candidate owned by the current v2 operation', async () => { + it('continues through the outbox while stale webhooks remain, then checkpoints the cleanup', async () => { + mockIsDeploymentOperationCurrent.mockResolvedValue(true) + mockGetDeploymentOperation.mockResolvedValue(operation({ status: 'active', completedAt: NOW })) + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + mockCleanupInactiveDeploymentWebhooks.mockResolvedValueOnce({ hasMore: true }) + const outboxContext = context() + + await expect(handler()(payload(), outboxContext)).resolves.toEqual({ + outcome: 'deferred', + reason: expect.any(String), + consumeAttempt: false, + }) + expect(outboxContext.checkpointPayload).not.toHaveBeenCalledWith( + expect.objectContaining({ + checkpoints: expect.objectContaining({ inactiveCleanupCompleted: true }), + }) + ) + + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + await expect(handler()(payload(), outboxContext)).resolves.toBeUndefined() + + expect(mockDeleteInactiveDeploymentSchedules).toHaveBeenCalledTimes(2) + expect(mockCleanupInactiveDeploymentWebhooks).toHaveBeenCalledTimes(2) + expect(outboxContext.checkpointPayload).toHaveBeenCalledWith( + expect.objectContaining({ + checkpoints: expect.objectContaining({ inactiveCleanupCompleted: true }), + }) + ) + }) + + it('stops legacy inactive cleanup as soon as its lease is aborted', async () => { + const controller = new AbortController() + controller.abort() + const cleanupHandler = + createWorkflowDeploymentOutboxHandlers()[ + WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_INACTIVE_SIDE_EFFECTS + ] + + await expect( + cleanupHandler( + { workflowId: 'workflow-1', activeDeploymentVersionId: 'version-2', userId: 'user-1' }, + context(controller) + ) + ).rejects.toMatchObject({ name: 'AbortError' }) + + expect(mockDeleteInactiveDeploymentSchedules).not.toHaveBeenCalled() + expect(mockCleanupInactiveDeploymentWebhooks).not.toHaveBeenCalled() + }) + + it('retires undeployed side effects by row and shields the candidate owned by the current v2 operation', async () => { queueTableRows(schemaMock.workflow, [ { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, ]) - queueTableRows(schemaMock.workflowDeploymentVersion, [{ isActive: false }]) queueTableRows(schemaMock.workflow, [{ isDeployed: true }]) - mockIsDeploymentVersionProtectedByCurrentOperation.mockResolvedValue(true) + mockGetProtectedDeploymentVersionId.mockResolvedValue('version-2') const cleanupHandler = createWorkflowDeploymentOutboxHandlers()[ WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_UNDEPLOYED_SIDE_EFFECTS ] - await cleanupHandler( - { + await expect( + cleanupHandler( + { + workflowId: 'workflow-1', + deploymentVersionIds: ['version-2'], + userId: 'user-1', + requestId: 'request-1', + }, + context() + ) + ).resolves.toBeUndefined() + + expect(mockDeleteInactiveDeploymentSchedules).toHaveBeenCalledWith({ + workflowId: 'workflow-1', + operationFence: undefined, + }) + expect(mockCleanupInactiveDeploymentWebhooks).toHaveBeenCalledWith( + expect.objectContaining({ workflowId: 'workflow-1', - deploymentVersionIds: ['version-2'], - userId: 'user-1', - requestId: 'request-1', - }, - context() + protectedDeploymentVersionId: 'version-2', + limit: 20, + }) ) - expect(mockCleanupWebhooksForWorkflow).not.toHaveBeenCalled() expect(mockCreateSchedulesForDeploy).not.toHaveBeenCalled() }) + + it('continues undeploy cleanup through the outbox before touching MCP tools', async () => { + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + mockCleanupInactiveDeploymentWebhooks.mockResolvedValueOnce({ hasMore: true }) + const cleanupHandler = + createWorkflowDeploymentOutboxHandlers()[ + WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_UNDEPLOYED_SIDE_EFFECTS + ] + + await expect( + cleanupHandler({ workflowId: 'workflow-1', userId: 'user-1' }, context()) + ).resolves.toEqual({ + outcome: 'deferred', + reason: expect.any(String), + consumeAttempt: false, + }) + + expect(mockNotifyMcpToolServers).not.toHaveBeenCalled() + }) }) diff --git a/apps/sim/lib/workflows/deployment-outbox.ts b/apps/sim/lib/workflows/deployment-outbox.ts index 62875063496..17aec8b99d5 100644 --- a/apps/sim/lib/workflows/deployment-outbox.ts +++ b/apps/sim/lib/workflows/deployment-outbox.ts @@ -3,10 +3,12 @@ import type { PrincipalActor } from '@sim/auth/principal' import { db, workflowDeploymentVersion, workflow as workflowTable } from '@sim/db' import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' -import { and, eq, ne } from 'drizzle-orm' +import { and, eq } from 'drizzle-orm' import { NextRequest } from 'next/server' import { env } from '@/lib/core/config/env' import { + continueOutboxHandler, + type DeferredOutboxHandlerResult, enqueueOutboxEvent, type OutboxEventContext, type OutboxHandler, @@ -24,6 +26,7 @@ import { } from '@/lib/mcp/workflow-mcp-sync' import { captureServerEvent } from '@/lib/posthog/server' import { + cleanupInactiveDeploymentWebhooks, cleanupWebhooksForWorkflow, prepareStableTriggerWebhooksForDeploy, saveTriggerWebhooksForDeploy, @@ -44,6 +47,7 @@ import { beginDeploymentOperationActivation, type DeploymentOperationGeneration, getDeploymentOperation, + getProtectedDeploymentVersionId, isDeploymentOperationCurrent, isDeploymentVersionProtectedByCurrentOperation, markDeploymentComponentReadiness, @@ -52,7 +56,11 @@ import { setDeploymentTxTimeouts, type WorkflowDeploymentOperation, } from '@/lib/workflows/persistence/deployment-operations' -import { createSchedulesForDeploy, deleteSchedulesForWorkflow } from '@/lib/workflows/schedules' +import { + createSchedulesForDeploy, + deleteInactiveDeploymentSchedules, + deleteSchedulesForWorkflow, +} from '@/lib/workflows/schedules' import { emitWorkflowDeployedEvent } from '@/lib/workspace-events/emitter' import type { BlockState } from '@/stores/workflows/workflow/types' @@ -77,6 +85,16 @@ export const DEPLOYMENT_READINESS_COMPONENTS = ['webhooks', 'schedules', 'mcp'] */ const DEPLOYMENT_PREPARATION_MAX_ATTEMPTS = 4 +/** + * Webhooks retired per outbox attempt when cleaning up inactive deployment + * versions. Each costs a provider call, so the batch keeps one attempt well + * inside the handler timeout; the handler continues through the outbox while + * rows remain. + */ +const INACTIVE_WEBHOOK_CLEANUP_BATCH_SIZE = 20 + +const INACTIVE_CLEANUP_CONTINUATION_REASON = 'Continuing inactive deployment side-effect cleanup' + interface DeploymentPreparationCheckpoints { webhooksPrepared?: boolean schedulesPrepared?: boolean @@ -132,7 +150,12 @@ interface SyncActiveSideEffectsPayload { interface CleanupUndeployedSideEffectsPayload { workflowId: string - deploymentVersionIds: string[] + /** + * Versions the undeploy retired. Cleanup finds stale rows from the versions' + * current state instead; kept for one release so events written by earlier + * pods still parse, and events written here still parse on them. + */ + deploymentVersionIds?: string[] userId: string requestId?: string } @@ -249,7 +272,7 @@ function createPrepareDeploymentHandler( return async (rawPayload, context) => { const payload = parsePrepareDeploymentV2Payload(rawPayload) try { - await prepareDeploymentOperation(payload, context, prepareWebhooks) + return await prepareDeploymentOperation(payload, context, prepareWebhooks) } catch (error) { const isFinalAttempt = context.attempts + 1 >= context.maxAttempts if (isNonRetryableDeploymentError(error) || isFinalAttempt) { @@ -305,7 +328,7 @@ async function prepareDeploymentOperation( payload: PrepareDeploymentV2Payload, context: OutboxEventContext, prepareWebhooks: PrepareDeploymentWebhooksHook -): Promise { +): Promise { context.signal.throwIfAborted() let operation = await getDeploymentOperation(payload) context.signal.throwIfAborted() @@ -338,7 +361,7 @@ async function prepareDeploymentOperation( * whatever else has started since, while only the fenced cleanup is * skipped. */ - await runPostActivationWork({ + return runPostActivationWork({ payload, operation, workflow: workflowRecord as Record, @@ -346,7 +369,6 @@ async function prepareDeploymentOperation( checkpoint, context, }) - return } if (operation.status !== 'preparing' && operation.status !== 'activating') return @@ -488,7 +510,7 @@ async function prepareDeploymentOperation( notifyMcpToolServers(affectedMcpServers) context.signal.throwIfAborted() - await runPostActivationWork({ + return runPostActivationWork({ payload, operation, workflow: workflowRecord as Record, @@ -517,6 +539,10 @@ async function prepareDeploymentOperation( * them on the same predicate would drop them for good in the window where a * newer generation exists but has not activated — this activation is still * the live one there, and nothing else will emit them. + * + * Inactive-version cleanup is bounded per attempt. While rows remain, the + * handler yields a continuation so the outbox re-runs it without spending an + * attempt; the notifications above are checkpointed and never repeat. */ async function runPostActivationWork(params: { payload: PrepareDeploymentV2Payload @@ -525,20 +551,21 @@ async function runPostActivationWork(params: { checkpoints: DeploymentPreparationCheckpoints checkpoint: (patch: Partial) => Promise context: OutboxEventContext -}): Promise { +}): Promise { await emitPostActivationSideEffects(params) await cleanupRetiredWebhooksForOperation({ payload: params.payload, workflow: params.workflow, context: params.context, }) - await cleanupInactiveDeploymentsForOperation({ + const cleanupComplete = await cleanupInactiveDeploymentsForOperation({ payload: params.payload, workflow: params.workflow, checkpoints: params.checkpoints, checkpoint: params.checkpoint, context: params.context, }) + return cleanupComplete ? undefined : continueOutboxHandler(INACTIVE_CLEANUP_CONTINUATION_REASON) } async function prepareReadinessComponent(params: { @@ -622,14 +649,19 @@ async function cleanupRetiredWebhooksForOperation(params: { }) } +/** + * Returns false while inactive-version cleanup still has rows to retire, so + * the caller yields a continuation instead of completing the event. A + * superseded attempt returns true: the newer generation owns the cleanup now. + */ async function cleanupInactiveDeploymentsForOperation(params: { payload: PrepareDeploymentV2Payload workflow: Record checkpoints: DeploymentPreparationCheckpoints checkpoint: (patch: Partial) => Promise context: OutboxEventContext -}): Promise { - if (params.checkpoints.inactiveCleanupCompleted) return +}): Promise { + if (params.checkpoints.inactiveCleanupCompleted) return true const operationFence = { workflowId: params.payload.workflowId, operationId: params.payload.operationId, @@ -644,18 +676,18 @@ async function cleanupInactiveDeploymentsForOperation(params: { return isCurrent } - if (!(await shouldContinue())) return - await cleanupInactiveDeploymentVersions({ + if (!(await shouldContinue())) return true + const { complete } = await cleanupInactiveDeploymentSideEffects({ workflowId: params.payload.workflowId, - activeDeploymentVersionId: params.payload.deploymentVersionId, workflow: params.workflow, - userId: params.payload.userId, requestId: params.payload.requestId, shouldContinue, operationFence, }) - if (!(await shouldContinue())) return + if (!(await shouldContinue())) return true + if (!complete) return false await params.checkpoint({ inactiveCleanupCompleted: true }) + return true } async function emitPostActivationSideEffects(params: { @@ -898,9 +930,10 @@ const syncActiveSideEffects = async (rawPayload: unknown): Promise => { }) } -const cleanupInactiveSideEffects = async (rawPayload: unknown): Promise => { +const cleanupInactiveSideEffects: OutboxHandler = async (rawPayload, context) => { const payload = parseCleanupInactiveSideEffectsPayload(rawPayload) const requestId = payload.requestId ?? generateRequestId() + context.signal.throwIfAborted() const [workflowRecord] = await db .select() .from(workflowTable) @@ -909,18 +942,19 @@ const cleanupInactiveSideEffects = async (rawPayload: unknown): Promise => if (!workflowRecord) return - await cleanupInactiveDeploymentVersions({ + const { complete } = await cleanupInactiveDeploymentSideEffects({ workflowId: payload.workflowId, - activeDeploymentVersionId: payload.activeDeploymentVersionId, workflow: workflowRecord as Record, - userId: payload.userId, requestId, + shouldContinue: unlessAborted(context.signal), }) + if (!complete) return continueOutboxHandler(INACTIVE_CLEANUP_CONTINUATION_REASON) } -const cleanupUndeployedSideEffects = async (rawPayload: unknown): Promise => { +const cleanupUndeployedSideEffects: OutboxHandler = async (rawPayload, context) => { const payload = parseCleanupUndeployedSideEffectsPayload(rawPayload) const requestId = payload.requestId ?? generateRequestId() + context.signal.throwIfAborted() const [workflowRecord] = await db .select() .from(workflowTable) @@ -930,47 +964,44 @@ const cleanupUndeployedSideEffects = async (rawPayload: unknown): Promise if (!workflowRecord) return const workflowData = workflowRecord as Record - for (const deploymentVersionId of payload.deploymentVersionIds) { - const [versionRow] = await db - .select({ isActive: workflowDeploymentVersion.isActive }) - .from(workflowDeploymentVersion) - .where( - and( - eq(workflowDeploymentVersion.workflowId, payload.workflowId), - eq(workflowDeploymentVersion.id, deploymentVersionId) - ) - ) - .limit(1) - - if (!versionRow || versionRow.isActive) continue - await cleanupDeploymentVersionIfInactive({ - workflowId: payload.workflowId, - workflow: workflowData, - userId: payload.userId, - requestId, - deploymentVersionId, - }) - } + const { complete } = await cleanupInactiveDeploymentSideEffects({ + workflowId: payload.workflowId, + workflow: workflowData, + requestId, + shouldContinue: unlessAborted(context.signal), + }) + if (!complete) return continueOutboxHandler(INACTIVE_CLEANUP_CONTINUATION_REASON) + context.signal.throwIfAborted() await cleanupNullVersionWebhooksIfStillUndeployed({ workflowId: payload.workflowId, workflow: workflowData, requestId, }) + context.signal.throwIfAborted() await removeMcpToolsIfStillUndeployed(payload.workflowId, requestId) } +/** Continuation gate for handlers without an operation fence: stops only when the lease aborts. */ +function unlessAborted(signal: AbortSignal): () => Promise { + return async () => { + signal.throwIfAborted() + return true + } +} + /** * Run inactive-version cleanup synchronously as part of the active-version sync, right * after the active version's webhooks/schedules are registered. * - * {@link cleanupInactiveDeploymentVersions} re-checks that each version is still inactive - * before tearing anything down, so it can never touch the now-active version. Running it - * inline — rather than only enqueueing it — closes the window where a lost + * {@link cleanupInactiveDeploymentSideEffects} only selects rows whose version is inactive and + * re-checks each webhook right before its delete, so it can never touch the now-active version. + * Running it inline — rather than only enqueueing it — closes the window where a lost * `CLEANUP_INACTIVE` outbox event leaves superseded webhooks behind as live-but-never-polled - * `is_active` orphans. The deferred event is kept as a fallback so cleanup still retries if - * the inline pass throws, without failing the already-succeeded registration. + * `is_active` orphans. The deferred event is kept as a fallback so cleanup still continues if + * the inline pass throws or has more rows than one bounded pass retires, without failing the + * already-succeeded registration. */ async function syncInactiveDeploymentCleanup(params: { workflowId: string @@ -980,57 +1011,64 @@ async function syncInactiveDeploymentCleanup(params: { requestId: string }): Promise { try { - await cleanupInactiveDeploymentVersions(params) + const { complete } = await cleanupInactiveDeploymentSideEffects({ + workflowId: params.workflowId, + workflow: params.workflow, + requestId: params.requestId, + }) + if (complete) return + logger.info( + `[${params.requestId}] Inline inactive-deployment cleanup has more rows; continuing through the outbox` + ) } catch (cleanupError) { logger.warn( `[${params.requestId}] Inline inactive-deployment cleanup failed; deferring to outbox retry`, cleanupError ) - await enqueueWorkflowInactiveDeploymentCleanup(db, { - workflowId: params.workflowId, - activeDeploymentVersionId: params.activeDeploymentVersionId, - userId: params.userId, - requestId: params.requestId, - }) } + await enqueueWorkflowInactiveDeploymentCleanup(db, { + workflowId: params.workflowId, + activeDeploymentVersionId: params.activeDeploymentVersionId, + userId: params.userId, + requestId: params.requestId, + }) } -async function cleanupInactiveDeploymentVersions(params: { +/** + * Retires schedules and webhooks still owned by inactive deployment versions + * of the workflow. Work is keyed by side-effect rows, never by versions, so a + * workflow deployed hundreds of times costs no more than one deployed twice. + * Schedules go in one fenced statement; webhooks need a provider call each + * and drain in bounded batches, with `complete: false` asking the caller to + * run again. `shouldContinue` gates every step and throws once the outbox + * lease is aborted. + */ +async function cleanupInactiveDeploymentSideEffects(params: { workflowId: string - activeDeploymentVersionId: string workflow: Record - userId: string requestId: string shouldContinue?: () => Promise operationFence?: DeploymentCleanupOperationFence -}): Promise { - if (params.shouldContinue && !(await params.shouldContinue())) return - const inactiveVersions = await db - .select({ id: workflowDeploymentVersion.id }) - .from(workflowDeploymentVersion) - .where( - and( - eq(workflowDeploymentVersion.workflowId, params.workflowId), - ne(workflowDeploymentVersion.id, params.activeDeploymentVersionId), - eq(workflowDeploymentVersion.isActive, false) - ) - ) +}): Promise<{ complete: boolean }> { + if (params.shouldContinue && !(await params.shouldContinue())) return { complete: false } - for (const version of inactiveVersions) { - if (params.shouldContinue && !(await params.shouldContinue())) return - if (await isDeploymentVersionProtectedByCurrentOperation(params.workflowId, version.id)) { - continue - } - await cleanupDeploymentVersionIfInactive({ - workflowId: params.workflowId, - workflow: params.workflow, - userId: params.userId, - requestId: params.requestId, - deploymentVersionId: version.id, - shouldContinue: params.shouldContinue, - operationFence: params.operationFence, - }) - } + const schedules = await deleteInactiveDeploymentSchedules({ + workflowId: params.workflowId, + operationFence: params.operationFence, + }) + if (schedules.status === 'superseded') return { complete: false } + + if (params.shouldContinue && !(await params.shouldContinue())) return { complete: false } + const protectedDeploymentVersionId = await getProtectedDeploymentVersionId(params.workflowId) + const { hasMore } = await cleanupInactiveDeploymentWebhooks({ + workflowId: params.workflowId, + workflow: params.workflow, + requestId: params.requestId, + protectedDeploymentVersionId, + limit: INACTIVE_WEBHOOK_CLEANUP_BATCH_SIZE, + shouldContinue: params.shouldContinue, + }) + return { complete: !hasMore } } async function cleanupDeploymentVersionIfInactive(params: { @@ -1458,7 +1496,7 @@ function parseCleanupUndeployedSideEffectsPayload( const record = parsePayloadRecord(payload) const workflowId = parseRequiredString(record.workflowId, 'workflowId') const userId = parseRequiredString(record.userId, 'userId') - const deploymentVersionIds = parseRequiredStringArray( + const deploymentVersionIds = parseOptionalStringArray( record.deploymentVersionIds, 'deploymentVersionIds' ) @@ -1467,7 +1505,12 @@ function parseCleanupUndeployedSideEffectsPayload( ? record.requestId : undefined - return { workflowId, deploymentVersionIds, userId, requestId } + return { + workflowId, + ...(deploymentVersionIds ? { deploymentVersionIds } : {}), + userId, + requestId, + } } function parseCleanupInactiveSideEffectsPayload( @@ -1509,12 +1552,13 @@ function parseRequiredPositiveInteger(value: unknown, fieldName: string): number return value } -function parseRequiredStringArray(value: unknown, fieldName: string): string[] { +function parseOptionalStringArray(value: unknown, fieldName: string): string[] | undefined { + if (value === undefined) return undefined if ( !Array.isArray(value) || value.some((item) => typeof item !== 'string' || item.length === 0) ) { - throw new Error(`Deployment outbox payload is missing ${fieldName}`) + throw new Error(`Deployment outbox payload has an invalid ${fieldName}`) } return value } diff --git a/apps/sim/lib/workflows/persistence/deployment-operations.test.ts b/apps/sim/lib/workflows/persistence/deployment-operations.test.ts index 757e54eb2e8..1a0ba58f210 100644 --- a/apps/sim/lib/workflows/persistence/deployment-operations.test.ts +++ b/apps/sim/lib/workflows/persistence/deployment-operations.test.ts @@ -21,6 +21,7 @@ vi.mock('@sim/utils/id', () => ({ import { activateDeploymentOperation, + getProtectedDeploymentVersionId, markDeploymentComponentReadiness, markDeploymentOperationFailed, prepareWorkflowDeployment, @@ -489,3 +490,30 @@ describe('deployment operation persistence', () => { expect(dbChainMockFns.update).not.toHaveBeenCalledWith(schemaMock.workflow) }) }) + +describe('getProtectedDeploymentVersionId', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + }) + + it('returns the version the in-flight current operation is preparing', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { deploymentVersionId: 'version-3', protocolVersion: 2, status: 'preparing' }, + ]) + + await expect(getProtectedDeploymentVersionId(WORKFLOW_ID)).resolves.toBe('version-3') + }) + + it('protects nothing once the latest operation is terminal', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { deploymentVersionId: 'version-3', protocolVersion: 2, status: 'active' }, + ]) + + await expect(getProtectedDeploymentVersionId(WORKFLOW_ID)).resolves.toBeNull() + }) + + it('protects nothing for a workflow without operations', async () => { + await expect(getProtectedDeploymentVersionId(WORKFLOW_ID)).resolves.toBeNull() + }) +}) diff --git a/apps/sim/lib/workflows/persistence/deployment-operations.ts b/apps/sim/lib/workflows/persistence/deployment-operations.ts index 9146f35b601..ed5b24ab244 100644 --- a/apps/sim/lib/workflows/persistence/deployment-operations.ts +++ b/apps/sim/lib/workflows/persistence/deployment-operations.ts @@ -70,6 +70,15 @@ export interface DeploymentOperationGeneration { generation: number } +/** + * Identifies the operation a fenced step belongs to, optionally narrowed to + * the version it targets and the statuses it may currently hold. + */ +export type DeploymentOperationFence = DeploymentOperationGeneration & { + deploymentVersionId?: string + statuses?: readonly DeploymentOperationStatus[] +} + export interface WorkflowDeploymentStatus { activeDeployment: { deploymentVersionId: string @@ -282,10 +291,7 @@ export async function getDeploymentOperation( * Confirms an operation still owns the workflow's latest generation. */ export async function isDeploymentOperationCurrent( - params: DeploymentOperationGeneration & { - deploymentVersionId?: string - statuses?: readonly DeploymentOperationStatus[] - }, + params: DeploymentOperationFence, executor: Pick = db ): Promise { const [latestOperation] = await executor @@ -322,6 +328,19 @@ export async function isDeploymentVersionProtectedByCurrentOperation( deploymentVersionId: string, executor: Pick = db ): Promise { + return (await getProtectedDeploymentVersionId(workflowId, executor)) === deploymentVersionId +} + +/** + * The deployment version the current operation is still preparing, or null + * once the latest operation is terminal. Cleanup must leave this version + * alone: it is inactive until cutover, yet its schedules and webhook + * candidates are live preparation state. + */ +export async function getProtectedDeploymentVersionId( + workflowId: string, + executor: Pick = db +): Promise { const [latestOperation] = await executor .select({ deploymentVersionId: workflowDeploymentOperation.deploymentVersionId, @@ -333,12 +352,15 @@ export async function isDeploymentVersionProtectedByCurrentOperation( .orderBy(desc(workflowDeploymentOperation.generation)) .limit(1) - return ( - latestOperation?.deploymentVersionId === deploymentVersionId && - latestOperation.protocolVersion === DEPLOYMENT_OPERATION_PROTOCOL_VERSION && - isDeploymentOperationStatus(latestOperation.status) && - IN_FLIGHT_STATUSES.includes(latestOperation.status) - ) + if ( + !latestOperation || + latestOperation.protocolVersion !== DEPLOYMENT_OPERATION_PROTOCOL_VERSION || + !isDeploymentOperationStatus(latestOperation.status) || + !IN_FLIGHT_STATUSES.includes(latestOperation.status) + ) { + return null + } + return latestOperation.deploymentVersionId } /** diff --git a/apps/sim/lib/workflows/schedules/deploy.test.ts b/apps/sim/lib/workflows/schedules/deploy.test.ts index afb5dc2abdc..62421cc8fdf 100644 --- a/apps/sim/lib/workflows/schedules/deploy.test.ts +++ b/apps/sim/lib/workflows/schedules/deploy.test.ts @@ -3,12 +3,21 @@ * * @vitest-environment node */ -import { dbChainMock, dbChainMockFns, resetDbChainMock, schemaMock } from '@sim/testing' +import { + dbChainMock, + dbChainMockFns, + flattenMockConditions, + resetDbChainMock, + schemaMock, +} from '@sim/testing' import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -const { mockRandomUUID } = vi.hoisted(() => ({ - mockRandomUUID: vi.fn(), -})) +const { mockRandomUUID, mockGetProtectedDeploymentVersionId, mockIsDeploymentOperationCurrent } = + vi.hoisted(() => ({ + mockRandomUUID: vi.fn(), + mockGetProtectedDeploymentVersionId: vi.fn(), + mockIsDeploymentOperationCurrent: vi.fn(), + })) vi.mock('@sim/db', () => ({ ...dbChainMock, ...schemaMock })) @@ -16,7 +25,17 @@ vi.mock('@/lib/webhooks/deploy', () => ({ cleanupWebhooksForWorkflow: vi.fn().mockResolvedValue(undefined), })) -import { createSchedulesForDeploy, deleteSchedulesForWorkflow } from './deploy' +vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ + getProtectedDeploymentVersionId: mockGetProtectedDeploymentVersionId, + isDeploymentOperationCurrent: mockIsDeploymentOperationCurrent, + setDeploymentTxTimeouts: vi.fn(), +})) + +import { + createSchedulesForDeploy, + deleteInactiveDeploymentSchedules, + deleteSchedulesForWorkflow, +} from './deploy' import type { BlockState } from './utils' import * as scheduleUtils from './utils' import { findScheduleBlocks, validateScheduleBlock, validateWorkflowSchedules } from './validation' @@ -795,3 +814,66 @@ describe('Schedule Deploy Utilities', () => { }) }) }) + +describe('deleteInactiveDeploymentSchedules', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + mockIsDeploymentOperationCurrent.mockResolvedValue(true) + mockGetProtectedDeploymentVersionId.mockResolvedValue(null) + }) + + it('deletes every schedule owned by an inactive version in one statement', async () => { + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1' }, { id: 'schedule-2' }]) + + await expect(deleteInactiveDeploymentSchedules({ workflowId: 'workflow-1' })).resolves.toEqual({ + status: 'deleted', + count: 2, + }) + + expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1) + const conditions = flattenMockConditions(dbChainMockFns.where.mock.calls.at(-1)?.[0]) + expect(conditions).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'inArray', + column: schemaMock.workflowSchedule.deploymentVersionId, + }), + expect.objectContaining({ type: 'isNull', column: schemaMock.workflowSchedule.archivedAt }), + ]) + ) + expect(conditions).not.toEqual( + expect.arrayContaining([expect.objectContaining({ type: 'ne' })]) + ) + }) + + it('shields the version an in-flight operation is preparing', async () => { + mockGetProtectedDeploymentVersionId.mockResolvedValue('version-3') + + await deleteInactiveDeploymentSchedules({ workflowId: 'workflow-1' }) + + const conditions = flattenMockConditions(dbChainMockFns.where.mock.calls.at(-1)?.[0]) + expect(conditions).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'ne', + left: schemaMock.workflowSchedule.deploymentVersionId, + right: 'version-3', + }), + ]) + ) + }) + + it('deletes nothing once a newer operation owns the workflow', async () => { + mockIsDeploymentOperationCurrent.mockResolvedValue(false) + + await expect( + deleteInactiveDeploymentSchedules({ + workflowId: 'workflow-1', + operationFence: { workflowId: 'workflow-1', operationId: 'operation-1', generation: 2 }, + }) + ).resolves.toEqual({ status: 'superseded' }) + + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/workflows/schedules/deploy.ts b/apps/sim/lib/workflows/schedules/deploy.ts index 1f2a1bfd662..475212ffd23 100644 --- a/apps/sim/lib/workflows/schedules/deploy.ts +++ b/apps/sim/lib/workflows/schedules/deploy.ts @@ -1,9 +1,15 @@ -import { db, workflowSchedule } from '@sim/db' +import { db, workflow, workflowDeploymentVersion, workflowSchedule } from '@sim/db' import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' -import { and, eq, inArray, isNull } from 'drizzle-orm' +import { and, eq, inArray, isNull, ne } from 'drizzle-orm' import type { DbOrTx } from '@/lib/db/types' +import { + type DeploymentOperationFence, + getProtectedDeploymentVersionId, + isDeploymentOperationCurrent, + setDeploymentTxTimeouts, +} from '@/lib/workflows/persistence/deployment-operations' import type { BlockState } from '@/lib/workflows/schedules/utils' import { findScheduleBlocks, validateScheduleBlock } from '@/lib/workflows/schedules/validation' @@ -203,3 +209,68 @@ export async function deleteSchedulesForWorkflow( : `Deleted all schedules for workflow ${workflowId}` ) } + +export type InactiveDeploymentScheduleCleanupResult = + | { status: 'deleted'; count: number } + | { status: 'superseded' } + +/** + * Deletes every schedule still owned by an inactive deployment version of the + * workflow in one fenced statement. Keyed by schedule rows rather than by + * versions, so the cost follows what is stale instead of how many times the + * workflow has been deployed. The version an in-flight operation is preparing + * is left alone: it is inactive until cutover, but its schedules are live + * preparation state. The workflow row lock serializes this with activation so + * `isActive` cannot flip underneath the delete. + */ +export async function deleteInactiveDeploymentSchedules(params: { + workflowId: string + /** When set, nothing is deleted once a newer operation has taken over the workflow. */ + operationFence?: DeploymentOperationFence +}): Promise { + return db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) + await tx + .select({ id: workflow.id }) + .from(workflow) + .where(eq(workflow.id, params.workflowId)) + .for('update') + if (params.operationFence && !(await isDeploymentOperationCurrent(params.operationFence, tx))) { + return { status: 'superseded' } + } + + const protectedDeploymentVersionId = await getProtectedDeploymentVersionId( + params.workflowId, + tx + ) + const inactiveVersionIds = tx + .select({ id: workflowDeploymentVersion.id }) + .from(workflowDeploymentVersion) + .where( + and( + eq(workflowDeploymentVersion.workflowId, params.workflowId), + eq(workflowDeploymentVersion.isActive, false) + ) + ) + const deleted = await tx + .delete(workflowSchedule) + .where( + and( + eq(workflowSchedule.workflowId, params.workflowId), + isNull(workflowSchedule.archivedAt), + inArray(workflowSchedule.deploymentVersionId, inactiveVersionIds), + protectedDeploymentVersionId + ? ne(workflowSchedule.deploymentVersionId, protectedDeploymentVersionId) + : undefined + ) + ) + .returning({ id: workflowSchedule.id }) + + if (deleted.length > 0) { + logger.info( + `Deleted ${deleted.length} schedule(s) owned by inactive deployments of workflow ${params.workflowId}` + ) + } + return { status: 'deleted', count: deleted.length } + }) +} diff --git a/apps/sim/lib/workflows/schedules/index.ts b/apps/sim/lib/workflows/schedules/index.ts index 67dd11fd2c8..026d35ecd8d 100644 --- a/apps/sim/lib/workflows/schedules/index.ts +++ b/apps/sim/lib/workflows/schedules/index.ts @@ -1,5 +1,7 @@ export { createSchedulesForDeploy, + deleteInactiveDeploymentSchedules, deleteSchedulesForWorkflow, + type InactiveDeploymentScheduleCleanupResult, } from './deploy' export { validateWorkflowSchedules } from './validation' From 225466f88050d7757695ab47acaf90b15ad40c69 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Wed, 2 Sep 2026 12:23:15 -0700 Subject: [PATCH 2/2] fix(deployments): fence webhook teardown on version inactivity and abort in undeploy cleanup --- apps/sim/lib/webhooks/deploy.test.ts | 16 +++++++++- apps/sim/lib/webhooks/deploy.ts | 14 ++++++--- .../lib/workflows/deployment-outbox.test.ts | 29 +++++++++++++++++++ apps/sim/lib/workflows/deployment-outbox.ts | 27 +++++------------ .../persistence/deployment-operations.ts | 27 +++++++++++++++++ 5 files changed, 89 insertions(+), 24 deletions(-) diff --git a/apps/sim/lib/webhooks/deploy.test.ts b/apps/sim/lib/webhooks/deploy.test.ts index 1423f20ddce..77ce7cc1642 100644 --- a/apps/sim/lib/webhooks/deploy.test.ts +++ b/apps/sim/lib/webhooks/deploy.test.ts @@ -35,10 +35,12 @@ vi.mock('@/lib/webhooks/utils.server', () => ({ vi.mock('@/lib/webhooks/pending-verification', () => ({ PendingWebhookVerificationTracker: vi.fn(), })) -const { mockIsDeploymentVersionProtected } = vi.hoisted(() => ({ +const { mockIsDeploymentVersionActive, mockIsDeploymentVersionProtected } = vi.hoisted(() => ({ + mockIsDeploymentVersionActive: vi.fn(), mockIsDeploymentVersionProtected: vi.fn(), })) vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ + isDeploymentVersionActive: mockIsDeploymentVersionActive, isDeploymentVersionProtectedByCurrentOperation: mockIsDeploymentVersionProtected, })) @@ -677,6 +679,7 @@ describe('cleanupInactiveDeploymentWebhooks', () => { } beforeEach(() => { + mockIsDeploymentVersionActive.mockResolvedValue(false) mockIsDeploymentVersionProtected.mockResolvedValue(false) }) @@ -734,6 +737,17 @@ describe('cleanupInactiveDeploymentWebhooks', () => { expect(dbChainMockFns.delete).not.toHaveBeenCalled() }) + it('leaves a row alone when its version was re-activated after the batch was selected', async () => { + queueTableRows(webhook, [staleWebhookRow('wh-1')]) + mockIsDeploymentVersionActive.mockResolvedValue(true) + + await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: true }) + + expect(mockIsDeploymentVersionActive).toHaveBeenCalledWith('workflow-1', 'version-1') + 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) diff --git a/apps/sim/lib/webhooks/deploy.ts b/apps/sim/lib/webhooks/deploy.ts index 141f03b6e8c..7543e0a0595 100644 --- a/apps/sim/lib/webhooks/deploy.ts +++ b/apps/sim/lib/webhooks/deploy.ts @@ -33,7 +33,10 @@ import { replaceSlackStreamAuthoringConfig, } from '@/lib/webhooks/slack-stream-config' import { findConflictingWebhookPathOwner } from '@/lib/webhooks/utils.server' -import { isDeploymentVersionProtectedByCurrentOperation } from '@/lib/workflows/persistence/deployment-operations' +import { + isDeploymentVersionActive, + isDeploymentVersionProtectedByCurrentOperation, +} from '@/lib/workflows/persistence/deployment-operations' import { buildCanonicalIndex, buildSubBlockValues, @@ -1373,9 +1376,11 @@ export interface InactiveDeploymentWebhookCleanupResult { * per row, so the work is bounded here and `hasMore` asks the caller to come * back; every finished row leaves the remaining set smaller, so repeated calls * converge. `protectedDeploymentVersionId` is the version an in-flight - * operation is preparing, inactive until cutover but live preparation state; - * each row is re-checked against the current operation right before its - * delete because that can change while the batch runs. + * operation is preparing, inactive until cutover but live preparation state. + * Each row is re-checked right before its provider call: the version must + * still be inactive and must not have become the current operation's + * candidate, since either can change while the batch runs and the fenced row + * delete that follows cannot undo provider teardown. */ export async function cleanupInactiveDeploymentWebhooks(params: { workflowId: string @@ -1433,6 +1438,7 @@ export async function cleanupInactiveDeploymentWebhooks(params: { shouldDeleteWebhook: async () => { if (shouldContinue && !(await shouldContinue())) return false if (!deploymentVersionId) return true + if (await isDeploymentVersionActive(workflowId, deploymentVersionId)) return false return !(await isDeploymentVersionProtectedByCurrentOperation( workflowId, deploymentVersionId diff --git a/apps/sim/lib/workflows/deployment-outbox.test.ts b/apps/sim/lib/workflows/deployment-outbox.test.ts index 92d4089ecb5..9cfa87dbce9 100644 --- a/apps/sim/lib/workflows/deployment-outbox.test.ts +++ b/apps/sim/lib/workflows/deployment-outbox.test.ts @@ -33,6 +33,7 @@ const { mockCleanupInactiveDeploymentWebhooks, mockDeleteInactiveDeploymentSchedules, mockGetProtectedDeploymentVersionId, + mockIsDeploymentVersionActive, mockTx, } = vi.hoisted(() => ({ mockPrepareWebhooks: vi.fn(), @@ -57,6 +58,7 @@ const { mockCleanupInactiveDeploymentWebhooks: vi.fn(), mockDeleteInactiveDeploymentSchedules: vi.fn(), mockGetProtectedDeploymentVersionId: vi.fn(), + mockIsDeploymentVersionActive: vi.fn(), mockTx: { select: vi.fn(), update: vi.fn(), execute: vi.fn() }, })) @@ -116,6 +118,7 @@ vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ getDeploymentOperation: mockGetDeploymentOperation, getProtectedDeploymentVersionId: mockGetProtectedDeploymentVersionId, isDeploymentOperationCurrent: mockIsDeploymentOperationCurrent, + isDeploymentVersionActive: mockIsDeploymentVersionActive, isDeploymentVersionProtectedByCurrentOperation: mockIsDeploymentVersionProtectedByCurrentOperation, markDeploymentComponentReadiness: mockMarkDeploymentComponentReadiness, @@ -233,6 +236,7 @@ describe('versioned deployment preparation outbox', () => { }) mockIsDeploymentOperationCurrent.mockResolvedValue(false) mockIsDeploymentVersionProtectedByCurrentOperation.mockResolvedValue(false) + mockIsDeploymentVersionActive.mockResolvedValue(false) mockGetProtectedDeploymentVersionId.mockResolvedValue(null) mockDeleteInactiveDeploymentSchedules.mockResolvedValue({ status: 'deleted', count: 0 }) mockCleanupInactiveDeploymentWebhooks.mockResolvedValue({ hasMore: false }) @@ -773,4 +777,29 @@ describe('versioned deployment preparation outbox', () => { expect(mockNotifyMcpToolServers).not.toHaveBeenCalled() }) + + it('lets a timed-out undeploy stop between null-version webhooks', async () => { + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + queueTableRows(schemaMock.workflow, [{ isDeployed: false }]) + const controller = new AbortController() + const cleanupHandler = + createWorkflowDeploymentOutboxHandlers()[ + WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_UNDEPLOYED_SIDE_EFFECTS + ] + + await expect( + cleanupHandler({ workflowId: 'workflow-1', userId: 'user-1' }, context(controller)) + ).resolves.toBeUndefined() + + expect(mockCleanupWebhooksForWorkflow).toHaveBeenCalledTimes(1) + const shouldDeleteWebhook = mockCleanupWebhooksForWorkflow.mock + .calls[0][6] as () => Promise + queueTableRows(schemaMock.workflow, [{ isDeployed: false }]) + await expect(shouldDeleteWebhook()).resolves.toBe(true) + + controller.abort() + await expect(shouldDeleteWebhook()).rejects.toMatchObject({ name: 'AbortError' }) + }) }) diff --git a/apps/sim/lib/workflows/deployment-outbox.ts b/apps/sim/lib/workflows/deployment-outbox.ts index 17aec8b99d5..f3462161392 100644 --- a/apps/sim/lib/workflows/deployment-outbox.ts +++ b/apps/sim/lib/workflows/deployment-outbox.ts @@ -49,6 +49,7 @@ import { getDeploymentOperation, getProtectedDeploymentVersionId, isDeploymentOperationCurrent, + isDeploymentVersionActive, isDeploymentVersionProtectedByCurrentOperation, markDeploymentComponentReadiness, markDeploymentOperationFailed, @@ -977,6 +978,7 @@ const cleanupUndeployedSideEffects: OutboxHandler = async (rawPayload, context) workflowId: payload.workflowId, workflow: workflowData, requestId, + signal: context.signal, }) context.signal.throwIfAborted() @@ -1193,25 +1195,6 @@ async function cleanupStaleDeploymentIfNeeded(params: { return false } -async function isDeploymentVersionActive( - workflowId: string, - deploymentVersionId: string -): Promise { - const [versionRow] = await db - .select({ id: workflowDeploymentVersion.id }) - .from(workflowDeploymentVersion) - .where( - and( - eq(workflowDeploymentVersion.workflowId, workflowId), - eq(workflowDeploymentVersion.id, deploymentVersionId), - eq(workflowDeploymentVersion.isActive, true) - ) - ) - .limit(1) - - return Boolean(versionRow) -} - async function removeMcpToolsIfStillUndeployed( workflowId: string, requestId: string @@ -1232,12 +1215,18 @@ async function removeMcpToolsIfStillUndeployed( notifyMcpToolServers(tools) } +/** + * The per-row gate also throws once the outbox lease aborts, so a timed-out + * undeploy stops between webhooks instead of overlapping its reaped retry. + */ async function cleanupNullVersionWebhooksIfStillUndeployed(params: { workflowId: string workflow: Record requestId: string + signal: AbortSignal }): Promise { const isStillUndeployed = async () => { + params.signal.throwIfAborted() const [workflowRecord] = await db .select({ isDeployed: workflowTable.isDeployed }) .from(workflowTable) diff --git a/apps/sim/lib/workflows/persistence/deployment-operations.ts b/apps/sim/lib/workflows/persistence/deployment-operations.ts index ed5b24ab244..3d5a95517a6 100644 --- a/apps/sim/lib/workflows/persistence/deployment-operations.ts +++ b/apps/sim/lib/workflows/persistence/deployment-operations.ts @@ -363,6 +363,33 @@ export async function getProtectedDeploymentVersionId( return latestOperation.deploymentVersionId } +/** + * True when the given deployment version is the workflow's active one. + * Cleanup re-checks this immediately before any provider teardown because a + * version can be re-activated between a batch being selected and its rows + * being processed, and the fenced row delete that follows cannot undo a + * provider call. + */ +export async function isDeploymentVersionActive( + workflowId: string, + deploymentVersionId: string, + executor: Pick = db +): Promise { + const [versionRow] = await executor + .select({ id: workflowDeploymentVersion.id }) + .from(workflowDeploymentVersion) + .where( + and( + eq(workflowDeploymentVersion.workflowId, workflowId), + eq(workflowDeploymentVersion.id, deploymentVersionId), + eq(workflowDeploymentVersion.isActive, true) + ) + ) + .limit(1) + + return Boolean(versionRow) +} + /** * Moves the current preparing generation into its activation phase. */