Skip to content

Commit b653f5e

Browse files
mzxchandramzxchandra
committed
fix(projects): close operator verification and contention gaps
1 parent 7f8eb18 commit b653f5e

6 files changed

Lines changed: 135 additions & 48 deletions

File tree

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

Lines changed: 22 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,9 @@ import { tmpdir } from 'node:os'
44
import { join } from 'node:path'
55
import { promisify } from 'node:util'
66
import journal from '@sim/db/migrations/meta/_journal.json'
7-
import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
7+
import { readTestDatabaseUrl, readTestRedisUrl } from '@sim/db/testing/test-infrastructure'
88
import { generateId } from '@sim/utils/id'
9+
import Redis from 'ioredis'
910
import postgres from 'postgres'
1011
import { afterAll, beforeAll, describe, expect, it } from 'vitest'
1112

@@ -17,10 +18,19 @@ const client = postgres(url.toString(), { max: 1, onnotice: () => {} })
1718
const directory = await mkdtemp(join(tmpdir(), 'project-repair-test-'))
1819
const manifest = join(directory, 'manifest.json')
1920
const report = join(directory, 'report.json')
21+
const redisUrl = readTestRedisUrl()
22+
if (!redisUrl) throw new Error('Archive repair integration requires TEST_REDIS_URL')
23+
const cache = new Redis(redisUrl)
24+
const subscriber = new Redis(redisUrl)
25+
const mcpServerId = generateId()
26+
const cacheKey = `mcp:tools:workspace:env:server:${mcpServerId}`
27+
const published: unknown[] = []
28+
subscriber.on('message', (_channel, message) => published.push(JSON.parse(message)))
2029
const environment = {
2130
...process.env,
2231
DATABASE_URL: url.toString(),
2332
MIGRATION_DATABASE_URL: url.toString(),
33+
REDIS_URL: redisUrl,
2434
}
2535
const execute = promisify(execFile)
2636
const run = (command: string) =>
@@ -42,6 +52,7 @@ const run = (command: string) =>
4252
)
4353

4454
beforeAll(async () => {
55+
await subscriber.subscribe('mcp:workflow_tools_changed')
4556
await admin.unsafe(`CREATE DATABASE "${name}"`)
4657
await client.unsafe(
4758
'CREATE EXTENSION vector; CREATE EXTENSION btree_gin; CREATE EXTENSION pg_trgm'
@@ -79,6 +90,8 @@ beforeAll(async () => {
7990
}, 120000)
8091

8192
afterAll(async () => {
93+
await cache.del(cacheKey)
94+
await Promise.all([cache.quit(), subscriber.quit()])
8295
await client.end({ timeout: 2 })
8396
await admin.unsafe(`DROP DATABASE IF EXISTS "${name}" WITH (FORCE)`)
8497
await admin.end()
@@ -95,10 +108,16 @@ describe('Operator archive repair against the full compatible schema', () => {
95108
await client`INSERT INTO webhook (id,workflow_id,provider,path,provider_config) VALUES ('hook','flow','gitlab','fixture-hook','{}')`
96109
await client`INSERT INTO chat (id,workflow_id,user_id,identifier,title) VALUES ('chat','flow','owner','repair-chat','Chat')`
97110
await client`INSERT INTO workflow_deployment_version (id,workflow_id,version,state,is_active) VALUES ('deployment','flow',1,'{}',true)`
98-
await client`INSERT INTO workflow_mcp_server (id,workspace_id,created_by,name,is_public) VALUES ('server','env','owner','Server',true)`
99-
await client`INSERT INTO workflow_mcp_tool (id,server_id,workflow_id,tool_name) VALUES ('tool','server','flow','fixture')`
111+
await client`INSERT INTO workflow_mcp_server (id,workspace_id,created_by,name,is_public) VALUES (${mcpServerId},'env','owner','Server',true)`
112+
await client`INSERT INTO workflow_mcp_tool (id,server_id,workflow_id,tool_name) VALUES ('tool',${mcpServerId},'flow','fixture')`
113+
await client`INSERT INTO mcp_servers (id,workspace_id,created_by,name,transport) VALUES (${mcpServerId},'env','owner','Cached server','streamable-http')`
114+
await cache.set(cacheKey, JSON.stringify({ tools: [], expiry: Date.now() + 60000 }), 'EX', 60)
100115
await run('plan')
101116
await expect(run('repair')).rejects.toMatchObject({ code: 1 })
117+
expect(await cache.get(cacheKey)).toBeNull()
118+
await expect
119+
.poll(() => published, { timeout: 2000 })
120+
.toContainEqual({ serverId: mcpServerId, workspaceId: 'env' })
102121
expect(JSON.parse(await readFile(report, 'utf8'))).toMatchObject({
103122
status: 'failed',
104123
repairsCompleted: [],

‎apps/sim/lib/workspaces/lifecycle.ts‎

Lines changed: 23 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -144,27 +144,31 @@ export async function finishEnvironmentArchive(
144144
options: { strictExternalCleanup?: boolean } = {}
145145
): Promise<void> {
146146
const { workspaceId } = effects
147-
await mapWithConcurrency(effects.workflows, ARCHIVE_NOTIFICATION_CONCURRENCY, (row) =>
148-
finishWorkflowArchive(row.id, workspaceId, row.serverIds, {
149-
requestId,
150-
strictExternalCleanup: options.strictExternalCleanup,
151-
}).catch((error) =>
152-
options.strictExternalCleanup
153-
? Promise.reject(error)
154-
: logger.warn(`[${requestId}] Post-archive notification failed for workflow ${row.id}`, {
147+
try {
148+
await mapWithConcurrency(effects.workflows, ARCHIVE_NOTIFICATION_CONCURRENCY, (row) =>
149+
finishWorkflowArchive(row.id, workspaceId, row.serverIds, {
150+
requestId,
151+
strictExternalCleanup: options.strictExternalCleanup,
152+
}).catch((error) =>
153+
options.strictExternalCleanup
154+
? Promise.reject(error)
155+
: logger.warn(`[${requestId}] Post-archive notification failed for workflow ${row.id}`, {
156+
error,
157+
})
158+
)
159+
)
160+
} finally {
161+
await mcpService.clearCache(workspaceId).catch(() => undefined)
162+
if (mcpPubSub) {
163+
for (const serverId of effects.serverIds) {
164+
try {
165+
mcpPubSub.publishWorkflowToolsChanged({ serverId, workspaceId })
166+
} catch (error) {
167+
logger.warn(`[${requestId}] MCP tools-changed publish failed for server ${serverId}`, {
155168
error,
156169
})
157-
)
158-
)
159-
await mcpService.clearCache(workspaceId).catch(() => undefined)
160-
if (!mcpPubSub) return
161-
for (const serverId of effects.serverIds) {
162-
try {
163-
mcpPubSub.publishWorkflowToolsChanged({ serverId, workspaceId })
164-
} catch (error) {
165-
logger.warn(`[${requestId}] MCP tools-changed publish failed for server ${serverId}`, {
166-
error,
167-
})
170+
}
171+
}
168172
}
169173
}
170174
}

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

Lines changed: 15 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -125,7 +125,7 @@ async function main() {
125125
if (values.help || !positionals.length) {
126126
process.stdout.write(`Operator-controlled Project preparation; never runs automatically on deployment.\n
127127
From apps/sim: bun --no-env-file scripts/backfill-projects.ts <command> [options]\n
128-
plan Read-only discovery; atomically writes a new --manifest PATH (never overwrites).\nrepair Repair only archived environments listed in --manifest; requires --report PATH.\napply Assign reviewed families from --manifest; requires --report PATH.\nverify Fresh database validation; include --manifest and --report to verify repair completion.\nstatus Read --report; does not connect or resume.\n
128+
plan Read-only discovery; atomically writes a new --manifest PATH (never overwrites).\nrepair Repair only archived environments listed in --manifest; requires --report PATH.\napply Assign reviewed families from --manifest; requires --report PATH.\nverify Validate the database and repair completion; requires --manifest and --report.\nstatus Read --report; does not connect or resume.\n
129129
Writes require MIGRATION_DATABASE_URL pointing directly to the primary and --ack-release-drained.\nDeploy #8830 and verify old servers/workers drained first. Keep the reviewed manifest and report\nin durable private job storage. Review conflicts and repairs before repair/apply. Repair requires\nthe app runtime environment (DATABASE_URL must target the same database) for provider cleanup.\n
130130
Defaults: --batch-size 50 (1–50), --max-batches 100 (1–10000), --seconds 60 (1–3600), --pause-ms 100 (1–60000).\nA budget limits scheduling. Transactions use 3s statement limits and a 5s total limit on PG17+,\notherwise a 5s idle-transaction limit on PG16. Reuse the same report\nto resume; lost reports can be recreated safely from the same manifest. Changed families need a\nnew plan/report. Complete verify plus deployment evidence is required before #8590 enforcement.\nExit: 0 complete, 2 paused/incomplete, 1 failure. Never run db:migrate to launch this tool.\n`)
131131
return
@@ -207,23 +207,21 @@ Defaults: --batch-size 50 (1–50), --max-batches 100 (1–10000), --seconds 60
207207
return
208208
}
209209
if (command === 'verify') {
210+
if (!manifestPath || !reportPath)
211+
throw new Error('Verification requires both --manifest and --report')
210212
const counts = await verifyProjectBackfill(sql)
211-
if (manifestPath || reportPath) {
212-
if (!manifestPath || !reportPath)
213-
throw new Error('Repair verification needs both --manifest and --report')
214-
const manifest = manifestSchema.parse(await readJson(manifestPath))
215-
const report = progressSchema.parse(await readJson(reportPath))
216-
const hash = createHash('sha256').update(JSON.stringify(manifest)).digest('hex')
217-
if (
218-
manifest.databaseId !== databaseId ||
219-
report.databaseId !== databaseId ||
220-
report.manifestHash !== hash
221-
)
222-
throw new Error('Verification artifacts do not match this database and manifest')
223-
counts.pendingCleanup = manifest.repairs.filter(
224-
(repair) => !report.repairsCompleted.includes(repair.workspaceId)
225-
).length
226-
}
213+
const manifest = manifestSchema.parse(await readJson(manifestPath))
214+
const report = progressSchema.parse(await readJson(reportPath))
215+
const hash = createHash('sha256').update(JSON.stringify(manifest)).digest('hex')
216+
if (
217+
manifest.databaseId !== databaseId ||
218+
report.databaseId !== databaseId ||
219+
report.manifestHash !== hash
220+
)
221+
throw new Error('Verification artifacts do not match this database and manifest')
222+
counts.pendingCleanup = manifest.repairs.filter(
223+
(repair) => !report.repairsCompleted.includes(repair.workspaceId)
224+
).length
227225
logger.info('Project verification', { databaseId, ...counts })
228226
if (Object.values(counts).some((count) => count !== 0)) process.exitCode = 2
229227
return

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

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -124,7 +124,11 @@ export async function discoverProjectBackfill(
124124
else groups.set(root, [row])
125125
}
126126
for (const [rootId, members] of groups) {
127-
if (members.every((row) => row.projectId !== null)) continue
127+
if (
128+
members.every((row) => row.projectId !== null) &&
129+
new Set(members.map((row) => row.projectId)).size === 1
130+
)
131+
continue
128132
const family = { rootId, members }
129133
try {
130134
validateFamily(family)

‎packages/db/migrations/0405_project_membership_enforcement.sql‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -149,7 +149,10 @@ DECLARE
149149
target_id text;
150150
BEGIN
151151
FOR target_id IN SELECT DISTINCT id FROM unnest(target_ids) AS ids(id) WHERE id IS NOT NULL ORDER BY id LOOP
152-
PERFORM pg_advisory_xact_lock(hashtextextended('project:' || target_id, 0));
152+
-- BEFORE row triggers already hold row locks; waiting here could invert the application lock order.
153+
IF NOT pg_try_advisory_xact_lock(hashtextextended('project:' || target_id, 0)) THEN
154+
RAISE EXCEPTION 'Project is changing; retry the operation' USING ERRCODE = '55P03';
155+
END IF;
153156
END LOOP;
154157
END;
155158
$$;
@@ -163,12 +166,11 @@ BEGIN
163166
IF TG_TABLE_NAME = 'project' THEN
164167
IF TG_OP <> 'INSERT' THEN previous_id := OLD.id; END IF;
165168
IF TG_OP <> 'DELETE' THEN next_id := NEW.id; END IF;
166-
PERFORM pg_advisory_xact_lock(hashtextextended('project:' || coalesce(next_id, previous_id), 0));
167169
ELSE
168170
IF TG_OP <> 'INSERT' THEN previous_id := OLD.project_id; END IF;
169171
IF TG_OP <> 'DELETE' THEN next_id := NEW.project_id; END IF;
170-
PERFORM project_contract_lock_projects(ARRAY[previous_id, next_id]);
171172
END IF;
173+
PERFORM project_contract_lock_projects(ARRAY[previous_id, next_id]);
172174
IF TG_OP = 'DELETE' THEN RETURN OLD; END IF;
173175
RETURN NEW;
174176
END;

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

Lines changed: 65 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,19 @@ describe('Project expand/backfill/contract against PostgreSQL', () => {
260260
).toEqual(first)
261261
expect(await sql`SELECT id FROM project`).toHaveLength(101)
262262
await run('verify')
263+
for (const artifacts of [[], ['--manifest', manifestPath], ['--report', reportPath]]) {
264+
await expect(
265+
promisify(execFile)(
266+
'bun',
267+
['--no-env-file', 'scripts/backfill-projects.ts', 'verify', ...artifacts],
268+
{
269+
cwd: new URL('../../../apps/sim/', import.meta.url),
270+
env: { ...process.env, MIGRATION_DATABASE_URL: url, DATABASE_URL: url },
271+
timeout: 30000,
272+
}
273+
)
274+
).rejects.toMatchObject({ code: 1 })
275+
}
263276
const manifest = JSON.parse(await readFile(manifestPath, 'utf8'))
264277
manifest.databaseId = projectBackfillDatabaseId('postgres://localhost/wrong_test')
265278
await writeFile(manifestPath, JSON.stringify(manifest))
@@ -692,6 +705,44 @@ describe('Project expand/backfill/contract against PostgreSQL', () => {
692705
})
693706
})
694707

708+
it.each([
709+
['Project update', "UPDATE project SET name = 'Contending update'"],
710+
['Project delete', 'DELETE FROM project'],
711+
['environment update', "UPDATE workspace SET archived_at = now() WHERE id = 'child'"],
712+
['environment delete', "DELETE FROM workspace WHERE id = 'child'"],
713+
])('rejects a direct %s promptly when its Project is changing', async (_operation, statement) => {
714+
await database(async (sql) => {
715+
await seed(sql)
716+
const held = createDeferred<void>()
717+
const release = createDeferred<void>()
718+
const writer = sql.begin(async (tx) => {
719+
await tx`UPDATE workflow SET archived_at = now() WHERE workspace_id = 'root'`
720+
await tx`UPDATE workspace SET archived_at = now() WHERE id = 'root'`
721+
held.resolve()
722+
await release.promise
723+
})
724+
await held.promise
725+
try {
726+
await expect(
727+
sql.begin(async (tx) => {
728+
await tx`SET LOCAL statement_timeout = '500ms'`
729+
await tx.unsafe(statement)
730+
})
731+
).rejects.toSatisfy((error: unknown) => getPostgresErrorCode(error) === '55P03')
732+
} finally {
733+
release.resolve()
734+
await writer
735+
}
736+
expect(
737+
await sql`SELECT 1 FROM workspace WHERE id = 'child' AND archived_at IS NULL`
738+
).toHaveLength(1)
739+
expect(await sql`SELECT 1 FROM project WHERE name = 'Contending update'`).toHaveLength(0)
740+
expect(Object.values(await verifyProjectBackfill(sql)).every((count) => count === 0)).toBe(
741+
true
742+
)
743+
})
744+
})
745+
695746
it.each(['read committed', 'repeatable read'] as const)(
696747
'prevents concurrent last-environment removal under %s',
697748
async (isolation) => {
@@ -712,6 +763,7 @@ describe('Project expand/backfill/contract against PostgreSQL', () => {
712763
const [connection] =
713764
await tx`SELECT pg_backend_pid() AS pid, count(*) FROM workspace WHERE archived_at IS NULL`
714765
started.resolve(connection.pid)
766+
await tx`SELECT pg_advisory_xact_lock(hashtextextended('project:' || project_id, 0)) FROM workspace WHERE id = 'child'`
715767
await tx`UPDATE workspace SET archived_at = now() WHERE id = 'child'`
716768
})
717769
.then(
@@ -736,7 +788,7 @@ describe('Project expand/backfill/contract against PostgreSQL', () => {
736788
await first
737789
}
738790
const failure = await second
739-
expect(['23514', '40001', '40P01']).toContain(getPostgresErrorCode(failure))
791+
expect(['23514', '40001']).toContain(getPostgresErrorCode(failure))
740792
expect(await sql`SELECT 1 FROM workspace WHERE archived_at IS NULL`).toHaveLength(1)
741793
})
742794
}
@@ -745,14 +797,17 @@ describe('Project expand/backfill/contract against PostgreSQL', () => {
745797
it('refuses enforcement when an archived Project retains an active workflow', async () => {
746798
await database(async (sql) => {
747799
await sql`INSERT INTO workspace (id, name, owner_id) VALUES ('root', 'Root', 'owner')`
800+
await sql`INSERT INTO project (id, name, owner_id, archived_at) VALUES ('existing', 'Existing', 'owner', now())`
801+
await sql`UPDATE workspace SET project_id = 'existing', archived_at = now()`
748802
await sql`INSERT INTO workflow VALUES ('flow', 'root', NULL)`
749-
await sql`UPDATE workspace SET archived_at = now()`
750-
await sql`UPDATE project SET archived_at = now()`
751-
await expect(prepareAndEnforce(sql)).rejects.toSatisfy(
803+
await expect(applyMigration(sql, migration)).rejects.toSatisfy(
752804
(error: unknown) => getPostgresErrorCode(error) === '55000'
753805
)
806+
expect(await sql`SELECT to_regclass('project_workspace')::text AS name`).toEqual([
807+
{ name: 'project_workspace' },
808+
])
754809
await sql`UPDATE workflow SET archived_at = now()`
755-
await prepareAndEnforce(sql)
810+
await applyMigration(sql, migration)
756811
})
757812
})
758813
it('groups fork families, preserves partial assignments, and retains archived and detached Projects on replay', async () => {
@@ -893,6 +948,11 @@ describe('Project expand/backfill/contract against PostgreSQL', () => {
893948
}
894949
}
895950
const before = await sql`SELECT id, project_id FROM workspace ORDER BY id`
951+
if (scenario === 'split') {
952+
expect(await discoverProjectBackfill(sql, 'fixture')).toMatchObject({
953+
conflicts: [{ id: 'root', reason: 'Fork family spans Projects' }],
954+
})
955+
}
896956
await expect(prepareAndEnforce(sql)).rejects.toSatisfy(
897957
(error: unknown) =>
898958
getPostgresErrorCode(error) === (scenario === 'oversized' ? '54000' : '55000')

0 commit comments

Comments
 (0)