11#!/usr/bin/env bun
22
33/**
4- * Deploys every workflow referenced by a table workflow group.
4+ * Deploys every mutable workflow referenced by a table workflow group.
55 *
66 * The script is idempotent and resumable: it pages over only workflows that do
77 * not have the full desired state, and each deployment uses a stable idempotency
@@ -52,6 +52,7 @@ export interface TableWorkflowDeploymentSummary {
5252 scanned : number
5353 deployed : number
5454 alreadyDeployed : number
55+ skippedLocked : number
5556}
5657
5758interface TableWorkflowDeploymentBackfillOptions {
@@ -158,6 +159,50 @@ async function getDatabase() {
158159 return db
159160}
160161
162+ function validateCandidatePage (
163+ candidates : TableWorkflowDeploymentCandidate [ ] ,
164+ afterWorkflowId : string ,
165+ limit : number
166+ ) : string | null {
167+ if ( candidates . length === 0 ) return null
168+ if ( candidates . length > limit ) {
169+ throw new Error ( 'Table workflow deployment store returned an oversized page' )
170+ }
171+
172+ const pageIds = new Set ( candidates . map ( ( candidate ) => candidate . workflowId ) )
173+ if ( pageIds . size !== candidates . length ) {
174+ throw new Error ( 'Table workflow deployment store returned duplicate workflow ids' )
175+ }
176+
177+ const lastWorkflowId = candidates . at ( - 1 ) ?. workflowId
178+ if ( ! lastWorkflowId || lastWorkflowId === afterWorkflowId ) {
179+ throw new Error ( 'Table workflow deployment store returned a non-advancing page' )
180+ }
181+ return lastWorkflowId
182+ }
183+
184+ async function assertOnlyLockedCandidatesRemain (
185+ store : TableWorkflowDeploymentStore ,
186+ lockedWorkflowIds : ReadonlySet < string >
187+ ) : Promise < void > {
188+ let afterWorkflowId = ''
189+ for ( ; ; ) {
190+ const candidates = await store . listCandidates ( afterWorkflowId , 1 )
191+ const lastWorkflowId = validateCandidatePage ( candidates , afterWorkflowId , 1 )
192+ if ( ! lastWorkflowId ) return
193+
194+ const unexpectedCandidate = candidates . find (
195+ ( candidate ) => ! lockedWorkflowIds . has ( candidate . workflowId )
196+ )
197+ if ( unexpectedCandidate ) {
198+ throw new Error (
199+ `Table workflow deployment backfill left workflow ${ unexpectedCandidate . workflowId } undeployed`
200+ )
201+ }
202+ afterWorkflowId = lastWorkflowId
203+ }
204+ }
205+
161206/** Ensures the table group references can be traversed without silently dropping corrupt data. */
162207async function assertTableWorkflowIntegrity ( ) : Promise < void > {
163208 const db = await getDatabase ( )
@@ -343,7 +388,7 @@ export async function deployTableWorkflow(
343388}
344389
345390/**
346- * Reaches and verifies that every table workflow has an active deployment.
391+ * Reaches and verifies that every mutable table workflow has an active deployment.
347392 */
348393export async function backfillTableWorkflowDeployments (
349394 store : TableWorkflowDeploymentStore ,
@@ -361,24 +406,15 @@ export async function backfillTableWorkflowDeployments(
361406 scanned : 0 ,
362407 deployed : 0 ,
363408 alreadyDeployed : 0 ,
409+ skippedLocked : 0 ,
364410 }
411+ const lockedWorkflowIds = new Set < string > ( )
365412 let afterWorkflowId = ''
366413
367414 for ( ; ; ) {
368415 const candidates = await store . listCandidates ( afterWorkflowId , batchSize )
369- if ( candidates . length === 0 ) break
370- if ( candidates . length > batchSize ) {
371- throw new Error ( 'Table workflow deployment store returned an oversized page' )
372- }
373-
374- const pageIds = new Set ( candidates . map ( ( candidate ) => candidate . workflowId ) )
375- if ( pageIds . size !== candidates . length ) {
376- throw new Error ( 'Table workflow deployment store returned duplicate workflow ids' )
377- }
378- const lastWorkflowId = candidates . at ( - 1 ) ?. workflowId
379- if ( ! lastWorkflowId || lastWorkflowId === afterWorkflowId ) {
380- throw new Error ( 'Table workflow deployment store returned a non-advancing page' )
381- }
416+ const lastWorkflowId = validateCandidatePage ( candidates , afterWorkflowId , batchSize )
417+ if ( ! lastWorkflowId ) break
382418
383419 for ( const candidate of candidates ) {
384420 summary . scanned += 1
@@ -391,6 +427,16 @@ export async function backfillTableWorkflowDeployments(
391427 } )
392428 const result = await deploy ( candidate )
393429 if ( ! result . success ) {
430+ if ( result . errorCode === 'locked' ) {
431+ lockedWorkflowIds . add ( candidate . workflowId )
432+ summary . skippedLocked += 1
433+ logger . warn ( 'Skipping locked workflow referenced by a table workflow group' , {
434+ workflowId : candidate . workflowId ,
435+ workspaceId : candidate . workspaceId ,
436+ reason : result . error ?? 'Workflow is locked' ,
437+ } )
438+ continue
439+ }
394440 throw new Error (
395441 `Failed to deploy table workflow ${ candidate . workflowId } : ${ result . error ?? 'deployment returned no error' } `
396442 )
@@ -413,12 +459,7 @@ export async function backfillTableWorkflowDeployments(
413459 }
414460
415461 await store . assertIntegrity ( )
416- const remaining = await store . listCandidates ( '' , 1 )
417- if ( remaining . length > 0 ) {
418- throw new Error (
419- `Table workflow deployment backfill left workflow ${ remaining [ 0 ] . workflowId } undeployed`
420- )
421- }
462+ await assertOnlyLockedCandidatesRemain ( store , lockedWorkflowIds )
422463
423464 return summary
424465}
0 commit comments