From e0a0d0d5200c36ecd7c267bd21aaf43d2f6d99f9 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Sun, 11 Oct 2026 22:28:57 +0200 Subject: [PATCH] An upload spread over groups is one release FTD posts one upload's files to different groups ("Knowsum-Loops_Are_My_ Religion-WEB-2026-FTD": par2 and part1 to a.b.music.misc, part2 to a.b.sounds.music, each article in one group only; releases 81825221 and 81825412): the catalog matched a chunk only to releases of its own group. A chunk without one there now joins a release of its key in another group whose window is within 3 hours, when none of its file names are in that release yet; a post of the same files to another group (other message-ids, same names) stays a release of its own, or its NZB would hold every file twice. Functional tests for both. Co-Authored-By: Claude Opus 5.5 --- app/services/batch_store.ts | 50 ++++++++++++++++++++++++++++ tests/functional/batch_store.spec.ts | 8 +++++ 2 files changed, 58 insertions(+) diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index 6d40d33..4cf461e 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -2,6 +2,7 @@ import db from '@adonisjs/lucid/services/db' import type { TransactionClientContract } from '@adonisjs/lucid/types/database' import { UPLOAD_GAP_MS, type EdgeBatch, type ProbeCandidate, type ReleaseChunk } from '#lib/edge/index' import ChunkPartitions from '#services/chunk_partitions' +import { decodeChunk } from '#lib/chunks/index' export class StaleBatch extends Error {} @@ -101,6 +102,54 @@ class BatchStore { }) } + /** + * An upload spread over groups (FTD: par2 and part1 to a.b.music.misc, + * part2 to a.b.sounds.music, each article in one group only; LOG.md + * 2026-10-11 (9)): a chunk without a release in its own group joins one of + * its key in another group whose window is within UPLOAD_GAP_MS, when none + * of its file names are in that release yet. A post of the same files to + * another group (other message-ids, the same names) stays a release of its + * own, or the NZB would hold every file twice. Reads the candidate's file + * names from its chunks; candidates are few. + */ + private async joinAcrossGroups( + trx: TransactionClientContract, + part: ReleaseChunk[], + windows: { idx: number; key: string; first: number | null | undefined; last: number | null | undefined }[], + groupId: number, + joined: Map + ) { + const open = windows.filter((w) => !joined.has(w.idx) && w.first) + if (open.length === 0) return + const { rows: candidates } = await trx.rawQuery( + `SELECT v.idx, r.id + FROM jsonb_to_recordset(?::jsonb) AS v(idx int, key text, first bigint, last bigint) + JOIN releases r ON r.identity_key = v.key AND r.group_id <> ? + WHERE r.posted_at <= to_timestamp((v.last + ?) / 1000.0) AND coalesce(r.last_posted_at, r.posted_at) >= to_timestamp((v.first - ?) / 1000.0) + ORDER BY v.idx, r.last_posted_at DESC NULLS LAST, r.id DESC`, + [JSON.stringify(open), groupId, UPLOAD_GAP_MS, UPLOAD_GAP_MS] + ) + if (candidates.length === 0) return + const ids: number[] = [...new Set(candidates.map((c: { id: string }) => Number(c.id)))] + const names = new Map>() + const { rows: chunks } = await trx.rawQuery(`SELECT release_id, payload FROM release_chunks WHERE release_id = ANY(?)`, [ids]) + for (const c of chunks) { + const set = names.get(Number(c.release_id)) ?? new Set() + for (const f of decodeChunk(c.payload)) set.add(f.name) + names.set(Number(c.release_id), set) + } + for (const c of candidates as { idx: number; id: string }[]) { + if (joined.has(c.idx)) continue + const have = names.get(Number(c.id)) ?? new Set() + const incoming = decodeChunk(part[c.idx].payload).map((f) => f.name) + if (incoming.every((name) => !have.has(name))) { + joined.set(c.idx, Number(c.id)) + for (const name of incoming) have.add(name) + names.set(Number(c.id), have) + } + } + } + /** * Release aggregates in the catalog, one write-once chunk per release, and * probe candidates (a batch's, or files the fragment merger completed). @@ -137,6 +186,7 @@ class BatchStore { [JSON.stringify(windows), groupId, UPLOAD_GAP_MS, UPLOAD_GAP_MS] ) const joined = new Map(matches.map((m: { idx: number; id: string }) => [m.idx, Number(m.id)])) + await this.joinAcrossGroups(trx, part, windows, groupId, joined) const row = (r: ReleaseChunk) => ({ name: r.name, source: r.source, diff --git a/tests/functional/batch_store.spec.ts b/tests/functional/batch_store.spec.ts index 4bfba0c..48f261d 100644 --- a/tests/functional/batch_store.spec.ts +++ b/tests/functional/batch_store.spec.ts @@ -128,6 +128,14 @@ test.group('BatchStore', (group) => { assert.equal(rows[0].identity_key, rows[1].identity_key) }) + test('an upload spread over groups (other files in another group) is one release', async ({ assert }) => { + await storeIn(GROUP, 1, upload(10, 'Some.Album-WEB-2026-FTD.part1.rar', 2, 'Some.Album-WEB-2026-FTD [2/3]', '2026-10-05T10:00:00Z')) + await storeIn('alt.binaries.other', 1, upload(10, 'Some.Album-WEB-2026-FTD.part2.rar', 2, 'Some.Album-WEB-2026-FTD [3/3]', '2026-10-05T10:00:30Z', 'alt.binaries.other')) + const rows = await releasesOf() + assert.lengthOf(rows, 1) + assert.equal(rows[0].files, 2) + }) + 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()