From 20910647fac02adcc786148ba5a104ae55c14bfa Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 00:21:17 +0200 Subject: [PATCH] Rebuild: the batch store (catalog upserts, write-once chunks, candidates, rule counts, fragments, resume point) A batch is stored in one transaction and only at its range's resume point, so a retried batch can't count twice. Garbage is judged before per-article: a dropped tool isn't worth reading bodies for either. Co-Authored-By: Claude Opus 5.5 --- REBUILD.md | 7 +- app/services/batch_store.ts | 116 +++++++++++++++++++++++++++ app/services/chunk_partitions.ts | 32 ++++++++ lib/edge/range_indexer.ts | 11 ++- lib/rules/rule_set.ts | 7 +- tests/functional/batch_store.spec.ts | 76 ++++++++++++++++++ 6 files changed, 242 insertions(+), 7 deletions(-) create mode 100644 app/services/batch_store.ts create mode 100644 app/services/chunk_partitions.ts create mode 100644 tests/functional/batch_store.spec.ts diff --git a/REBUILD.md b/REBUILD.md index 5e8d563..1432e7e 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -65,8 +65,11 @@ dropping garbage; 28 TB of compressed overview traffic. Throughput goal: 3. **Chunk store** (bulk, append-only). One record per release per batch: its files' names, sizes and segment lists, compactly encoded (template - codec for message-ids, zstd). Written once, never updated; a re-run of a - range replaces that range's chunks (no per-file dedup). ~3 TB. + codec for message-ids, zstd). Written once, never updated. A batch is + stored in one transaction and only when it starts at its range's resume + point, so a batch sent twice or late is refused and the catalog's sums + stay right (no per-file dedup); a recovery run for a retracted rule only + adds what that rule had dropped. ~3 TB. 4. **Release catalog** (hot, Postgres on NVMe). One row per release: key, name, poster, groups, posted_at, file count, size, segments, expected diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts new file mode 100644 index 0000000..1598886 --- /dev/null +++ b/app/services/batch_store.ts @@ -0,0 +1,116 @@ +import db from '@adonisjs/lucid/services/db' +import type { EdgeBatch } from '#lib/edge/index' +import ChunkPartitions from '#services/chunk_partitions' + +export class StaleBatch extends Error {} + +export interface StoredBatch { + releases: number + newReleases: number + candidates: number + fragments: number +} + +/** + * Stores one batch of a leased range (REBUILD.md, parts 3–7), in one + * transaction: release aggregates in the catalog, one chunk per release, + * probe candidates, rule counts and samples, edge fragments, and the + * range's resume point. A batch that doesn't start at the resume point was + * sent twice or too late, and is refused (StaleBatch), so the sums stay right. + */ +class BatchStore { + async store(rangeId: number, nodeId: number, batch: EdgeBatch): Promise { + await ChunkPartitions.ensure() + return db.transaction(async (trx) => { + const range = await trx.from('ranges').where('id', rangeId).forUpdate().first() + if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new StaleBatch(`Range ${rangeId} is not leased to this node`) + if (Number(range.next_id) !== batch.from) throw new StaleBatch(`Range ${rangeId} is at ${range.next_id}, the batch starts at ${batch.from}`) + const result: StoredBatch = { releases: batch.releases.length, newReleases: 0, candidates: 0, fragments: batch.fragments.length } + + // Releases, sorted by key so concurrent batches lock them in the same order. + const releases = [...batch.releases].sort((a, b) => a.key.localeCompare(b.key)) + const ids = new Map() + for (let i = 0; i < releases.length; i += 1000) { + const chunk = releases.slice(i, i + 1000) + const { rows } = await trx.rawQuery( + `INSERT INTO releases (release_key, name, source, poster, groups, posted_at, file_total, files, segments, bytes, chunks, first_article, last_article, readable, has_par2) + SELECT v.key, v.name, v.source, v.poster, v.groups, to_timestamp(v.posted_at / 1000.0), v.file_total, v.files, v.segments, v.bytes, 1, v.first_article, v.last_article, v.readable, v.has_par2 + FROM jsonb_to_recordset(?::jsonb) AS v(key text, name text, source text, poster text, groups jsonb, posted_at 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 UPDATE SET + files = releases.files + excluded.files, + segments = releases.segments + excluded.segments, + bytes = releases.bytes + excluded.bytes, + chunks = releases.chunks + 1, + posted_at = least(releases.posted_at, excluded.posted_at), + file_total = coalesce(releases.file_total, excluded.file_total), + first_article = least(releases.first_article, excluded.first_article), + last_article = greatest(releases.last_article, excluded.last_article), + readable = releases.readable OR excluded.readable, + has_par2 = releases.has_par2 OR excluded.has_par2, + last_seen_at = now() + RETURNING id, release_key, (xmax = 0) AS inserted`, + [ + JSON.stringify( + chunk.map((r) => ({ + key: r.key, name: r.name, source: r.source, poster: r.poster, groups: r.groups, posted_at: r.postedAt, file_total: r.fileTotal, + files: r.files, segments: r.segments, bytes: r.bytes, first_article: r.firstArticle, last_article: r.lastArticle, readable: r.readable, has_par2: r.hasPar2, + })) + ), + ] + ) + for (const row of rows) { + ids.set(row.release_key, Number(row.id)) + if (row.inserted) result.newReleases++ + } + } + + // One write-once chunk per release. + for (let i = 0; i < releases.length; i += 500) { + const rows = releases.slice(i, i + 500).map((r) => ({ + release_id: ids.get(r.key)!, + range_id: rangeId, + batch_seq: batch.from, + files: r.files, + segments: r.segments, + payload: r.payload, + })) + if (rows.length > 0) await trx.insertQuery().table('release_chunks').multiInsert(rows) + } + + const candidates = batch.candidates.filter((c) => ids.has(c.releaseKey)) + for (let i = 0; i < candidates.length; i += 2000) { + const rows = candidates.slice(i, i + 2000).map((c) => ({ release_id: ids.get(c.releaseKey)!, kind: c.kind, message_id: c.messageId, file: c.file })) + const added = await trx.knexQuery().table('probe_candidates').insert(rows).onConflict(['release_id', 'kind', 'message_id']).ignore().returning('id') + result.candidates += added.length + } + + if (batch.counts.length > 0) { + await trx.rawQuery( + `INSERT INTO rule_counts (range_id, rule, key, articles, files) + SELECT ?, v.rule, v.key, v.articles, v.files FROM jsonb_to_recordset(?::jsonb) AS v(rule text, key text, articles bigint, files bigint) + ON CONFLICT (range_id, rule, key) DO UPDATE SET articles = rule_counts.articles + excluded.articles, files = rule_counts.files + excluded.files`, + [rangeId, JSON.stringify(batch.counts)] + ) + } + + for (const s of batch.samples) { + await trx.insertQuery().table('rule_samples').insert({ + range_id: rangeId, rule: s.rule, rule_key: s.ruleKey, release_key: s.key, name: s.name, files: s.files, segments: s.segments, payload: s.payload, + }) + } + + for (let i = 0; i < batch.fragments.length; i += 2000) { + const rows = batch.fragments.slice(i, i + 2000).map((f) => ({ range_id: rangeId, file_key: f.fileKey, article_number: f.articleNumber, segment: JSON.stringify(f) })) + await trx.insertQuery().table('fragments').multiInsert(rows) + } + + await trx + .from('ranges') + .where('id', rangeId) + .update({ next_id: batch.next, headers: db.raw('headers + ?', [batch.headers]), heartbeat_at: new Date() }) + return result + }) + } +} + +export default new BatchStore() diff --git a/app/services/chunk_partitions.ts b/app/services/chunk_partitions.ts new file mode 100644 index 0000000..db0dd02 --- /dev/null +++ b/app/services/chunk_partitions.ts @@ -0,0 +1,32 @@ +import db from '@adonisjs/lucid/services/db' +import env from '#start/env' + +/** Months ahead that always have a release_chunks partition. */ +const AHEAD = 2 + +/** + * release_chunks is partitioned by month of writing (REBUILD.md, decision a). + * Inserts fail without a partition for now(), so the current and the next + * months' partitions are created before they are needed; checked once per + * process and month. + */ +class ChunkPartitions { + private ensuredFor: string | null = null + + async ensure(now = new Date()) { + const month = now.toISOString().slice(0, 7) + if (this.ensuredFor === month) return + const tablespace = env.get('CHUNK_TABLESPACE') + for (let i = 0; i <= AHEAD; i++) { + const from = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + i, 1)) + const to = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + i + 1, 1)) + const name = `release_chunks_${from.toISOString().slice(0, 7).replace('-', '')}` + await db.rawQuery( + `CREATE TABLE IF NOT EXISTS ${name} PARTITION OF release_chunks FOR VALUES FROM ('${from.toISOString()}') TO ('${to.toISOString()}')${tablespace ? ` TABLESPACE ${tablespace}` : ''}` + ) + } + this.ensuredFor = month + } +} + +export default new ChunkPartitions() diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts index f38f593..a408f0d 100644 --- a/lib/edge/range_indexer.ts +++ b/lib/edge/range_indexer.ts @@ -20,6 +20,9 @@ export interface ReleaseChunk { bytes: number firstArticle: number lastArticle: number + /** A file with a readable name; a par2 file. */ + readable: boolean + hasPar2: boolean payload: Buffer } @@ -63,6 +66,8 @@ export interface Fragment { } export interface EdgeBatch { + /** The range's resume point this batch starts at. */ + from: number /** First article number not covered yet. */ next: number headers: number @@ -193,8 +198,8 @@ export class RangeIndexer { } } - private empty(next: number): EdgeBatch { - return { next, headers: 0, releases: [], samples: [], candidates: [], counts: [], fragments: [] } + private empty(from: number): EdgeBatch { + return { from, next: from, headers: 0, releases: [], samples: [], candidates: [], counts: [], fragments: [] } } } @@ -247,6 +252,8 @@ function chunkOf(key: string, files: AggregatedFile[]): ReleaseCh bytes, firstArticle, lastArticle, + readable: chunkFiles.some((f) => !isRandomFile(f.name)), + hasPar2: chunkFiles.some((f) => /\.par2$/i.test(f.name)), payload: encodeChunk(chunkFiles), } } diff --git a/lib/rules/rule_set.ts b/lib/rules/rule_set.ts index b3a3dce..e46b5f3 100644 --- a/lib/rules/rule_set.ts +++ b/lib/rules/rule_set.ts @@ -53,17 +53,18 @@ export class RuleSet { /** * A release's verdict from its files: kept as a whole when any file has a - * readable name; else dropped as per-article random subjects when every - * file has few parts, as garbage when its signature or tool is dropped. + * readable name; else dropped as garbage when its signature or tool is + * dropped, as per-article random subjects when every file has few parts. */ judge(releaseKey: string, files: { name: string; totalParts: number; groups: string[]; firstMessageId: string }[]): Verdict { if (files.some((f) => !isRandomFile(f.name))) return { rule: null } const first = files[0] const tool = uploadTool(first.firstMessageId) const sample = explored(releaseKey, this.rules.samplePercent) - if (files.every((f) => f.totalParts <= this.rules.perArticleMaxParts)) return { rule: 'per-article', key: tool, sample } + // Garbage first: a dropped tool or signature isn't worth reading bodies for either. const group = first.groups[0] ?? '' if (this.tools.has(tool) || this.signatures.has(signatureKey(tool, group))) return { rule: 'garbage', key: `${tool} ${group}`, sample } + if (files.every((f) => f.totalParts <= this.rules.perArticleMaxParts)) return { rule: 'per-article', key: tool, sample } return { rule: null } } } diff --git a/tests/functional/batch_store.spec.ts b/tests/functional/batch_store.spec.ts new file mode 100644 index 0000000..55a0ee8 --- /dev/null +++ b/tests/functional/batch_store.spec.ts @@ -0,0 +1,76 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import BatchStore, { StaleBatch } from '#services/batch_store' +import { RangeIndexer } from '#lib/edge/index' +import { decodeChunk } from '#lib/chunks/index' +import { DEFAULT_RULES } from '#lib/rules/index' +import type { Overview } from '#lib/nntp/index' + +const GROUP = 'alt.binaries.test' +let id = 0 + +function post(at: number, name: string, parts: number, counter: string, tool: 'ngPost' | 'plain' = 'plain'): Overview[] { + return Array.from({ length: parts }, (_, i) => { + id++ + const mid = tool === 'ngPost' ? `${id.toString(16).padStart(32, '0')}@ngPost` : `art${id}@test` + return [at + i, { 'subject': `${counter} - "${name}" yEnc (${i + 1}/${parts})`, 'from': 'p ', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', 'message-id': `<${mid}>`, ':bytes': '700000', 'xref': `news.example ${GROUP}:${at + i}` }] as Overview + }) +} + +async function leasedRange() { + const [group] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id') + const [node] = await db.table('nodes').insert({ name: 'test-node' }).returning('id') + const [range] = await db + .table('ranges') + .insert({ group_id: group.id, start_id: 1, end_id: 1_000_000, next_id: 1, status: 'leased', node_id: node.id, rule_version: 1 }) + .returning('id') + return { rangeId: Number(range.id), nodeId: Number(node.id) } +} + +function indexer() { + return new RangeIndexer({ group: GROUP, start: 1, end: 1_000_000, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } }, edgeArticles: 1000 }) +} + +test.group('BatchStore', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + + test('stores releases, their chunks, probe candidates, rule counts and fragments, and moves the resume point', async ({ assert }) => { + const { rangeId, nodeId } = await leasedRange() + const ix = indexer() + ix.add([ + ...post(100_000, 'Some.Show.S01E01.1080p.WEB.part1.rar', 2, 'Some.Show.S01E01 [1/2]'), + ...post(100_010, 'Some.Show.S01E01.1080p.WEB.par2', 1, 'Some.Show.S01E01 [2/2]'), + ...post(100_020, 'Hs7dKq2LmZ9xWv3RtYp1Qa.part01.rar', 3, '[1/1]', 'ngPost'), + ...post(10, 'Edge.Cut.2020.part1.rar', 3, 'Edge.Cut [1/1]').slice(0, 1), + ], 500_000) + const first = ix.take() + const stored = await BatchStore.store(rangeId, nodeId, first) + assert.include(stored, { releases: 1, newReleases: 1, candidates: 2, fragments: 0 }) + + const release = await db.from('releases').first() + assert.include(release, { files: 2, segments: '3', chunks: 1, readable: true, has_par2: true, file_total: 2 }) + const chunk = await db.from('release_chunks').where('release_id', release.id).first() + assert.deepEqual(decodeChunk(chunk.payload).map((f) => f.name).sort(), ['Some.Show.S01E01.1080p.WEB.par2', 'Some.Show.S01E01.1080p.WEB.part1.rar']) + assert.deepEqual((await db.from('probe_candidates').orderBy('kind')).map((c) => c.kind), ['archive', 'par2']) + assert.deepEqual((await db.from('rule_counts')).map((c) => [c.rule, c.key, Number(c.articles)]), [['garbage', 'ngPost alt.binaries.test', 3]]) + assert.equal(Number((await db.from('ranges').where('id', rangeId).first()).next_id), 500_000) + + // The same batch again (a retry after a lost reply) is refused, and changes nothing. + await assert.rejects(() => BatchStore.store(rangeId, nodeId, first), StaleBatch) + assert.equal((await db.from('releases').first()).files, 2) + + // The end of the range: the cut-off file near the start edge is kept as a fragment. + ix.finish() + const last = ix.take() + assert.include(await BatchStore.store(rangeId, nodeId, last), { fragments: 1 }) + assert.equal(Number((await db.from('ranges').where('id', rangeId).first()).next_id), 1_000_001) + }) + + test('a range leased to another node is refused', async ({ assert }) => { + const { rangeId, nodeId } = await leasedRange() + const ix = indexer() + ix.add(post(100_000, 'A.Release.2020.mkv', 1, 'A [1/1]'), 200_000) + await assert.rejects(() => BatchStore.store(rangeId, nodeId + 1, ix.take()), /not leased to this node/) + }) +})