From c9e990924c77c2d1985c096ed706e701d9e39c63 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 21 Jul 2026 23:17:50 -0700 Subject: [PATCH] improvement(deployments): shorten lock windows and add tx safety timeouts on the deploy path --- .../cron/cleanup-stale-executions/route.ts | 75 ++++++- .../lib/webhooks/registration-store.test.ts | 34 +-- apps/sim/lib/webhooks/registration-store.ts | 198 +++++++++++------- .../lib/workflows/deployment-outbox.test.ts | 1 + apps/sim/lib/workflows/deployment-outbox.ts | 4 + .../persistence/deployment-operations.test.ts | 10 +- .../persistence/deployment-operations.ts | 53 ++++- 7 files changed, 273 insertions(+), 102 deletions(-) diff --git a/apps/sim/app/api/cron/cleanup-stale-executions/route.ts b/apps/sim/app/api/cron/cleanup-stale-executions/route.ts index 47505d2a94c..e58d9a037d4 100644 --- a/apps/sim/app/api/cron/cleanup-stale-executions/route.ts +++ b/apps/sim/app/api/cron/cleanup-stale-executions/route.ts @@ -1,8 +1,9 @@ import { asyncJobs, db } from '@sim/db' -import { tableJobs, workflowExecutionLogs } from '@sim/db/schema' +import { tableJobs, workflowDeploymentOperation, workflowExecutionLogs } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' -import { and, eq, inArray, lt, sql } from 'drizzle-orm' +import { and, eq, exists, gt, inArray, lt, sql } from 'drizzle-orm' +import { alias } from 'drizzle-orm/pg-core' import { type NextRequest, NextResponse } from 'next/server' import { verifyCronAuth } from '@/lib/auth/internal' import { JOB_RETENTION_HOURS, JOB_STATUS } from '@/lib/core/async-jobs' @@ -17,6 +18,14 @@ const STALE_THRESHOLD_MINUTES = Math.ceil(STALE_THRESHOLD_MS / 60000) const MAX_INT32 = 2_147_483_647 /** Terminal table-jobs older than this are pruned; only the latest job per table is ever read. */ const TABLE_JOB_RETENTION_HOURS = 24 +/** + * Terminal deployment operations older than this are pruned. Every reader of + * this table is latest-generation-only, and idempotency keys only need to + * survive a client retry window, so 30 days is generous. + */ +const DEPLOYMENT_OPERATION_RETENTION_DAYS = 30 +const DEPLOYMENT_OPERATION_PRUNE_BATCH_SIZE = 2000 +const DEPLOYMENT_OPERATION_PRUNE_MAX_BATCHES = 10 export const GET = withRouteHandler(async (request: NextRequest) => { try { @@ -221,6 +230,64 @@ export const GET = withRouteHandler(async (request: NextRequest) => { }) } + /** + * Prune terminal deployment operations past retention. HARD INVARIANT: + * the newest-generation row per workflow must always survive — the next + * deploy computes `generation = MAX(generation) + 1`, and the webhook + * registration store fences rows with lt/gt comparisons against stored + * generations, so generation reuse after a full wipe would permanently + * wedge that workflow's deployments. The `exists(newer)` predicate + * guarantees the max-generation row is never eligible; the status filter + * keeps in-flight rows (an outbox worker may still hold their fence). + */ + let deploymentOperationsPruned = 0 + try { + const deploymentOpRetention = new Date( + Date.now() - DEPLOYMENT_OPERATION_RETENTION_DAYS * 24 * 60 * 60 * 1000 + ) + const newerOperation = alias(workflowDeploymentOperation, 'newer_operation') + for (let batch = 0; batch < DEPLOYMENT_OPERATION_PRUNE_MAX_BATCHES; batch++) { + const prunable = db + .select({ id: workflowDeploymentOperation.id }) + .from(workflowDeploymentOperation) + .where( + and( + inArray(workflowDeploymentOperation.status, ['active', 'failed', 'superseded']), + lt(workflowDeploymentOperation.completedAt, deploymentOpRetention), + exists( + db + .select({ id: newerOperation.id }) + .from(newerOperation) + .where( + and( + eq(newerOperation.workflowId, workflowDeploymentOperation.workflowId), + gt(newerOperation.generation, workflowDeploymentOperation.generation) + ) + ) + ) + ) + ) + .limit(DEPLOYMENT_OPERATION_PRUNE_BATCH_SIZE) + + const deleted = await db + .delete(workflowDeploymentOperation) + .where(inArray(workflowDeploymentOperation.id, prunable)) + .returning({ id: workflowDeploymentOperation.id }) + + deploymentOperationsPruned += deleted.length + if (deleted.length < DEPLOYMENT_OPERATION_PRUNE_BATCH_SIZE) break + } + if (deploymentOperationsPruned > 0) { + logger.info( + `Pruned ${deploymentOperationsPruned} old deployment operations (retention: ${DEPLOYMENT_OPERATION_RETENTION_DAYS}d)` + ) + } + } catch (error) { + logger.error('Failed to prune old deployment operations:', { + error: toError(error).message, + }) + } + return NextResponse.json({ success: true, executions: { @@ -239,6 +306,10 @@ export const GET = withRouteHandler(async (request: NextRequest) => { tableJobs: { staleMarkedFailed: staleTableJobsMarkedFailed, }, + deploymentOperations: { + pruned: deploymentOperationsPruned, + retentionDays: DEPLOYMENT_OPERATION_RETENTION_DAYS, + }, }) } catch (error) { logger.error('Error in stale execution cleanup job:', error) diff --git a/apps/sim/lib/webhooks/registration-store.test.ts b/apps/sim/lib/webhooks/registration-store.test.ts index 204580b899b..bcb54a43e79 100644 --- a/apps/sim/lib/webhooks/registration-store.test.ts +++ b/apps/sim/lib/webhooks/registration-store.test.ts @@ -25,11 +25,18 @@ vi.mock('@sim/db', () => ({ vi.mock('drizzle-orm', () => ({ and: (...conditions: Condition[]) => ({ kind: 'and', conditions }), eq: (column: unknown, value: unknown) => ({ kind: 'eq', column, value }), + exists: (subquery: unknown) => ({ kind: 'exists', subquery }), gt: (column: unknown, value: unknown) => ({ kind: 'gt', column, value }), inArray: (column: unknown, value: unknown) => ({ kind: 'inArray', column, value }), isNull: (column: unknown) => ({ kind: 'isNull', column }), lt: (column: unknown, value: unknown) => ({ kind: 'lt', column, value }), lte: (column: unknown, value: unknown) => ({ kind: 'lte', column, value }), + notExists: (subquery: unknown) => ({ kind: 'notExists', subquery }), + sql: (strings: TemplateStringsArray, ...values: unknown[]) => ({ + kind: 'sql', + strings: [...strings], + values, + }), })) vi.mock('@/lib/webhooks/provider-subscriptions', () => ({ @@ -42,6 +49,7 @@ vi.mock('@/lib/webhooks/path-claims', () => ({ vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ isDeploymentOperationCurrent: mockIsDeploymentOperationCurrent, + setDeploymentTxTimeouts: vi.fn(), })) import type { DbOrTx } from '@sim/workflow-persistence/types' @@ -182,22 +190,22 @@ describe('activateWebhookRegistrations', () => { await activateWebhookRegistrations(tx, FENCE) - expect(updates).toHaveLength(3) - expect(updates[0].payload).toEqual( - expect.objectContaining({ registrationStatus: 'retired', isActive: false }) + expect(updates).toHaveLength(2) + /** + * Retire + repoint fold into one generation-conditional statement over + * the active rows: every mutated column is a CASE keyed on the fence + * generation, and the WHERE covers both phases via lte. + */ + expect(updates[0].payload.registrationStatus).toEqual(expect.objectContaining({ kind: 'sql' })) + expect(updates[0].payload.deploymentVersionId).toEqual( + expect.objectContaining({ kind: 'sql', values: expect.arrayContaining(['version-3']) }) ) - expect(updates[0].payload.archivedAt).toBeInstanceOf(Date) - expect(JSON.stringify(updates[0].condition)).toContain('"lt"') + expect(updates[0].payload.isActive).toEqual(expect.objectContaining({ kind: 'sql' })) + expect(updates[0].payload.archivedAt).toEqual(expect.objectContaining({ kind: 'sql' })) + expect(updates[0].payload.updatedAt).toBeInstanceOf(Date) + expect(JSON.stringify(updates[0].condition)).toContain('"lte"') expect(updates[1].payload).toEqual( - expect.objectContaining({ - deploymentVersionId: 'version-3', - isActive: true, - archivedAt: null, - }) - ) - - expect(updates[2].payload).toEqual( expect.objectContaining({ registrationStatus: 'active', deploymentVersionId: 'version-3', diff --git a/apps/sim/lib/webhooks/registration-store.ts b/apps/sim/lib/webhooks/registration-store.ts index 2754a89de94..eeeb1e350f6 100644 --- a/apps/sim/lib/webhooks/registration-store.ts +++ b/apps/sim/lib/webhooks/registration-store.ts @@ -1,9 +1,14 @@ import { db } from '@sim/db' -import { webhook, workflow, workflowDeploymentVersion } from '@sim/db/schema' +import { + webhook, + workflow, + workflowDeploymentOperation, + workflowDeploymentVersion, +} from '@sim/db/schema' import { generateShortId } from '@sim/utils/id' import { isPlainRecord } from '@sim/utils/object' import type { DbOrTx } from '@sim/workflow-persistence/types' -import { and, eq, gt, inArray, isNull, lt, lte } from 'drizzle-orm' +import { and, eq, exists, gt, inArray, isNull, lt, lte, notExists, sql } from 'drizzle-orm' import { claimWebhookPath } from '@/lib/webhooks/path-claims' import { projectDesiredWebhookProviderConfig } from '@/lib/webhooks/provider-subscriptions' import { @@ -12,7 +17,10 @@ import { } from '@/lib/webhooks/registration-identity' import { planWebhookRegistrationReconciliation } from '@/lib/webhooks/registration-reconciliation' import type { DeploymentOperationStatus } from '@/lib/workflows/deployment-lifecycle' -import { isDeploymentOperationCurrent } from '@/lib/workflows/persistence/deployment-operations' +import { + isDeploymentOperationCurrent, + setDeploymentTxTimeouts, +} from '@/lib/workflows/persistence/deployment-operations' export type WebhookRegistrationRow = typeof webhook.$inferSelect export type WebhookRegistrationStatus = 'active' | 'candidate' | 'retired' | 'orphaned' @@ -216,6 +224,7 @@ export async function prepareWebhookRegistrationIntents(input: { desired: readonly DesiredWebhookRegistrationIntent[] }): Promise { return db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) await assertCurrentOperation(tx, input.fence, ['preparing']) await adoptLegacyActiveRows(tx, input.fence) @@ -398,35 +407,82 @@ export async function prepareWebhookRegistrationIntents(input: { }) } -/** Checkpoints provider-managed candidate state under the operation and generation fences. */ +/** + * Predicate asserting the fence's operation is still the workflow's current + * preparing attempt, evaluated inside the same statement that carries it. + * Mirrors `isDeploymentOperationCurrent` (exact operation identity + no newer + * generation) without a separate round trip or the workflow-row lock. + */ +function currentPreparingOperationPredicate(fence: WebhookRegistrationOperationFence) { + return and( + exists( + db + .select({ id: workflowDeploymentOperation.id }) + .from(workflowDeploymentOperation) + .where( + and( + eq(workflowDeploymentOperation.id, fence.operationId), + eq(workflowDeploymentOperation.workflowId, fence.workflowId), + eq(workflowDeploymentOperation.generation, fence.generation), + eq(workflowDeploymentOperation.deploymentVersionId, fence.deploymentVersionId), + eq(workflowDeploymentOperation.status, 'preparing') + ) + ) + ), + notExists( + db + .select({ id: workflowDeploymentOperation.id }) + .from(workflowDeploymentOperation) + .where( + and( + eq(workflowDeploymentOperation.workflowId, fence.workflowId), + gt(workflowDeploymentOperation.generation, fence.generation) + ) + ) + ) + ) +} + +/** + * Checkpoints provider-managed candidate state under the operation and + * generation fences. + * + * Runs as ONE fenced UPDATE — no transaction and no workflow-row lock. This + * is the hottest write on the deployment path (it follows every external + * provider call), so it must not hold locks across client round trips. The + * row fence (`candidate` + exact generation) is the authoritative guard: any + * newer preparation either adopts the row (bumping its generation) or orphans + * it (flipping its status) before touching provider state, so a stale + * checkpoint can never match. The operation-currency predicate rejects writes + * from superseded attempts up front. + */ export async function checkpointWebhookCandidate(input: { fence: WebhookRegistrationOperationFence webhookId: string providerConfig: Record prepared?: boolean }): Promise { - return db.transaction(async (tx) => { - await assertCurrentOperation(tx, input.fence, ['preparing']) - const now = new Date() - const [updated] = await tx - .update(webhook) - .set({ - providerConfig: input.providerConfig, - ...(input.prepared === false ? {} : { preparedAt: now }), - updatedAt: now, - }) - .where( - and( - eq(webhook.id, input.webhookId), - eq(webhook.workflowId, input.fence.workflowId), - eq(webhook.registrationStatus, 'candidate'), - eq(webhook.registrationGeneration, input.fence.generation) - ) + assertOperationGeneration(input.fence.generation) + const now = new Date() + const [updated] = await db + .update(webhook) + .set({ + providerConfig: input.providerConfig, + ...(input.prepared === false ? {} : { preparedAt: now }), + updatedAt: now, + }) + .where( + and( + eq(webhook.id, input.webhookId), + eq(webhook.workflowId, input.fence.workflowId), + eq(webhook.registrationStatus, 'candidate'), + eq(webhook.registrationGeneration, input.fence.generation), + currentPreparingOperationPredicate(input.fence) ) - .returning() - if (!updated) throw new StaleWebhookRegistrationOperationError() - return updated - }) + ) + .returning() + if (!updated) throw new StaleWebhookRegistrationOperationError() + return updated } /** @@ -469,36 +525,32 @@ export async function activateWebhookRegistrations( .limit(1) if (newerRows.length > 0) throw new StaleWebhookRegistrationOperationError() + /** + * Retire (generation < fence) and repoint (generation = fence) touch + * disjoint subsets of the active rows, so they fold into one statement. + * Promotion of candidates stays a SEPARATE statement below: promoting a + * candidate inserts a live entry into the non-deferrable partial unique + * index `webhook_active_registration_unique` for the same (workflowId, + * blockId) the retired row is vacating, and a single statement offers no + * ordering guarantee between those two row versions — retire must be fully + * applied first. + */ const now = new Date() + const isCurrentGeneration = sql`${webhook.registrationGeneration} = ${fence.generation}` await tx .update(webhook) .set({ - registrationStatus: 'retired', - isActive: false, - archivedAt: now, + registrationStatus: sql`CASE WHEN ${isCurrentGeneration} THEN ${webhook.registrationStatus} ELSE 'retired' END`, + deploymentVersionId: sql`CASE WHEN ${isCurrentGeneration} THEN ${fence.deploymentVersionId}::text ELSE ${webhook.deploymentVersionId} END`, + isActive: sql`${isCurrentGeneration}`, + archivedAt: sql`CASE WHEN ${isCurrentGeneration} THEN NULL ELSE ${now.toISOString()}::timestamp END`, updatedAt: now, }) .where( and( eq(webhook.workflowId, fence.workflowId), eq(webhook.registrationStatus, 'active'), - lt(webhook.registrationGeneration, fence.generation) - ) - ) - - await tx - .update(webhook) - .set({ - deploymentVersionId: fence.deploymentVersionId, - isActive: true, - archivedAt: null, - updatedAt: now, - }) - .where( - and( - eq(webhook.workflowId, fence.workflowId), - eq(webhook.registrationStatus, 'active'), - eq(webhook.registrationGeneration, fence.generation) + lte(webhook.registrationGeneration, fence.generation) ) ) @@ -525,6 +577,7 @@ export async function listRetiredWebhookRegistrationsForCleanup( input: WebhookRegistrationOperationFence & { limit?: number } ): Promise { return db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) await assertCurrentOperation(tx, input, ['active']) return tx .select() @@ -542,7 +595,10 @@ export async function listRetiredWebhookRegistrationsForCleanup( } /** - * Reloads an exact cleanup snapshot. A row advanced or reused by a newer generation is rejected. + * Reloads an exact cleanup snapshot. A row advanced or reused by a newer + * generation is rejected. A plain fenced SELECT suffices: cleanup actions run + * after this returns (so a row lock could not protect them anyway), and the + * delete that follows is independently fenced on the same predicates. */ export async function getWebhookCleanupSnapshotIfCurrent(input: { workflowId: string @@ -550,21 +606,19 @@ export async function getWebhookCleanupSnapshotIfCurrent(input: { expectedGeneration: number statuses: readonly WebhookRegistrationStatus[] }): Promise { - return db.transaction(async (tx) => { - const [row] = await tx - .select() - .from(webhook) - .where( - and( - eq(webhook.workflowId, input.workflowId), - eq(webhook.id, input.webhookId), - eq(webhook.registrationGeneration, input.expectedGeneration), - inArray(webhook.registrationStatus, input.statuses) - ) + const [row] = await db + .select() + .from(webhook) + .where( + and( + eq(webhook.workflowId, input.workflowId), + eq(webhook.id, input.webhookId), + eq(webhook.registrationGeneration, input.expectedGeneration), + inArray(webhook.registrationStatus, input.statuses) ) - .for('update') - return row ?? null - }) + ) + .limit(1) + return row ?? null } /** Deletes an externally cleaned row while preserving sticky path ownership. */ @@ -574,18 +628,16 @@ export async function deleteWebhookRegistrationAfterCleanup(input: { expectedGeneration: number statuses: readonly WebhookRegistrationStatus[] }): Promise { - return db.transaction(async (tx) => { - const [deleted] = await tx - .delete(webhook) - .where( - and( - eq(webhook.workflowId, input.workflowId), - eq(webhook.id, input.webhookId), - eq(webhook.registrationGeneration, input.expectedGeneration), - inArray(webhook.registrationStatus, input.statuses) - ) + const [deleted] = await db + .delete(webhook) + .where( + and( + eq(webhook.workflowId, input.workflowId), + eq(webhook.id, input.webhookId), + eq(webhook.registrationGeneration, input.expectedGeneration), + inArray(webhook.registrationStatus, input.statuses) ) - .returning({ id: webhook.id }) - return Boolean(deleted) - }) + ) + .returning({ id: webhook.id }) + return Boolean(deleted) } diff --git a/apps/sim/lib/workflows/deployment-outbox.test.ts b/apps/sim/lib/workflows/deployment-outbox.test.ts index 29bbc7a6a3d..72881739136 100644 --- a/apps/sim/lib/workflows/deployment-outbox.test.ts +++ b/apps/sim/lib/workflows/deployment-outbox.test.ts @@ -150,6 +150,7 @@ vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ markDeploymentComponentReadiness: mockMarkDeploymentComponentReadiness, markDeploymentOperationFailed: mockMarkDeploymentOperationFailed, recordDeploymentOperationRetry: mockRecordDeploymentOperationRetry, + setDeploymentTxTimeouts: vi.fn(), })) vi.mock('@/lib/workflows/schedules', () => ({ diff --git a/apps/sim/lib/workflows/deployment-outbox.ts b/apps/sim/lib/workflows/deployment-outbox.ts index 59dbf525f88..ed58b1d040a 100644 --- a/apps/sim/lib/workflows/deployment-outbox.ts +++ b/apps/sim/lib/workflows/deployment-outbox.ts @@ -48,6 +48,7 @@ import { markDeploymentComponentReadiness, markDeploymentOperationFailed, recordDeploymentOperationRetry, + setDeploymentTxTimeouts, type WorkflowDeploymentOperation, } from '@/lib/workflows/persistence/deployment-operations' import { createSchedulesForDeploy, deleteSchedulesForWorkflow } from '@/lib/workflows/schedules' @@ -1022,6 +1023,7 @@ async function deleteSchedulesForDeploymentIfInactive(params: { operationFence?: DeploymentCleanupOperationFence }): Promise { return db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) await tx .select({ id: workflowTable.id }) .from(workflowTable) @@ -1202,6 +1204,7 @@ async function createSchedulesIfStillActive(params: { blocks: Record }) { return db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) const [workflowRecord] = await tx .select({ id: workflowTable.id }) .from(workflowTable) @@ -1250,6 +1253,7 @@ async function pruneWorkflowGroupOutputsIfStillActive(params: { requestId: string }): Promise { await db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) const [workflowRecord] = await tx .select({ id: workflowTable.id }) .from(workflowTable) diff --git a/apps/sim/lib/workflows/persistence/deployment-operations.test.ts b/apps/sim/lib/workflows/persistence/deployment-operations.test.ts index 5692b0e8a4b..757e54eb2e8 100644 --- a/apps/sim/lib/workflows/persistence/deployment-operations.test.ts +++ b/apps/sim/lib/workflows/persistence/deployment-operations.test.ts @@ -433,19 +433,19 @@ describe('deployment operation persistence', () => { expect(result.success).toBe(true) expect(dbChainMockFns.update.mock.calls.map(([table]) => table)).toEqual([ - schemaMock.workflowDeploymentVersion, schemaMock.workflowDeploymentVersion, schemaMock.workflow, schemaMock.workflowDeploymentOperation, ]) - expect(dbChainMockFns.set).toHaveBeenNthCalledWith(1, { isActive: false }) - expect(dbChainMockFns.set).toHaveBeenNthCalledWith(2, { isActive: true }) + expect(dbChainMockFns.set).toHaveBeenNthCalledWith(1, { + isActive: expect.objectContaining({ values: expect.arrayContaining(['version-2']) }), + }) expect(dbChainMockFns.set).toHaveBeenNthCalledWith( - 3, + 2, expect.objectContaining({ isDeployed: true, deployedAt: expect.any(Date) }) ) expect(dbChainMockFns.set).toHaveBeenNthCalledWith( - 4, + 3, expect.objectContaining({ status: 'active', completedAt: expect.any(Date), diff --git a/apps/sim/lib/workflows/persistence/deployment-operations.ts b/apps/sim/lib/workflows/persistence/deployment-operations.ts index f2964dfd5e0..9146f35b601 100644 --- a/apps/sim/lib/workflows/persistence/deployment-operations.ts +++ b/apps/sim/lib/workflows/persistence/deployment-operations.ts @@ -3,7 +3,7 @@ import { generateId } from '@sim/utils/id' import type { DbOrTx } from '@sim/workflow-persistence/types' import type { WorkflowState } from '@sim/workflow-types/workflow' import type { InferSelectModel } from 'drizzle-orm' -import { and, desc, eq, inArray, sql } from 'drizzle-orm' +import { and, desc, eq, inArray, or, sql } from 'drizzle-orm' import { canTransitionDeploymentOperation, createDeploymentReadiness, @@ -114,6 +114,28 @@ type ResolveOperationTarget = (context: PrepareOperationContext) => const IN_FLIGHT_STATUSES: DeploymentOperationStatus[] = ['preparing', 'activating'] +const DEPLOYMENT_TX_STATEMENT_TIMEOUT_MS = 30_000 +const DEPLOYMENT_TX_LOCK_TIMEOUT_MS = 5_000 +const DEPLOYMENT_TX_IDLE_TIMEOUT_MS = 30_000 + +/** + * Applies transaction-scoped Postgres safety timeouts in one round trip. + * + * Deployment transactions hold the workflow-row lock plus webhook and + * operation row locks across several statements; the idle timeout is the + * backstop that stops a wedged client from pinning those locks indefinitely, + * and `lock_timeout` keeps a waiter from inheriting the full statement clock. + * `set_config(..., true)` is `SET LOCAL`-scoped, so it clears at + * COMMIT/ROLLBACK and is safe under PgBouncer transaction pooling. + */ +export async function setDeploymentTxTimeouts(tx: DbOrTx): Promise { + await tx.execute( + sql`SELECT set_config('statement_timeout', ${`${DEPLOYMENT_TX_STATEMENT_TIMEOUT_MS}ms`}, true), + set_config('lock_timeout', ${`${DEPLOYMENT_TX_LOCK_TIMEOUT_MS}ms`}, true), + set_config('idle_in_transaction_session_timeout', ${`${DEPLOYMENT_TX_IDLE_TIMEOUT_MS}ms`}, true)` + ) +} + /** * Creates an inactive immutable snapshot and a preparing deployment attempt. */ @@ -326,6 +348,7 @@ export async function beginDeploymentOperationActivation( params: DeploymentOperationGeneration ): Promise { return db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) const operation = await lockCurrentOperation(tx, params) if (!operation.success) return operation @@ -438,6 +461,7 @@ export async function activateDeploymentOperation( params: ActivateDeploymentOperationParams ): Promise { return db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) const operationResult = await lockCurrentOperation(tx, params) if (!operationResult.success) return operationResult @@ -539,18 +563,25 @@ export async function activateDeploymentOperation( } } + /** + * Single-statement cutover: only rows whose isActive value actually + * changes are touched (the previously active row and the target), instead + * of rewriting every version row of the workflow on each deploy. There is + * no unique index on (workflowId, isActive), so the one-statement flip has + * no constraint-ordering hazard. + */ await tx .update(workflowDeploymentVersion) - .set({ isActive: false }) - .where(eq(workflowDeploymentVersion.workflowId, operation.workflowId)) - - await tx - .update(workflowDeploymentVersion) - .set({ isActive: true }) + .set({ + isActive: sql`${workflowDeploymentVersion.id} = ${operation.deploymentVersionId}`, + }) .where( and( eq(workflowDeploymentVersion.workflowId, operation.workflowId), - eq(workflowDeploymentVersion.id, operation.deploymentVersionId) + or( + eq(workflowDeploymentVersion.isActive, true), + eq(workflowDeploymentVersion.id, operation.deploymentVersionId) + ) ) ) @@ -818,7 +849,11 @@ async function prepareOperation( return { success: true, operation, reused: false } } - return params.tx ? executePrepare(params.tx) : db.transaction(executePrepare) + if (params.tx) return executePrepare(params.tx) + return db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) + return executePrepare(tx) + }) } async function lockCurrentOperation(