From 68eb9a44fdfbb87f8c4f5958581c08013bc5a290 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 5 Oct 2026 10:41:18 +0200 Subject: [PATCH] Edge aggregation: seed:range indexes history one row per file A seeding node fetches a range over one NNTP connection, parses headers with the live rules, groups them into files in memory (lib/aggregate) and stores per complete file its release, the file and a completed binary (SeedService), instead of one parts row per article. What doesn't complete within the window goes through the regular ingest path as fragments. `--dry-run` only fetches and aggregates, to measure a node. Co-Authored-By: Claude Opus 5.5 --- app/services/SeedService.ts | 107 +++++++++++++++++++++++ commands/SeedRange.ts | 155 ++++++++++++++++++++++++++++++++++ lib/aggregate/index.ts | 86 +++++++++++++++++++ tests/functional/seed.spec.ts | 47 +++++++++++ tests/unit/aggregate.spec.ts | 39 +++++++++ 5 files changed, 434 insertions(+) create mode 100644 app/services/SeedService.ts create mode 100644 commands/SeedRange.ts create mode 100644 lib/aggregate/index.ts create mode 100644 tests/functional/seed.spec.ts create mode 100644 tests/unit/aggregate.spec.ts diff --git a/app/services/SeedService.ts b/app/services/SeedService.ts new file mode 100644 index 0000000..4ad0d8f --- /dev/null +++ b/app/services/SeedService.ts @@ -0,0 +1,107 @@ +import db from '@adonisjs/lucid/services/db' +import type { AggregatedFile } from '#lib/aggregate/index' +import IngestService, { type IngestSegment } from '#services/IngestService' +import FileStore from '#services/FileStore' +import SfvService from '#services/SfvService' +import ContentService from '#services/ContentService' +import QueueService from '#services/QueueService' + +export interface SeedStoreResult { + files: number + /** Files that existed already (same file_key, e.g. from live indexing or a crosspost). */ + existing: number + releases: number + inspections: number +} + +/** + * Stores files an edge node aggregated from headers (CONCEPT.md D37): their + * releases, the files with their segment lists, and one completed binary per + * file, as worker:file would have left it, so par2 naming, inspection and GC + * treat them like any other file. A batch is a few statements, not one per + * article. + */ +class SeedService { + async store(files: AggregatedFile[]): Promise { + const result: SeedStoreResult = { files: 0, existing: 0, releases: 0, inspections: 0 } + if (files.length === 0) return result + const knex = db.connection().getWriteClient() + const now = new Date() + + // A file's identity and release come from its first segment, its date from the earliest. + const firstOf = (file: AggregatedFile) => file.segments[0] + const postedOf = (file: AggregatedFile) => { + const dates = file.segments.map((s) => s.postedAt?.getTime()).filter((t): t is number => Number.isFinite(t)) + return dates.length > 0 ? new Date(Math.min(...dates)) : undefined + } + const releaseIds = await IngestService.upsertReleases( + knex, + files.map((file) => ({ ...firstOf(file), postedAt: postedOf(file) })), + now + ) + result.releases = releaseIds.size + + const rows = files.map((file) => { + const first = firstOf(file) + const postedAt = postedOf(file) + return { + file_key: file.fileKey, + filename: first.name, + poster: first.poster, + date: (postedAt ?? now).getTime(), + parts: file.segments.length, + segments: file.segments.map((s) => ({ number: s.number, id: s.messageId, size: s.bytes })), + groups: JSON.stringify(first.groups), + release_id: first.release ? (releaseIds.get(first.release.key) ?? null) : null, + size: file.segments.reduce((sum, s) => sum + s.bytes, 0), + created_at: now, + updated_at: now, + } + }) + const inserted = await FileStore.insert(rows) + result.files = inserted.length + result.existing = rows.length - inserted.length + + const fileIds = new Map() + for (const row of await db.from('files').whereIn('file_key', rows.map((r) => r.file_key)).select('id', 'file_key')) { + fileIds.set(row.file_key, Number(row.id)) + } + // The completed binary worker:file would have left behind; an existing one just learns its file. + const binaries = rows.map((row, i) => ({ + file_key: row.file_key, + poster: row.poster, + subject: firstOf(files[i]).subject, + name: row.filename, + total_parts: firstOf(files[i]).totalParts, + groups: row.groups, + posted_at: postedOf(files[i]) ?? null, + first_seen_at: now, + last_seen_at: now, + completed_at: now, + release_id: row.release_id, + file_id: fileIds.get(row.file_key) ?? null, + })) + binaries.sort((a, b) => a.file_key.localeCompare(b.file_key)) + await knex('binaries') + .insert(binaries) + .onConflict('file_key') + .merge({ file_id: knex.raw('coalesce(binaries.file_id, excluded.file_id)') as any }) + + // As worker:file does for a new file: an SFV posted earlier may list it (D34), then inspection. + const rowById = new Map(rows.map((row) => [fileIds.get(row.file_key), row])) + const candidates: { name: string; data: { fileId: number }; opts: { jobId: string } }[] = [] + for (const file of inserted) { + const row = rowById.get(file.id) + if (!row) continue + await SfvService.claim({ id: file.id, name: row.filename, poster: row.poster, date: row.date, release_id: row.release_id }) + if (ContentService.isCandidate(row.filename)) { + candidates.push({ name: 'process-collection', data: { fileId: file.id }, opts: { jobId: `file-${file.id}` } }) + } + } + if (candidates.length > 0) await QueueService.collectionQueue.addBulk(candidates) + result.inspections = candidates.length + return result + } +} + +export default new SeedService() diff --git a/commands/SeedRange.ts b/commands/SeedRange.ts new file mode 100644 index 0000000..765db54 --- /dev/null +++ b/commands/SeedRange.ts @@ -0,0 +1,155 @@ +import { BaseCommand, args, flags } from '@adonisjs/core/ace' +import { CommandOptions } from '@adonisjs/core/types/ace' +import NntpService from '#services/NntpService' +import IngestService, { segmentFromHeader, type IngestSegment } from '#services/IngestService' +import StagingService from '#services/StagingService' +import SeedService from '#services/SeedService' +import QueueService from '#services/QueueService' +import { isDeferred } from '#services/DeferredService' +import { FileAggregator } from '#lib/aggregate/index' + +/** + * Edge aggregation (CONCEPT.md D37): indexes a range of a group's history + * over one NNTP connection, grouping headers into files in memory and + * storing one row per file. Files cut by the range edges, or that never + * complete, go through the regular ingest path as fragments. Header-only, + * like the backfill: obfuscated subjects are counted, not downloaded. + * + * node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000 + * node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000 --dry-run + */ +export default class SeedRange extends BaseCommand { + public static commandName = 'seed:range' + public static description = "Indexes a range of a group's headers by aggregating them into files on this node." + + public static options: CommandOptions = { + startApp: true, + } + + @args.string({ description: 'Group' }) + declare group: string + + @flags.string({ description: 'First article number' }) + declare start: string + + @flags.number({ description: 'Articles to index', default: 1_000_000 }) + declare count: number + + @flags.number({ description: 'Articles per XOVER', default: 20_000 }) + declare chunk: number + + @flags.number({ description: 'Batches without a new part before an open file is given up as fragments', default: 5 }) + declare idle: number + + @flags.boolean({ description: 'Fetch and aggregate only; store nothing' }) + declare dryRun: boolean + + public async run() { + const first = Number(this.start) + const last = first + this.count - 1 + if (!Number.isSafeInteger(first) || first < 1) throw new Error(`--start must be an article number, got ${this.start}`) + const aggregator = new FileAggregator() + const stats = { headers: 0, segments: 0, obfuscated: 0, deferred: 0, complete: 0, files: 0, existing: 0, fragments: 0, inspections: 0 } + const time = { fetch: 0, parse: 0, store: 0 } + const cpuStart = process.cpuUsage() + const started = Date.now() + + const conn = await NntpService.acquire() + let failed = false + try { + await NntpService.timed(conn.group(this.group), `GROUP ${this.group}`) + const fetch = async (from: number) => { + const to = Math.min(from + this.chunk - 1, last) + const t = Date.now() + const reply = await NntpService.timed(conn.xover(from, to), `XOVER ${this.group} ${from}-${to}`) + time.fetch += Date.now() - t + return reply.overviews.filter(([n, h]: [number | null, Record]) => n !== null && h?.['message-id']) + } + + // One XOVER in flight while the previous batch is parsed and stored. + let from = first + const prefetch = (at: number) => { + const promise = fetch(at) + // Awaited after the current batch is stored; until then a failure must not count as unhandled. + promise.catch(() => {}) + return promise + } + let next: Promise<[number, Record][]> | null = prefetch(from) + while (next) { + const overviews = await next + from += this.chunk + next = from <= last ? prefetch(from) : null + + let t = Date.now() + const segments: IngestSegment[] = [] + for (const [articleNumber, header] of overviews) { + const segment = segmentFromHeader(header, this.group, articleNumber) + if (segment) segments.push(segment) + else if (isDeferred(header)) stats.deferred++ + else stats.obfuscated++ + } + stats.headers += overviews.length + stats.segments += segments.length + const complete = aggregator.add(segments) + const fragments = aggregator.evict(this.idle) + stats.complete += complete.length + stats.fragments += fragments.length + time.parse += Date.now() - t + + t = Date.now() + if (!this.dryRun) await this.store(complete, fragments, stats) + time.store += Date.now() - t + this.progress(stats, started) + } + + const rest = aggregator.evict(0) + stats.fragments += rest.length + if (!this.dryRun) await this.store([], rest, stats) + } catch (error) { + failed = true + throw error + } finally { + NntpService.release(conn, failed) + await NntpService.shutdown() + await QueueService.closeAll() + } + + const seconds = (Date.now() - started) / 1000 + const cpu = process.cpuUsage(cpuStart) + this.logger.info( + `${this.group} ${first}-${last}: ${stats.headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s ` + + `(${Math.round(stats.headers / seconds).toLocaleString('en')}/s). ` + + `${stats.segments.toLocaleString('en')} readable, ${stats.obfuscated.toLocaleString('en')} obfuscated, ${stats.deferred.toLocaleString('en')} deferred. ` + + `${stats.complete.toLocaleString('en')} complete files (${stats.files.toLocaleString('en')} new, ${stats.existing.toLocaleString('en')} known), ` + + `${stats.fragments.toLocaleString('en')} fragment segments, ${stats.inspections.toLocaleString('en')} queued for inspection.` + ) + this.logger.info( + `Time: fetch ${(time.fetch / 1000).toFixed(0)} s (overlapped), parse+aggregate ${(time.parse / 1000).toFixed(0)} s, ` + + `store ${(time.store / 1000).toFixed(0)} s; node CPU ${((cpu.user + cpu.system) / 1e6).toFixed(0)} s.` + ) + } + + private async store(complete: Parameters[0], fragments: IngestSegment[], stats: Record) { + const stored = await SeedService.store(complete) + stats.files += stored.files + stats.existing += stored.existing + stats.inspections += stored.inspections + if (fragments.length === 0) return + // The regular path: staging (D35) for lone segments, binaries and parts for the rest. + const { ingest } = StagingService.enabled(this.group) + ? await StagingService.filter(fragments, { counters: StagingService.countersEnabled(this.group) }) + : { ingest: fragments } + await IngestService.ingest(ingest) + } + + private lastReport = 0 + private progress(stats: Record, started: number) { + if (Date.now() - this.lastReport < 30_000) return + this.lastReport = Date.now() + const seconds = (Date.now() - started) / 1000 + this.logger.info( + `${stats.headers.toLocaleString('en')} headers (${Math.round(stats.headers / seconds).toLocaleString('en')}/s), ` + + `${stats.complete.toLocaleString('en')} files, ${stats.fragments.toLocaleString('en')} fragment segments.` + ) + } +} diff --git a/lib/aggregate/index.ts b/lib/aggregate/index.ts new file mode 100644 index 0000000..e25b56b --- /dev/null +++ b/lib/aggregate/index.ts @@ -0,0 +1,86 @@ +/** + * Edge aggregation (CONCEPT.md D37): groups header segments into files in + * memory, so a seeding node writes one row per file instead of one per + * article. A file is done once every part number from 1 to its total has + * an article; what never completes within the node's window (an upload cut + * by the range edges, a random subject per article) leaves as fragments + * for the regular ingest path, where it can meet its other parts. + */ + +/** What the aggregator needs of a segment (IngestSegment has all of it). */ +export interface AggregatableSegment { + fileKey: string + number: number + totalParts: number + messageId: string + bytes: number +} + +export interface AggregatedFile { + fileKey: string + /** One segment per part number, in part order; the first article seen wins (a later one is a repost). */ + segments: S[] +} + +interface OpenFile { + parts: Map + totalParts: number + /** Batch in which a segment last arrived. */ + touched: number +} + +export class FileAggregator { + private open = new Map>() + private batch = 0 + /** Segments with a part number outside 1..total: never part of a complete file. */ + private stray: S[] = [] + + /** Adds one batch of segments; returns the files they completed. */ + add(segments: S[]): AggregatedFile[] { + this.batch++ + const touched = new Set() + for (const segment of segments) { + if (segment.number < 1 || segment.number > segment.totalParts) { + this.stray.push(segment) + continue + } + let file = this.open.get(segment.fileKey) + if (!file) { + file = { parts: new Map(), totalParts: segment.totalParts, touched: this.batch } + this.open.set(segment.fileKey, file) + } + if (!file.parts.has(segment.number)) file.parts.set(segment.number, segment) + file.touched = this.batch + touched.add(segment.fileKey) + } + + const complete: AggregatedFile[] = [] + for (const key of touched) { + const file = this.open.get(key)! + if (file.parts.size < file.totalParts) continue + this.open.delete(key) + complete.push({ fileKey: key, segments: [...file.parts.values()].sort((a, b) => a.number - b.number) }) + } + return complete + } + + /** + * Gives up on files that got no segment in the last `idleBatches` batches + * (all open files when 0) and returns their segments, plus stray ones. + */ + evict(idleBatches = 0): S[] { + const out = this.stray + this.stray = [] + for (const [key, file] of this.open) { + if (idleBatches > 0 && this.batch - file.touched < idleBatches) continue + this.open.delete(key) + out.push(...file.parts.values()) + } + return out + } + + /** Files still waiting for parts. */ + get openFiles(): number { + return this.open.size + } +} diff --git a/tests/functional/seed.spec.ts b/tests/functional/seed.spec.ts new file mode 100644 index 0000000..241e47a --- /dev/null +++ b/tests/functional/seed.spec.ts @@ -0,0 +1,47 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import { segmentFromHeader, type IngestSegment } from '#services/IngestService' +import SeedService from '#services/SeedService' +import QueueService from '#services/QueueService' +import { FileAggregator } from '#lib/aggregate/index' +import { segmentsOf } from '#services/FileStore' + +function header(subject: string, n: number) { + return { + 'subject': subject, + 'from': 'seeder ', + 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', + 'message-id': ``, + ':bytes': '700000', + } +} + +test.group('SeedService', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + group.teardown(() => QueueService.closeAll()) + + test('stores aggregated files with their release and a completed binary', async ({ assert }) => { + const aggregator = new FileAggregator() + const subjects = [1, 2].map((i) => `Seed.Test.S01E01.1080p - [1/2] - "Seed.Test.S01E01.1080p.part${i}.mkv" yEnc (%d/3)`) + const segments = subjects.flatMap((s) => [1, 2, 3].map((n) => segmentFromHeader(header(s.replace('%d', String(n)), n), 'alt.binaries.test', n)!)) + const files = aggregator.add(segments) + assert.lengthOf(files, 2) + + const result = await SeedService.store(files) + assert.include(result, { files: 2, existing: 0, releases: 1 }) + + const stored = await db.from('files').where('filename', 'Seed.Test.S01E01.1080p.part1.mkv').first() + assert.equal(stored.parts, 3) + assert.deepEqual(segmentsOf(stored).map((s) => s.number), [1, 2, 3]) + const release = await db.from('releases').where('id', stored.release_id).first() + assert.equal(release.name, segments[0].release!.name) // named as live indexing would + const binary = await db.from('binaries').where('file_key', stored.file_key).first() + assert.equal(Number(binary.file_id), Number(stored.id)) + assert.isNotNull(binary.completed_at) + + // The same files again (a crosspost, a second run): nothing new. + const again = await SeedService.store(files) + assert.include(again, { files: 0, existing: 2 }) + }) +}) diff --git a/tests/unit/aggregate.spec.ts b/tests/unit/aggregate.spec.ts new file mode 100644 index 0000000..e1f7c60 --- /dev/null +++ b/tests/unit/aggregate.spec.ts @@ -0,0 +1,39 @@ +import { test } from '@japa/runner' +import { FileAggregator, type AggregatableSegment } from '#lib/aggregate/index' + +const seg = (fileKey: string, number: number, totalParts: number, messageId = `${fileKey}-${number}@test`): AggregatableSegment => ({ + fileKey, + number, + totalParts, + messageId, + bytes: 100, +}) + +test.group('aggregate / FileAggregator', () => { + test('completes a file once every part arrived, across batches and out of order', ({ assert }) => { + const aggregator = new FileAggregator() + assert.deepEqual(aggregator.add([seg('a', 2, 3), seg('b', 1, 1)]).map((f) => f.fileKey), ['b']) + assert.equal(aggregator.openFiles, 1) + const [done] = aggregator.add([seg('a', 3, 3), seg('a', 1, 3)]) + assert.equal(done.fileKey, 'a') + assert.deepEqual(done.segments.map((s) => s.number), [1, 2, 3]) + assert.equal(aggregator.openFiles, 0) + }) + + test('keeps the first article of a part number (a repost) and drops impossible numbers', ({ assert }) => { + const aggregator = new FileAggregator() + const [done] = aggregator.add([seg('a', 1, 2, 'first@test'), seg('a', 1, 2, 'repost@test'), seg('a', 5, 2), seg('a', 2, 2)]) + assert.deepEqual(done.segments.map((s) => s.messageId), ['first@test', 'a-2@test']) + assert.deepEqual(aggregator.evict().map((s) => s.number), [5]) + }) + + test('gives up idle files as fragments, all of them at the end', ({ assert }) => { + const aggregator = new FileAggregator() + aggregator.add([seg('old', 1, 3)]) + aggregator.add([seg('new', 1, 3)]) + aggregator.add([seg('new', 2, 3)]) + assert.deepEqual(aggregator.evict(2).map((s) => s.fileKey), ['old']) + assert.deepEqual(aggregator.evict(0).map((s) => s.number), [1, 2]) + assert.equal(aggregator.openFiles, 0) + }) +})