From 97bdf8e982d399d63fc3b4d01da858aea045d213 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 5 Oct 2026 11:11:41 +0200 Subject: [PATCH] Seeding node: gzip overviews, batched SFV lookups, fragments beside fetching MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - lib/nntp/xover_client: a minimal one-connection client for seed:range with XFEATURE COMPRESS GZIP (a zlib stream after "224 … [COMPRESS=GZIP]", then a "." line; 420 for an empty range). About 6x less header traffic. Overviews come out with the field names nntp-js uses. - SfvService.claimAll: one sfv_entries lookup per stored batch. - Fragments go through the regular ingest path beside fetching (at most two batches in flight); a failure fails the run at the end. Co-Authored-By: Claude Opus 5.5 --- app/services/SeedService.ts | 15 ++- app/services/SfvService.ts | 15 +++ commands/SeedRange.ts | 60 +++++++--- lib/nntp/xover_client.ts | 189 ++++++++++++++++++++++++++++++++ tests/unit/xover_client.spec.ts | 76 +++++++++++++ 5 files changed, 329 insertions(+), 26 deletions(-) create mode 100644 lib/nntp/xover_client.ts create mode 100644 tests/unit/xover_client.spec.ts diff --git a/app/services/SeedService.ts b/app/services/SeedService.ts index 4ad0d8f..4187912 100644 --- a/app/services/SeedService.ts +++ b/app/services/SeedService.ts @@ -89,15 +89,14 @@ class SeedService { // 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 added = inserted.flatMap((file) => { 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}` } }) - } - } + return row ? [{ id: file.id, name: row.filename, poster: row.poster, date: row.date, release_id: row.release_id }] : [] + }) + await SfvService.claimAll(added) + const candidates = added + .filter((file) => ContentService.isCandidate(file.name)) + .map((file) => ({ 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 diff --git a/app/services/SfvService.ts b/app/services/SfvService.ts index 3507135..41513a5 100644 --- a/app/services/SfvService.ts +++ b/app/services/SfvService.ts @@ -52,6 +52,21 @@ class SfvService { } /** A new file an SFV already listed joins that SFV's release. Returns the release, or null. */ + /** + * claim() for a batch of new files: one lookup for all their names, the + * per-file work only for the few an SFV lists (seeding stores thousands + * of files per batch; CONCEPT.md D37). + */ + async claimAll(files: { id: number; name: string; poster?: string | null; date?: number | null; release_id?: number | null }[]): Promise { + if (files.length === 0) return 0 + const listed = new Set( + (await db.from('sfv_entries').whereIn('name', [...new Set(files.map((f) => f.name))]).select('name')).map((e) => e.name as string) + ) + let moved = 0 + for (const file of files) if (listed.has(file.name) && (await this.claim(file)) !== null) moved++ + return moved + } + async claim(file: { id: number; name: string; poster?: string | null; date?: number | null; release_id?: number | null }): Promise { const entries = await db.from('sfv_entries').where('name', file.name) const entry = entries.find((e) => this.sameUpload(file, e.poster, e.posted_at ? new Date(e.posted_at) : null)) diff --git a/commands/SeedRange.ts b/commands/SeedRange.ts index 765db54..8d3554b 100644 --- a/commands/SeedRange.ts +++ b/commands/SeedRange.ts @@ -1,6 +1,7 @@ import { BaseCommand, args, flags } from '@adonisjs/core/ace' import { CommandOptions } from '@adonisjs/core/types/ace' -import NntpService from '#services/NntpService' +import nntpConfig from '#config/nntp' +import { XoverClient, type Overview } from '#lib/nntp/xover_client' import IngestService, { segmentFromHeader, type IngestSegment } from '#services/IngestService' import StagingService from '#services/StagingService' import SeedService from '#services/SeedService' @@ -10,7 +11,8 @@ 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 + * over one NNTP connection (gzip-compressed overviews when the server has + * them), 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. @@ -44,6 +46,13 @@ export default class SeedRange extends BaseCommand { @flags.boolean({ description: 'Fetch and aggregate only; store nothing' }) declare dryRun: boolean + @flags.boolean({ description: 'Ask the server for gzip-compressed overviews', default: true }) + declare compress: boolean + + /** Fragment batches being stored the regular way, off the node's critical path. */ + private fragmentWork = new Set>() + private fragmentError?: unknown + public async run() { const first = Number(this.start) const last = first + this.count - 1 @@ -54,16 +63,24 @@ export default class SeedRange extends BaseCommand { const cpuStart = process.cpuUsage() const started = Date.now() - const conn = await NntpService.acquire() - let failed = false + const client = await XoverClient.connect({ + host: nntpConfig.host, + port: nntpConfig.port, + secure: nntpConfig.secure, + user: nntpConfig.user, + password: nntpConfig.password?.release(), + timeoutMs: nntpConfig.commandTimeoutMs, + compress: this.compress, + }) + this.logger.info(`Connected${client.compressed ? ', overviews gzip-compressed' : ''}.`) try { - await NntpService.timed(conn.group(this.group), `GROUP ${this.group}`) + await client.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}`) + const overviews = await client.xover(from, to) time.fetch += Date.now() - t - return reply.overviews.filter(([n, h]: [number | null, Record]) => n !== null && h?.['message-id']) + return overviews.filter(([n, h]) => Number.isFinite(n) && h['message-id']) } // One XOVER in flight while the previous batch is parsed and stored. @@ -74,7 +91,7 @@ export default class SeedRange extends BaseCommand { promise.catch(() => {}) return promise } - let next: Promise<[number, Record][]> | null = prefetch(from) + let next: Promise | null = prefetch(from) while (next) { const overviews = await next from += this.chunk @@ -105,12 +122,10 @@ export default class SeedRange extends BaseCommand { const rest = aggregator.evict(0) stats.fragments += rest.length if (!this.dryRun) await this.store([], rest, stats) - } catch (error) { - failed = true - throw error + await Promise.all(this.fragmentWork) + if (this.fragmentError) throw this.fragmentError } finally { - NntpService.release(conn, failed) - await NntpService.shutdown() + client.close() await QueueService.closeAll() } @@ -135,11 +150,20 @@ export default class SeedRange extends BaseCommand { 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) + // The regular path: staging (D35) for lone segments, binaries and parts for the rest. It runs + // beside fetching; with two batches in flight already, the node waits for one to finish. + while (this.fragmentWork.size >= 2) await Promise.race([...this.fragmentWork].map((w) => w.catch(() => {}))) + if (this.fragmentError) throw this.fragmentError + const work = (async () => { + const { ingest } = StagingService.enabled(this.group) + ? await StagingService.filter(fragments, { counters: StagingService.countersEnabled(this.group) }) + : { ingest: fragments } + await IngestService.ingest(ingest) + })() + this.fragmentWork.add(work) + work + .catch((error) => (this.fragmentError ??= error)) + .finally(() => this.fragmentWork.delete(work)) } private lastReport = 0 diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts new file mode 100644 index 0000000..c54de44 --- /dev/null +++ b/lib/nntp/xover_client.ts @@ -0,0 +1,189 @@ +import net from 'node:net' +import tls from 'node:tls' +import { unzipSync } from 'node:zlib' + +export interface XoverClientOptions { + host: string + port: number + user?: string + password?: string + secure?: boolean + /** A command whose reply stalls this long fails, and the connection with it. */ + timeoutMs?: number + /** Ask for gzip-compressed overview data (XFEATURE COMPRESS GZIP), when the server has it. */ + compress?: boolean +} + +export type Overview = [number, Record] + +/** The overview fields in RFC 3977 order, named as nntp-js names them (segmentFromHeader reads these). */ +const FIELDS = ['subject', 'from', 'date', 'message-id', 'references', ':bytes', ':lines', 'xref'] +const TERMINATOR = Buffer.from('\r\n.\r\n') +const DOT_LINE = Buffer.from('.\r\n') + +/** + * A minimal NNTP client for seeding nodes (CONCEPT.md D37): one connection, + * GROUP and XOVER only, with gzip-compressed overviews. nntp-js can't read + * those: after "224 … [COMPRESS=GZIP]" the lines come as one zlib stream, + * followed by a plain "." line; an empty range is a 420 without body. + * About six times less traffic per header (LOG.md 2026-10-05). + */ +export class XoverClient { + private chunks: Buffer[] = [] + private length = 0 + private wake?: () => void + private failure?: Error + compressed = false + + private constructor(private sock: net.Socket) { + sock.on('data', (chunk: Buffer) => { + this.chunks.push(chunk) + this.length += chunk.length + this.wake?.() + }) + const fail = (error: Error) => { + this.failure ??= error + this.wake?.() + } + sock.on('error', fail) + sock.on('close', () => fail(new Error('NNTP connection closed'))) + sock.on('timeout', () => { + fail(new Error('NNTP reply timed out')) + sock.destroy() + }) + } + + static async connect(options: XoverClientOptions): Promise { + const sock = options.secure + ? tls.connect({ host: options.host, port: options.port, servername: options.host }) + : net.connect({ host: options.host, port: options.port }) + sock.setTimeout(options.timeoutMs ?? 120_000) + const client = new XoverClient(sock) + try { + await client.expect(await client.line(), ['200', '201']) + if (options.user) { + const reply = await client.command(`AUTHINFO USER ${options.user}`) + if (reply.startsWith('381')) await client.expect(await client.command(`AUTHINFO PASS ${options.password ?? ''}`), ['281']) + else await client.expect(reply, ['281']) + } + if (options.compress) client.compressed = (await client.command('XFEATURE COMPRESS GZIP')).startsWith('290') + } catch (error) { + client.close() + throw error + } + return client + } + + async group(name: string): Promise<{ count: number; low: number; high: number }> { + const reply = await this.command(`GROUP ${name}`) + await this.expect(reply, ['211']) + const [, count, low, high] = reply.split(' ') + return { count: Number(count), low: Number(low), high: Number(high) } + } + + /** Overviews of articles `from`..`to` in the current group; an empty range gives []. */ + async xover(from: number, to: number): Promise { + const reply = await this.command(`XOVER ${from}-${to}`) + if (reply.startsWith('420') || reply.startsWith('423')) return [] + await this.expect(reply, ['224']) + const data = reply.includes('COMPRESS=GZIP') ? await this.compressedBody() : await this.plainBody() + return parseOverviews(data) + } + + close() { + try { + this.sock.end('QUIT\r\n') + } catch { + // already closed + } + this.sock.destroy() + } + + private async command(line: string): Promise { + this.sock.write(`${line}\r\n`) + return this.line() + } + + private async expect(reply: string, codes: string[]) { + if (!codes.some((code) => reply.startsWith(code))) throw new Error(`NNTP: unexpected reply "${reply}"`) + } + + private buffer(): Buffer { + if (this.chunks.length > 1) this.chunks = [Buffer.concat(this.chunks, this.length)] + return this.chunks[0] ?? Buffer.alloc(0) + } + + private take(n: number): Buffer { + const all = this.buffer() + const out = all.subarray(0, n) + const rest = all.subarray(n) + this.chunks = rest.length > 0 ? [rest] : [] + this.length = rest.length + return out + } + + private async more() { + if (this.failure) throw this.failure + await new Promise((resolve) => (this.wake = resolve)) + this.wake = undefined + if (this.failure && this.length === 0) throw this.failure + } + + private async line(): Promise { + while (true) { + const at = this.buffer().indexOf('\r\n') + if (at !== -1) return this.take(at + 2).subarray(0, at).toString('latin1') + await this.more() + } + } + + /** A plain multi-line body, up to "\r\n.\r\n", dot-unstuffed. */ + private async plainBody(): Promise { + let searched = 0 + while (true) { + const all = this.buffer() + // An empty body is just ".\r\n". + if (all.length >= 3 && searched === 0 && all.subarray(0, 3).equals(DOT_LINE)) { + this.take(3) + return '' + } + const at = all.indexOf(TERMINATOR, Math.max(0, searched - TERMINATOR.length)) + if (at !== -1) return this.take(at + TERMINATOR.length).subarray(0, at + 2).toString('utf8').replace(/^\.\./gm, '.') + searched = all.length + await this.more() + } + } + + /** A zlib/gzip stream followed by a "." line: complete once it ends in ".\r\n" and inflates. */ + private async compressedBody(): Promise { + while (true) { + const all = this.buffer() + if (all.length > DOT_LINE.length && all.subarray(all.length - DOT_LINE.length).equals(DOT_LINE)) { + try { + const text = unzipSync(all.subarray(0, all.length - DOT_LINE.length)).toString('utf8') + this.take(all.length) + return text + } catch { + // ".\r\n" inside the compressed data: wait for more. + } + } + await this.more() + } + } +} + +/** Overview lines (tab-separated, after the article number) as nntp-js returns them. */ +export function parseOverviews(data: string): Overview[] { + const out: Overview[] = [] + for (const line of data.split('\r\n')) { + if (!line) continue + const [number, ...tokens] = line.split('\t') + const fields: Record = {} + for (let i = 0; i < FIELDS.length && i < tokens.length; i++) { + // Full fields after the defaults carry their name: "Xref: host group:number". + fields[FIELDS[i]] = i >= 7 ? tokens[i].replace(/^[^:\s]+:\s*/, '') : tokens[i] + } + out.push([Number.parseInt(number, 10), fields]) + } + return out +} diff --git a/tests/unit/xover_client.spec.ts b/tests/unit/xover_client.spec.ts new file mode 100644 index 0000000..c00bfe0 --- /dev/null +++ b/tests/unit/xover_client.spec.ts @@ -0,0 +1,76 @@ +import { test } from '@japa/runner' +import net from 'node:net' +import { deflateSync } from 'node:zlib' +import { XoverClient, parseOverviews } from '#lib/nntp/xover_client' + +const LINES = [ + '101\tShow.S01E01 - "a.mkv" yEnc (1/2)\tposter \tMon, 05 Oct 2026 10:00:00 +0000\t\t\t700000\t5000\tXref: news.example a.b.test:101', + '102\tShow.S01E01 - "a.mkv" yEnc (2/2)\tposter \tMon, 05 Oct 2026 10:00:01 +0000\t\t\t650000\t4600\tXref: news.example a.b.test:102 a.b.other:7', +] + +/** A tiny NNTP server: plain replies, gzip after XFEATURE, 420 for 1-1, replies split into small chunks. */ +function server(): Promise<{ port: number; close: () => void }> { + return new Promise((resolve) => { + const srv = net.createServer((sock) => { + let gzip = false + const send = (data: Buffer) => { + for (let i = 0; i < data.length; i += 7) sock.write(data.subarray(i, i + 7)) + } + sock.write('200 welcome\r\n') + let pending = '' + sock.on('data', (chunk) => { + pending += chunk.toString() + let at: number + while ((at = pending.indexOf('\r\n')) !== -1) { + const line = pending.slice(0, at) + pending = pending.slice(at + 2) + if (line.startsWith('AUTHINFO USER')) sock.write('381 more\r\n') + else if (line.startsWith('AUTHINFO PASS')) sock.write('281 ok\r\n') + else if (line === 'XFEATURE COMPRESS GZIP') (gzip = true), sock.write('290 Feature Enabled\r\n') + else if (line.startsWith('GROUP')) sock.write('211 2 101 102 a.b.test\r\n') + else if (line === 'XOVER 1-1') sock.write('420 No Articles Selected\r\n') + else if (line.startsWith('XOVER')) { + const body = Buffer.from(LINES.join('\r\n') + '\r\n') + if (gzip) { + sock.write('224 Overview Information Follows [COMPRESS=GZIP]\r\n') + send(Buffer.concat([deflateSync(body), Buffer.from('.\r\n')])) + } else { + sock.write('224 Overview Information Follows\r\n') + send(Buffer.concat([body, Buffer.from('.\r\n')])) + } + } else if (line === 'QUIT') sock.end('205 bye\r\n') + } + }) + }) + srv.listen(0, '127.0.0.1', () => resolve({ port: (srv.address() as net.AddressInfo).port, close: () => srv.close() })) + }) +} + +test.group('nntp / XoverClient', () => { + test('parses overview lines into the fields nntp-js gives', ({ assert }) => { + const [[number, fields]] = parseOverviews(LINES[0] + '\r\n') + assert.equal(number, 101) + assert.equal(fields['message-id'], '') + assert.equal(fields[':bytes'], '700000') + assert.equal(fields.xref, 'news.example a.b.test:101') + }) + + for (const compress of [false, true]) { + test(`fetches overviews ${compress ? 'gzip-compressed' : 'plain'}, and an empty range`, async ({ assert }) => { + const { port, close } = await server() + const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress }) + try { + assert.equal(client.compressed, compress) + assert.deepEqual(await client.group('a.b.test'), { count: 2, low: 101, high: 102 }) + const overviews = await client.xover(101, 102) + assert.deepEqual(overviews.map(([n]) => n), [101, 102]) + assert.equal(overviews[1][1].subject, 'Show.S01E01 - "a.mkv" yEnc (2/2)') + assert.deepEqual(await client.xover(1, 1), []) + assert.lengthOf(await client.xover(101, 102), 2) + } finally { + client.close() + close() + } + }) + } +})