fix(platform): queue serializable task-comment writes per task

packages/shared/src/db/serializable.test.ts

@@ -6,6 +6,7 @@ import {
   markRetryQueueKey,
   RETRY_QUEUE_LOCK_CLASS,
   retryQueueKeyOf,
+  retryQueueKeysOf,
   transactSerializable,
   type SerializableTransactionRunner,
 } from './serializable.ts';
@@ -138,7 +139,10 @@ interface Statement {
  * `unsafe` both log the statement; `unsafe` answers with the command tag a
  * server would (overridable for COMMIT), `release` counts.
  */
-function createReserved(commitAnswer = 'COMMIT'): {
+function createReserved(
+  commitAnswer = 'COMMIT',
+  failOn?: (text: string, values: unknown[]) => Error | undefined,
+): {
   reserved: ReservedSql;
   statements: Statement[];
   released: () => number;
@@ -155,7 +159,10 @@ function createReserved(commitAnswer = 'COMMIT'): {
       .replace(/\s+/g, ' ')
       .trim();
     statements.push({ text, values });
-    return Promise.resolve([]);
+    const failure = failOn?.(text, values);
+    return failure === undefined
+      ? Promise.resolve([])
+      : Promise.reject(failure);
   };
   tag.unsafe = (text: string) => {
     statements.push({ text, values: [] });
@@ -181,7 +188,10 @@ function createReserved(commitAnswer = 'COMMIT'): {
  */
 function createQueueRunner(
   failures: Error[],
-  options: { commitAnswer?: string } = {},
+  options: {
+    commitAnswer?: string;
+    failOn?: (text: string, values: unknown[]) => Error | undefined;
+  } = {},
 ): {
   runner: SerializableTransactionRunner;
   beginAttempts: () => number;
@@ -195,7 +205,7 @@ function createQueueRunner(
       callback: (tx: TransactionSql) => Promise<T>,
     ): Promise<T> => runner.begin(isolation, callback),
     reserve: () => {
-      const reservation = createReserved(options.commitAnswer);
+      const reservation = createReserved(options.commitAnswer, options.failOn);
       reservations.push(reservation);
       return Promise.resolve(reservation.reserved);
     },
@@ -383,3 +393,131 @@ describe('transactSerializable — retry queues', () => {
     ]);
   });
 });
+
+describe('markRetryQueueKey — nested keys', () => {
+  it('prepends an outer mark so the list reads outermost first', () => {
+    const error = markRetryQueueKey(
+      markRetryQueueKey(sqlstateError('40001'), 'audit-chain:org_1'),
+      'task-comment:t_1',
+    );
+    expect(retryQueueKeysOf(error)).toEqual([
+      'task-comment:t_1',
+      'audit-chain:org_1',
+    ]);
+    expect(retryQueueKeyOf(error)).toBe('task-comment:t_1');
+  });
+
+  it('never adds a key twice', () => {
+    const error = markRetryQueueKey(
+      markRetryQueueKey(sqlstateError('40001'), 'k'),
+      'k',
+    );
+    expect(retryQueueKeysOf(error)).toEqual(['k']);
+    expect(retryQueueKeysOf(sqlstateError('40001'))).toEqual([]);
+    expect(retryQueueKeysOf(undefined)).toEqual([]);
+  });
+});
+
+describe('transactSerializable — nested retry queues', () => {
+  it('locks every key in order before BEGIN and unlocks in reverse', async () => {
+    const { runner, beginAttempts, reservations } = createQueueRunner([
+      markRetryQueueKey(
+        markRetryQueueKey(sqlstateError('40001'), 'audit-chain:org_1'),
+        'task-comment:t_1',
+      ),
+    ]);
+    const result = await transactSerializable(
+      runner,
+      async (tx) => {
+        await tx`SELECT 1`;
+        return 'ok';
+      },
+      { sleep: noSleep },
+    );
+    expect(result).toBe('ok');
+    expect(beginAttempts()).toBe(1);
+    const statements = reservations[0]?.statements ?? [];
+    expect(texts(statements)).toEqual([
+      'SELECT pg_advisory_lock(?, hashtext(?))',
+      'SELECT pg_advisory_lock(?, hashtext(?))',
+      'BEGIN ISOLATION LEVEL SERIALIZABLE',
+      'SELECT 1',
+      'COMMIT',
+      'SELECT pg_advisory_unlock(?, hashtext(?))',
+      'SELECT pg_advisory_unlock(?, hashtext(?))',
+    ]);
+    expect(statements.map((s) => s.values[1])).toEqual([
+      'task-comment:t_1',
+      'audit-chain:org_1',
+      undefined,
+      undefined,
+      undefined,
+      'audit-chain:org_1',
+      'task-comment:t_1',
+    ]);
+    expect(reservations[0]?.released()).toBe(1);
+  });
+});
+
+describe('transactSerializable — nested retry queues, failure paths', () => {
+  const nested = (): Error =>
+    markRetryQueueKey(
+      markRetryQueueKey(sqlstateError('40001'), 'audit-chain:org_1'),
+      'task-comment:t_1',
+    );
+
+  it('releases the keys it already holds when a later key cannot be locked', async () => {
+    const cancelled = sqlstateError('57014');
+    const { runner, beginAttempts, reservations } = createQueueRunner(
+      [nested()],
+      {
+        failOn: (text, values) =>
+          text.startsWith('SELECT pg_advisory_lock') &&
+          values[1] === 'audit-chain:org_1'
+            ? cancelled
+            : undefined,
+      },
+    );
+    await expect(
+      transactSerializable(runner, () => Promise.resolve('never'), {
+        sleep: noSleep,
+      }),
+    ).rejects.toBe(cancelled);
+    expect(beginAttempts()).toBe(1);
+    // A cancelled statement is not transient: no second attempt of any kind.
+    expect(reservations).toHaveLength(1);
+    const statements = reservations[0]?.statements ?? [];
+    expect(texts(statements)).toEqual([
+      'SELECT pg_advisory_lock(?, hashtext(?))',
+      'SELECT pg_advisory_lock(?, hashtext(?))',
+      'SELECT pg_advisory_unlock(?, hashtext(?))',
+    ]);
+    expect(statements[2]?.values[1]).toBe('task-comment:t_1');
+    expect(reservations[0]?.released()).toBe(1);
+  });
+
+  it('keeps the fuller key list when a queued attempt is re-marked with a subset', async () => {
+    const { runner, reservations } = createQueueRunner([nested()]);
+    let queuedCalls = 0;
+    const result = await transactSerializable(
+      runner,
+      () => {
+        queuedCalls += 1;
+        // The first queued attempt loses to a writer outside the inner
+        // queue and is marked with the outer key alone.
+        return queuedCalls === 1
+          ? Promise.reject(
+              markRetryQueueKey(sqlstateError('40001'), 'task-comment:t_1'),
+            )
+          : Promise.resolve('ok');
+      },
+      { sleep: noSleep },
+    );
+    expect(result).toBe('ok');
+    expect(reservations).toHaveLength(2);
+    const locks = (reservations[1]?.statements ?? [])
+      .filter((s) => s.text.startsWith('SELECT pg_advisory_lock'))
+      .map((s) => s.values[1]);
+    expect(locks).toEqual(['task-comment:t_1', 'audit-chain:org_1']);
+  });
+});

packages/shared/src/db/serializable.ts

@@ -72,6 +72,16 @@ function jitteredSleep(ms: number): Promise<void> {
  * snapshot post-dates the writer it lost to, and other appenders that take
  * the transaction-level lock on the same key inside their own transaction
  * queue behind it instead of bumping the row underneath it.
+ *
+ * Marks nest. A callback wrapped in another queue (a task comment inside a
+ * task's queue, whose audit write sits inside the org's chain-head queue)
+ * marks inner-first, and each outer mark is PREPENDED, so the list reads in
+ * the callback's own acquisition order — task, then chain head. The queued
+ * retry takes the session locks in exactly that order and unlocks in
+ * reverse; every first attempt takes its transaction-level locks in the same
+ * order, so a queued retry and a first attempt never hold the two keys in
+ * opposite orders. A retry queued on the outer key alone would drop the
+ * inner one and lose there again.
  */
 
 const RETRY_QUEUE_KEY = Symbol.for('tale.db.retryQueueKey');
@@ -87,26 +97,42 @@ export const RETRY_QUEUE_LOCK_CLASS = 72_085_002;
 
 /**
  * Mark a serialization failure so the retry queues on `key` (see the retry
- * queues note). Returns the same error for `throw markRetryQueueKey(e, k)`.
+ * queues note). An outer mark is prepended to the keys an inner callback
+ * already put on the error; a key already present is not added twice.
+ * Returns the same error for `throw markRetryQueueKey(e, k)`.
  */
 export function markRetryQueueKey<E>(error: E, key: string): E {
   if (isSerializationFailure(error)) {
-    Object.defineProperty(error, RETRY_QUEUE_KEY, {
-      value: key,
-      enumerable: false,
-      configurable: true,
-    });
+    const keys = retryQueueKeysOf(error);
+    if (!keys.includes(key)) {
+      Object.defineProperty(error, RETRY_QUEUE_KEY, {
+        value: [key, ...keys],
+        enumerable: false,
+        configurable: true,
+      });
+    }
   }
   return error;
 }
 
-/** The queue key a failure was marked with, if any. */
-export function retryQueueKeyOf(error: unknown): string | undefined {
+/**
+ * The queue keys a failure was marked with, outermost first (the order the
+ * queued retry locks them in); empty when unmarked.
+ */
+export function retryQueueKeysOf(error: unknown): readonly string[] {
   if (error === null || typeof error !== 'object') {
-    return undefined;
+    return [];
   }
-  const key: unknown = Reflect.get(error, RETRY_QUEUE_KEY);
-  return typeof key === 'string' ? key : undefined;
+  const keys: unknown = Reflect.get(error, RETRY_QUEUE_KEY);
+  return Array.isArray(keys) &&
+    keys.every((key): key is string => typeof key === 'string')
+    ? keys
+    : [];
+}
+
+/** The outermost queue key a failure was marked with, if any. */
+export function retryQueueKeyOf(error: unknown): string | undefined {
+  return retryQueueKeysOf(error)[0];
 }
 
 /** Postgres accepts unquoted savepoint names of this shape only. */
@@ -154,20 +180,28 @@ function transactionOver(reserved: ReservedSql): TransactionSql {
 }
 
 /**
- * One serializable attempt queued on `key`: session advisory lock → BEGIN →
- * callback → COMMIT → unlock, all on one reserved connection.
+ * One serializable attempt queued on `keys`: session advisory locks in that
+ * order → BEGIN → callback → COMMIT → unlocks in reverse, all on one
+ * reserved connection. A key that cannot be locked releases the ones
+ * already held before the connection goes back to the pool — a session
+ * lock outlives the reservation, and a pooled connection still holding one
+ * would block that key for everyone until the connection dies.
  */
 async function beginQueued<T>(
   reserve: () => Promise<ReservedSql>,
-  key: string,
+  keys: readonly string[],
   callback: (tx: TransactionSql) => Promise<T>,
 ): Promise<T> {
   const reserved = await reserve();
   try {
-    await reserved`
-      SELECT pg_advisory_lock(${RETRY_QUEUE_LOCK_CLASS}, hashtext(${key}))
-    `;
+    const locked: string[] = [];
     try {
+      for (const key of keys) {
+        await reserved`
+          SELECT pg_advisory_lock(${RETRY_QUEUE_LOCK_CLASS}, hashtext(${key}))
+        `;
+        locked.push(key);
+      }
       await reserved.unsafe('BEGIN ISOLATION LEVEL SERIALIZABLE');
       let result: T;
       try {
@@ -190,36 +224,46 @@ async function beginQueued<T>(
       }
       return result;
     } finally {
-      await reserved`
-        SELECT pg_advisory_unlock(${RETRY_QUEUE_LOCK_CLASS}, hashtext(${key}))
-      `.catch((unlockError: unknown) => {
-        // The lock is session-scoped: it dies with the connection anyway.
-        console.warn(
-          `[db] advisory unlock after a queued retry failed: ${String(unlockError)}`,
-        );
-      });
+      for (const key of locked.reverse()) {
+        await reserved`
+          SELECT pg_advisory_unlock(${RETRY_QUEUE_LOCK_CLASS}, hashtext(${key}))
+        `.catch((unlockError: unknown) => {
+          // The lock is session-scoped: it dies with the connection anyway.
+          console.warn(
+            `[db] advisory unlock after a queued retry failed: ${String(unlockError)}`,
+          );
+        });
+      }
     }
   } finally {
     reserved.release();
   }
 }
 
+/** True when every key in `keys` is already in `of`. */
+function isSubsetOf(
+  keys: readonly string[],
+  of: readonly string[] | undefined,
+): boolean {
+  return of !== undefined && keys.every((key) => of.includes(key));
+}
+
 /**
  * Open a SERIALIZABLE transaction and execute `callback`, retrying the entire
  * transaction on serialization failures (40001/40P01) and on transient
  * connection faults. Each retry runs in a fresh transaction; a retry after a
- * failure marked with a queue key runs queued on that key (see above).
+ * failure marked with queue keys runs queued on those keys (see above).
  */
 export function transactSerializable<T>(
   sql: SerializableTransactionRunner,
   callback: (tx: TransactionSql) => Promise<T>,
   options: RetryOptions = {},
 ): Promise<T> {
-  let queueKey: string | undefined;
+  let queueKeys: readonly string[] | undefined;
   const reserve = sql.reserve?.bind(sql);
   const attempt = (): Promise<T> =>
-    queueKey !== undefined && reserve !== undefined
-      ? beginQueued(reserve, queueKey, callback)
+    queueKeys !== undefined && reserve !== undefined
+      ? beginQueued(reserve, queueKeys, callback)
       : sql.begin('isolation level serializable', callback);
   const isTransient =
     options.isTransient ??
@@ -232,7 +276,14 @@ export function transactSerializable<T>(
     sleep: jitteredSleep,
     ...options,
     isTransient: (error) => {
-      queueKey = retryQueueKeyOf(error) ?? queueKey;
+      // A queued attempt that loses to a writer outside its inner queues is
+      // re-marked with fewer keys than it held; the held list is a superset
+      // in the same acquisition order, so keep it — dropping a key is how a
+      // retry loses at that resource again.
+      const keys = retryQueueKeysOf(error);
+      if (keys.length > 0 && !isSubsetOf(keys, queueKeys)) {
+        queueKeys = keys;
+      }
       return isTransient(error);
     },
   });