Skip to content

Commit 1e4499e

Browse files
mzxchandramzxchandra
committed
fix(projects): integrate reviewed repair and fixture corrections
2 parents 639c8ca + d9743bb commit 1e4499e

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)