fix(platform): claim message slots until a deadline, not a fixed count

services/platform/backend/domains/chat/append-message.test.ts

@@ -19,7 +19,7 @@ vi.mock('../../core/lib/providers/catalog_fetch.ts', () => ({
 }));
 vi.mock('../../jobs/enqueue.ts', () => ({ addJobInTx: vi.fn() }));
 
-import { MESSAGE_SLOT_ATTEMPTS } from '../threads/store.ts';
+import { MESSAGE_SLOT_CLAIM_DEADLINE_MS } from '../threads/store.ts';
 import { appendMessageRow } from './store.ts';
 
 /** A `sql` whose INSERTs answer from `outcomes` in order (an empty array is a
@@ -78,12 +78,61 @@ describe('appendMessageRow — claiming a unique slot', () => {
     expect(insertsOf(statements)).toBe(3);
   });
 
-  it('fails loudly, and writes nothing else, once the attempts are spent', async () => {
+  it('keeps re-claiming through a burst larger than any fixed count', async () => {
+    const lostRaces = Array.from({ length: 40 }, () => []);
+    const { sql, statements } = fakeSql([
+      ...lostRaces,
+      [{ id: 'm-41', order: 40 }],
+    ]);
+    const pauses: number[] = [];
+    await expect(
+      appendMessageRow(sql, MESSAGE, {
+        sleep: (ms) => {
+          pauses.push(ms);
+          return Promise.resolve();
+        },
+      }),
+    ).resolves.toEqual({ id: 'm-41', sequence: 40 });
+    expect(insertsOf(statements)).toBe(41);
+    // One jittered pause per lost race, never longer than the cap.
+    expect(pauses).toHaveLength(40);
+    expect(Math.max(...pauses)).toBeLessThanOrEqual(30);
+  });
+
+  it('lets an error from the claim through unchanged, after one insert', async () => {
+    // Under SERIALIZABLE the lost race surfaces as a 40001 and the enclosing
+    // transactSerializable reruns the transaction; the claim must not retry
+    // or swallow it.
+    const boom = Object.assign(new Error('could not serialize access'), {
+      code: '40001',
+    });
+    const statements: string[] = [];
+    const tag = (strings: TemplateStringsArray): Promise<unknown[]> => {
+      statements.push(strings.join('?'));
+      return Promise.reject(boom);
+    };
+    Object.assign(tag, { json: (value: unknown) => value });
+    // oxlint-disable-next-line typescript/no-unsafe-type-assertion -- only the tag call and `json` are exercised
+    const sql = tag as unknown as Sql;
+    await expect(
+      appendMessageRow(sql, MESSAGE, { sleep: () => Promise.resolve() }),
+    ).rejects.toBe(boom);
+    expect(insertsOf(statements)).toBe(1);
+  });
+
+  it('fails loudly, and writes nothing else, once the deadline is spent', async () => {
     const { sql, statements } = fakeSql([]);
-    await expect(appendMessageRow(sql, MESSAGE)).rejects.toThrow(
-      /no free slot/,
+    // Each clock read advances 4 s: the 10 s budget is gone at the third claim.
+    let clock = 0;
+    await expect(
+      appendMessageRow(sql, MESSAGE, {
+        now: () => (clock += 4_000),
+        sleep: () => Promise.resolve(),
+      }),
+    ).rejects.toThrow(
+      `no free slot within ${MESSAGE_SLOT_CLAIM_DEADLINE_MS} ms (3 attempts)`,
     );
-    expect(insertsOf(statements)).toBe(MESSAGE_SLOT_ATTEMPTS);
+    expect(insertsOf(statements)).toBe(3);
     expect(statements.some((text) => text.includes('UPDATE'))).toBe(false);
   });
 });

services/platform/backend/domains/chat/store.ts

@@ -13,7 +13,7 @@ import { toJson } from '../../db/sql.ts';
 import { addJobInTx } from '../../jobs/enqueue.ts';
 import { resolveOrgSlug } from '../../lib/org-config.ts';
 import { incrementUsageLedger } from '../governance/service.ts';
-import { MESSAGE_SLOT_ATTEMPTS } from '../threads/store.ts';
+import { claimMessageSlot, type SlotClaimOptions } from '../threads/store.ts';
 
 /**
  * The Postgres-backed ports the turn pipeline writes through — the 0.5 twin
@@ -47,16 +47,12 @@ export async function appendMessageRow(
     truncation?: { droppedMessages: number };
     status?: string;
   },
+  slot: SlotClaimOptions = {},
 ): Promise<{ id: string; sequence: number }> {
   // The slot is UNIQUE: two turns appending to one thread at once both read
   // the same max, and the one the index refuses re-claims the next slot on
   // a fresh statement instead of tying the winner's ordering.
-  let row: { id: string; order: number } | undefined;
-  for (
-    let attempt = 0;
-    row === undefined && attempt < MESSAGE_SLOT_ATTEMPTS;
-    attempt += 1
-  ) {
+  const row = await claimMessageSlot(async () => {
     const rows = await sql<{ id: string; order: number }[]>`
       INSERT INTO app.messages (
         thread_id, org_id, "order", step_order, role, parts, text, model,
@@ -77,13 +73,8 @@ export async function appendMessageRow(
       ON CONFLICT (thread_id, "order", step_order) DO NOTHING
       RETURNING id, "order"
     `;
-    row = rows[0];
-  }
-  if (!row) {
-    throw new Error(
-      `message insert failed: no free slot after ${MESSAGE_SLOT_ATTEMPTS} attempts`,
-    );
-  }
+    return rows[0]; // undefined: the slot went to a concurrent append
+  }, slot);
   // A turn just wrote to the thread; keep its list ordering fresh. An
   // assistant row also stamps the unread watermark; activity on a hidden
   // branch surfaces on its ROOT — the row the sidebar shows for the lineage.

services/platform/backend/domains/threads/store.test.ts

@@ -4,13 +4,13 @@
  * A message slot is unique: an append that loses the race for `max+1` gets
  * no row back from `ON CONFLICT DO NOTHING` and must claim the next slot on
  * a fresh statement — never land on the winner's slot, never surface the
- * lost race as an error while attempts remain.
+ * lost race as an error while the deadline remains.
  */
 
 import type { TransactionSql } from 'postgres';
 import { describe, expect, it } from 'vitest';
 
-import { MESSAGE_SLOT_ATTEMPTS, saveMessage } from './store.ts';
+import { MESSAGE_SLOT_CLAIM_DEADLINE_MS, saveMessage } from './store.ts';
 
 /** A `tx` whose INSERTs answer from `outcomes` in order (an empty array is a
  * lost race), recording every statement so the retry count is observable. */
@@ -74,10 +74,54 @@ describe('saveMessage — claiming a unique slot', () => {
     expect(insertsOf(statements)).toBe(2);
   });
 
-  it('gives up only after the bounded number of lost races', async () => {
+  it('outlasts a burst larger than any fixed count of lost races', async () => {
+    const lostRaces = Array.from({ length: 40 }, () => []);
+    const { tx, statements } = fakeTx([
+      ...lostRaces,
+      [{ id: 'm-41', order: 40 }],
+    ]);
+    await expect(
+      saveMessage(tx, ARGS, { sleep: () => Promise.resolve() }),
+    ).resolves.toEqual({ messageId: 'm-41', order: 40 });
+    expect(insertsOf(statements)).toBe(41);
+  });
+
+  it('lets an error from the claim through unchanged, after one insert', async () => {
+    // A 40001 under SERIALIZABLE belongs to transactSerializable's rerun;
+    // the claim must neither retry nor swallow it.
+    const boom = Object.assign(new Error('could not serialize access'), {
+      code: '40001',
+    });
+    const statements: string[] = [];
+    const tag = (strings: TemplateStringsArray): Promise<unknown[]> => {
+      statements.push(strings.join('?'));
+      return Promise.reject(boom);
+    };
+    Object.assign(tag, { json: (value: unknown) => value });
+    // oxlint-disable-next-line typescript/no-unsafe-type-assertion -- only the tag call and `json` are exercised
+    const tx = tag as unknown as TransactionSql;
+    await expect(
+      saveMessage(tx, ARGS, { sleep: () => Promise.resolve() }),
+    ).rejects.toBe(boom);
+    expect(insertsOf(statements)).toBe(1);
+    expect(statements.some((text) => text.includes('UPDATE app.threads'))).toBe(
+      false,
+    );
+  });
+
+  it('gives up only when the deadline is spent', async () => {
     const { tx, statements } = fakeTx([]);
-    await expect(saveMessage(tx, ARGS)).rejects.toThrow(/no free slot/);
-    expect(insertsOf(statements)).toBe(MESSAGE_SLOT_ATTEMPTS);
+    // Each clock read advances 4 s: the 10 s budget is gone at the third claim.
+    let clock = 0;
+    await expect(
+      saveMessage(tx, ARGS, {
+        now: () => (clock += 4_000),
+        sleep: () => Promise.resolve(),
+      }),
+    ).rejects.toThrow(
+      `no free slot within ${MESSAGE_SLOT_CLAIM_DEADLINE_MS} ms (3 attempts)`,
+    );
+    expect(insertsOf(statements)).toBe(3);
     expect(statements.some((text) => text.includes('UPDATE app.threads'))).toBe(
       false,
     );

services/platform/backend/domains/threads/store.ts

@@ -64,16 +64,77 @@ export interface SaveMessageArgs {
 }
 
 /**
- * How many times an appender re-claims the next (order, step) slot after
- * another appender took the one it computed. The slot is UNIQUE
+ * How long an appender keeps re-claiming the next (order, step) slot while
+ * concurrent appends take the ones it computes. The slot is UNIQUE
  * (`messages_thread_slot`), so a lost race is refused at the index rather
- * than landing two rows on one slot; under READ COMMITTED each attempt is a
- * fresh statement that sees the winner's row, so one retry is the norm and a
- * handful is generous. Under SERIALIZABLE the same conflict surfaces as a
- * serialization failure and `transactSerializable` reruns the whole
- * transaction instead, so this loop never spins there.
+ * than landing two rows on one slot; under READ COMMITTED each claim is a
+ * fresh statement that sees the winner's row. Every round has at least one
+ * winner among the appenders racing for a thread (exactly one when they run
+ * in lockstep), so a burst of N appenders lands within N rounds — which is
+ * why the budget is wall-clock and never a count: a count is defeated by any
+ * burst larger than itself. Under
+ * SERIALIZABLE the same conflict surfaces as a serialization failure and
+ * `transactSerializable` reruns the whole transaction instead, so the loop
+ * never spins there.
  */
-export const MESSAGE_SLOT_ATTEMPTS = 8;
+export const MESSAGE_SLOT_CLAIM_DEADLINE_MS = 10_000;
+
+/**
+ * The longest pause between two claims of one appender: enough to break the
+ * lockstep, small enough that the last member of a large burst waits well
+ * under a second in total.
+ */
+const MESSAGE_SLOT_BACKOFF_CAP_MS = 20;
+
+export interface SlotClaimOptions {
+  /** Wall-clock budget for the whole claim; the default is the constant above. */
+  deadlineMs?: number;
+  /** Clock and sleep, injectable so tests exhaust the budget deterministically. */
+  now?: () => number;
+  sleep?: (ms: number) => Promise<void>;
+}
+
+/**
+ * Jittered exponential pause after the `attempt`-th lost race (1 ms, 2 ms,
+ * 4 ms … capped at {@link MESSAGE_SLOT_BACKOFF_CAP_MS}): the losers of one
+ * round must not re-collide in lockstep.
+ */
+function slotBackoffMs(attempt: number): number {
+  const base = Math.min(MESSAGE_SLOT_BACKOFF_CAP_MS, 2 ** (attempt - 1));
+  return base * (0.5 + Math.random());
+}
+
+function defaultSleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+/**
+ * Run `claim` — one `INSERT … ON CONFLICT DO NOTHING RETURNING` — until it
+ * lands a row (`undefined` = the computed slot went to a concurrent append).
+ * Gives up only when {@link SlotClaimOptions.deadlineMs} is spent, naming the
+ * budget and the rounds it took; nothing else is written meanwhile.
+ */
+export async function claimMessageSlot<T>(
+  claim: () => Promise<T | undefined>,
+  options: SlotClaimOptions = {},
+): Promise<T> {
+  const deadlineMs = options.deadlineMs ?? MESSAGE_SLOT_CLAIM_DEADLINE_MS;
+  const now = options.now ?? Date.now;
+  const sleep = options.sleep ?? defaultSleep;
+  const deadline = now() + deadlineMs;
+  for (let attempt = 1; ; attempt += 1) {
+    const row = await claim();
+    if (row !== undefined) {
+      return row;
+    }
+    if (now() >= deadline) {
+      throw new Error(
+        `message insert failed: no free slot within ${deadlineMs} ms (${attempt} attempts)`,
+      );
+    }
+    await sleep(slotBackoffMs(attempt));
+  }
+}
 
 /**
  * Append a message as the next turn: claims `max(order)+1` with
@@ -85,8 +146,9 @@ export const MESSAGE_SLOT_ATTEMPTS = 8;
 export async function saveMessage(
   tx: TransactionSql,
   args: SaveMessageArgs,
+  slot: SlotClaimOptions = {},
 ): Promise<{ messageId: string; order: number }> {
-  for (let attempt = 0; attempt < MESSAGE_SLOT_ATTEMPTS; attempt += 1) {
+  const row = await claimMessageSlot(async () => {
     const rows = await tx<{ id: string; order: number }[]>`
       INSERT INTO app.messages (
         thread_id, org_id, "order", step_order, role, parts, text, author_id,
@@ -101,17 +163,13 @@ export async function saveMessage(
       ON CONFLICT (thread_id, "order", step_order) DO NOTHING
       RETURNING id, "order"
     `;
-    const row = rows[0];
-    if (row === undefined) continue; // the slot went to a concurrent append
-    await tx`
-      UPDATE app.threads SET updated_at_ms = ${Date.now()}
-      WHERE id = ${args.threadId}
-    `;
-    return { messageId: row.id, order: row.order };
-  }
-  throw new Error(
-    `message insert failed: no free slot after ${MESSAGE_SLOT_ATTEMPTS} attempts`,
-  );
+    return rows[0]; // undefined: the slot went to a concurrent append
+  }, slot);
+  await tx`
+    UPDATE app.threads SET updated_at_ms = ${Date.now()}
+    WHERE id = ${args.threadId}
+  `;
+  return { messageId: row.id, order: row.order };
 }
 
 /** The most messages one read may ask for, on either lane below. */

services/platform/backend/integration-check.ts

@@ -5234,7 +5234,9 @@ async function checkMessageSlots(
       thread_id, org_id, user_id, chat_type, status, created_at_ms
     ) VALUES (${threadId}, ${orgId}, ${userId}, 'assistant', 'active', ${now})
   `;
-  const APPENDS = 12;
+  // Larger than the retry count the loop once carried (8): every round has
+  // one winner, so a burst needs as many rounds as appenders.
+  const APPENDS = 32;
   const outcomes = await Promise.allSettled(
     Array.from({ length: APPENDS }, (_, i) =>
       appendMessageRow(sql, {