From 23c82f51434bff9b2e3d9514d0e845852acfcd94 Mon Sep 17 00:00:00 2001 From: larryro <371767072@qq.com> Date: Sun, 6 Sep 2026 15:40:33 +0800 Subject: [PATCH 1/6] fix(platform): queue serializable task-comment writes per task MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every comment on a task bumps the same rows — the task's comment_count and the discussion thread's next message slot — so serializable commenters that overlap all lose but the first, and a plain retry loses again whenever another commits first. `transactSerializable` gives up after five attempts, a count any burst larger than itself defeats; the audit chain head had the same storm (#3246) and the message slot under READ COMMITTED had its count-bound twin (#3267). Comment appends and deletes now run through `queuedOnTask`: the transaction-level advisory lock on `task-comment:` queues a first attempt behind a retry that holds the key as a session lock from before its BEGIN, and a 40001/40P01 raised anywhere in the write is marked with the key so the caller's next attempt takes that session lock first. A commenter therefore wastes at most one attempt; READ COMMITTED callers pay only the lock. Edits touch no shared row and stay as they were. The integration proof gains a lane: twelve serializable commenters on one task, overlap widened, all land with one comment_count of twelve, twelve distinct message slots, and between thirteen and twenty-four attempts — every loser retried exactly once, queued. --- .../backend/domains/tasks/comments.ts | 95 ++++++++++++++++--- .../platform/backend/integration-check.ts | 80 ++++++++++++++++ 2 files changed, 161 insertions(+), 14 deletions(-) diff --git a/services/platform/backend/domains/tasks/comments.ts b/services/platform/backend/domains/tasks/comments.ts index 03726c9675..23bf240669 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,74 @@ 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`). */ +function taskCommentQueueKey(taskId: string): string { + return `task-comment:${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. Under contention a writer wastes at most one + * attempt. Plain READ COMMITTED callers pay only the lock, which orders the + * task's comments and marks nothing. + */ +async function queuedOnTask( tx: TransactionSql, - auth: ProjectAuthContext, - args: { - taskId: string; - body: string; - bodyByLocale?: Record; - author?: CommentAuthor; - }, -): Promise<{ + taskId: string, + work: () => Promise, +): Promise { + const queueKey = taskCommentQueueKey(taskId); + try { + await tx` + SELECT pg_advisory_xact_lock(${RETRY_QUEUE_LOCK_CLASS}, hashtext(${queueKey})) + `; + return await work(); + } catch (error) { + throw markRetryQueueKey(error, queueKey); + } +} + +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 @@ -496,12 +550,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: Awaited>, +): 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/integration-check.ts b/services/platform/backend/integration-check.ts index ee0e177b9a..384c6cb3dc 100644 --- a/services/platform/backend/integration-check.ts +++ b/services/platform/backend/integration-check.ts @@ -5345,6 +5345,85 @@ 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)`, + ); +} + async function checkRateLimitShapes( sql: Sql, base: string, @@ -44136,6 +44215,7 @@ async function main(): Promise { ], ['checkSmallDomains', () => checkSmallDomains(sql, baseUrl, authCtx)], ['checkMessageSlots', () => checkMessageSlots(sql, authCtx)], + ['checkTaskCommentBurst', () => checkTaskCommentBurst(sql, authCtx)], [ 'checkRateLimitShapes', () => checkRateLimitShapes(sql, baseUrl, authCtx), From 9dba83b38ca1c7cbe3ac42d1a2ba85ee21e2ae1d Mon Sep 17 00:00:00 2001 From: larryro <371767072@qq.com> Date: Sun, 6 Sep 2026 16:34:20 +0800 Subject: [PATCH 2/6] fix(shared): queue serializable retries on every key a failure carries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A retry queued on one key drops the others: a task comment that lost at the org's audit chain head was re-marked with the task key by its outer queue, so its retry took the task's session lock alone, met the other tasks' commenters at the chain head again and lost there — the storm #3246 removed came back for cross-task bursts. Marks now nest. `markRetryQueueKey` prepends an outer key to the keys an inner callback already put on the error (never twice), so the list reads in the callback's own acquisition order; `beginQueued` takes the session locks in that order before BEGIN and unlocks in reverse; the retry loop adopts the whole list whenever a marked failure arrives. Every first attempt takes its transaction-level locks in the same order, so a queued retry and a first attempt never hold two keys in opposite orders. `retryQueueKeysOf` exposes the list; `retryQueueKeyOf` keeps answering the outermost key. --- packages/shared/src/db/serializable.test.ts | 66 ++++++++++++++ packages/shared/src/db/serializable.ts | 96 ++++++++++++++------- 2 files changed, 132 insertions(+), 30 deletions(-) diff --git a/packages/shared/src/db/serializable.test.ts b/packages/shared/src/db/serializable.test.ts index f2b95bdd36..ad74d7a91e 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'; @@ -383,3 +384,68 @@ 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); + }); +}); diff --git a/packages/shared/src/db/serializable.ts b/packages/shared/src/db/serializable.ts index 5557219ff7..39073dffd0 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,19 +180,24 @@ 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. */ 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[] = []; + for (const key of keys) { + await reserved` + SELECT pg_advisory_lock(${RETRY_QUEUE_LOCK_CLASS}, hashtext(${key})) + `; + locked.push(key); + } try { await reserved.unsafe('BEGIN ISOLATION LEVEL SERIALIZABLE'); let result: T; @@ -190,14 +221,16 @@ 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(); @@ -208,18 +241,18 @@ async function beginQueued( * 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 +265,10 @@ export function transactSerializable( sleep: jitteredSleep, ...options, isTransient: (error) => { - queueKey = retryQueueKeyOf(error) ?? queueKey; + const keys = retryQueueKeysOf(error); + if (keys.length > 0) { + queueKeys = keys; + } return isTransient(error); }, }); From b53b7a0374dedc832931e27aade598eb31a8c8b4 Mon Sep 17 00:00:00 2001 From: larryro <371767072@qq.com> Date: Sun, 6 Sep 2026 16:34:22 +0800 Subject: [PATCH 3/6] fix(platform): take the task comment key before the overdue nudge's claim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The overdue ladder's level-2 rung stamps the task row and then comments on the task, while every commenter takes the task's comment key first and writes the row second — two lock orders for one pair of resources, which Postgres resolves after deadlock_timeout with a 40P01 that the READ COMMITTED commenters (agents, workflows, REST) surface as an error. The claim now takes the key as its first statement, the order everyone else uses; the unit test pins key-before-claim. --- .../domains/tasks/date-notifications.test.ts | 16 ++++++++++++++-- .../backend/domains/tasks/date-notifications.ts | 13 ++++++++++--- 2 files changed, 24 insertions(+), 5 deletions(-) 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); From de4857a7a0cbaefd527e3a20ffb566bb2cd5aad4 Mon Sep 17 00:00:00 2001 From: larryro <371767072@qq.com> Date: Sun, 6 Sep 2026 16:34:24 +0800 Subject: [PATCH 4/6] fix(platform): keep the chain-head mark under the task comment queue `queuedOnTask` marks a serialization failure with the task key on top of the chain-head key the audit write inside it already carries, and the shared retry queue now honours both, so a retry holds the task AND the org's chain head from before its BEGIN. The lock statement is exported (`lockTaskCommentQueue`) for transactions that write the task row before commenting, the meta row shape has a name, and unit tests pin the contract: lock first, task key in front of an inner audit key, other failures untouched. The integration proof gains the cross-task lane: twelve serializable commenters on twelve tasks of one org meet only at the chain head, all land, the chain verifies, and attempts stay within twice the burst. --- .../domains/tasks/comments.queue.test.ts | 112 ++++++++++++++++++ .../backend/domains/tasks/comments.ts | 49 +++++--- .../platform/backend/integration-check.ts | 79 ++++++++++++ 3 files changed, 224 insertions(+), 16 deletions(-) create mode 100644 services/platform/backend/domains/tasks/comments.queue.test.ts 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 23bf240669..a10d55fe12 100644 --- a/services/platform/backend/domains/tasks/comments.ts +++ b/services/platform/backend/domains/tasks/comments.ts @@ -97,10 +97,25 @@ async function ensureTaskDiscussionThread( } /** Retry-queue key for one task's discussion (see `queuedOnTask`). */ -function taskCommentQueueKey(taskId: string): string { +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, + 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 @@ -113,23 +128,23 @@ function taskCommentQueueKey(taskId: string): string { * 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. Under contention a writer wastes at most one - * attempt. Plain READ COMMITTED callers pay only the lock, which orders the - * task's comments and marks nothing. + * 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. */ -async function queuedOnTask( +export async function queuedOnTask( tx: TransactionSql, taskId: string, work: () => Promise, ): Promise { - const queueKey = taskCommentQueueKey(taskId); try { - await tx` - SELECT pg_advisory_xact_lock(${RETRY_QUEUE_LOCK_CLASS}, hashtext(${queueKey})) - `; + await lockTaskCommentQueue(tx, taskId); return await work(); } catch (error) { - throw markRetryQueueKey(error, queueKey); + throw markRetryQueueKey(error, taskCommentQueueKey(taskId)); } } @@ -415,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; @@ -567,7 +584,7 @@ async function removeTaskComment( tx: TransactionSql, auth: ProjectAuthContext, messageId: string, - meta: Awaited>, + meta: CommentMeta, ): Promise { const task = await loadTaskOrThrow(tx, meta.taskId, auth.organizationId); const project = await loadProjectOrThrow(tx, task.projectId); diff --git a/services/platform/backend/integration-check.ts b/services/platform/backend/integration-check.ts index 384c6cb3dc..a20fd58344 100644 --- a/services/platform/backend/integration-check.ts +++ b/services/platform/backend/integration-check.ts @@ -5424,6 +5424,81 @@ async function checkTaskCommentBurst( ); } +/** + * 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, @@ -44216,6 +44291,10 @@ 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), From 3b127927201497cb60e9a70beecc3827ce365765 Mon Sep 17 00:00:00 2001 From: larryro <371767072@qq.com> Date: Sun, 6 Sep 2026 16:51:17 +0800 Subject: [PATCH 5/6] fix(shared): release the queue locks already held when a later key fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The key list made `beginQueued` take several session locks in a row, but the loop sat outside the unlock guard: a failure on the second key (a cancelled statement, lock-table exhaustion) skipped the unlocks and returned the connection to the pool still holding the first key — every later transaction-level lock on that key would then wait until that pooled connection died. The loop now runs inside the guard, so the keys already held are unlocked in reverse on every exit and the connection is released clean. The retry loop also keeps its key list when a marked failure carries a subset of it: a queued attempt that loses to a writer outside its inner queues is re-marked with the outer key alone, and dropping the inner key is how a retry loses at that resource again. --- packages/shared/src/db/serializable.test.ts | 78 +++++++++++++++++++-- packages/shared/src/db/serializable.ts | 30 +++++--- 2 files changed, 96 insertions(+), 12 deletions(-) diff --git a/packages/shared/src/db/serializable.test.ts b/packages/shared/src/db/serializable.test.ts index ad74d7a91e..04e92f3d6d 100644 --- a/packages/shared/src/db/serializable.test.ts +++ b/packages/shared/src/db/serializable.test.ts @@ -139,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; @@ -156,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: [] }); @@ -182,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; @@ -196,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); }, @@ -449,3 +458,64 @@ describe('transactSerializable — nested retry queues', () => { 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); + 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 39073dffd0..c778dc15ac 100644 --- a/packages/shared/src/db/serializable.ts +++ b/packages/shared/src/db/serializable.ts @@ -182,7 +182,10 @@ function transactionOver(reserved: ReservedSql): TransactionSql { /** * One serializable attempt queued on `keys`: session advisory locks in that * order → BEGIN → callback → COMMIT → unlocks in reverse, all on one - * reserved connection. + * 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, @@ -192,13 +195,13 @@ async function beginQueued( const reserved = await reserve(); try { const locked: string[] = []; - for (const key of keys) { - await reserved` - SELECT pg_advisory_lock(${RETRY_QUEUE_LOCK_CLASS}, hashtext(${key})) - `; - locked.push(key); - } 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 { @@ -243,6 +246,13 @@ async function beginQueued( * connection faults. Each retry runs in a fresh transaction; a retry after a * failure marked with queue keys runs queued on those keys (see above). */ +function isSubsetOf( + keys: readonly string[], + of: readonly string[] | undefined, +): boolean { + return of !== undefined && keys.every((key) => of.includes(key)); +} + export function transactSerializable( sql: SerializableTransactionRunner, callback: (tx: TransactionSql) => Promise, @@ -265,8 +275,12 @@ export function transactSerializable( sleep: jitteredSleep, ...options, isTransient: (error) => { + // 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) { + if (keys.length > 0 && !isSubsetOf(keys, queueKeys)) { queueKeys = keys; } return isTransient(error); From 1aced44cc1368b7bb450d0419366caa0a34cd52a Mon Sep 17 00:00:00 2001 From: larryro <371767072@qq.com> Date: Sun, 6 Sep 2026 16:59:05 +0800 Subject: [PATCH 6/6] refactor(shared): keep the retry loop's doc on the retry loop The subset helper had landed between transactSerializable's JSDoc and the function; it now sits above with its own line. The cancelled-lock test also asserts that no second reservation happens, so a future widening of the transient table is caught here rather than trusted. --- packages/shared/src/db/serializable.test.ts | 2 ++ packages/shared/src/db/serializable.ts | 13 +++++++------ 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/packages/shared/src/db/serializable.test.ts b/packages/shared/src/db/serializable.test.ts index 04e92f3d6d..cab1ac5157 100644 --- a/packages/shared/src/db/serializable.test.ts +++ b/packages/shared/src/db/serializable.test.ts @@ -484,6 +484,8 @@ describe('transactSerializable — nested retry queues, failure paths', () => { }), ).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(?))', diff --git a/packages/shared/src/db/serializable.ts b/packages/shared/src/db/serializable.ts index c778dc15ac..487c424a92 100644 --- a/packages/shared/src/db/serializable.ts +++ b/packages/shared/src/db/serializable.ts @@ -240,12 +240,7 @@ async function beginQueued( } } -/** - * 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 queue keys runs queued on those keys (see above). - */ +/** True when every key in `keys` is already in `of`. */ function isSubsetOf( keys: readonly string[], of: readonly string[] | undefined, @@ -253,6 +248,12 @@ function isSubsetOf( 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 queue keys runs queued on those keys (see above). + */ export function transactSerializable( sql: SerializableTransactionRunner, callback: (tx: TransactionSql) => Promise,