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 <noreply@anthropic.com>

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<string, number>(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<string, number>(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<string, number>(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))!))
+        }
       }
     }