diff --git a/services/platform/backend/domains/chat/append-message.test.ts b/services/platform/backend/domains/chat/append-message.test.ts index 7ed730ca02..e40589c5ba 100644 --- a/services/platform/backend/domains/chat/append-message.test.ts +++ b/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 => { + 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); }); }); diff --git a/services/platform/backend/domains/chat/store.ts b/services/platform/backend/domains/chat/store.ts index 3df517dcb0..adbdd82a4f 100644 --- a/services/platform/backend/domains/chat/store.ts +++ b/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. diff --git a/services/platform/backend/domains/threads/store.test.ts b/services/platform/backend/domains/threads/store.test.ts index 2bc8dd0c8e..5e760533f6 100644 --- a/services/platform/backend/domains/threads/store.test.ts +++ b/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 => { + 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, ); diff --git a/services/platform/backend/domains/threads/store.ts b/services/platform/backend/domains/threads/store.ts index f45583322d..5bea03693a 100644 --- a/services/platform/backend/domains/threads/store.ts +++ b/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; +} + +/** + * 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 { + 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( + claim: () => Promise, + options: SlotClaimOptions = {}, +): Promise { + 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. */ diff --git a/services/platform/backend/integration-check.ts b/services/platform/backend/integration-check.ts index ee0e177b9a..6c9c876234 100644 --- a/services/platform/backend/integration-check.ts +++ b/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, {