diff --git a/packages/shared/src/db/serializable.test.ts b/packages/shared/src/db/serializable.test.ts index f2b95bdd36..cab1ac5157 100644 --- a/packages/shared/src/db/serializable.test.ts +++ b/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, ): Promise => 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']); + }); +}); diff --git a/packages/shared/src/db/serializable.ts b/packages/shared/src/db/serializable.ts index 5557219ff7..487c424a92 100644 --- a/packages/shared/src/db/serializable.ts +++ b/packages/shared/src/db/serializable.ts @@ -72,6 +72,16 @@ function jitteredSleep(ms: number): Promise { * 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(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( reserve: () => Promise, - key: string, + keys: readonly string[], callback: (tx: TransactionSql) => Promise, ): Promise { 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( } 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( sql: SerializableTransactionRunner, callback: (tx: TransactionSql) => Promise, options: RetryOptions = {}, ): Promise { - let queueKey: string | undefined; + let queueKeys: readonly string[] | undefined; const reserve = sql.reserve?.bind(sql); const attempt = (): Promise => - 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( 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); }, }); diff --git a/services/platform/backend/domains/tasks/comments.queue.test.ts b/services/platform/backend/domains/tasks/comments.queue.test.ts new file mode 100644 index 0000000000..b53e884b63 --- /dev/null +++ b/services/platform/backend/domains/tasks/comments.queue.test.ts @@ -0,0 +1,112 @@ +// @vitest-environment node + +/** + * The task comment queue's contract with the shared retry queue: the task + * key is locked before anything else, a serialization failure inside the + * write carries the task key in front of any key an inner write (the org's + * audit chain head) already put on it, and other failures pass untouched. + */ + +import { + RETRY_QUEUE_LOCK_CLASS, + markRetryQueueKey, + retryQueueKeysOf, +} from '@tale/shared/db/serializable'; +import type { TransactionSql } from 'postgres'; +import { describe, expect, it } from 'vitest'; + +import { + lockTaskCommentQueue, + queuedOnTask, + taskCommentQueueKey, +} from './comments.ts'; + +interface Statement { + text: string; + values: unknown[]; +} + +function fakeTx(): { tx: TransactionSql; statements: Statement[] } { + const statements: Statement[] = []; + const tag = (strings: TemplateStringsArray, ...values: unknown[]) => { + const text = strings + .reduce( + (acc, part, index) => + `${acc}${part}${index < values.length ? '?' : ''}`, + '', + ) + .replace(/\s+/g, ' ') + .trim(); + statements.push({ text, values }); + return Promise.resolve([]); + }; + // oxlint-disable-next-line typescript/no-unsafe-type-assertion -- only the tag call is exercised + return { tx: tag as unknown as TransactionSql, statements }; +} + +function sqlstateError(code: string): Error { + const error: Error & { code?: string } = new Error(`sqlstate ${code}`); + error.code = code; + return error; +} + +describe('queuedOnTask', () => { + it("takes the task's queue lock before the work runs", async () => { + const { tx, statements } = fakeTx(); + const order: string[] = []; + const result = await queuedOnTask(tx, 't_1', () => { + order.push(`work after ${statements.length} statement(s)`); + return Promise.resolve('done'); + }); + expect(result).toBe('done'); + expect(statements).toEqual([ + { + text: 'SELECT pg_advisory_xact_lock(?, hashtext(?))', + values: [RETRY_QUEUE_LOCK_CLASS, taskCommentQueueKey('t_1')], + }, + ]); + expect(order).toEqual(['work after 1 statement(s)']); + }); + + it('marks a serialization failure with the task key', async () => { + const { tx } = fakeTx(); + const failure = await queuedOnTask(tx, 't_1', () => + Promise.reject(sqlstateError('40001')), + ).catch((error: unknown) => error); + expect(retryQueueKeysOf(failure)).toEqual([taskCommentQueueKey('t_1')]); + }); + + it('keeps an inner audit mark behind the task key', async () => { + const { tx } = fakeTx(); + const failure = await queuedOnTask(tx, 't_1', () => + Promise.reject( + markRetryQueueKey(sqlstateError('40001'), 'audit-chain:org_1'), + ), + ).catch((error: unknown) => error); + expect(retryQueueKeysOf(failure)).toEqual([ + taskCommentQueueKey('t_1'), + 'audit-chain:org_1', + ]); + }); + + it('leaves other failures unmarked and untouched', async () => { + const { tx } = fakeTx(); + const boom = sqlstateError('23505'); + const failure = await queuedOnTask(tx, 't_1', () => + Promise.reject(boom), + ).catch((error: unknown) => error); + expect(failure).toBe(boom); + expect(retryQueueKeysOf(failure)).toEqual([]); + }); +}); + +describe('lockTaskCommentQueue', () => { + it('is the same lock statement the queue takes', async () => { + const { tx, statements } = fakeTx(); + await lockTaskCommentQueue(tx, 't_9'); + expect(statements[0]?.values).toEqual([ + RETRY_QUEUE_LOCK_CLASS, + taskCommentQueueKey('t_9'), + ]); + }); +}); diff --git a/services/platform/backend/domains/tasks/comments.ts b/services/platform/backend/domains/tasks/comments.ts index 03726c9675..a10d55fe12 100644 --- a/services/platform/backend/domains/tasks/comments.ts +++ b/services/platform/backend/domains/tasks/comments.ts @@ -1,3 +1,7 @@ +import { + markRetryQueueKey, + RETRY_QUEUE_LOCK_CLASS, +} from '@tale/shared/db/serializable'; import type { Sql, TransactionSql } from 'postgres'; import { z } from 'zod'; @@ -92,24 +96,89 @@ async function ensureTaskDiscussionThread( return rows[0]?.discussionThreadId ?? threadId; } -/** Append one comment (message + lockstep meta + count + activity + audit). - * `bodyByLocale` is the same text written natively per language (the - * workflow `task.comment` native and the automated date nudge carry it); the - * reader picks their locale and falls back to `body`. */ -export async function addTaskComment( +/** Retry-queue key for one task's discussion (see `queuedOnTask`). */ +export function taskCommentQueueKey(taskId: string): string { + return `task-comment:${taskId}`; +} + +/** + * The transaction-level lock every comment write takes FIRST. A transaction + * that writes the task row for another reason and then comments (the + * overdue nudge's claim) must take it before that write too, or it holds the + * row while a commenter holds the key — a deadlock pair. + */ +export async function lockTaskCommentQueue( tx: TransactionSql, - auth: ProjectAuthContext, - args: { - taskId: string; - body: string; - bodyByLocale?: Record; - author?: CommentAuthor; - }, -): Promise<{ + taskId: string, +): Promise { + await tx` + SELECT pg_advisory_xact_lock(${RETRY_QUEUE_LOCK_CLASS}, hashtext(${taskCommentQueueKey(taskId)})) + `; +} + +/** + * Run one comment write queued on its task. Every comment on a task bumps + * the same rows — the task's `comment_count` and the discussion thread's + * next message slot — so serializable writers that overlap all lose but the + * first, and a plain retry loses again whenever another commits first: the + * storm `withRetry`'s five attempts cannot outlast. Two pieces make a retry + * deterministic instead (the audit chain head's `lockChainHead` is the + * twin): the transaction-level advisory lock on the task's queue key, taken + * before the write's first read, queues this transaction behind a retry that + * holds the same key as a session lock from before its BEGIN (see + * `transactSerializable`); and a 40001/40P01 raised anywhere in `work` is + * marked with the key, which is what makes the caller's next attempt take + * that session lock first. The audit write inside `work` marks a loss at the + * org's chain head with its own key; marks nest, so that retry queues on the + * task AND the chain head, in that order (the retry-queue note in + * `@tale/shared/db/serializable`). Under contention on this task's rows a + * writer wastes at most one attempt. Plain READ COMMITTED callers pay only + * the lock, which orders the task's comments and marks nothing. + */ +export async function queuedOnTask( + tx: TransactionSql, + taskId: string, + work: () => Promise, +): Promise { + try { + await lockTaskCommentQueue(tx, taskId); + return await work(); + } catch (error) { + throw markRetryQueueKey(error, taskCommentQueueKey(taskId)); + } +} + +interface AddedTaskComment { messageId: string; threadId: string; unresolvedMentionTokens: string[]; -}> { +} + +interface AddTaskCommentArgs { + taskId: string; + body: string; + bodyByLocale?: Record; + author?: CommentAuthor; +} + +/** Append one comment (message + lockstep meta + count + activity + audit), + * queued on its task (`queuedOnTask`). `bodyByLocale` is the same text + * written natively per language (the workflow `task.comment` native and the + * automated date nudge carry it); the reader picks their locale and falls + * back to `body`. */ +export function addTaskComment( + tx: TransactionSql, + auth: ProjectAuthContext, + args: AddTaskCommentArgs, +): Promise { + return queuedOnTask(tx, args.taskId, () => appendTaskComment(tx, auth, args)); +} + +async function appendTaskComment( + tx: TransactionSql, + auth: ProjectAuthContext, + args: AddTaskCommentArgs, +): Promise { const task = await loadTaskOrThrow(tx, args.taskId, auth.organizationId); const project = await loadProjectOrThrow(tx, task.projectId); // Commenting is READ-level (0.4 `addTaskComment*`): anyone who can see @@ -361,15 +430,17 @@ export async function listTaskComments( }; } -async function loadCommentMeta( - tx: TransactionSql | Sql, - messageId: string, -): Promise<{ +interface CommentMeta { taskId: string; authorType: string; authorId: string; mentions: ResolvedMention[] | null; -}> { +} + +async function loadCommentMeta( + tx: TransactionSql | Sql, + messageId: string, +): Promise { const rows = await tx< { taskId: string; @@ -496,12 +567,25 @@ export async function editTaskComment( }); } +/** Delete one comment, queued on its task (`queuedOnTask`): the count it + * decrements is the same hot row every append bumps. */ export async function deleteTaskComment( tx: TransactionSql, auth: ProjectAuthContext, messageId: string, ): Promise { const meta = await loadCommentMeta(tx, messageId); + await queuedOnTask(tx, meta.taskId, () => + removeTaskComment(tx, auth, messageId, meta), + ); +} + +async function removeTaskComment( + tx: TransactionSql, + auth: ProjectAuthContext, + messageId: string, + meta: CommentMeta, +): Promise { const task = await loadTaskOrThrow(tx, meta.taskId, auth.organizationId); const project = await loadProjectOrThrow(tx, task.projectId); assertTaskWritable(project, auth); diff --git a/services/platform/backend/domains/tasks/date-notifications.test.ts b/services/platform/backend/domains/tasks/date-notifications.test.ts index 98c77aa12c..6652d9c63e 100644 --- a/services/platform/backend/domains/tasks/date-notifications.test.ts +++ b/services/platform/backend/domains/tasks/date-notifications.test.ts @@ -2,14 +2,17 @@ import type { Sql, TransactionSql } from 'postgres'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import { notifyUser } from '../collab/service.ts'; -import { addTaskComment } from './comments.ts'; +import { addTaskComment, lockTaskCommentQueue } from './comments.ts'; import { enforceTaskDatesForOrg, OVERDUE_NUDGE_BODY, } from './date-notifications.ts'; vi.mock('../collab/service.ts', () => ({ notifyUser: vi.fn() })); -vi.mock('./comments.ts', () => ({ addTaskComment: vi.fn() })); +vi.mock('./comments.ts', () => ({ + addTaskComment: vi.fn(), + lockTaskCommentQueue: vi.fn(), +})); /** * The date ladder's delivery contract: each rung claims and announces a row @@ -124,6 +127,7 @@ describe('enforceTaskDatesForOrg — each row is claimed and announced in its ow it('the overdue nudge is posted on the row’s transaction, in every locale', async () => { let claimTx: unknown; + const order: string[] = []; const { sql } = fakeSql((text, values) => { if ( text.startsWith('SELECT t2.id FROM app.tasks t2') && @@ -136,11 +140,16 @@ describe('enforceTaskDatesForOrg — each row is claimed and announced in its ow if ( text.startsWith('UPDATE app.tasks SET sla_level = ?, sla_level_at_ms') ) { + order.push('claim'); const id = values.find((value) => value === 't-late'); return id === undefined ? [] : [sweepRow('t-late', { newLevel: 2 })]; } return []; }); + vi.mocked(lockTaskCommentQueue).mockImplementation((_tx, taskId) => { + order.push(`lock ${taskId}`); + return Promise.resolve(); + }); vi.mocked(addTaskComment).mockImplementation((tx) => { claimTx = tx; return Promise.resolve({ @@ -163,6 +172,9 @@ describe('enforceTaskDatesForOrg — each row is claimed and announced in its ow }); // On the transaction that stamped the level, not a sibling one. expect(claimTx).toBe(sql); + // The task's comment key comes BEFORE the row is written — the order + // every commenter uses, so the claim never deadlocks with a comment. + expect(order).toEqual(['lock t-late', 'claim']); expect(Object.keys(OVERDUE_NUDGE_BODY).sort()).toEqual(['de', 'en', 'fr']); }); diff --git a/services/platform/backend/domains/tasks/date-notifications.ts b/services/platform/backend/domains/tasks/date-notifications.ts index 5b6ea82239..b3851aad8d 100644 --- a/services/platform/backend/domains/tasks/date-notifications.ts +++ b/services/platform/backend/domains/tasks/date-notifications.ts @@ -2,7 +2,7 @@ import type { Sql, TransactionSql } from 'postgres'; import { resolveDateNotifyAudience } from '../../core/tasks/date_notification_recipients.ts'; import { notifyUser } from '../collab/service.ts'; -import { addTaskComment } from './comments.ts'; +import { addTaskComment, lockTaskCommentQueue } from './comments.ts'; /** * Task date enforcement — the 0.5 twin of 0.4's @@ -330,7 +330,13 @@ export async function enforceTaskDatesForOrg( const announced = await claimAndAnnounce( sql, id, - (tx) => tx<(SweepRow & { newLevel: number })[]>` + async (tx) => { + // Level 2 comments on the task, and every commenter takes the + // task's comment key BEFORE it writes the task row — take it first + // here too, or this claim holds the row while a commenter holds the + // key and the two deadlock. + await lockTaskCommentQueue(tx, id); + return tx<(SweepRow & { newLevel: number })[]>` UPDATE app.tasks SET sla_level = ${targetLevel}, sla_level_at_ms = ${now} FROM app.projects p @@ -341,7 +347,8 @@ export async function enforceTaskDatesForOrg( AND ${notTerminal} RETURNING ${tx.unsafe(SWEEP_COLUMNS.replaceAll('t.', 'app.tasks.'))}, app.tasks.sla_level AS "newLevel" - `, + `; + }, async (tx, row) => { if (row.newLevel <= 2) { await postOverdueNudge(tx, organizationId, row.taskId); diff --git a/services/platform/backend/integration-check.ts b/services/platform/backend/integration-check.ts index ee0e177b9a..a20fd58344 100644 --- a/services/platform/backend/integration-check.ts +++ b/services/platform/backend/integration-check.ts @@ -5345,6 +5345,160 @@ async function checkMessageSlots( * fixed window filled for the current period) and restored afterwards so the * rest of the suite keeps its budgets. */ +/** + * A burst of serializable commenters on ONE task: every comment bumps the + * task's comment_count and claims the discussion thread's next slot, so each + * overlapping transaction but the first loses; the task-scoped retry queue in + * `addTaskComment` (`queuedOnTask`) lands every one within a single retry. + */ +async function checkTaskCommentBurst( + sql: Sql, + ctx: { orgId: string; userId: string }, +): Promise { + const { orgId, userId } = ctx; + const { addTaskComment } = await import('./domains/tasks/comments.ts'); + const now = Date.now(); + const projectRows = await sql<{ id: string }[]>` + INSERT INTO app.projects (org_id, name, created_by, created_at_ms, + updated_at_ms) + VALUES (${orgId}, 'Comment burst project', ${userId}, ${now}, ${now}) + RETURNING id + `; + const projectId = projectRows[0]?.id ?? ''; + const taskRows = await sql<{ id: string }[]>` + INSERT INTO app.tasks ( + org_id, project_id, title, status, rank, created_by, created_by_type, + outputs, created_at_ms, updated_at_ms + ) VALUES ( + ${orgId}, ${projectId}, 'Comment burst task', 'todo', 'q0', ${userId}, + 'user', ${sql.json([])}, ${now}, ${now} + ) RETURNING id + `; + const taskId = taskRows[0]?.id ?? ''; + const auth = { + organizationId: orgId, + userId, + role: 'owner', + teamIds: [] as string[], + }; + const COMMENTERS = 12; + let attempts = 0; + const outcomes = await Promise.allSettled( + Array.from({ length: COMMENTERS }, (_, index) => + transactSerializable(sql, async (tx) => { + attempts += 1; + // Fix the snapshot first, then let every commenter overlap: each + // one's reads are now stale for all but the first committer. + await tx`SELECT 1`; + await sleep(50); + return addTaskComment(tx, auth, { + taskId, + body: `burst comment ${index}`, + }); + }), + ), + ); + const failures = outcomes.filter((o) => o.status === 'rejected'); + const taskAfter = await sql< + { commentCount: number; threadId: string | null }[] + >` + SELECT comment_count AS "commentCount", + discussion_thread_id AS "threadId" + FROM app.tasks WHERE id = ${taskId} + `; + const threadId = taskAfter[0]?.threadId ?? ''; + const slots = await sql<{ order: number }[]>` + SELECT "order" FROM app.messages WHERE thread_id = ${threadId} + ORDER BY "order" + `; + const distinctOrders = new Set(slots.map((row) => row.order)).size; + record( + 'tasks: a burst of serializable commenters on one task all land', + failures.length === 0 && + taskAfter[0]?.commentCount === COMMENTERS && + slots.length === COMMENTERS && + distinctOrders === COMMENTERS && + attempts > COMMENTERS && + attempts <= COMMENTERS * 2, + `landed=${slots.length}/${COMMENTERS} rejected=${failures.length}${failures.length > 0 ? ` (${failures.map((f) => errorText(f.reason)).join('; ')})` : ''} commentCount=${String(taskAfter[0]?.commentCount)} (want ${COMMENTERS}) distinctOrders=${distinctOrders} attempts=${attempts} (want >${COMMENTERS} — losers retried — and ≤${COMMENTERS * 2}: each lost at most once, queued)`, + ); +} + +/** + * A burst of serializable commenters on DIFFERENT tasks of one org: the task + * keys never collide, so every one of them meets the others at the org's + * audit chain head and all but the first lose there. The loss is marked with + * the chain-head key inside the task's queue, and the marks nest, so each + * retry holds the task key AND the chain-head key from before its BEGIN — + * a retry queued on the task key alone would lose at the head again. + */ +async function checkTaskCommentCrossTaskBurst( + sql: Sql, + ctx: { orgId: string; userId: string }, +): Promise { + const { orgId, userId } = ctx; + const { addTaskComment } = await import('./domains/tasks/comments.ts'); + const { verifyAuditChain } = await import('./domains/audit_logs/verify.ts'); + const now = Date.now(); + const projectRows = await sql<{ id: string }[]>` + INSERT INTO app.projects (org_id, name, created_by, created_at_ms, + updated_at_ms) + VALUES (${orgId}, 'Cross-task burst project', ${userId}, ${now}, ${now}) + RETURNING id + `; + const projectId = projectRows[0]?.id ?? ''; + const COMMENTERS = 12; + const taskIds: string[] = []; + for (let index = 0; index < COMMENTERS; index += 1) { + const rows = await sql<{ id: string }[]>` + INSERT INTO app.tasks ( + org_id, project_id, title, status, rank, created_by, created_by_type, + outputs, created_at_ms, updated_at_ms + ) VALUES ( + ${orgId}, ${projectId}, ${`Cross-task burst ${index}`}, 'todo', + ${`r${index}`}, ${userId}, 'user', ${sql.json([])}, ${now}, ${now} + ) RETURNING id + `; + taskIds.push(rows[0]?.id ?? ''); + } + const auth = { + organizationId: orgId, + userId, + role: 'owner', + teamIds: [] as string[], + }; + let attempts = 0; + const outcomes = await Promise.allSettled( + taskIds.map((taskId, index) => + transactSerializable(sql, async (tx) => { + attempts += 1; + await tx`SELECT 1`; + await sleep(50); + return addTaskComment(tx, auth, { + taskId, + body: `cross-task comment ${index}`, + }); + }), + ), + ); + const failures = outcomes.filter((o) => o.status === 'rejected'); + const counts = await sql<{ commentCount: number }[]>` + SELECT comment_count AS "commentCount" FROM app.tasks + WHERE project_id = ${projectId} + `; + const landed = counts.filter((row) => row.commentCount === 1).length; + const chain = await verifyAuditChain(sql, orgId); + record( + "tasks: a burst of serializable commenters across one org's tasks all land", + failures.length === 0 && + landed === COMMENTERS && + chain.valid && + attempts > COMMENTERS && + attempts <= COMMENTERS * 2, + `landed=${landed}/${COMMENTERS} rejected=${failures.length}${failures.length > 0 ? ` (${failures.map((f) => errorText(f.reason)).join('; ')})` : ''} chainValid=${chain.valid} attempts=${attempts} (want >${COMMENTERS} — losers at the chain head retried — and ≤${COMMENTERS * 2}: each lost at most once, queued on task + chain head)`, + ); +} + async function checkRateLimitShapes( sql: Sql, base: string, @@ -44136,6 +44290,11 @@ async function main(): Promise { ], ['checkSmallDomains', () => checkSmallDomains(sql, baseUrl, authCtx)], ['checkMessageSlots', () => checkMessageSlots(sql, authCtx)], + ['checkTaskCommentBurst', () => checkTaskCommentBurst(sql, authCtx)], + [ + 'checkTaskCommentCrossTaskBurst', + () => checkTaskCommentCrossTaskBurst(sql, authCtx), + ], [ 'checkRateLimitShapes', () => checkRateLimitShapes(sql, baseUrl, authCtx),