From 508d8b5723aef4ca05597a63e70a90fec086d37b Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 14:10:48 +0200 Subject: [PATCH] Rebuild: one advisory lock per group when writing releases, not per key A picture group's batch has 30-55k release keys; an advisory lock each ran Postgres out of shared memory (max_locks_per_transaction), and those batches failed with HTTP 500 (the ranges went back, nothing stored). A chunk only matches releases of its range's group, so one lock per group serializes exactly the writers that could both start an upload. Co-Authored-By: Claude Opus 5.5 --- app/services/batch_store.ts | 9 +++++---- app/services/fragment_merger.ts | 2 +- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index 0a9a965..4a910e8 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -97,8 +97,8 @@ class BatchStore { * probe candidates (a batch's, or files the fragment merger completed). * A chunk joins the release of its key whose articles the range's group * owns and whose posting window is within UPLOAD_GAP_MS of its own, else - * starts one: re-uploads are releases of their own (decision d). Keys are - * locked in order, so concurrent writers don't deadlock or both start one. + * starts one: re-uploads are releases of their own (decision d). The + * group is locked, so concurrent writers don't both start one. */ async writeReleases( trx: TransactionClientContract, @@ -111,11 +111,12 @@ class BatchStore { const ids = new Map() const slot = (key: string, upload: number) => `${key}\n${upload}` const { group_id: groupId } = await trx.from('ranges').where('id', rangeId).select('group_id').firstOrFail() + // One writer per group at a time: a chunk only matches releases of its group, and a lock per key + // ran out of shared memory (a picture group's batch has 50k keys; LOG.md 2026-10-09 (12)). + await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended('releases:' || ?::text, 0))`, [groupId]) const sorted = [...releases].sort((a, b) => a.key.localeCompare(b.key) || a.upload - b.upload) for (let i = 0; i < sorted.length; i += 1000) { const part = sorted.slice(i, i + 1000) - const keys = [...new Set(part.map((r) => r.key))] - await trx.rawQuery(`SELECT pg_advisory_xact_lock(hashtextextended(k, 0)) FROM unnest(?::text[]) AS k ORDER BY k`, [keys]) const windows = part.map((r, idx) => ({ idx, key: r.key, first: r.postedAt, last: r.postedLast ?? r.postedAt })) const { rows: matches } = await trx.rawQuery( `SELECT DISTINCT ON (v.idx) v.idx, r.id diff --git a/app/services/fragment_merger.ts b/app/services/fragment_merger.ts index a803792..df708fc 100644 --- a/app/services/fragment_merger.ts +++ b/app/services/fragment_merger.ts @@ -89,7 +89,7 @@ class FragmentMerger { } await ChunkPartitions.ensure() await db.transaction(async (trx) => { - // In key order, as batches lock keys (no deadlocks). + // In key order (batches lock their group; the merger only ever waits on one batch at a time). releases.sort((a, b) => a.chunk.key.localeCompare(b.chunk.key) || a.chunk.upload - b.chunk.upload) for (const { rangeId, chunk, candidates } of releases) await BatchStore.writeReleases(trx, rangeId, 0, [chunk], candidates) const done = [...completed.flatMap((c) => c.ids), ...incomplete.flatMap((c) => c.ids)]