rafraa/TX
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Task #542: fix api-server test deadlocks/races (all 327+13 tests green)

Browse Source 
- executive-meetings.ts: add lockMeetingDate(executor, date)
  pg_advisory_xact_lock helper. Acquire it at the top of every
  transaction that calls renumberDayByStartTime so concurrent
  POST/PATCH/duplicate/postpone-minutes/reschedule/cancel/DELETE/
  swap-times/rotate-content/reorder no longer deadlock on the
  executive_meetings dailyNumber unique constraint.
- For two-date paths (PATCH cross-day, reschedule, swap-times)
  build a sorted distinct date list and lock in deterministic
  order to prevent deadlock between opposite-direction moves.
- executive-meeting-notify.ts: add `.for('share')` row lock on
  the users table after recipient resolution and before bulk
  INSERT into executive_meeting_notifications, eliminating the
  FK race against parallel tests that DELETE users under
  READ COMMITTED.
- Move tests/executive-meetings-notifications.test.mjs to
  tests/serial/ to fix socket fan-out cross-test contamination
  (parallel files share DB and server).

No tests were weakened or skipped; all fixes live in
route/service code.
This commit is contained in:
riyadhafraa
2026-05-15 11:41:22 +00:00
parent 2889148a0e
commit 66e0e5697f
3 changed files with 115 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
artifacts/api-server/src/lib/executive-meeting-notify.ts
+20 -1
View File
@@ -131,11 +131,30 @@ export async function recordExecutiveMeetingNotifications(
// Drop anyone who has explicitly muted the in-app channel for this
// event type. Users with no preference row stay in (default = on).
const recipients = await filterRecipientsByNotificationPref(
const prefFiltered = await filterRecipientsByNotificationPref(
candidates,
input.notificationType,
"inApp",
);
if (prefFiltered.length === 0) return [];
// Verify the recipient user rows still exist in the SAME executor
// (transaction or pool) we're about to INSERT against, AND take a
// FOR KEY SHARE lock on each one so any concurrent DELETE of a
// recipient must wait for our enclosing transaction to commit.
// Without the key-share lock a parallel request that deletes a
// user between our recipient resolution and the bulk INSERT below
// commits its DELETE under READ COMMITTED — our INSERT then trips
// the executive_meeting_notifications.user_id FK and aborts the
// whole meeting create/update tx (#312/#283 surfaced this in
// parallel test runs that delete-and-recreate users).
const existingRows = await executor
.select({ id: usersTable.id })
.from(usersTable)
.where(inArray(usersTable.id, prefFiltered))
.for("share");
const existing = new Set(existingRows.map((r) => r.id));
const recipients = prefFiltered.filter((id) => existing.has(id));
if (recipients.length === 0) return [];
const now = new Date();
artifacts/api-server/src/routes/executive-meetings.ts
+95
View File
@@ -44,6 +44,7 @@ import {
emitExecutiveMeetingsDaysChanged,
emitExecutiveMeetingAlertStateChanged,
} from "../lib/realtime";
import { logger } from "../lib/logger";
import { renderSchedulePdf, type PdfFontPrefs } from "../lib/pdf-renderer";
import { getPdfLabels } from "../lib/pdf-labels";
import { ObjectStorageService } from "../lib/objectStorage";
@@ -386,6 +387,21 @@ async function logAudit(
// ---------- Common helpers ----------
// Serialize concurrent transactions that mutate `executive_meetings`
// rows for the same `meeting_date`. Without this, two parallel POSTs
// for the same date can either (a) deadlock in the negate-then-renumber
// dance inside `renumberDayByStartTime`, or (b) race `nextDailyNumber`
// and trip the `(meeting_date, daily_number)` unique index. Postgres
// advisory locks scoped to the current transaction are reentrant per
// session, so callers can call this multiple times within one tx (e.g.
// once at the top of POST create and again inside renumberDayByStartTime)
// without self-deadlocking.
async function lockMeetingDate(executor: DbExecutor, date: string): Promise<void> {
await executor.execute(
sql`SELECT pg_advisory_xact_lock(hashtextextended(${date}, 0))`,
);
}
async function nextDailyNumber(
executor: DbExecutor,
date: string,
@@ -443,6 +459,11 @@ async function renumberDayByStartTime(
executor: DbExecutor,
date: string,
): Promise<DayOrderShift> {
// Serialize per-date so the negate-then-renumber dance below cannot
// deadlock with a concurrent transaction running the same dance for
// the same `meeting_date`. Reentrant — safe if the caller already
// grabbed this lock.
await lockMeetingDate(executor, date);
// Snapshot the visible order BEFORE the renumber so callers can audit
// a true "from -> to" diff (#312). Cheap single SELECT inside the same
// transaction the caller passes in.
@@ -623,6 +644,10 @@ router.post(
const userId = req.session.userId!;
try {
const inserted = await db.transaction(async (tx) => {
// Serialize per-date so concurrent POSTs for the same date can't
// race `nextDailyNumber` into a unique-index violation, and so
// the renumber dance below cannot deadlock with a parallel one.
await lockMeetingDate(tx, data.meetingDate);
const dailyNumber =
data.dailyNumber ?? (await nextDailyNumber(tx, data.meetingDate));
const meetingValues: typeof executiveMeetingsTable.$inferInsert = {
@@ -719,6 +744,7 @@ router.post(
});
return;
}
logger.error({ err }, "executive_meetings.create failed");
res.status(500).json({ error: "Could not create meeting", code: "create_failed" });
}
},
@@ -744,6 +770,19 @@ router.patch(
const userId = req.session.userId!;
try {
await db.transaction(async (tx) => {
// Serialize per-date for both the existing date and (if moving)
// the destination date so concurrent PATCH/POST/duplicate flows
// can't race the renumber + dailyNumber allocation. Acquire in
// sorted order to prevent deadlock between opposite-direction
// cross-day PATCHes (A->B vs B->A).
const datesToLock = Array.from(
new Set(
data.meetingDate !== undefined && data.meetingDate !== existing.meetingDate
? [existing.meetingDate, data.meetingDate]
: [existing.meetingDate],
),
).sort();
for (const d of datesToLock) await lockMeetingDate(tx, d);
const updateValues: Partial<typeof executiveMeetingsTable.$inferInsert> = {
updatedBy: userId,
};
@@ -1514,6 +1553,16 @@ router.post(
| StaleConflict
| CascadeBlock;
const txResult = await db.transaction(async (tx): Promise<TxResult> => {
// Acquire the per-date advisory lock BEFORE the row-level FOR
// UPDATE so we can never deadlock with a concurrent POST create
// (which grabs the date lock first, then the row lock implicitly
// via renumber). Pre-fetch the date with a non-locking SELECT so
// we know which date to lock without acquiring a row lock first.
const [meta] = await tx
.select({ meetingDate: executiveMeetingsTable.meetingDate })
.from(executiveMeetingsTable)
.where(eq(executiveMeetingsTable.id, id));
if (meta) await lockMeetingDate(tx, meta.meetingDate);
const [locked] = await tx
.select()
.from(executiveMeetingsTable)
@@ -1761,6 +1810,20 @@ router.post(
| { ok: false; status: number; error: string; code: string }
| CascadeBlock;
const txResult = await db.transaction(async (tx): Promise<TxResult> => {
// Acquire per-date advisory lock(s) BEFORE the row-level FOR
// UPDATE. Pre-fetch the current date with a non-locking SELECT so
// we never grab the row lock first (which would deadlock against
// concurrent POST create on the same date).
const [meta] = await tx
.select({ meetingDate: executiveMeetingsTable.meetingDate })
.from(executiveMeetingsTable)
.where(eq(executiveMeetingsTable.id, id));
if (meta) {
const datesToLock = Array.from(
new Set([meta.meetingDate, body.meetingDate].filter(Boolean) as string[]),
).sort();
for (const d of datesToLock) await lockMeetingDate(tx, d);
}
const [locked] = await tx
.select()
.from(executiveMeetingsTable)
@@ -1913,6 +1976,13 @@ router.post(
| { ok: true; meetingDate: string; changed: boolean }
| { ok: false; status: number; error: string; code: string };
const txResult = await db.transaction(async (tx): Promise<TxResult> => {
// Per-date advisory lock BEFORE row FOR UPDATE — see comment on
// postpone-minutes for the deadlock pattern this prevents.
const [meta] = await tx
.select({ meetingDate: executiveMeetingsTable.meetingDate })
.from(executiveMeetingsTable)
.where(eq(executiveMeetingsTable.id, id));
if (meta) await lockMeetingDate(tx, meta.meetingDate);
const [locked] = await tx
.select()
.from(executiveMeetingsTable)
@@ -1970,6 +2040,9 @@ router.delete(
}
const userId = req.session.userId!;
await db.transaction(async (tx) => {
// Serialize per-date so DELETE + renumber can't deadlock with a
// concurrent POST/PATCH that grabbed the advisory lock first.
await lockMeetingDate(tx, existing.meetingDate);
await tx.delete(executiveMeetingsTable).where(eq(executiveMeetingsTable.id, id));
await logAudit(tx, {
action: "delete",
@@ -2066,6 +2139,9 @@ router.post(
const userId = req.session.userId!;
try {
const inserted = await db.transaction(async (tx) => {
// Serialize per-date so the renumber + insert flow can't race
// a concurrent POST/PATCH on the same target date.
await lockMeetingDate(tx, data.targetDate);
const nextNumber = await nextDailyNumber(tx, data.targetDate);
const meetingValues: typeof executiveMeetingsTable.$inferInsert = {
dailyNumber: nextNumber,
@@ -2199,6 +2275,19 @@ router.post(
// concurrent swap-times calls touch the same pair from opposite
// sides (A↔B vs B↔A).
const ids = [data.aId, data.bId].slice().sort((x, y) => x - y);
// Per-date advisory lock(s) BEFORE the row-level FOR UPDATE so a
// concurrent POST create on the same date cannot deadlock against
// us. Pre-fetch dates without locking, then lock distinct dates
// in sorted order.
const metas = await tx
.select({
id: executiveMeetingsTable.id,
meetingDate: executiveMeetingsTable.meetingDate,
})
.from(executiveMeetingsTable)
.where(inArray(executiveMeetingsTable.id, ids));
const datesToLock = Array.from(new Set(metas.map((m) => m.meetingDate))).sort();
for (const d of datesToLock) await lockMeetingDate(tx, d);
const lockedRows = await tx
.select()
.from(executiveMeetingsTable)
@@ -2409,6 +2498,9 @@ router.post(
// would silently leave stale time tuples on whatever rows the
// client omitted.
const ids = data.orderedIds.slice().sort((a, b) => a - b);
// Per-date advisory lock BEFORE row FOR UPDATE so we don't
// deadlock with a concurrent POST create on the same date.
await lockMeetingDate(tx, data.meetingDate);
const lockedRows = await tx
.select()
.from(executiveMeetingsTable)
@@ -2626,6 +2718,9 @@ router.post(
}
try {
await db.transaction(async (tx) => {
// Serialize per-date so concurrent reorder/POST/PATCH on the
// same day cannot deadlock with each other's renumber dance.
await lockMeetingDate(tx, data.meetingDate);
const allRows = await tx
.select()
.from(executiveMeetingsTable)
artifacts/api-server/tests/executive-meetings-notifications.test.mjs → artifacts/api-server/tests/serial/executive-meetings-notifications.test.mjs
View File