Skip to content

Commit d7ec406

Browse files
mzxchandramzxchandra
authored
fix(db): optimize Project backfill and repair archived workflows (#8962)
* fix(db): reduce Project backfill round trips and scheduling limits * fix(db): repair archived workspace workflows during Project backfill * fix(db): retry busy Project archive repairs
1 parent 2b5dbf7 commit d7ec406

4 files changed

Lines changed: 445 additions & 71 deletions

File tree

‎packages/db/maintenance/project-backfill.ts‎

Lines changed: 90 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -457,20 +457,28 @@ export async function tryProjectBackfillLocks(tx: TransactionSql, keys: string[]
457457
if (!result?.acquired) throw new ProjectBackfillBusy('Project backfill target is busy')
458458
}
459459

460-
/** Writes at most 50 independent families, or one complete fork family, in a bounded transaction. */
460+
/** New Projects need no Project-wide evidence reads; existing Projects retain a smaller batch. */
461+
export function projectBackfillBatchSize(family: ProjectBackfillFamily): number {
462+
return family.projectMembers || family.members.some((row) => effectiveProjectId(row) !== null)
463+
? 50
464+
: 200
465+
}
466+
467+
/** Writes bounded independent singleton batches, or one complete fork family, atomically. */
461468
export async function assignProjectBackfillBatch(sql: Sql, families: ProjectBackfillFamily[]) {
462469
if (
463470
!families.length ||
464-
families.length > 50 ||
471+
families.some((family) => families.length > projectBackfillBatchSize(family)) ||
465472
(families.length > 1 && families.some((f) => f.members.length !== 1))
466473
)
467-
throw new Error('Expected 1–50 singletons or one complete fork family')
474+
throw new Error(
475+
'Expected up to 200 new-Project singletons, 50 existing-Project singletons, or one complete fork family'
476+
)
468477
for (const family of families) validateFamily(family)
469478
return sql.begin(async (tx) => {
470479
await tx`SELECT set_config(CASE WHEN current_setting('transaction_timeout', true) IS NULL
471-
THEN 'idle_in_transaction_session_timeout' ELSE 'transaction_timeout' END, '5s', true)`
472-
await tx`SET LOCAL statement_timeout = '3s'`
473-
await tx`SET LOCAL lock_timeout = '250ms'`
480+
THEN 'idle_in_transaction_session_timeout' ELSE 'transaction_timeout' END, '5s', true),
481+
set_config('statement_timeout', '3s', true), set_config('lock_timeout', '250ms', true)`
474482
const legacy = await hasLegacyMemberships(tx)
475483
await assertProjectColumnAuthority(tx)
476484
const roots = families.map((f) => f.rootId).sort(compareStrings)
@@ -538,15 +546,80 @@ export async function assignProjectBackfillBatch(sql: Sql, families: ProjectBack
538546
'Reviewed workspace scope, owner, archive state or assignment changed; rediscover'
539547
)
540548
}
541-
const children =
542-
await tx`SELECT id FROM workspace WHERE forked_from_workspace_id = ANY(${ids}) AND NOT id = ANY(${ids}) LIMIT 1`
543-
if (children.length)
549+
const [validation] = await tx`SELECT
550+
EXISTS (SELECT 1 FROM workspace WHERE forked_from_workspace_id = ANY(${ids})
551+
AND NOT id = ANY(${ids})) AS children,
552+
EXISTS (SELECT 1 FROM workflow f JOIN workspace w ON w.id = f.workspace_id
553+
WHERE w.id = ANY(${ids}) AND w.archived_at IS NOT NULL AND f.archived_at IS NULL) AS invalid_archive`
554+
if (validation.children)
544555
throw new ProjectBackfillConflict('Reviewed fork family gained descendants; rediscover')
545-
const invalidArchive =
546-
await tx`SELECT f.id FROM workflow f JOIN workspace w ON w.id = f.workspace_id
547-
WHERE w.id = ANY(${ids}) AND w.archived_at IS NOT NULL AND f.archived_at IS NULL LIMIT 1`
548-
if (invalidArchive.length)
556+
if (validation.invalid_archive)
549557
throw new ProjectBackfillConflict('Archive repair is required before assignment')
558+
const evidenceRequests = families.flatMap((family) => {
559+
const projectIds = [
560+
...new Set(
561+
family.members.map((row) => {
562+
const current = byId.get(row.id)
563+
if (!current)
564+
throw new ProjectBackfillConflict('Reviewed workspace disappeared; rediscover')
565+
return effectiveProjectId(current)
566+
})
567+
),
568+
].filter((id): id is string => id !== null)
569+
if (projectIds.length > 1) throw new ProjectBackfillConflict('Family now spans Projects')
570+
return projectIds.length
571+
? [
572+
{
573+
root_id: family.rootId,
574+
project_id: projectIds[0],
575+
reviewed_ids: family.projectMembers?.map((row) => row.id) ?? [],
576+
},
577+
]
578+
: []
579+
})
580+
const existingProjectIds = [...new Set(evidenceRequests.map((request) => request.project_id))]
581+
const projectRows = existingProjectIds.length
582+
? await tx<
583+
{
584+
id: string
585+
owner_id: string
586+
organization_id: string | null
587+
archived_at: string | null
588+
}[]
589+
>`SELECT id, owner_id, organization_id, archived_at::text FROM project
590+
WHERE id = ANY(${existingProjectIds}) ORDER BY id COLLATE "C" FOR NO KEY UPDATE NOWAIT`
591+
: []
592+
const existingProjects = new Map(projectRows.map((row) => [row.id, row]))
593+
const evidence = evidenceRequests.length
594+
? await tx<(ProjectBackfillWorkspace & { rootId: string })[]>`
595+
SELECT request.root_id AS "rootId", member.*
596+
FROM jsonb_to_recordset(${JSON.stringify(evidenceRequests)}::text::jsonb)
597+
AS request(root_id text, project_id text, reviewed_ids text[])
598+
CROSS JOIN LATERAL (
599+
SELECT w.id, w.owner_id AS "ownerId", w.organization_id AS "organizationId", w.archived_at::text AS "archivedAt",
600+
w.forked_from_workspace_id AS "parentId", w.project_id AS "projectId", ${legacyAssignment(tx, legacy)} AS "legacyProjectId"
601+
FROM workspace w WHERE w.id IN (
602+
SELECT id FROM workspace WHERE project_id = request.project_id
603+
${
604+
legacy
605+
? tx`UNION
606+
SELECT legacy_member.id FROM project_workspace pw JOIN workspace legacy_member ON legacy_member.id = pw.workspace_id
607+
WHERE pw.project_id = request.project_id AND legacy_member.project_id IS NULL`
608+
: tx``
609+
}
610+
UNION SELECT unnest(request.reviewed_ids)
611+
)
612+
ORDER BY w.id COLLATE "C" LIMIT 1001 FOR NO KEY UPDATE NOWAIT
613+
) member
614+
ORDER BY request.root_id COLLATE "C", member.id COLLATE "C"
615+
`
616+
: []
617+
const evidenceByRoot = new Map<string, ProjectBackfillWorkspace[]>()
618+
for (const row of evidence) {
619+
const rows = evidenceByRoot.get(row.rootId) ?? []
620+
rows.push(row)
621+
evidenceByRoot.set(row.rootId, rows)
622+
}
550623
const inserts: {
551624
id: string
552625
name: string
@@ -574,16 +647,8 @@ export async function assignProjectBackfillBatch(sql: Sql, families: ProjectBack
574647
: null
575648
let projectId = projectIds[0]
576649
if (projectId) {
577-
const [existing] =
578-
await tx`SELECT owner_id, organization_id, archived_at::text FROM project WHERE id = ${projectId} FOR NO KEY UPDATE NOWAIT`
579-
const reviewedIds = family.projectMembers?.map((row) => row.id) ?? []
580-
const evidenceRows = await tx<ProjectBackfillWorkspace[]>`
581-
SELECT w.id, w.owner_id AS "ownerId", w.organization_id AS "organizationId", w.archived_at::text AS "archivedAt",
582-
w.forked_from_workspace_id AS "parentId", w.project_id AS "projectId", ${legacyAssignment(tx, legacy)} AS "legacyProjectId"
583-
FROM workspace w WHERE coalesce(w.project_id, ${legacyAssignment(tx, legacy)}) = ${projectId}
584-
OR w.id = ANY(${reviewedIds})
585-
ORDER BY w.id COLLATE "C" LIMIT 1001 FOR NO KEY UPDATE NOWAIT
586-
`
650+
const existing = existingProjects.get(projectId)
651+
const evidenceRows = evidenceByRoot.get(family.rootId) ?? []
587652
const environments = evidenceRows.filter((row) => effectiveProjectId(row) === projectId)
588653
if (!existing || evidenceRows.length > MAX_FAMILY)
589654
throw new ProjectBackfillConflict(
@@ -644,6 +709,8 @@ export async function assignProjectBackfillBatch(sql: Sql, families: ProjectBack
644709
)
645710
await tx`UPDATE project SET organization_id = ${root.organizationId},
646711
archived_at = ${projectArchiveAt}::text::timestamp, updated_at = now() WHERE id = ${projectId}`
712+
existing.organization_id = root.organizationId
713+
existing.archived_at = projectArchiveAt
647714
}
648715
} else {
649716
projectId = generateId()

‎packages/db/maintenance/project-repairs.ts‎

Lines changed: 62 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,8 @@
1-
import type { ProjectArchiveRepair } from '@sim/db/maintenance/project-backfill'
1+
import {
2+
type ProjectArchiveRepair,
3+
tryProjectBackfillLocks,
4+
} from '@sim/db/maintenance/project-backfill'
5+
import { assertProjectColumnAuthority } from '@sim/db/maintenance/project-rollout'
26
import type { Sql } from 'postgres'
37

48
/** Runner-managed recovery state, like script_migrations; never an application ownership table. */
@@ -30,3 +34,60 @@ export async function countPendingProjectArchiveRepairs(sql: Sql): Promise<numbe
3034
await sql`SELECT count(*)::int AS count FROM public.project_backfill_archive_repairs WHERE completed_at IS NULL`
3135
return row.count
3236
}
37+
38+
/**
39+
* Repairs workflow archive state using database-only lifecycle effects. Provider-backed
40+
* subscriptions still require the application repair path and its durable cleanup receipt.
41+
*/
42+
export async function archiveProjectRepairWorkflows(
43+
sql: Sql,
44+
repair: ProjectArchiveRepair
45+
): Promise<void> {
46+
if (!repair.workflowIds.length || repair.workflowIds.length > 1000)
47+
throw new Error('Expected a bounded workflow archive repair')
48+
await sql.begin(async (tx) => {
49+
await tx`SELECT set_config(CASE WHEN current_setting('transaction_timeout', true) IS NULL
50+
THEN 'idle_in_transaction_session_timeout' ELSE 'transaction_timeout' END, '5s', true),
51+
set_config('statement_timeout', '3s', true), set_config('lock_timeout', '250ms', true)`
52+
await assertProjectColumnAuthority(tx)
53+
await tryProjectBackfillLocks(tx, [`project-backfill:${repair.workspaceId}`])
54+
const [workspace] = await tx`SELECT archived_at::text AS archived_at FROM workspace
55+
WHERE id = ${repair.workspaceId} FOR NO KEY UPDATE NOWAIT`
56+
if (!workspace || workspace.archived_at !== repair.archivedAt)
57+
throw new Error('Archived workspace changed; rediscover before repair')
58+
const workflows = await tx`SELECT id, workspace_id FROM workflow
59+
WHERE id = ANY(${repair.workflowIds}) OR (workspace_id = ${repair.workspaceId} AND archived_at IS NULL)
60+
ORDER BY id COLLATE "C" FOR NO KEY UPDATE NOWAIT`
61+
if (
62+
workflows.some(
63+
(row) => row.workspace_id !== repair.workspaceId || !repair.workflowIds.includes(row.id)
64+
)
65+
)
66+
throw new Error('Archive repair workflow membership changed; rediscover')
67+
const subscriptions =
68+
await tx`SELECT id FROM webhook WHERE workflow_id = ANY(${repair.workflowIds})
69+
AND (provider IS DISTINCT FROM 'generic'
70+
OR json_typeof(provider_config->'previousSubscription'->'provider') = 'string') LIMIT 1`
71+
if (subscriptions.length)
72+
throw new Error(
73+
'Archive repair requires external provider cleanup; use backfill-projects.ts plan/repair'
74+
)
75+
await tx`UPDATE workflow_deployment_operation SET status = 'superseded', completed_at = now(), updated_at = now()
76+
WHERE workflow_id = ANY(${repair.workflowIds}) AND status IN ('preparing', 'activating')`
77+
await tx`DELETE FROM webhook_path_claim WHERE workflow_id = ANY(${repair.workflowIds})`
78+
await tx`UPDATE workflow_schedule SET archived_at = ${repair.archivedAt}::text::timestamp, updated_at = now(),
79+
status = 'disabled', next_run_at = NULL, last_queued_at = NULL
80+
WHERE workflow_id = ANY(${repair.workflowIds}) AND archived_at IS NULL`
81+
await tx`UPDATE webhook SET archived_at = ${repair.archivedAt}::text::timestamp, updated_at = now(), is_active = false
82+
WHERE workflow_id = ANY(${repair.workflowIds}) AND archived_at IS NULL`
83+
await tx`UPDATE chat SET archived_at = ${repair.archivedAt}::text::timestamp, updated_at = now(), is_active = false
84+
WHERE workflow_id = ANY(${repair.workflowIds}) AND archived_at IS NULL`
85+
await tx`UPDATE workflow_mcp_tool SET archived_at = ${repair.archivedAt}::text::timestamp, updated_at = now()
86+
WHERE workflow_id = ANY(${repair.workflowIds}) AND archived_at IS NULL`
87+
await tx`UPDATE workflow_deployment_version SET is_active = false
88+
WHERE workflow_id = ANY(${repair.workflowIds}) AND is_active = true`
89+
await tx`UPDATE workflow SET archived_at = ${repair.archivedAt}::text::timestamp, updated_at = now(),
90+
is_deployed = false, is_public_api = false
91+
WHERE id = ANY(${repair.workflowIds}) AND archived_at IS NULL`
92+
})
93+
}

‎packages/db/script-migrations/0032_project_membership.ts‎

Lines changed: 43 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -4,12 +4,16 @@ import {
44
discoverProjectBackfill,
55
ProjectBackfillBusy,
66
ProjectBackfillConflict,
7+
projectBackfillBatchSize,
78
projectBackfillDatabaseId,
89
readProjectGroupingReviews,
910
verifyProjectBackfill,
1011
} from '@sim/db/maintenance/project-backfill'
1112
import { enforceProjectMembership } from '@sim/db/maintenance/project-enforcement'
12-
import { countPendingProjectArchiveRepairs } from '@sim/db/maintenance/project-repairs'
13+
import {
14+
archiveProjectRepairWorkflows,
15+
countPendingProjectArchiveRepairs,
16+
} from '@sim/db/maintenance/project-repairs'
1317
import { switchProjectMembershipAuthority } from '@sim/db/maintenance/project-rollout'
1418
import type { ScriptMigration } from '@sim/db/script-migrations/types'
1519
import { createLogger } from '@sim/logger'
@@ -22,7 +26,6 @@ function retryable(error: unknown): boolean {
2226
const failure = getTransientDatabaseFailure(error)
2327
return failure === 'capacity' || failure === 'conflict'
2428
}
25-
const MAX_DURATION_MS = 20 * 60_000
2629

2730
/** Commits bounded assignments before validation and contraction; retry discovers only remaining work. */
2831
export const projectMembershipMigration: ScriptMigration = {
@@ -60,8 +63,8 @@ export const projectMembershipMigration: ScriptMigration = {
6063
throw new Error('Reviewed Project migration requires its database URL')
6164
const databaseId = reviewUrl ? projectBackfillDatabaseId(reviewUrl) : 'migration-runner'
6265
const reviews = reviewPath ? await readProjectGroupingReviews(reviewPath, databaseId) : []
63-
const manifest = await discoverProjectBackfill(sql, databaseId, reviews)
64-
if (manifest.conflicts.length || manifest.repairs.length) {
66+
let manifest = await discoverProjectBackfill(sql, databaseId, reviews)
67+
if (manifest.conflicts.length) {
6568
logger.error('Project preparation requires operator remediation', {
6669
conflicts: manifest.conflicts.slice(0, 50),
6770
conflictCount: manifest.conflicts.length,
@@ -71,25 +74,53 @@ export const projectMembershipMigration: ScriptMigration = {
7174
'Resolve Project conflicts and archive/provider repairs with backfill-projects.ts plan/repair before retrying'
7275
)
7376
}
74-
const pending = manifest.families.map((family) => ({ family, attempts: 0 }))
77+
if (manifest.repairs.length) {
78+
for (const repair of manifest.repairs) {
79+
for (let attempt = 0; ; attempt++) {
80+
await assertSession()
81+
try {
82+
await archiveProjectRepairWorkflows(sql, repair)
83+
break
84+
} catch (error) {
85+
if (!(error instanceof ProjectBackfillBusy || retryable(error)) || attempt >= 2)
86+
throw error
87+
await sleep(backoffWithJitter(attempt + 1, null, { baseMs: 250, maxMs: 1000 }))
88+
}
89+
}
90+
}
91+
logger.info('Archived workflows in archived workspaces', {
92+
workspaces: manifest.repairs.length,
93+
workflows: manifest.repairs.reduce(
94+
(count, repair) => count + repair.workflowIds.length,
95+
0
96+
),
97+
})
98+
manifest = await discoverProjectBackfill(sql, databaseId, reviews)
99+
if (manifest.conflicts.length || manifest.repairs.length)
100+
throw new ProjectBackfillConflict(
101+
'Project state changed during archive repair; rediscover'
102+
)
103+
}
104+
const pending = manifest.families
105+
.map((family) => ({ family, attempts: 0 }))
106+
.sort((a, b) => {
107+
const priority = (family: typeof a.family) =>
108+
family.members.length === 1 && projectBackfillBatchSize(family) === 200 ? 0 : 1
109+
return priority(a.family) - priority(b.family)
110+
})
75111
const failed: string[] = []
76-
const started = performance.now()
77-
let batchSize = 50
78112
let assigned = 0
79113
let batches = 0
80114
for (let position = 0; position < pending.length; ) {
81-
if (performance.now() - started > MAX_DURATION_MS || batches >= 10_000) {
82-
throw new Error(
83-
'Project backfill scheduling budget exhausted; rerun to resume committed progress'
84-
)
85-
}
86115
await assertSession()
87116
const items = [pending[position++]]
117+
const batchSize = projectBackfillBatchSize(items[0].family)
88118
if (items[0].family.members.length === 1 && items[0].attempts === 0) {
89119
while (
90120
items.length < batchSize &&
91121
position < pending.length &&
92122
pending[position].family.members.length === 1 &&
123+
projectBackfillBatchSize(pending[position].family) === batchSize &&
93124
pending[position].attempts === 0
94125
) {
95126
items.push(pending[position++])
@@ -104,7 +135,6 @@ export const projectMembershipMigration: ScriptMigration = {
104135
)
105136
assigned += result.assigned
106137
const elapsedMs = performance.now() - batchStarted
107-
if (elapsedMs > 250) batchSize = Math.max(1, Math.floor(batchSize / 2))
108138
logger.info('Project migration batch committed', {
109139
...result,
110140
assigned,
@@ -131,7 +161,6 @@ export const projectMembershipMigration: ScriptMigration = {
131161
})
132162
await sleep(backoffWithJitter(items[0].attempts, null, { baseMs: 250, maxMs: 1000 }))
133163
}
134-
await sleep(100)
135164
}
136165
if (failed.length) {
137166
logger.error('Project families still need retry or reconciliation', {

0 commit comments

Comments
 (0)