Skip to content

Commit d9743bb

Browse files
mzxchandramzxchandra
committed
fix(projects): preserve backfill repair and resume guarantees
Lock and verify effective legacy memberships before materializing reviewed assignments and applying the shared detach policy. Reject resume reports whose code hash differs before changing their checkpoint. Isolate local pubsub subscriber failures so healthy archive subscribers still receive notifications. Extend existing CLI and real-database integration coverage for mixed memberships, concurrent and stale connector changes, unchanged mismatched reports, replay, and local subscriber delivery.
1 parent ecfdf8b commit d9743bb

5 files changed

Lines changed: 135 additions & 59 deletions

File tree

‎apps/sim/lib/events/pubsub.ts‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -162,9 +162,16 @@ class LocalPubSubChannel<T> implements PubSubChannel<T> {
162162
}
163163

164164
subscribe(handler: (event: T) => void): () => void {
165-
this.emitter.on(this.config.channel, handler)
165+
const isolatedHandler = (event: T) => {
166+
try {
167+
handler(event)
168+
} catch (error) {
169+
logger.error(`Error in ${this.config.label} handler:`, error)
170+
}
171+
}
172+
this.emitter.on(this.config.channel, isolatedHandler)
166173
return () => {
167-
this.emitter.off(this.config.channel, handler)
174+
this.emitter.off(this.config.channel, isolatedHandler)
168175
}
169176
}
170177

‎apps/sim/lib/projects/__integration__/backfill-repair.integration.ts‎

Lines changed: 85 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -182,61 +182,91 @@ describe('Operator archive repair against the full compatible schema', () => {
182182
60000
183183
)
184184

185-
it('repairs an explicitly reviewed detached family without changing environments and replays a lost report', async () => {
186-
const artifacts = {
187-
manifest: join(directory, 'detach-manifest.json'),
188-
report: join(directory, 'detach-report.json'),
189-
}
190-
try {
191-
await client`INSERT INTO "user" (id,name,email,email_verified,created_at,updated_at)
185+
it.each(['column', 'mixed'] as const)(
186+
'repairs an explicitly reviewed detached family with %s membership and replays a lost report',
187+
async (membership) => {
188+
const artifacts = {
189+
manifest: join(directory, `detach-${membership}-manifest.json`),
190+
report: join(directory, `detach-${membership}-report.json`),
191+
}
192+
try {
193+
await client`INSERT INTO "user" (id,name,email,email_verified,created_at,updated_at)
192194
VALUES ('detach-owner','Owner','detach@fixture.test',true,now(),now())`
193-
await client`INSERT INTO project (id,name,owner_id) VALUES ('detach-project','Keep context','detach-owner')`
194-
await client`INSERT INTO workspace (id,name,owner_id,billed_account_user_id,project_id)
195+
await client`INSERT INTO project (id,name,owner_id) VALUES ('detach-project','Keep context','detach-owner')`
196+
await client`INSERT INTO workspace (id,name,owner_id,billed_account_user_id,project_id)
195197
VALUES ('retained-root','Retained','detach-owner','detach-owner','detach-project'),
196198
('detached-root','Detached','detach-owner','detach-owner','detach-project')`
197-
await client`INSERT INTO workspace (id,name,owner_id,billed_account_user_id,project_id,forked_from_workspace_id)
199+
await client`INSERT INTO workspace (id,name,owner_id,billed_account_user_id,project_id,forked_from_workspace_id)
198200
VALUES ('detached-child','Child','detach-owner','detach-owner','detach-project','detached-root')`
199-
await expect(run('plan', artifacts)).rejects.toMatchObject({ code: 2 })
200-
const plan = JSON.parse(await readFile(artifacts.manifest, 'utf8'))
201-
plan.groupings[0].decision = 'detach'
202-
plan.groupings[0].detachRootId = 'detached-root'
203-
await writeFile(artifacts.manifest, JSON.stringify(plan))
204-
const before =
205-
await client`SELECT id,owner_id,organization_id,archived_at,forked_from_workspace_id FROM workspace ORDER BY id`
206-
const holder = postgres(url.toString(), { max: 1 })
207-
try {
208-
await holder.begin(async (tx) => {
209-
await tx`SELECT pg_advisory_xact_lock(hashtextextended('project:detach-project',0))`
210-
await expect(run('repair', artifacts)).rejects.toMatchObject({ code: 2 })
211-
expect(await client`SELECT id FROM project WHERE owner_id='detach-owner'`).toHaveLength(1)
212-
})
201+
if (membership === 'mixed') {
202+
await client`INSERT INTO project_workspace (project_id,workspace_id)
203+
VALUES ('detach-project','retained-root'),('detach-project','detached-child')`
204+
await client`UPDATE workspace SET project_id=NULL WHERE id IN ('retained-root','detached-child')`
205+
}
206+
await expect(run('plan', artifacts)).rejects.toMatchObject({ code: 2 })
207+
const plan = JSON.parse(await readFile(artifacts.manifest, 'utf8'))
208+
plan.groupings[0].decision = 'detach'
209+
plan.groupings[0].detachRootId = 'detached-root'
210+
await writeFile(artifacts.manifest, JSON.stringify(plan))
211+
const before =
212+
await client`SELECT id,owner_id,organization_id,archived_at,forked_from_workspace_id FROM workspace ORDER BY id`
213+
const holder = postgres(url.toString(), { max: 1 })
214+
try {
215+
await holder.begin(async (tx) => {
216+
await tx`SELECT pg_advisory_xact_lock(hashtextextended('project:detach-project',0))`
217+
await expect(run('repair', artifacts)).rejects.toMatchObject({ code: 2 })
218+
expect(await client`SELECT id FROM project WHERE owner_id='detach-owner'`).toHaveLength(
219+
1
220+
)
221+
})
222+
if (membership === 'mixed') {
223+
await client`INSERT INTO project (id,name,owner_id) VALUES ('changed-project','Changed','detach-owner')`
224+
await holder.begin(async (tx) => {
225+
await tx`UPDATE project_workspace SET project_id='changed-project' WHERE workspace_id='retained-root'`
226+
await expect(run('repair', artifacts)).rejects.toMatchObject({ code: 2 })
227+
expect(
228+
await client`SELECT project_id FROM workspace WHERE id='retained-root'`
229+
).toEqual([{ project_id: null }])
230+
expect(
231+
await client`SELECT id FROM project WHERE id=${plan.groupings[0].destinationProjectId}`
232+
).toHaveLength(0)
233+
})
234+
await expect(run('repair', artifacts)).rejects.toMatchObject({ code: 1 })
235+
expect(await client`SELECT project_id FROM workspace WHERE id='retained-root'`).toEqual(
236+
[{ project_id: null }]
237+
)
238+
await client`UPDATE project_workspace SET project_id='detach-project' WHERE workspace_id='retained-root'`
239+
await client`DELETE FROM project WHERE id='changed-project'`
240+
}
241+
} finally {
242+
await holder.end()
243+
}
244+
await client`UPDATE workspace SET archived_at='2026-05-01' WHERE id='detached-child'`
245+
await expect(run('repair', artifacts)).rejects.toMatchObject({ code: 1 })
246+
expect(await client`SELECT id FROM project WHERE owner_id='detach-owner'`).toHaveLength(1)
247+
await client`UPDATE workspace SET archived_at=NULL WHERE id='detached-child'`
248+
await expect(run('repair', artifacts)).rejects.toMatchObject({ code: 2 })
249+
const destination = plan.groupings[0].destinationProjectId
250+
expect(
251+
await client`SELECT id FROM workspace WHERE project_id = ${destination} ORDER BY id`
252+
).toEqual([{ id: 'detached-child' }, { id: 'detached-root' }])
253+
expect(await client`SELECT id,name FROM project WHERE id = 'detach-project'`).toEqual([
254+
{ id: 'detach-project', name: 'Keep context' },
255+
])
256+
expect(
257+
await client`SELECT id,owner_id,organization_id,archived_at,forked_from_workspace_id FROM workspace ORDER BY id`
258+
).toEqual(before)
259+
await rm(artifacts.report)
260+
await expect(run('repair', artifacts)).rejects.toMatchObject({ code: 2 })
261+
expect(await client`SELECT id FROM project WHERE owner_id = 'detach-owner'`).toHaveLength(2)
213262
} finally {
214-
await holder.end()
263+
await client`DELETE FROM workspace WHERE owner_id = 'detach-owner'`
264+
await client`DELETE FROM project WHERE owner_id = 'detach-owner'`
265+
await client`DELETE FROM "user" WHERE id = 'detach-owner'`
215266
}
216-
await client`UPDATE workspace SET archived_at='2026-05-01' WHERE id='detached-child'`
217-
await expect(run('repair', artifacts)).rejects.toMatchObject({ code: 1 })
218-
expect(await client`SELECT id FROM project WHERE owner_id='detach-owner'`).toHaveLength(1)
219-
await client`UPDATE workspace SET archived_at=NULL WHERE id='detached-child'`
220-
await expect(run('repair', artifacts)).rejects.toMatchObject({ code: 2 })
221-
const destination = plan.groupings[0].destinationProjectId
222-
expect(
223-
await client`SELECT id FROM workspace WHERE project_id = ${destination} ORDER BY id`
224-
).toEqual([{ id: 'detached-child' }, { id: 'detached-root' }])
225-
expect(await client`SELECT id,name FROM project WHERE id = 'detach-project'`).toEqual([
226-
{ id: 'detach-project', name: 'Keep context' },
227-
])
228-
expect(
229-
await client`SELECT id,owner_id,organization_id,archived_at,forked_from_workspace_id FROM workspace ORDER BY id`
230-
).toEqual(before)
231-
await rm(artifacts.report)
232-
await expect(run('repair', artifacts)).rejects.toMatchObject({ code: 2 })
233-
expect(await client`SELECT id FROM project WHERE owner_id = 'detach-owner'`).toHaveLength(2)
234-
} finally {
235-
await client`DELETE FROM workspace WHERE owner_id = 'detach-owner'`
236-
await client`DELETE FROM project WHERE owner_id = 'detach-owner'`
237-
await client`DELETE FROM "user" WHERE id = 'detach-owner'`
238-
}
239-
}, 60000)
267+
},
268+
60000
269+
)
240270

241271
it('keeps a Project active when another active environment still uses legacy membership', async () => {
242272
const artifacts = {
@@ -265,7 +295,7 @@ describe('Operator archive repair against the full compatible schema', () => {
265295
}
266296
}, 60000)
267297

268-
it('completes provider cleanup despite a throwing local MCP subscriber', async () => {
298+
it('notifies healthy local MCP subscribers and completes cleanup after another subscriber throws', async () => {
269299
const serverId = generateId()
270300
try {
271301
await client`INSERT INTO "user" (id,name,email,email_verified,created_at,updated_at)
@@ -288,13 +318,17 @@ describe('Operator archive repair against the full compatible schema', () => {
288318
import { repairArchivedProjectEnvironment } from '@/lib/projects/backfill-repair';
289319
import { mcpPubSub } from '@/lib/mcp/pubsub';
290320
import { db, dbReplica } from '@sim/db';
321+
import assert from 'node:assert/strict';
291322
import postgres from 'postgres';
292323
const client = postgres(process.env.MIGRATION_DATABASE_URL, {max:1});
293324
const unsubscribe = mcpPubSub.onWorkflowToolsChanged(() => { throw new Error('fixture subscriber failed'); });
325+
const received = [];
326+
const unsubscribeHealthy = mcpPubSub.onWorkflowToolsChanged((event) => received.push(event));
294327
try {
295328
await repairArchivedProjectEnvironment(client, {workspaceId:'local-env', archivedAt:'2026-05-01 00:00:00', workflowIds:['local-flow']}, 'local-subscriber');
329+
assert.ok(received.some((event) => event.serverId === '${serverId}' && event.workspaceId === 'local-env'), 'Healthy subscriber missed the archive notification');
296330
} finally {
297-
unsubscribe(); mcpPubSub.dispose(); await client.end();
331+
unsubscribe(); unsubscribeHealthy(); mcpPubSub.dispose(); await client.end();
298332
await Promise.all([...new Set([db.$client,dbReplica.$client])].map(client=>client.end()));
299333
}
300334
`,

‎apps/sim/lib/projects/backfill-repair.ts‎

Lines changed: 27 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -153,11 +153,11 @@ export async function repairProjectGrouping(
153153
!review.roots.includes(rootId) ||
154154
review.destinationProjectId === review.projectId ||
155155
review.members.length > 1000 ||
156-
review.members.some((row) => row.projectId !== review.projectId) ||
156+
review.members.some((row) => (row.projectId ?? row.legacyProjectId) !== review.projectId) ||
157157
projectGroupingEvidence(review.members) !== review.evidence
158158
)
159159
throw new ProjectBackfillConflict(
160-
'Detach requires an exact reviewed, fully assigned Project; rediscover after apply'
160+
'Detach requires an exact reviewed Project grouping; rediscover'
161161
)
162162
const expected = new Map(review.members.map((row) => [row.id, row]))
163163
const root = expected.get(rootId)
@@ -185,14 +185,32 @@ export async function repairProjectGrouping(
185185
]
186186
if (!(await tryAcquireAdvisoryXactLocks(tx, 'project_backfill', keys)))
187187
throw new ProjectBackfillBusy('Reviewed Project grouping is busy')
188+
const reviewedIds = review.members.map((row) => row.id)
189+
await tx.execute(sql`SELECT id FROM workspace
190+
WHERE ${inArray(schema.workspace.id, reviewedIds)}
191+
ORDER BY id COLLATE "C" FOR NO KEY UPDATE NOWAIT`)
192+
const [legacy] = await tx.execute<{ present: boolean }>(
193+
sql`SELECT to_regclass('public.project_workspace') IS NOT NULL AS present`
194+
)
195+
if (legacy?.present) {
196+
await tx.execute(sql`SELECT pw.workspace_id FROM project_workspace pw
197+
JOIN workspace ON workspace.id = pw.workspace_id
198+
WHERE ${inArray(schema.workspace.id, reviewedIds)} AND workspace.project_id IS NULL
199+
ORDER BY pw.workspace_id COLLATE "C" FOR SHARE OF pw NOWAIT`)
200+
}
201+
const legacyAssignment = legacy?.present
202+
? sql`(SELECT pw.project_id FROM project_workspace pw WHERE pw.workspace_id = workspace.id)`
203+
: sql`NULL::text`
188204
const current = await tx.execute<ProjectBackfillWorkspace>(sql`
189205
SELECT id, forked_from_workspace_id AS "parentId", owner_id AS "ownerId",
190-
organization_id AS "organizationId", archived_at::text AS "archivedAt", project_id AS "projectId", NULL::text AS "legacyProjectId"
191-
FROM workspace WHERE project_id IN (${review.projectId},${review.destinationProjectId})
206+
organization_id AS "organizationId", archived_at::text AS "archivedAt", project_id AS "projectId", ${legacyAssignment} AS "legacyProjectId"
207+
FROM workspace WHERE coalesce(project_id, ${legacyAssignment}) IN (${review.projectId},${review.destinationProjectId})
192208
ORDER BY id COLLATE "C" LIMIT 1001 FOR NO KEY UPDATE NOWAIT`)
193209
if (current.length > 1000)
194210
throw new ProjectBackfillConflict('Project grouping exceeds repair limit')
195-
const destination = current.filter((row) => row.projectId === review.destinationProjectId)
211+
const destination = current.filter(
212+
(row) => (row.projectId ?? row.legacyProjectId) === review.destinationProjectId
213+
)
196214
if (destination.length) {
197215
if (
198216
projectGroupingEvidence(
@@ -220,6 +238,10 @@ export async function repairProjectGrouping(
220238
throw new ProjectBackfillConflict(
221239
'Reviewed Project ownership or scope requires reconciliation before detach'
222240
)
241+
await tx
242+
.update(schema.workspace)
243+
.set({ projectId: review.projectId })
244+
.where(and(inArray(schema.workspace.id, reviewedIds), isNull(schema.workspace.projectId)))
223245
await splitForkProject(tx, rootId, review.destinationProjectId)
224246
})
225247
}

‎apps/sim/scripts/backfill-projects.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -323,13 +323,16 @@ The registered migration backfills remaining assignments, verifies, then enforce
323323
}
324324
if (progress.manifestHash !== manifestHash || progress.databaseId !== databaseId)
325325
throw new Error('Report does not match the reviewed manifest')
326+
if (progress.codeHash !== codeHash)
327+
throw new Error(
328+
'Report belongs to different backfill code; use the original checkout or a new report'
329+
)
326330
if (
327331
progress.nextIndex > manifest.families.length ||
328332
progress.deferred.some((index) => index >= manifest.families.length)
329333
)
330334
throw new Error('Invalid report cursor')
331335
progress.runId = generateId()
332-
progress.codeHash = codeHash
333336
progress.status = 'running'
334337
const started = performance.now()
335338
let attempts = 0

‎packages/db/scripts/project-contract.integration.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -613,6 +613,16 @@ describe('Project expand/backfill/contract against PostgreSQL', () => {
613613
projectsCreated: 50,
614614
status: 'paused',
615615
})
616+
const mismatched = JSON.stringify({ ...report, codeHash: 'another-checkout' })
617+
await writeFile(reportPath, mismatched)
618+
await expect(run('apply', ['--ack-release-drained'])).rejects.toMatchObject({
619+
code: 1,
620+
stderr: expect.stringContaining('Report belongs to different backfill code'),
621+
})
622+
expect(await readFile(reportPath, 'utf8')).toBe(mismatched)
623+
expect(await sql`SELECT id FROM project`).toHaveLength(50)
624+
expect(await sql`SELECT id FROM workspace WHERE project_id IS NOT NULL`).toHaveLength(50)
625+
await writeFile(reportPath, JSON.stringify(report))
616626
for (const [status, exitCode] of [
617627
['running', 2],
618628
['paused', 2],

0 commit comments

Comments
 (0)