Apply Apex concurrency and historical ownership fixes to storage

Preserve reviewed privacy rules and published stack history.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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.
 

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();
   }
 }
 

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<void>> = [];
 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<unknown> | 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);
+
 });

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