From 27a26c9614d24cf57707da6a69d79d220d427b31 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Sat, 10 Oct 2026 23:22:49 +0200 Subject: [PATCH] Rebuild: a release key taken by another group's writer gets the next number Writers lock their group only, so two groups' batches can number the same identity key at once (a spam post crossposted to 67 groups: duplicate key ...:65, HTTP 500, the batch retried). Releases are now inserted with ON CONFLICT (release_key) DO NOTHING, and those skipped take the next number (up to 20 tries). Co-Authored-By: Claude Opus 5.5 --- app/services/batch_store.ts | 40 +++++++++++++++++++--------- tests/functional/batch_store.spec.ts | 11 ++++++++ 2 files changed, 38 insertions(+), 13 deletions(-) diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index 11a5b6b..6d40d33 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -185,22 +185,36 @@ class BatchStore { // The first upload of a key keeps the key; later ones are key:2, key:3, … const { rows: counts } = await trx.rawQuery(`SELECT identity_key, count(*) AS n FROM releases WHERE identity_key = ANY(?) GROUP BY 1`, [[...new Set(fresh.map((r) => r.key))]]) const taken = new Map(counts.map((c: { identity_key: string; n: string }) => [c.identity_key, Number(c.n)])) - const inserts = fresh.map((r) => { + const numbered = (r: ReleaseChunk) => { const n = (taken.get(r.key) ?? 0) + 1 taken.set(r.key, n) return { release_key: n === 1 ? r.key : `${r.key}:${n}`, key: r.key, upload: r.upload, ...row(r) } - }) - const { rows } = await trx.rawQuery( - `INSERT INTO releases (release_key, identity_key, group_id, name, source, poster, groups, posted_at, last_posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2, complete) - SELECT v.release_key, v.key, ?, v.name, v.source, v.poster, v.groups, to_timestamp(v.first / 1000.0), to_timestamp(v.last / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, - v.first_article, v.last_article, v.readable, v.has_par2, CASE WHEN v.file_total IS NULL THEN NULL ELSE v.files >= v.file_total END - FROM jsonb_to_recordset(?::jsonb) AS v(release_key text, key text, name text, source text, poster text, groups jsonb, first bigint, last bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean) - RETURNING id, release_key`, - [groupId, JSON.stringify(inserts)] - ) - const byReleaseKey = new Map(rows.map((x: { id: string; release_key: string }) => [x.release_key, Number(x.id)])) - for (const r of inserts) ids.set(slot(r.key, r.upload), byReleaseKey.get(r.release_key)!) - out.newReleases += rows.length + } + let inserts = fresh.map(numbered) + // Writers of other groups number the same key at the same time (a spam post crossposted to 67 + // groups; only the group is locked): a number taken meanwhile is skipped, and the next one tried. + for (let attempt = 0; inserts.length > 0; attempt++) { + if (attempt === 20) throw new Error(`No free release key for ${inserts[0].key}`) + const { rows } = await trx.rawQuery( + `INSERT INTO releases (release_key, identity_key, group_id, name, source, poster, groups, posted_at, last_posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2, complete) + SELECT v.release_key, v.key, ?, v.name, v.source, v.poster, v.groups, to_timestamp(v.first / 1000.0), to_timestamp(v.last / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, + v.first_article, v.last_article, v.readable, v.has_par2, CASE WHEN v.file_total IS NULL THEN NULL ELSE v.files >= v.file_total END + FROM jsonb_to_recordset(?::jsonb) AS v(release_key text, key text, name text, source text, poster text, groups jsonb, first bigint, last bigint, file_total int, files int, segments bigint, bytes bigint, first_article bigint, last_article bigint, readable boolean, has_par2 boolean) + ON CONFLICT (release_key) DO NOTHING + RETURNING id, release_key`, + [groupId, JSON.stringify(inserts)] + ) + const byReleaseKey = new Map(rows.map((x: { id: string; release_key: string }) => [x.release_key, Number(x.id)])) + const missed: typeof inserts = [] + for (const r of inserts) { + const id = byReleaseKey.get(r.release_key) + if (id === undefined) missed.push(r) + else ids.set(slot(r.key, r.upload), id) + } + out.newReleases += rows.length + const chunks = new Map(fresh.map((r) => [slot(r.key, r.upload), r])) + inserts = missed.map((r) => numbered(chunks.get(slot(r.key, r.upload))!)) + } } } diff --git a/tests/functional/batch_store.spec.ts b/tests/functional/batch_store.spec.ts index c36abed..4bfba0c 100644 --- a/tests/functional/batch_store.spec.ts +++ b/tests/functional/batch_store.spec.ts @@ -127,4 +127,15 @@ test.group('BatchStore', (group) => { assert.lengthOf(rows, 2) assert.equal(rows[0].identity_key, rows[1].identity_key) }) + + test('a release key taken meanwhile by another group\'s writer is skipped for the next number', async ({ assert }) => { + await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:00:00Z')) + const [first] = await releasesOf() + // As if another group's batch numbered the key at the same moment: key:2 exists, not counted for this key. + const [g] = await db.table('groups').insert({ name: 'alt.binaries.third', priority: 3 }).returning('id') + await db.table('releases').insert({ release_key: `${first.identity_key}:2`, identity_key: 'someone-else', group_id: g.id, name: 'x', source: 'subject', poster: 'p', groups: JSON.stringify([]), files: 0, segments: 0, bytes: 0, chunks: 0 }) + await storeIn('alt.binaries.other', 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:01:00Z', 'alt.binaries.other')) + const keys = (await releasesOf()).filter((r) => r.identity_key === first.identity_key).map((r) => r.release_key) + assert.deepEqual(keys, [first.identity_key, `${first.identity_key}:3`]) + }) })