From 364ff40aaba2cf78a1347153d31b661039df781e Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 6 Oct 2026 20:54:39 -0500 Subject: [PATCH] Apply Apex concurrency and historical ownership fixes to storage Preserve reviewed privacy rules and published stack history. Co-Authored-By: Paperclip --- doc/DATABASE.md | 7 +- packages/db/src/client.ts | 102 ++++++++++-------- ...rtbeat-run-issue-privacy-migration.test.ts | 54 +++++++++- .../migrations/0312_private_task_access.sql | 38 ++++--- 4 files changed, 137 insertions(+), 64 deletions(-) diff --git a/doc/DATABASE.md b/doc/DATABASE.md index f9b856501b9..faff78e6ecf 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -226,11 +226,14 @@ When authoring migrations or one-time backfills: - Split schema changes, index creation, and data backfill into separate phases so each step has clear locking and rollback behavior. - Treat the `check:migrations` CI gate as the enforcement backstop for these rules. If it flags a migration, rewrite the migration or add a suppression comment with the indexed predicate, batch bound, and reason the remaining scan is safe. -Private-task migrations `0311` and `0312` are explicitly allowlisted in the +Private-task migrations `0312` and `0313` are explicitly allowlisted in the Paperclip executor to run outside a file-wide transaction. Their idempotent keyset batches commit every 1,000 rows, and migration history is recorded only when all batches finish. The executor repairs invalid concurrent indexes on -retry. Bootstrap uses the same executor; other migrations remain transactional +retry. A reserved connection holds a session advisory lock across all batch +commits, so concurrent migrators recheck history only after the preceding runner +finishes. Privacy triggers are replaced atomically. Bootstrap uses the same +executor; other migrations remain transactional per file. Apply these migrations through `pnpm db:migrate` using a direct connection, before enabling the new server and UI. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 5961fceb904..d306e03db86 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -482,61 +482,75 @@ async function applyPendingMigrationsManually( journalEntries.map((entry) => [entry.fileName, normalizeFolderMillis(entry.folderMillis)]), ); - const sql = createUtilitySql(url); + const pool = createUtilitySql(url); try { - const { migrationTableSchema, columnNames } = await ensureMigrationJournalTable(sql); - const qualifiedTable = `${quoteIdentifier(migrationTableSchema)}.${quoteIdentifier(DRIZZLE_MIGRATIONS_TABLE)}`; + const sql = await pool.reserve(); + try { + // A session lock survives each batch COMMIT. Reserve the connection so + // pool rotation cannot release it while another migrator is waiting. + await sql`SELECT pg_advisory_lock(hashtextextended('paperclip:migrations', 0))`; + const { migrationTableSchema, columnNames } = await ensureMigrationJournalTable(sql); + const qualifiedTable = `${quoteIdentifier(migrationTableSchema)}.${quoteIdentifier(DRIZZLE_MIGRATIONS_TABLE)}`; - for (const migrationFile of orderedPendingMigrations) { - const migrationContent = await readMigrationFileContent(migrationFile); - const hash = createHash("sha256").update(migrationContent).digest("hex"); - const existingEntry = await migrationHistoryEntryExists( - sql, - qualifiedTable, - columnNames, - migrationFile, - hash, - ); - if (existingEntry) continue; - - const applyMigration = async () => { - for (const statement of splitMigrationStatements(migrationContent)) { - if (MIGRATIONS_WITH_BATCH_COMMITS.has(migrationFile)) { - // A cancelled concurrent build leaves an invalid index. IF NOT - // EXISTS alone would skip it on retry and journal an incomplete index. - const index = statement.replace(/^\s*--.*$/gm, "").trim() - .match(/^CREATE (?:UNIQUE )?INDEX CONCURRENTLY IF NOT EXISTS "([A-Za-z_][A-Za-z0-9_]*)"/i); - if (index) { - const invalid = await sql<{ invalid: boolean }[]>` - SELECT NOT i.indisvalid AS invalid FROM pg_index i - JOIN pg_class c ON c.oid = i.indexrelid - JOIN pg_namespace n ON n.oid = c.relnamespace - WHERE n.nspname = 'public' AND c.relname = ${index[1]}`; - if (invalid[0]?.invalid) { - await sql.unsafe(`DROP INDEX CONCURRENTLY ${quoteIdentifier("public")}.${quoteIdentifier(index[1])}`); - } - } - } - await sql.unsafe(statement); - } - - await recordMigrationHistoryEntry( + for (const migrationFile of orderedPendingMigrations) { + const migrationContent = await readMigrationFileContent(migrationFile); + const hash = createHash("sha256").update(migrationContent).digest("hex"); + const existingEntry = await migrationHistoryEntryExists( sql, qualifiedTable, columnNames, migrationFile, hash, - folderMillisByFileName.get(migrationFile) ?? Date.now(), ); - }; - if (MIGRATIONS_WITH_BATCH_COMMITS.has(migrationFile)) { - await applyMigration(); - } else { - await runInTransaction(sql, applyMigration); + if (existingEntry) continue; + + const applyMigration = async () => { + for (const statement of splitMigrationStatements(migrationContent)) { + if (MIGRATIONS_WITH_BATCH_COMMITS.has(migrationFile)) { + // A cancelled concurrent build leaves an invalid index. IF NOT + // EXISTS alone would skip it on retry and journal an incomplete index. + const index = statement.replace(/^\s*--.*$/gm, "").trim() + .match(/^CREATE (?:UNIQUE )?INDEX CONCURRENTLY IF NOT EXISTS "([A-Za-z_][A-Za-z0-9_]*)"/i); + if (index) { + const invalid = await sql<{ invalid: boolean }[]>` + SELECT NOT i.indisvalid AS invalid FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = 'public' AND c.relname = ${index[1]}`; + if (invalid[0]?.invalid) { + await sql.unsafe(`DROP INDEX CONCURRENTLY ${quoteIdentifier("public")}.${quoteIdentifier(index[1])}`); + } + } + } + await sql.unsafe(statement); + } + + await recordMigrationHistoryEntry( + sql, + qualifiedTable, + columnNames, + migrationFile, + hash, + folderMillisByFileName.get(migrationFile) ?? Date.now(), + ); + }; + if (MIGRATIONS_WITH_BATCH_COMMITS.has(migrationFile)) { + await applyMigration(); + } else { + await runInTransaction(sql, applyMigration); + } } + } finally { + try { + await sql`SELECT pg_advisory_unlock(hashtextextended('paperclip:migrations', 0))`; + } catch { + // Preserve the migration error if the session died; PostgreSQL releases + // its lock on disconnect, and pool.end() closes any remaining connection. + } + sql.release(); } } finally { - await sql.end(); + await pool.end(); } } diff --git a/packages/db/src/heartbeat-run-issue-privacy-migration.test.ts b/packages/db/src/heartbeat-run-issue-privacy-migration.test.ts index 00a8ae78afb..856a05cb135 100644 --- a/packages/db/src/heartbeat-run-issue-privacy-migration.test.ts +++ b/packages/db/src/heartbeat-run-issue-privacy-migration.test.ts @@ -11,6 +11,7 @@ import { const MIGRATION_FILE = "0312_private_task_access.sql"; const cleanups: Array<() => Promise> = []; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +if (!embeddedPostgresSupport.supported) console.warn(`Private-task migration checks unavailable: ${embeddedPostgresSupport.reason}`); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; async function migrationHash() { @@ -71,7 +72,22 @@ describeEmbeddedPostgres("heartbeat run issue privacy migration", () => { await sql`INSERT INTO project_access_members (company_id, project_id, subject_type, subject_id) VALUES (${companyId}, ${legacyProjectId}, 'user', 'ordinary-reader')`; + const agentProjectId = randomUUID(), runProjectId = randomUUID(), preferredProjectId = randomUUID(); + await sql`UPDATE heartbeat_runs SET responsible_user_id = 'run-owner' WHERE id = ${secondRunId}`; + await sql`INSERT INTO projects (id, company_id, name, visibility) VALUES + (${agentProjectId}, ${companyId}, 'Agent API project', 'private'), + (${runProjectId}, ${companyId}, 'Run-backed project', 'private'), + (${preferredProjectId}, ${companyId}, 'Recorded owner project', 'private')`; + await sql`INSERT INTO activity_log (company_id, actor_type, actor_id, responsible_user_id, run_id, action, entity_type, entity_id) VALUES + (${companyId}, 'agent', ${agentId}, 'activity-owner', NULL, 'project.created', 'project', ${agentProjectId}), + (${companyId}, 'agent', ${agentId}, NULL, ${secondRunId}, 'project.created', 'project', ${runProjectId}), + (${companyId}, 'agent', ${agentId}, 'activity-owner', ${secondRunId}, 'project.created', 'project', ${preferredProjectId})`; + await applyPendingMigrations(database.connectionString); + for (const [projectId, owner] of [[agentProjectId, 'activity-owner'], [runProjectId, 'run-owner'], [preferredProjectId, 'activity-owner']]) { + expect(await sql`SELECT privacy_owner_user_id FROM projects WHERE id = ${projectId}`).toEqual([{ privacy_owner_user_id: owner }]); + expect(await sql`SELECT subject_id FROM project_access_members WHERE project_id = ${projectId}`).toEqual([{ subject_id: owner }]); + } const [legacyProject] = await sql`SELECT privacy_owner_user_id FROM projects WHERE id = ${legacyProjectId}`; expect(legacyProject.privacy_owner_user_id).toBe('original-owner'); expect(await sql`SELECT subject_id FROM project_access_members WHERE project_id = ${legacyProjectId} ORDER BY subject_id`) @@ -200,6 +216,7 @@ describeEmbeddedPostgres("heartbeat run issue privacy migration", () => { await sql`DELETE FROM drizzle.__drizzle_migrations WHERE hash = ${hash}`; await sql`SELECT pg_advisory_lock(736319937)`; const migration = applyPendingMigrations(database.connectionString).then(() => null, error => error); + let contender: Promise | undefined; try { // The second batch is blocked while another session can see the first // committed batch and write a row whose lock has already been released. @@ -208,6 +225,15 @@ describeEmbeddedPostgres("heartbeat run issue privacy migration", () => { WHERE company_id = ${companyId} AND scope_kind = 'issue'`; return row.n; }, { timeout: 10_000 }).toBe(1000); + // The session lock survives batch commits and blocks another migrator. + contender = applyPendingMigrations(database.connectionString).then(() => null, error => error); + await expect.poll(async () => { + const [row] = await sql`SELECT count(*)::int AS n FROM pg_stat_activity + WHERE datname = current_database() AND wait_event = 'advisory' + AND strpos(query, 'paperclip:migrations') > 0`; + return row.n; + }, { timeout: 10_000 }).toBe(1); + expect(await sql`SELECT hash FROM drizzle.__drizzle_migrations WHERE hash = ${hash}`).toHaveLength(0); await sql`SET lock_timeout = '500ms'`; await sql`UPDATE heartbeat_runs SET context_snapshot = context_snapshot || '{"probe":"visible"}'::jsonb WHERE id = ${runId(1)}`; @@ -216,12 +242,15 @@ describeEmbeddedPostgres("heartbeat run issue privacy migration", () => { } finally { await sql`SELECT pg_advisory_unlock(736319937)`; await sql`RESET lock_timeout`; + await migration; + await contender; } expect(await migration).toMatchObject({ message: "fixture interrupted second batch" }); + expect(await contender).toMatchObject({ message: "fixture interrupted second batch" }); expect(await sql`SELECT hash FROM drizzle.__drizzle_migrations WHERE hash = ${hash}`).toHaveLength(0); await sql`DROP TRIGGER zzz_interrupt_privacy_batch_fixture ON heartbeat_runs`; await sql`DROP FUNCTION interrupt_privacy_batch_fixture()`; - await applyPendingMigrations(database.connectionString); + await Promise.all([applyPendingMigrations(database.connectionString), applyPendingMigrations(database.connectionString)]); const [finished] = await sql`SELECT count(*)::int AS n FROM heartbeat_runs WHERE company_id = ${companyId} AND scope_kind = 'issue' AND issue_id = ${issueId}`; expect(finished.n).toBe(1001); @@ -230,4 +259,27 @@ describeEmbeddedPostgres("heartbeat run issue privacy migration", () => { expect(await sql`SELECT hash FROM drizzle.__drizzle_migrations WHERE hash = ${hash}`).toHaveLength(1); }, 30_000); + it("keeps all privacy triggers installed when replacement is interrupted", async () => { + const database = await startEmbeddedPostgresTestDatabase("paperclip-privacy-trigger-atomic-"); + cleanups.push(database.cleanup); + const sql = postgres(database.connectionString, { max: 1, onnotice: () => {} }); + cleanups.push(async () => sql.end()); + const content = await fs.promises.readFile(new URL(`./migrations/${MIGRATION_FILE}`, import.meta.url), "utf8"); + const replacements = content.split("--> statement-breakpoint").filter(statement => /DROP TRIGGER IF EXISTS/.test(statement)); + expect(replacements).toHaveLength(4); + for (const statement of replacements) { + const name = statement.match(/DROP TRIGGER IF EXISTS ([A-Za-z_]+)/)![1]; + const interrupted = statement.replace("CREATE TRIGGER", "RAISE EXCEPTION 'fixture interrupted trigger replacement';\n CREATE TRIGGER"); + await expect(sql.unsafe(interrupted)).rejects.toThrow("fixture interrupted trigger replacement"); + expect(await sql`SELECT tgname FROM pg_trigger WHERE tgname = ${name} AND NOT tgisinternal`).toEqual([{ tgname: name }]); + } + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await sql`INSERT INTO companies (id, name, issue_prefix) VALUES (${companyId}, 'Atomic triggers', 'ATR')`; + await sql`INSERT INTO agents (id, company_id, name, adapter_type) VALUES (${agentId}, ${companyId}, 'Runner', 'process')`; + await sql`INSERT INTO issues (id, company_id, title, visibility) VALUES (${issueId}, ${companyId}, 'Private', 'private')`; + expect(await sql`INSERT INTO heartbeat_runs (company_id, agent_id, status, context_snapshot) + VALUES (${companyId}, ${agentId}, 'succeeded', ${sql.json({ taskId: issueId })}) RETURNING issue_id, scope_kind`) + .toEqual([{ issue_id: issueId, scope_kind: "issue" }]); + }, 30_000); + }); diff --git a/packages/db/src/migrations/0312_private_task_access.sql b/packages/db/src/migrations/0312_private_task_access.sql index 10470881f0d..480a3387400 100644 --- a/packages/db/src/migrations/0312_private_task_access.sql +++ b/packages/db/src/migrations/0312_private_task_access.sql @@ -132,10 +132,11 @@ BEGIN RETURN NEW; END $$; --> statement-breakpoint -DROP TRIGGER IF EXISTS heartbeat_runs_set_scope_kind ON heartbeat_runs; ---> statement-breakpoint -CREATE TRIGGER heartbeat_runs_set_scope_kind BEFORE INSERT OR UPDATE OF issue_id, native_issue_id, context_snapshot, scope_kind - ON heartbeat_runs FOR EACH ROW EXECUTE FUNCTION set_heartbeat_run_scope_kind(); +DO $$ BEGIN + DROP TRIGGER IF EXISTS heartbeat_runs_set_scope_kind ON heartbeat_runs; + CREATE TRIGGER heartbeat_runs_set_scope_kind BEFORE INSERT OR UPDATE OF issue_id, native_issue_id, context_snapshot, scope_kind + ON heartbeat_runs FOR EACH ROW EXECUTE FUNCTION set_heartbeat_run_scope_kind(); +END $$; --> statement-breakpoint -- Indexed primary-key keyset batches visit each historical run once. Missing -- source tasks become issue-scoped tombstones, never company-visible run content. @@ -180,10 +181,11 @@ BEGIN RETURN NEW; END $$; --> statement-breakpoint -DROP TRIGGER IF EXISTS execution_workspaces_privacy_sources ON execution_workspaces; ---> statement-breakpoint -CREATE TRIGGER execution_workspaces_privacy_sources BEFORE INSERT OR UPDATE OF source_issue_id, metadata - ON execution_workspaces FOR EACH ROW EXECUTE FUNCTION retain_workspace_privacy_sources(); +DO $$ BEGIN + DROP TRIGGER IF EXISTS execution_workspaces_privacy_sources ON execution_workspaces; + CREATE TRIGGER execution_workspaces_privacy_sources BEFORE INSERT OR UPDATE OF source_issue_id, metadata + ON execution_workspaces FOR EACH ROW EXECUTE FUNCTION retain_workspace_privacy_sources(); +END $$; --> statement-breakpoint CREATE OR REPLACE FUNCTION record_issue_workspace_privacy_source() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN @@ -194,10 +196,11 @@ BEGIN RETURN NEW; END $$; --> statement-breakpoint -DROP TRIGGER IF EXISTS issues_record_workspace_privacy_source ON issues; ---> statement-breakpoint -CREATE TRIGGER issues_record_workspace_privacy_source AFTER INSERT OR UPDATE OF execution_workspace_id - ON issues FOR EACH ROW EXECUTE FUNCTION record_issue_workspace_privacy_source(); +DO $$ BEGIN + DROP TRIGGER IF EXISTS issues_record_workspace_privacy_source ON issues; + CREATE TRIGGER issues_record_workspace_privacy_source AFTER INSERT OR UPDATE OF execution_workspace_id + ON issues FOR EACH ROW EXECUTE FUNCTION record_issue_workspace_privacy_source(); +END $$; --> statement-breakpoint DO $$ DECLARE cursor_id uuid := '00000000-0000-0000-0000-000000000000'; next_id uuid; BEGIN LOOP @@ -227,10 +230,11 @@ BEGIN END; $$ LANGUAGE plpgsql; --> statement-breakpoint -DROP TRIGGER IF EXISTS workspace_operations_keep_privacy_sources ON workspace_operations; ---> statement-breakpoint -CREATE TRIGGER workspace_operations_keep_privacy_sources BEFORE INSERT OR UPDATE ON workspace_operations -FOR EACH ROW EXECUTE FUNCTION paperclip_keep_operation_privacy_sources(); +DO $$ BEGIN + DROP TRIGGER IF EXISTS workspace_operations_keep_privacy_sources ON workspace_operations; + CREATE TRIGGER workspace_operations_keep_privacy_sources BEFORE INSERT OR UPDATE ON workspace_operations + FOR EACH ROW EXECUTE FUNCTION paperclip_keep_operation_privacy_sources(); +END $$; --> statement-breakpoint DO $$ DECLARE cursor_id uuid := '00000000-0000-0000-0000-000000000000'; batch_ids uuid[]; @@ -259,7 +263,7 @@ BEGIN ) batch; EXIT WHEN batch_ids IS NULL; UPDATE projects p SET privacy_owner_user_id = COALESCE(p.personal_owner_user_id, ( - SELECT CASE WHEN a.actor_type = 'user' THEN a.actor_id ELSE r.responsible_user_id END + SELECT CASE WHEN a.actor_type = 'user' THEN a.actor_id ELSE coalesce(a.responsible_user_id, r.responsible_user_id) END FROM activity_log a LEFT JOIN heartbeat_runs r ON r.id = a.run_id AND r.company_id = a.company_id WHERE a.company_id = p.company_id AND a.entity_type = 'project' AND a.entity_id = p.id::text AND a.action = 'project.created' ORDER BY a.created_at, a.id LIMIT 1