migration-runner: extend advisory-lock guard to webhook-triggers, routines, workflow-deploy-source

docs/package-migrations.md

@@ -72,7 +72,8 @@ package onto the transactional pattern — see below):
 1. **Self-contained, transactional** — `@corbits/chat`, `@corbits/notify`,
    `@corbits/webhook-triggers`, `@corbits/routines`, `@corbits/insights`,
    `@corbits/skills`, `@corbits/bench`, `@corbits/preferences`,
-   `@corbits/inference-catalog`, `@corbits/evals`, `@corbits/access-policy`.
+   `@corbits/inference-catalog`, `@corbits/evals`, `@corbits/access-policy`,
+   `@corbits/workflow-deploy-source`.
    The package's `src/migrations.ts` owns only a literal `{ name, sql }[]`
    array and a thin `applyXMigrations(databaseUrl)` wrapper; the mechanics —
    schema/ledger bootstrap, the transactional apply loop, and the advisory

packages/migration-runner/README.md

@@ -25,11 +25,13 @@ name rather than the name itself, so two distinct table names could in
 principle hash to the same key — two packages would then serialize their
 boot-time migrations against each other instead of running in parallel, a
 liveness cost (one waits its turn) and never a correctness one (each still
-applies to its own schema and ledger). The six current ledger tables
+applies to its own schema and ledger). The nine current ledger tables
 (`access_policy_migrations`, `bench_migrations`, `evals_migrations`,
 `inference_catalog_migrations`, `insights_migrations`,
-`preferences_migrations`) do not collide — verified against a live
-`hashtext()`. Confirm the same before naming a seventh.
+`preferences_migrations`, `webhook_triggers_migrations`,
+`routine_migrations`, `workflow_deploy_source_migrations`) do not collide —
+verified pairwise against a live PostgreSQL 17.11 `hashtext()`. Confirm the
+same before naming a tenth.
 
 ## What it does not change
 

packages/routines/package.json

@@ -17,6 +17,7 @@
   },
   "dependencies": {
     "@corbits/folded-run-one-shot": "workspace:*",
+    "@corbits/migration-runner": "workspace:*",
     "@corbits/slug": "workspace:*",
     "cronstrue": "^3.24.0",
     "@corbits/workflow-catalog": "workspace:*",

packages/routines/src/migrations.ts

@@ -4,13 +4,17 @@
 // this package's half of the "mount + migrations is the entire install
 // story" install contract. Bookkeeping is its own ledger table, never
 // the platform's drizzle journal, so this package's migration history
-// stays extractable on its own.
-import postgres from "postgres";
+// stays extractable on its own. Mechanics (schema/ledger bootstrap,
+// transactional apply, the advisory lock across concurrent hub
+// replicas) live in @corbits/migration-runner — this file owns only
+// the domain SQL.
+import {
+  applyPackageMigrations,
+  type ApplyPackageMigrationsReport,
+  type PackageMigration,
+} from "@corbits/migration-runner";
 
-export interface RoutineMigration {
-  name: string;
-  sql: string;
-}
+export type RoutineMigration = PackageMigration;
 
 export const routineMigrations: readonly RoutineMigration[] = [
   {
@@ -119,18 +123,7 @@ export const routineMigrations: readonly RoutineMigration[] = [
 const SCHEMA = "routines";
 const LEDGER_TABLE = "routine_migrations";
 
-function quoteIdentifier(name: string): string {
-  return `"${name.replace(/"/g, '""')}"`;
-}
-
-function quoteQualified(schema: string, name: string): string {
-  return `${quoteIdentifier(schema)}.${quoteIdentifier(name)}`;
-}
-
-export interface ApplyRoutineMigrationsReport {
-  applied: string[];
-  alreadyApplied: string[];
-}
+export type ApplyRoutineMigrationsReport = ApplyPackageMigrationsReport;
 
 /**
  * Apply `routineMigrations` against `databaseUrl`, idempotently: a
@@ -141,40 +134,11 @@ export interface ApplyRoutineMigrationsReport {
 export async function applyRoutineMigrations(
   databaseUrl: string,
 ): Promise<ApplyRoutineMigrationsReport> {
-  const sql = postgres(databaseUrl, { max: 1, onnotice: () => undefined });
-  try {
-    await sql.unsafe(`CREATE SCHEMA IF NOT EXISTS ${quoteIdentifier(SCHEMA)}`);
-
-    await sql.unsafe(
-      `CREATE TABLE IF NOT EXISTS ${quoteQualified(SCHEMA, LEDGER_TABLE)} (` +
-        `name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`,
-    );
-    const rows = await sql.unsafe(
-      `SELECT name FROM ${quoteQualified(SCHEMA, LEDGER_TABLE)}`,
-    );
-    const alreadyApplied = new Set(rows.map((row) => String(row["name"])));
-    const applied: string[] = [];
-    for (const migration of routineMigrations) {
-      if (alreadyApplied.has(migration.name)) continue;
-      try {
-        await sql.begin(async (tx) => {
-          await tx.unsafe(migration.sql);
-          await tx.unsafe(
-            `INSERT INTO ${quoteQualified(SCHEMA, LEDGER_TABLE)} (name) VALUES ($1)`,
-            [migration.name],
-          );
-        });
-        applied.push(migration.name);
-      } catch (error) {
-        throw new Error(
-          `@corbits/routines migration ${JSON.stringify(migration.name)} failed: ` +
-            `${error instanceof Error ? error.message : String(error)}`,
-          { cause: error },
-        );
-      }
-    }
-    return { applied, alreadyApplied: [...alreadyApplied] };
-  } finally {
-    await sql.end();
-  }
+  return applyPackageMigrations({
+    databaseUrl,
+    schema: SCHEMA,
+    ledgerTable: LEDGER_TABLE,
+    migrations: routineMigrations,
+    packageLabel: "@corbits/routines",
+  });
 }

packages/routines/test/migrations.test.ts

@@ -126,3 +126,71 @@ describeIfDb("applyRoutineMigrations", () => {
     }
   });
 });
+
+// Separate database from the suites above: two replicas racing the same
+// ledger must not collide with the idempotency test's own already-applied
+// rows, and must start from a schema that has never seen this migration
+// set before.
+describeIfDb("applyRoutineMigrations concurrency", () => {
+  const scratchUrl = scratchUrlFor(
+    databaseUrl ?? "postgres://localhost:5432/unused",
+  ).replace("_routine_migrations_test", "_routine_migrations_concurrent_test");
+  const scratchDatabase = new URL(scratchUrl).pathname.replace(/^\//, "");
+
+  beforeAll(async () => {
+    const maintenanceUrl = new URL(scratchUrl);
+    maintenanceUrl.pathname = "/postgres";
+    const maintenance = postgres(maintenanceUrl.toString(), {
+      max: 1,
+      onnotice: () => undefined,
+    });
+    try {
+      await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`);
+      await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`);
+    } finally {
+      await maintenance.end();
+    }
+  }, 20000);
+
+  afterAll(async () => {
+    const maintenanceUrl = new URL(scratchUrl);
+    maintenanceUrl.pathname = "/postgres";
+    const maintenance = postgres(maintenanceUrl.toString(), {
+      max: 1,
+      onnotice: () => undefined,
+    });
+    try {
+      await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`);
+    } finally {
+      await maintenance.end();
+    }
+  }, 20000);
+
+  test("two replicas booting concurrently both complete without either crashing on a duplicate ledger insert", async () => {
+    const [first, second] = await Promise.all([
+      applyRoutineMigrations(scratchUrl),
+      applyRoutineMigrations(scratchUrl),
+    ]);
+
+    const appliedNames = [...first.applied, ...second.applied].sort();
+    expect(new Set(appliedNames).size).toBe(appliedNames.length);
+
+    const sql = postgres(scratchUrl, { max: 1, onnotice: () => undefined });
+    try {
+      const ledgerRows = await sql.unsafe(
+        `SELECT name FROM "routines"."routine_migrations" ORDER BY name`,
+      );
+      const ledgerNames = ledgerRows.map((row) => String(row["name"]));
+      expect(new Set(ledgerNames).size).toBe(ledgerNames.length);
+      expect(
+        [
+          ...appliedNames,
+          ...first.alreadyApplied,
+          ...second.alreadyApplied,
+        ].sort(),
+      ).toEqual([...ledgerNames, ...ledgerNames].sort());
+    } finally {
+      await sql.end();
+    }
+  }, 10000);
+});

packages/webhook-triggers/package.json

@@ -16,6 +16,7 @@
   "dependencies": {
     "@corbits/error-sink": "workspace:*",
     "@corbits/folded-runs": "workspace:*",
+    "@corbits/migration-runner": "workspace:*",
     "@intx/db": "workspace:*",
     "@intx/hub-api": "workspace:*",
     "@intx/hub-common": "0.3.0",

packages/webhook-triggers/src/migrations.ts

@@ -5,13 +5,17 @@
 // install story, mirroring `@corbits/chat`'s `migrations.ts`.
 // Bookkeeping is deliberately its own table, never the platform's
 // drizzle journal, so this package's migration history stays
-// extractable on its own.
-import postgres from "postgres";
+// extractable on its own. Mechanics (schema/ledger bootstrap,
+// transactional apply, the advisory lock across concurrent hub
+// replicas) live in @corbits/migration-runner — this file owns only
+// the domain SQL.
+import {
+  applyPackageMigrations,
+  type ApplyPackageMigrationsReport,
+  type PackageMigration,
+} from "@corbits/migration-runner";
 
-export interface WebhookTriggersMigration {
-  name: string;
-  sql: string;
-}
+export type WebhookTriggersMigration = PackageMigration;
 
 /**
  * Ordered, explicit migration set for the table declared in
@@ -47,26 +51,15 @@ export const webhookTriggersMigrations: readonly WebhookTriggersMigration[] = [
   },
 ];
 
-// Bookkeeping table for this package's own migrations. Named
-// distinctly from the platform's setup ledger and from any drizzle
-// journal, so extracting `@corbits/webhook-triggers` out of this repo
-// never has to disentangle its history from the platform's. Lives in
-// the package's own `webhook_triggers` schema, like the table it owns.
+// Named distinctly from the platform's setup ledger and from any
+// drizzle journal, so extracting `@corbits/webhook-triggers` out of
+// this repo never has to disentangle its history from the platform's.
+// Lives in the package's own `webhook_triggers` schema, like the
+// table it owns.
 const SCHEMA = "webhook_triggers";
 const LEDGER_TABLE = "webhook_triggers_migrations";
 
-function quoteIdentifier(name: string): string {
-  return `"${name.replace(/"/g, '""')}"`;
-}
-
-function quoteQualified(schema: string, name: string): string {
-  return `${quoteIdentifier(schema)}.${quoteIdentifier(name)}`;
-}
-
-export interface ApplyWebhookTriggersMigrationsReport {
-  applied: string[];
-  alreadyApplied: string[];
-}
+export type ApplyWebhookTriggersMigrationsReport = ApplyPackageMigrationsReport;
 
 /**
  * Apply `webhookTriggersMigrations` against `databaseUrl`,
@@ -78,40 +71,11 @@ export interface ApplyWebhookTriggersMigrationsReport {
 export async function applyWebhookTriggersMigrations(
   databaseUrl: string,
 ): Promise<ApplyWebhookTriggersMigrationsReport> {
-  const sql = postgres(databaseUrl, { max: 1, onnotice: () => undefined });
-  try {
-    await sql.unsafe(`CREATE SCHEMA IF NOT EXISTS ${quoteIdentifier(SCHEMA)}`);
-
-    await sql.unsafe(
-      `CREATE TABLE IF NOT EXISTS ${quoteQualified(SCHEMA, LEDGER_TABLE)} (` +
-        `name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`,
-    );
-    const rows = await sql.unsafe(
-      `SELECT name FROM ${quoteQualified(SCHEMA, LEDGER_TABLE)}`,
-    );
-    const alreadyApplied = new Set(rows.map((row) => String(row["name"])));
-    const applied: string[] = [];
-    for (const migration of webhookTriggersMigrations) {
-      if (alreadyApplied.has(migration.name)) continue;
-      try {
-        await sql.begin(async (tx) => {
-          await tx.unsafe(migration.sql);
-          await tx.unsafe(
-            `INSERT INTO ${quoteQualified(SCHEMA, LEDGER_TABLE)} (name) VALUES ($1)`,
-            [migration.name],
-          );
-        });
-        applied.push(migration.name);
-      } catch (error) {
-        throw new Error(
-          `@corbits/webhook-triggers migration ${JSON.stringify(migration.name)} failed: ` +
-            `${error instanceof Error ? error.message : String(error)}`,
-          { cause: error },
-        );
-      }
-    }
-    return { applied, alreadyApplied: [...alreadyApplied] };
-  } finally {
-    await sql.end();
-  }
+  return applyPackageMigrations({
+    databaseUrl,
+    schema: SCHEMA,
+    ledgerTable: LEDGER_TABLE,
+    migrations: webhookTriggersMigrations,
+    packageLabel: "@corbits/webhook-triggers",
+  });
 }