fix(profiles,users): serialize backfill setup with OIDC provisioning transactions

packages/profiles/AGENTS.md

@@ -90,10 +90,15 @@ import { smrtProfilesGenerateBioPrompt } from '@happyvertical/smrt-profiles';
   `@happyvertical/smrt-profiles/internal/oidc-provisioning` subpath instead of
   duplicating adapter probing, locking, or retry policy, and supply both exact
   issuer/subject and normalized-email lock keys in deterministic order. The
-  coordinator additionally serializes all SQLite/DuckDB provisioning
-  transactions per database URL because those adapters cannot overlap
-  unrelated root transactions safely and retries bounded PostgreSQL
-  deadlock/serialization failures. New OIDC Profiles use per-profile,
+  coordinator additionally serializes every root-handle statement it owns per
+  database URL on SQLite/DuckDB — shared `_smrt_backfills` initialization, the
+  provisioning transaction, and the post-commit rebind — because those adapters
+  multiplex one native connection and cannot overlap unrelated root
+  transactions safely. **Never overlap two statements on one such handle,
+  transaction-bound or not** — no `Promise.all` over reads, not even
+  primary-key rebinds — because DuckDB fails the losing prepared statement or
+  aborts the process outright. It retries
+  bounded PostgreSQL deadlock/serialization failures. New OIDC Profiles use per-profile,
   non-semantic slugs so duplicate display names never invoke natural-key upsert.
   Caller-owned transactions never execute `_smrt_backfills` DDL; paths that
   perform canonical email lookup or reservation require the table to already

packages/profiles/src/__tests__/oidc-provisioning-coordinator.test.ts

@@ -0,0 +1,217 @@
+import { getDatabase } from '@happyvertical/sql';
+import { describe, expect, it } from 'vitest';
+import { createProfileFromOidc } from '../auth/index.js';
+import { coordinateOidcProvisioning } from '../auth/oidcProvisioningCoordinator';
+import { ProfileTypeCollection } from '../collections/ProfileTypeCollection.js';
+import { backfillProfileEmailKeys } from '../migrations/backfillProfileEmailKeys.js';
+
+type DatabaseInterface = Awaited<ReturnType<typeof getDatabase>>;
+
+interface Deferred {
+  promise: Promise<void>;
+  resolve: () => void;
+}
+
+function createDeferred(): Deferred {
+  let resolve!: () => void;
+  const promise = new Promise<void>((settle) => {
+    resolve = settle;
+  });
+  return { promise, resolve };
+}
+
+/** Give a blocked flow every chance to reach its first statement. */
+async function settleEventLoop(): Promise<void> {
+  for (let tick = 0; tick < 5; tick += 1) {
+    await new Promise<void>((resolve) => setTimeout(resolve, 5));
+  }
+}
+
+describe('coordinateOidcProvisioning shared-connection ordering', () => {
+  /**
+   * #2352: SQLite and DuckDB root handles multiplex one native connection, so
+   * any coordinator statement issued while a concurrent flow holds that
+   * connection's transaction races it. DuckDB reports the loser as
+   * `Failed to execute prepared statement` or aborts the worker outright, and
+   * the shared `_smrt_backfills` initialization used to run before the adapter
+   * lock. Every statement the coordinator owns must run inside that lock.
+   */
+  it('issues no statement on a shared DuckDB root handle while another flow holds its provisioning transaction', async () => {
+    const db = (await getDatabase({
+      type: 'duckdb',
+      url: ':memory:',
+      __smrtSkipVitestSchemaPreparation: true,
+    })) as DatabaseInterface;
+    const statementsDuringTransaction: string[] = [];
+    let openTransactions = 0;
+
+    const observeRoot = (database: DatabaseInterface): DatabaseInterface =>
+      new Proxy(database, {
+        get(target, property, receiver) {
+          if (property === 'query') {
+            return async (sql: string, ...params: unknown[]) => {
+              if (openTransactions > 0) {
+                // Record instead of executing. Really racing the connection
+                // would corrupt it and hide the ordering defect behind an
+                // adapter error.
+                statementsDuringTransaction.push(
+                  sql.replace(/\s+/gu, ' ').trim(),
+                );
+                throw new Error(
+                  'A coordinator statement raced an open provisioning transaction.',
+                );
+              }
+              return target.query(sql, ...params);
+            };
+          }
+          if (property === 'transaction') {
+            const transaction = target.transaction;
+            if (!transaction) return undefined;
+            return async <T>(
+              callback: (tx: DatabaseInterface) => Promise<T>,
+            ): Promise<T> => {
+              openTransactions += 1;
+              try {
+                return await transaction.call(target, callback);
+              } finally {
+                openTransactions -= 1;
+              }
+            };
+          }
+          const value = Reflect.get(target, property, receiver);
+          return typeof value === 'function' ? value.bind(target) : value;
+        },
+      }) as DatabaseInterface;
+
+    const firstTransactionOpen = createDeferred();
+    const releaseFirstTransaction = createDeferred();
+    const coordinate = (
+      lockKey: string,
+      provision: () => Promise<string>,
+    ): Promise<string> =>
+      coordinateOidcProvisioning<string>({
+        db: observeRoot(db),
+        lockKeys: [lockKey],
+        isRaceConflict: () => false,
+        createTransactionError: (message, cause) =>
+          new Error(message, cause === undefined ? undefined : { cause }),
+        createConcurrencyError: (cause) =>
+          new Error(
+            'Concurrent provisioning did not converge.',
+            cause === undefined ? undefined : { cause },
+          ),
+        provision,
+      });
+
+    let first: Promise<string> | undefined;
+    let second: Promise<string> | undefined;
+    try {
+      first = coordinate('identity:first', async () => {
+        firstTransactionOpen.resolve();
+        await releaseFirstTransaction.promise;
+        return 'first';
+      });
+      first.catch(() => undefined);
+      // Racing the flow itself turns an early failure into that failure
+      // rather than a hook timeout waiting for a signal that never arrives.
+      await Promise.race([firstTransactionOpen.promise, first]);
+
+      // The second flow must block on the adapter lock rather than touch the
+      // connection the first flow's transaction owns. Mark it handled up
+      // front: when this guard catches a regression the flow rejects while
+      // nothing awaits it yet, and the unhandled rejection would drown the
+      // assertion that actually names the offending statement.
+      second = coordinate('identity:second', async () => 'second');
+      second.catch(() => undefined);
+      await settleEventLoop();
+      expect(statementsDuringTransaction).toEqual([]);
+
+      releaseFirstTransaction.resolve();
+      await expect(Promise.all([first, second])).resolves.toEqual([
+        'first',
+        'second',
+      ]);
+      expect(statementsDuringTransaction).toEqual([]);
+    } finally {
+      releaseFirstTransaction.resolve();
+      await Promise.allSettled([first, second]);
+      await db.close?.();
+    }
+  });
+
+  /**
+   * #2352: the ordering guard above only covers statements the coordinator
+   * itself owns. A single flow can just as easily overlap its own statements
+   * — the post-commit rebind used `Promise.all` over two or three primary-key
+   * reads — and on one native connection that is the same defect. Counting
+   * in-flight statements catches both without racing: an overlap is a
+   * structural property of the call, not a timing accident.
+   */
+  it('never overlaps two statements on one shared DuckDB connection during concurrent provisioning', async () => {
+    const db = (await getDatabase({
+      type: 'duckdb',
+      url: ':memory:',
+    })) as DatabaseInterface;
+    await ProfileTypeCollection.create({ db });
+    await backfillProfileEmailKeys(db);
+
+    let inFlight = 0;
+    let maxInFlight = 0;
+    const overlapped: string[] = [];
+    const observe = (database: DatabaseInterface): DatabaseInterface =>
+      new Proxy(database, {
+        get(target, property, receiver) {
+          if (property === 'query') {
+            return async (sql: string, ...params: unknown[]) => {
+              inFlight += 1;
+              maxInFlight = Math.max(maxInFlight, inFlight);
+              if (inFlight > 1) {
+                overlapped.push(sql.replace(/\s+/gu, ' ').trim());
+              }
+              try {
+                return await target.query(sql, ...params);
+              } finally {
+                inFlight -= 1;
+              }
+            };
+          }
+          if (property === 'transaction') {
+            const transaction = target.transaction;
+            if (!transaction) return undefined;
+            return async <T>(
+              callback: (tx: DatabaseInterface) => Promise<T>,
+            ): Promise<T> =>
+              transaction.call(target, (tx) => callback(observe(tx)));
+          }
+          const value = Reflect.get(target, property, receiver);
+          return typeof value === 'function' ? value.bind(target) : value;
+        },
+      }) as DatabaseInterface;
+
+    const sharedClaims = {
+      sub: 'overlap-guard-subject',
+      iss: 'https://auth.example.com',
+      email_verified: true,
+      name: 'Overlap Guard',
+    };
+    try {
+      await Promise.all([
+        createProfileFromOidc(
+          { ...sharedClaims, email: 'overlap-guard-first@example.com' },
+          'example',
+          { db: observe(db) },
+        ),
+        createProfileFromOidc(
+          { ...sharedClaims, email: 'overlap-guard-second@example.com' },
+          'example',
+          { db: observe(db) },
+        ),
+      ]);
+
+      expect(overlapped).toEqual([]);
+      expect(maxInFlight).toBe(1);
+    } finally {
+      await db.close?.();
+    }
+  });
+});

packages/profiles/src/auth/oidcProvisioningCoordinator.ts

@@ -82,15 +82,19 @@ export interface OidcProvisioningCoordinatorOptions<T> {
 export async function coordinateOidcProvisioning<T>(
   options: OidcProvisioningCoordinatorOptions<T>,
 ): Promise<T> {
-  // Establish the shared backfill table outside the provisioning transaction.
-  // Transaction-bound callers are handled below via savepoints and must not
-  // leak DDL into their caller's transaction.
   const rootBackfillTableReady =
     typeof options.db.beginTransaction === 'function';
-  if (rootBackfillTableReady) {
-    await new BackfillTracker({ db: options.db }).initialize();
-  }
   return withProvisioningLocks(options.db, options.lockKeys, async () => {
+    // Establish the shared backfill table outside the provisioning
+    // transaction but inside the coordinator locks. Transaction-bound callers
+    // are handled below via savepoints and must not leak DDL into their
+    // caller's transaction, and SQLite and DuckDB root handles multiplex one
+    // native connection: a statement issued here while a concurrent flow
+    // holds that connection's transaction races it, and DuckDB then fails the
+    // in-flight prepared statement outright (#2352).
+    if (rootBackfillTableReady) {
+      await new BackfillTracker({ db: options.db }).initialize();
+    }
     const result = await retryProvisioning(options, () =>
       withProvisioningTransaction(
         options.db,

packages/profiles/src/auth/resolveIdentity.ts

@@ -519,10 +519,13 @@ async function rebindOidcProfileResult<
   const [profile, oidcIdentity] = await withSystemContext(async () => {
     const profiles = await ProfileCollection.create({ db: rootDb });
     const identities = await OidcIdentityCollection.create({ db: rootDb });
-    return Promise.all([
-      profiles.get({ id: profileId }),
-      identities.get({ id: identityId }),
-    ]);
+    // Read one statement at a time. SQLite and DuckDB handles multiplex a
+    // single native connection, so overlapping statements on it fail the
+    // losing prepared statement or abort the process outright (#2352). These
+    // are primary-key reads, so the sequential cost is one extra round trip.
+    const rebound = await profiles.get({ id: profileId });
+    const reboundIdentity = await identities.get({ id: identityId });
+    return [rebound, reboundIdentity] as const;
   });
   if (!profile || !oidcIdentity) {
     throw new Error(

packages/users/AGENTS.md

@@ -295,9 +295,14 @@ the package root).
   private `oidc_profile_email_reservations.email_key`, `User.emailKey`, and the
   unique `User.profileId`; local callbacks acquire exact issuer/subject and normalized
   email locks in deterministic order so changed email claims also serialize.
-  SQLite and DuckDB callbacks additionally serialize transactions per database
-  URL because one adapter cannot safely overlap unrelated root transactions;
-  owner-authorized DuckDB callbacks use that same root-handle serialization;
+  SQLite and DuckDB callbacks additionally serialize every root-handle statement
+  the coordinator owns per database URL — `_smrt_backfills` initialization, the
+  transaction, and the post-commit rebind — because one adapter multiplexes a
+  single native connection and cannot safely overlap unrelated root
+  transactions. Never overlap two statements on one such handle, inside a
+  transaction or not: rebind and owner/email candidate reads are sequential,
+  never `Promise.all`. Owner-authorized DuckDB callbacks use that same
+  root-handle serialization;
   PostgreSQL deadlock and serialization errors use a bounded transaction retry.
   Newly provisioned Profiles use non-semantic per-profile slugs so equal IdP
   display names cannot trigger a natural-key upsert;

packages/users/src/collections/UserCollection.ts

@@ -640,12 +640,15 @@ export class UserCollection extends SmrtCollection<User> {
        LIMIT 2${lockClause}`,
       profileId,
     );
-    const owners = await Promise.all(
-      result.rows.map((row) =>
-        typeof row.id === 'string' ? this.get({ id: row.id }) : null,
-      ),
-    );
-    return owners.filter((owner): owner is User => owner !== null);
+    // Sequential by design: this runs on the caller's transaction handle, and
+    // SQLite/DuckDB multiplex one native connection per handle (#2352).
+    const owners: User[] = [];
+    for (const row of result.rows) {
+      if (typeof row.id !== 'string') continue;
+      const owner = await this.get({ id: row.id });
+      if (owner !== null && owner !== undefined) owners.push(owner);
+    }
+    return owners;
   }
 
   private async validateProfileOwnerAuthorization(
@@ -723,11 +726,12 @@ export class UserCollection extends SmrtCollection<User> {
       const users = await UserCollection.create({ db: rootDb });
       const profiles = await ProfileCollection.create({ db: rootDb });
       const identities = await OidcIdentityCollection.create({ db: rootDb });
-      const [user, profile, oidcIdentity] = await Promise.all([
-        users.get({ id: userId }),
-        profiles.get({ id: profileId }),
-        identities.get({ id: identityId }),
-      ]);
+      // Read one statement at a time. SQLite and DuckDB handles multiplex a
+      // single native connection, so overlapping statements on it fail the
+      // losing prepared statement or abort the process outright (#2352).
+      const user = await users.get({ id: userId });
+      const profile = await profiles.get({ id: profileId });
+      const oidcIdentity = await identities.get({ id: identityId });
       if (!user || !profile || !oidcIdentity) {
         throw new OidcProvisioningError(
           'concurrency_conflict',
@@ -805,12 +809,15 @@ export class UserCollection extends SmrtCollection<User> {
     for (const row of result.rows) {
       this.assertUserEmailKeyCurrent(row, email);
     }
-    const users = await Promise.all(
-      result.rows.map((row) =>
-        typeof row.id === 'string' ? this.get({ id: row.id }) : null,
-      ),
-    );
-    return users.filter((user): user is User => user !== null);
+    // Sequential by design: this runs on the caller's transaction handle, and
+    // SQLite/DuckDB multiplex one native connection per handle (#2352).
+    const matched: User[] = [];
+    for (const row of result.rows) {
+      if (typeof row.id !== 'string') continue;
+      const user = await this.get({ id: row.id });
+      if (user !== null && user !== undefined) matched.push(user);
+    }
+    return matched;
   }
 
   /** Require the deploy-time backfill marker before indexed identity reads. */