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`]) + }) })