diff --git a/app/controllers/seed_controller.ts b/app/controllers/seed_controller.ts index 8e244fc..8ab224b 100644 --- a/app/controllers/seed_controller.ts +++ b/app/controllers/seed_controller.ts @@ -1,6 +1,7 @@ import type { HttpContext } from '@adonisjs/core/http' import SeedOrchestrator, { LeaseLost } from '#services/SeedOrchestrator' import SeedFetchPool from '#services/SeedFetchPool' +import RemoteContentService, { JobLost } from '#services/RemoteContentService' import type { SeedBatch } from '#lib/seed/wire' /** @@ -71,11 +72,77 @@ export default class SeedController { }) } + /** + * POST /api/v1/seed/inspect/lease { max } → 200 { jobs } or 204: content + * inspection and archive walks for the node (CONCEPT.md D41). + */ + async inspectLease({ seedNode, request, response }: HttpContext) { + const max = Math.min(10, Math.max(1, Number(request.input('max')) || 1)) + const jobs = await RemoteContentService.lease(seedNode.id, max) + return jobs.length > 0 ? { jobs } : response.noContent() + } + + /** + * POST /api/v1/seed/inspect/done { queue, jobId, token, result } where + * result is the probe as a JSON string (the body parser would turn the + * empty strings inside it into nulls). + */ + async inspectDone({ request, response }: HttpContext) { + const { queue, jobId, token, result } = request.only(['queue', 'jobId', 'token', 'result']) + if (!RemoteContentService.isQueue(queue) || typeof jobId !== 'string' || typeof token !== 'string' || typeof result !== 'string') { + return response.badRequest({ error: 'Expected { queue, jobId, token, result: JSON string }' }) + } + return this.job(response, () => RemoteContentService.finish(queue, jobId, token, JSON.parse(result))) + } + + /** POST /api/v1/seed/inspect/failed { queue, jobId, token, error } */ + async inspectFailed({ request, response }: HttpContext) { + const { queue, jobId, token, error } = request.only(['queue', 'jobId', 'token', 'error']) + if (!RemoteContentService.isQueue(queue) || typeof jobId !== 'string' || typeof token !== 'string') { + return response.badRequest({ error: 'Expected { queue, jobId, token, error }' }) + } + return this.job(response, () => RemoteContentService.fail(queue, jobId, token, String(error ?? 'unknown error'))) + } + + /** POST /api/v1/seed/inspect/release { queue, jobId, token }: a stopping node gives a job back. */ + async inspectRelease({ request, response }: HttpContext) { + const { queue, jobId, token } = request.only(['queue', 'jobId', 'token']) + if (!RemoteContentService.isQueue(queue) || typeof jobId !== 'string' || typeof token !== 'string') { + return response.badRequest({ error: 'Expected { queue, jobId, token }' }) + } + return this.job(response, () => RemoteContentService.release(queue, jobId, token)) + } + + /** GET /api/v1/seed/articles?id= → the raw article body; 410 when the server doesn't have it. */ + async article({ request, response }: HttpContext) { + const id = request.input('id') + if (typeof id !== 'string' || id.length === 0 || id.length > 250) return response.badRequest({ error: 'Expected ?id=' }) + try { + const body = await RemoteContentService.article(id) + if (!body) return response.gone({ error: 'No such article' }) + response.header('content-type', 'application/octet-stream') + return response.send(body) + } catch (error: any) { + return response.badGateway({ error: `NNTP: ${error.message}` }) + } + } + /** GET /api/v1/seed/status */ async status() { return SeedOrchestrator.status() } + /** 409 when the job's lock is gone (it was given to another node meanwhile). */ + private async job(response: HttpContext['response'], action: () => Promise) { + try { + await action() + return { ok: true } + } catch (error) { + if (error instanceof JobLost) return response.conflict({ error: error.message }) + throw error + } + } + /** 409 when the range was given to another node meanwhile; the node should drop it. */ private async leased(response: HttpContext['response'], action: () => Promise) { try { diff --git a/app/services/ArchiveService.ts b/app/services/ArchiveService.ts index 00de756..6223247 100644 --- a/app/services/ArchiveService.ts +++ b/app/services/ArchiveService.ts @@ -1,12 +1,7 @@ -import { execFile } from 'node:child_process' -import { mkdtemp, rm, writeFile } from 'node:fs/promises' -import { tmpdir } from 'node:os' -import path from 'node:path' -import { promisify } from 'node:util' import db from '@adonisjs/lucid/services/db' import logger from '@adonisjs/core/services/logger' import IngestService from '#services/IngestService' -import { storedEntryData, type RarEntry, type RarInfo } from '#lib/rar/index' +import type { RarInfo } from '#lib/rar/index' import { fileStem, isGenericName, rarReleaseIdentity } from '#lib/release/index' /** Stored per file and release; a listing of thousands of files is not useful. */ @@ -71,43 +66,6 @@ class ArchiveService { } } -const run = promisify(execFile) - -/** Tools that print one archive entry to stdout, tried in order. */ -const EXTRACTORS: [string, (archive: string, entry: string) => string[]][] = [ - ['bsdtar', (archive, entry) => ['-xOf', archive, entry]], - ['unrar', (archive, entry) => ['p', '-inul', '-p-', '--', archive, entry]], -] - -/** - * The unpacked data of one entry of a complete (single-volume) archive: - * copied out when stored, else unpacked with bsdtar (libarchive, in the - * Docker image) or unrar. Null when that is not possible. - */ -export async function extractEntry(archive: Buffer, entry: RarEntry, maxBytes = 64 * 1024 * 1024): Promise { - const stored = storedEntryData(archive, entry) - if (stored) return stored - if (entry.encrypted || entry.directory || entry.splitBefore || entry.splitAfter) return null - if (entry.size !== null && entry.size > maxBytes) return null - - const dir = await mkdtemp(path.join(tmpdir(), 'rar-')) - try { - const file = path.join(dir, 'archive.rar') - await writeFile(file, archive) - for (const [tool, args] of EXTRACTORS) { - try { - const { stdout } = await run(tool, args(file, entry.name), { encoding: 'buffer', maxBuffer: maxBytes, timeout: 60_000 }) - if (stdout.length > 0) return stdout - } catch (error: any) { - if (error.code !== 'ENOENT') logger.warn(`${tool} could not extract "${entry.name}": ${error.message}`) - } - } - return null - } finally { - await rm(dir, { recursive: true, force: true }) - } -} - /** * A release name from an archive listing: the top-level directory shared by * all entries when it reads like a release name, otherwise the stem of the diff --git a/app/services/ArchiveWalkService.ts b/app/services/ArchiveWalkService.ts index 29becf2..67c4d44 100644 --- a/app/services/ArchiveWalkService.ts +++ b/app/services/ArchiveWalkService.ts @@ -1,13 +1,12 @@ import db from '@adonisjs/lucid/services/db' -import logger from '@adonisjs/core/services/logger' import env from '#start/env' import NntpService from '#services/NntpService' import QueueService from '#services/QueueService' import { segmentsOf } from '#services/FileStore' import { looksRandom, releaseNameFromContents } from '#services/ArchiveService' -import { decodeSegment, YencCrcError, YencError, type DecodedSegment } from '#lib/yenc/index' import { rarVolume } from '#lib/content/index' -import { walkRar, type RarInfo, type VolumeReader } from '#lib/rar/index' +import type { RarEntry, RarInfo } from '#lib/rar/index' +import { probeWalk, type WalkProbe, type WalkTask } from '#lib/content/probe' /** Stored per release, as for a first-volume listing (ArchiveService). */ const MAX_ENTRIES = 500 @@ -55,33 +54,50 @@ class ArchiveWalkService { await QueueService.archiveWalkQueue.add('walk', { releaseId }, { jobId: `walk-${releaseId}`, delay: WALK_DELAY_MS }) } - async walk(releaseId: number): Promise { - const release = await db.from('releases').where('id', releaseId).select('id', 'name', 'source', 'size', 'contents', 'encrypted').first() - if (!release) return { releaseId, skipped: 'missing' } + /** + * The database half before a walk (D41): the release's volumes, or the + * outcome when there's nothing to walk. + */ + async prepare(releaseId: number): Promise<{ task: WalkTask } | { outcome: WalkOutcome }> { + const release = await db.from('releases').where('id', releaseId).select('id', 'size', 'contents').first() + if (!release) return { outcome: { releaseId, skipped: 'missing' } } const before = Array.isArray(release.contents) ? release.contents : [] const listed = before.reduce((sum: number, e: any) => sum + Number(e.size ?? 0), 0) - if (!this.worthWalking(before.length, listed, Number(release.size))) return this.done(releaseId, { releaseId, skipped: 'covered' }) - + if (!this.worthWalking(before.length, listed, Number(release.size))) { + return { outcome: await this.done(releaseId, { releaseId, skipped: 'covered' }) } + } const volumes = await this.volumes(releaseId) - if (!volumes) return this.done(releaseId, { releaseId, skipped: 'volumes' }) + if (!volumes) return { outcome: await this.done(releaseId, { releaseId, skipped: 'volumes' }) } + return { + task: { + kind: 'walk', + releaseId, + volumes: volumes.map((file) => segmentsOf(file).map(({ number, id }) => ({ number, id }))), + maxSegments: this.maxSegments, + }, + } + } - const reader = new SegmentReader(volumes, this.maxSegments) - const walked = await walkRar(reader) + /** The database half after a walk: the fuller listing, the password flag and a better name. */ + async apply(releaseId: number, walked: WalkProbe): Promise { + const release = await db.from('releases').where('id', releaseId).select('id', 'name', 'source', 'contents', 'encrypted').first() + if (!release) return { releaseId, skipped: 'missing' } + const before = Array.isArray(release.contents) ? release.contents : [] const outcome: WalkOutcome = { releaseId, before: before.length, entries: walked.entries.length, complete: walked.complete, stopped: walked.stopped, - segments: reader.fetched, + segments: walked.segments, } if (walked.entries.length > before.length) { - const contents = walked.entries.slice(0, MAX_ENTRIES).map(({ stored: _s, dataOffset: _d, ...entry }) => entry) + const contents = walked.entries.slice(0, MAX_ENTRIES) const update: Record = { contents: JSON.stringify(contents), encrypted: Boolean(release.encrypted) || walked.headersEncrypted || walked.entries.some((e) => e.encrypted), } - const better = releaseNameFromContents({ entries: walked.entries }) + const better = releaseNameFromContents({ entries: walked.entries as RarEntry[] }) // The rule of ArchiveService.record: random and post-subject names give way, a spot's title stays. if (better && better !== release.name && release.source !== 'spot' && (looksRandom(release.name) || release.source === 'post')) { update.name = better @@ -92,6 +108,13 @@ class ArchiveWalkService { return this.done(releaseId, outcome) } + /** Walks a release here, on the inspection pool's connections when inspection isn't waiting for one. */ + async walk(releaseId: number): Promise { + const prepared = await this.prepare(releaseId) + if ('outcome' in prepared) return prepared.outcome + return this.apply(releaseId, await probeWalk(prepared.task, fetchLow)) + } + private async done(releaseId: number, outcome: WalkOutcome) { await db.from('releases').where('id', releaseId).update({ contents_walked_at: new Date() }) return outcome @@ -120,82 +143,6 @@ class ArchiveWalkService { } } -/** - * Reads volumes a segment at a time, finding the segment for an offset from - * the part size (yEnc posters use one size per set) and correcting from the - * `=ypart` offsets when a guess is off. - */ -export class SegmentReader implements VolumeReader { - fetched = 0 - private partSize = 0 - private readonly sizes = new Map() - private readonly cache = new Map() - - constructor( - private readonly files: any[], - private readonly maxSegments: number, - private readonly fetchBody: (messageId: string) => Promise = fetchLow - ) {} - - get volumes() { - return this.files.length - } - - size(v: number) { - return this.sizes.get(v) ?? null - } - - async read(v: number, offset: number, more: boolean): Promise { - const count = segmentsOf(this.files[v]).length - let index = this.partSize ? Math.min(count - 1, Math.floor(offset / this.partSize)) : 0 - for (let tries = 0; tries < 4; tries++) { - const segment = await this.segment(v, index) - if (!segment) return null - const end = segment.offset + segment.data.length - if (offset >= segment.offset && offset < end) { - const data = segment.data.subarray(offset - segment.offset) - if (!more || index + 1 >= count) return data - const next = await this.segment(v, index + 1) - return next ? Buffer.concat([data, next.data]) : null - } - const step = Math.floor((offset - segment.offset) / Math.max(1, segment.data.length)) - const moved = Math.max(0, Math.min(count - 1, index + (step === 0 ? (offset < segment.offset ? -1 : 1) : step))) - if (moved === index) return null - index = moved - } - return null - } - - private async segment(v: number, index: number): Promise { - const key = `${v}:${index}` - const cached = this.cache.get(key) - if (cached) return cached - const part = segmentsOf(this.files[v])[index] - if (!part?.id || this.fetched >= this.maxSegments) return null - this.fetched++ - let body: Buffer - try { - body = await this.fetchBody(part.id) - } catch (error: any) { - logger.debug(`Archive walk: segment ${index} of file ${this.files[v].id} unavailable: ${error.message}`) - return null - } - let segment: DecodedSegment - try { - segment = decodeSegment(body) - } catch (error) { - if (!(error instanceof YencError) || error instanceof YencCrcError) return null - segment = decodeSegment(body, { lenientSize: true }) - } - this.sizes.set(v, segment.meta.begin.size) - if (!this.partSize && segment.meta.part) this.partSize = segment.meta.part.end - segment.meta.part.begin + 1 - // Walks move forward: the last few segments are enough. - if (this.cache.size >= 4) this.cache.delete(this.cache.keys().next().value!) - this.cache.set(key, segment) - return segment - } -} - /** One article body on a pooled connection, behind inspection's waiting jobs. */ async function fetchLow(messageId: string): Promise { const conn = await NntpService.acquire(true) diff --git a/app/services/ContentService.ts b/app/services/ContentService.ts index 496b415..747ace2 100644 --- a/app/services/ContentService.ts +++ b/app/services/ContentService.ts @@ -1,24 +1,22 @@ -import { createHash } from 'node:crypto' import db from '@adonisjs/lucid/services/db' +import logger from '@adonisjs/core/services/logger' import NntpService from '#services/NntpService' import Par2Service, { type Par2Result } from '#services/Par2Service' -import ArchiveService, { extractEntry, looksRandom, type ArchiveResult } from '#services/ArchiveService' +import ArchiveService, { looksRandom, type ArchiveResult } from '#services/ArchiveService' import NameService from '#services/NameService' -import { decodeSegment, YencCrcError, YencError, YencFile } from '#lib/yenc/index' +import { decodeSegment } from '#lib/yenc/index' import NzbImportService, { type NzbImportResult } from '#services/NzbImportService' import { isArchiveStart, isLaterVolume, NICOVIDEO_FILE, NICOVIDEO_RELEASE, - nicovideoInfo, nicovideoReleaseName, - sniffContentType, - zstdPrefix, type ContentType, type NicovideoInfo, } from '#lib/content/index' -import { parseRarHeaders, type RarEntry, type RarInfo } from '#lib/rar/index' +import { chunkMd5, probeInspection, type InspectProbe, type InspectTask, type TaskSegment } from '#lib/content/probe' +import type { Segment } from '#lib/nzb/index' import { fileStem, releaseIdentity } from '#lib/release/index' import IngestService from '#services/IngestService' import { looksLikeFilename } from '#lib/subject/index' @@ -28,8 +26,6 @@ import SfvService from '#services/SfvService' import ArchiveWalkService from '#services/ArchiveWalkService' import { segmentsOf } from '#services/FileStore' -const FIRST_CHUNK = 16 * 1024 - /** * MD5 of the first 16 KiB of a file, as par2 records it, from the body of * the file's first part. Undefined when that part is shorter than 16 KiB but @@ -39,11 +35,6 @@ export function firstChunkMd5(body: Buffer, fileSize: number): string | undefine return chunkMd5(decodeSegment(body).data, fileSize) } -function chunkMd5(data: Buffer, fileSize: number): string | undefined { - if (data.length < Math.min(FIRST_CHUNK, fileSize)) return undefined - return createHash('md5').update(data.subarray(0, FIRST_CHUNK)).digest('hex') -} - export interface ContentResult { par2?: Par2Result archive?: ArchiveResult @@ -66,6 +57,8 @@ export interface ContentResult { * - its fingerprint (size + first-16-KiB MD5) matches it to a par2 entry, * - par2 files are recognised by content and name every file of their set, * - RAR headers give the release grouping, name and contents. + * Inspection is prepare (database) → probe (downloads and parsing, + * lib/content/probe; here or on a seeding node, D41) → apply (database). */ class ContentService { /** Random or missing names, and named first RAR volumes (for contents and password flag). */ @@ -128,89 +121,41 @@ class ContentService { return name } - /** Downloads a whole (small) file whose first segment is already decoded. */ - private async readWhole(first: Buffer, file: any): Promise { - const parts = segmentsOf(file) - const bytes = parts.reduce((sum, p) => sum + (p.size ?? 0), 0) - if (parts.length > WHOLE_MAX_PARTS || bytes > WHOLE_MAX_BYTES) return null - if (parts.length === 1) return first - - let assembled: YencFile | undefined - const conn = await NntpService.acquire() - let failed = false - try { - for (const part of parts) { - const segment = decodeSegment(await NntpService.fetchBody(conn, part.id), { lenientSize: true }) - assembled ??= YencFile.fromSegment(segment) - assembled.add(segment) - } - } catch (error) { - failed = true - throw error - } finally { - NntpService.release(conn, failed) - } - return assembled!.getBuffer() - } - /** - * Downloads a small archive that lists NZBs and imports each of them - * (CONCEPT.md D28): NZBs are posted inside RARs too, e.g. for obfuscated - * uploads by other posters. + * The database half before a probe (CONCEPT.md D41): what to inspect for + * a file, or null when there's nothing to do. A later 7z volume becomes its + * archive's first volume, once all volumes are in. */ - private async importArchiveNzbs(file: any, first: Buffer): Promise { - const archive = await this.readWhole(first, file) - const info = archive && parseRarHeaders(archive) - if (!archive || !info) return [] - const entries = info.entries.filter(isNzbEntry) - const results: NzbImportResult[] = [] - for (const entry of entries) { - const xml = await extractEntry(archive, entry) - if (!xml) continue - const imported = await NzbImportService.import(file, xml.toString('utf8'), { - entryName: entry.name, - moveContainer: entries.length === 1, - }) - if (imported) results.push(imported) - } - return results - } - - async inspect(fileId: number): Promise { + async prepare(fileId: number): Promise { const file = await db.from('files').where('id', fileId).first() - const binary = await db.from('binaries').where('file_id', fileId).first() if (!file || !this.isCandidate(file.filename)) return null - // A later 7z volume: list the set from its first volume once all volumes are in. if ((sevenZipVolume(String(file.real_name ?? file.filename)) ?? 0) > 1) { const volumes = await SevenZipService.volumes(file) if (!volumes || volumes[0].archive) return null - return this.inspect(Number(volumes[0].id)) + return this.prepare(Number(volumes[0].id)) } - const first = segmentsOf(file)[0] - if (!first?.id) return null + const segments = segmentsOf(file) + if (!segments[0]?.id) return null + // A 7z listing reads the last volume's tail (sniffing decides whether it's 7z: random names too). + const volumes = await SevenZipService.volumes(file) + const last = volumes && volumes.length > 1 ? segmentsOf(volumes[volumes.length - 1]) : segments + return { + kind: 'inspect', + fileId: Number(file.id), + name: String(file.real_name ?? file.filename), + segments: taskSegments(segments), + sevenZip: volumes ? { volumes: volumes.length, last: taskSegments(last, true) } : null, + } + } - const conn = await NntpService.acquire() - let failed = false - let body: Buffer - try { - body = await NntpService.fetchBody(conn, first.id) - } catch (error) { - failed = true - throw error - } finally { - NntpService.release(conn, failed) - } - // Some tools declare a size that leaves out a header (nzbe); the CRC still has to match. - let segment - try { - segment = decodeSegment(body) - } catch (error) { - if (!(error instanceof YencError) || error instanceof YencCrcError) throw error - segment = decodeSegment(body, { lenientSize: true }) - } - const size = segment.meta.begin.size - const contentType = sniffContentType(segment.data) - await db.from('files').where('id', fileId).update({ content_type: contentType }) + /** The database half after a probe: names, groupings, listings and flags from what it found. */ + async apply(task: InspectTask, probe: InspectProbe): Promise { + const result: ContentResult = { named: 0, contentType: probe.contentType } + const file = await db.from('files').where('id', task.fileId).first() + if (!file) return result + const binary = await db.from('binaries').where('file_id', task.fileId).first() + const { contentType } = probe + await db.from('files').where('id', file.id).update({ content_type: contentType }) // Fingerprint, so par2 sets (now or later) can name this file exactly. if (binary) { @@ -218,13 +163,12 @@ class ContentService { .from('binaries') .where('id', binary.id) .update({ - size: db.raw('coalesce(size, ?)', [size]), - md5_16k: db.raw('coalesce(md5_16k, ?)', [chunkMd5(segment.data, size) ?? null]), + size: db.raw('coalesce(size, ?)', [probe.size]), + md5_16k: db.raw('coalesce(md5_16k, ?)', [probe.md5_16k]), }) } - const result: ContentResult = { named: 0, contentType } - result.yencName = await this.nameFromYenc(file, binary, segment.meta.begin.name) + result.yencName = await this.nameFromYenc(file, binary, probe.yencName ?? '') if (contentType === 'nzbe' || contentType === 'rclone' || contentType === 'yencbin') { // An encrypted NZB, rclone crypt (names and data encrypted) or yEncBin's container: nothing // to read, and useless without the key. @@ -234,53 +178,105 @@ class ContentService { } return result } - const nicovideo = NICOVIDEO_FILE.exec(String(file.real_name ?? file.filename)) + const nicovideo = NICOVIDEO_FILE.exec(task.name) if (nicovideo) { - const info = nicovideoInfo(await zstdPrefix(segment.data)) - if (info) result.nicovideo = await this.nameNicovideo(file, info, nicovideo[1]) + if (probe.nicovideo) result.nicovideo = await this.nameNicovideo(file, probe.nicovideo, nicovideo[1]) return result } - if (/\.sfv$/i.test(String(file.real_name ?? file.filename))) { - const text = await this.readWhole(segment.data, file) - if (text) result.sfvMoved = (await SfvService.attach(file, text.toString('latin1'))) ?? undefined + if (/\.sfv$/i.test(task.name)) { + if (typeof probe.text === 'string') result.sfvMoved = (await SfvService.attach(file, probe.text)) ?? undefined return result } if (contentType === 'nzb') { - const xml = await this.readWhole(segment.data, file) - if (xml) result.nzb = (await NzbImportService.import(file, xml.toString('utf8'))) ?? undefined + if (typeof probe.text === 'string') result.nzb = (await NzbImportService.import(file, probe.text)) ?? undefined return result } - const set = await Par2Service.read(segment.data, segmentsOf(file)) - if (set && binary) { - result.par2 = await Par2Service.attach(binary, set) - } else if (contentType === '7z') { - const info = await SevenZipService.list(file, segment.data) - if (info) result.archive = await ArchiveService.record(file, binary, info) - } else { - const info = parseRarHeaders(segment.data) - if (info) { - result.archive = await ArchiveService.record(file, binary, info) - // The rest of the listing, later and at low priority (D40). - await ArchiveWalkService.queueAfterInspection(result.archive.releaseId ?? null, info) - if (hasNzbEntries(info)) result.archiveNzbs = await this.importArchiveNzbs(file, segment.data) + if (probe.par2 && binary) { + result.par2 = await Par2Service.attach(binary, probe.par2) + } else if (probe.archive && probe.archive.format === '7z') { + result.archive = await ArchiveService.record(file, binary, probe.archive) + } else if (probe.archive) { + result.archive = await ArchiveService.record(file, binary, probe.archive) + // The rest of the listing, later and at low priority (D40). + await ArchiveWalkService.queueAfterInspection(result.archive.releaseId ?? null, probe.archive) + if (probe.archiveNzbs?.length) { + result.archiveNzbs = [] + for (const { entryName, xml } of probe.archiveNzbs) { + const imported = await NzbImportService.import(file, xml, { entryName, moveContainer: (probe.archiveNzbEntries ?? 0) === 1 }) + if (imported) result.archiveNzbs.push(imported) + } } } if (binary) result.named = await NameService.resolve([binary.id]) return result } + + /** Inspects a file here: prepare, probe on the inspection pool's connections, apply. */ + async inspect(fileId: number): Promise { + const task = await this.prepare(fileId) + if (!task) return null + const probe = await probeInspection(task, fetchPooled, (message) => logger.warn(message)) + return this.apply(task, probe) + } } -/** NZBs that can be read from a single, unencrypted archive. */ -function hasNzbEntries(info: RarInfo): boolean { - return !info.volume && !info.headersEncrypted && info.entries.some(isNzbEntry) +/** A task's segment list: all of a small file's; of a big one, the first and the last few (a 7z tail). */ +function taskSegments(segments: Segment[], tailOnly = false): TaskSegment[] { + const list = segments.map(({ number, id, size }) => ({ number, id, size })) + if (list.length <= TASK_SEGMENTS) return list + return tailOnly ? list.slice(-TASK_SEGMENTS) : [list[0], ...list.slice(-TASK_SEGMENTS)] } -function isNzbEntry(entry: RarEntry): boolean { - return /\.nzb$/i.test(entry.name) && !entry.encrypted && !entry.directory +/** More than whole files (40 parts) and par2 files (20) are read; a 7z tail of 16 MB fits easily. */ +const TASK_SEGMENTS = 40 + +/** One article body on a pooled connection. */ +async function fetchPooled(messageId: string): Promise { + const conn = await NntpService.acquire() + let failed = false + try { + return await NntpService.timed(NntpService.fetchBody(conn, messageId), `BODY ${messageId}`) + } catch (error: any) { + // A missing article (430) leaves the connection fine. + failed = !/^4\d\d/.test(String(error.message)) + throw error + } finally { + NntpService.release(conn, failed) + } } -/** Small posted files (NZBs) that are read completely. */ -const WHOLE_MAX_PARTS = 40 -const WHOLE_MAX_BYTES = 25 * 1024 * 1024 +/** + * One line per finding, as worker:collection logs them (and the seed API + * for nodes' results). + */ +export function describeContent(filename: string, content: ContentResult | null): string[] { + const lines: string[] = [] + if (content?.par2) { + const { par2 } = content + lines.push(`Par2 "${filename}": release "${par2.releaseName}", ${par2.listed} files listed.`) + } else if (content?.archive) { + const { info, releaseRenamed, grouped } = content.archive + const flags = [info.headersEncrypted && 'encrypted headers', info.entries.some((e) => e.encrypted) && 'encrypted files'] + .filter(Boolean) + .join(', ') + lines.push( + `Archive "${filename}": ${info.format}, ${info.entries.length} entries${flags ? ` (${flags})` : ''}` + + (grouped ? ', grouped into a release' : '') + + (releaseRenamed ? `, release renamed to "${releaseRenamed}"` : '') + ) + } + if (content?.named) lines.push(`"${filename}" got its real name from par2.`) + if (content?.contentType === 'nzbe') lines.push(`"${filename}" is an encrypted NZB container.`) + if (content?.contentType === 'rclone') lines.push(`"${filename}" is rclone-encrypted.`) + if (content?.yencName) lines.push(`"${filename}" is "${content.yencName}" (yEnc name).`) + if (content?.nicovideo) lines.push(`"${filename}" → "${content.nicovideo}".`) + if (content?.sfvMoved) lines.push(`SFV "${filename}" moved ${content.sfvMoved} files into its release.`) + if (content?.contentType === 'yencbin') lines.push(`"${filename}" is a yEncBin encrypted container.`) + if (content?.nzb) { + const n = content.nzb + lines.push(`NZB "${filename}": ${n.listed} files (${n.known} known, ${n.added} added) → "${n.releaseName}".`) + } + return lines +} export default new ContentService() diff --git a/app/services/Par2Service.ts b/app/services/Par2Service.ts index 5507fe8..c2925d9 100644 --- a/app/services/Par2Service.ts +++ b/app/services/Par2Service.ts @@ -1,15 +1,10 @@ import db from '@adonisjs/lucid/services/db' -import type { Segment } from '#lib/nzb/index' import logger from '@adonisjs/core/services/logger' -import NntpService from '#services/NntpService' import IngestService, { PAR2_MATCH_WINDOW_SECONDS } from '#services/IngestService' -import { decodeSegment, YencFile } from '#lib/yenc/index' -import { isPar2, parsePar2, type Par2Set } from '#lib/par2/index' +import type { Par2Set } from '#lib/par2/index' import { par2ReleaseIdentity, par2ReleaseName } from '#lib/release/index' import NameService, { WEAK_RELEASE_SQL, isBetterName } from '#services/NameService' -/** Par2 index files are small; recovery volumes can be large and are not needed. */ -const MAX_PARTS = 20 /** Size-only matching below this is too ambiguous. */ const SIZE_ONLY_MIN = 1024 * 1024 @@ -32,34 +27,6 @@ export interface Par2Result { * body-derived binaries carry their exact size. See CONCEPT.md (D16). */ class Par2Service { - /** - * Reads a par2 file whose first segment is already decoded, downloading the - * remaining parts if there are any. Returns null when it is not par2. - */ - async read(first: Buffer, segments: Segment[]): Promise { - if (!isPar2(first)) return null - const ids = segments.map((part) => part.id) - if (ids.length > MAX_PARTS) return null - if (ids.length === 1) return parsePar2(first) - - let assembled: YencFile | undefined - const conn = await NntpService.acquire() - let failed = false - try { - for (const id of ids) { - const segment = decodeSegment(await NntpService.fetchBody(conn, id)) - assembled ??= YencFile.fromSegment(segment) - assembled.add(segment) - } - } catch (error) { - failed = true - throw error - } finally { - NntpService.release(conn, failed) - } - return parsePar2(assembled!.getBuffer()) - } - async attach(binary: any, set: Par2Set): Promise { const result = await db.transaction(async (trx) => { // Exact: size and first-16-KiB MD5 of a listed file. Only these may pull diff --git a/app/services/RemoteContentService.ts b/app/services/RemoteContentService.ts new file mode 100644 index 0000000..23b3110 --- /dev/null +++ b/app/services/RemoteContentService.ts @@ -0,0 +1,159 @@ +import { randomUUID } from 'node:crypto' +import { Job, Worker } from 'bullmq' +import logger from '@adonisjs/core/services/logger' +import queueConfig from '#config/queue' +import QueueService from '#services/QueueService' +import NntpService from '#services/NntpService' +import ContentService, { describeContent } from '#services/ContentService' +import ArchiveWalkService from '#services/ArchiveWalkService' +import type { ContentTask, InspectProbe, InspectTask, WalkProbe } from '#lib/content/probe' + +/** Inspection first; archive walks (D40) only when it has nothing waiting. */ +const QUEUES = ['collection-queue', 'archive-walk-queue'] as const +type QueueName = (typeof QUEUES)[number] + +/** A node has this long to post a job's result; after that the job goes back to the queue (BullMQ's stalled check). */ +const LOCK_MS = 10 * 60 * 1000 + +export interface RemoteJob { + queue: QueueName + jobId: string + /** The job's lock: whoever holds it may complete the job. */ + token: string + task: ContentTask +} + +export class JobLost extends Error {} + +/** + * Content inspection and archive walks on seeding nodes (CONCEPT.md D41). + * Nodes take jobs from the same queues worker:collection works on: a lease + * locks a job (BullMQ's manual processing) and gives the node a task that + * was prepared from the database; the node probes it, fetching articles + * through the relay (articles()), and posts the result, which is applied + * here. A job whose node goes quiet is unlocked after LOCK_MS and picked up + * again by whoever comes next. + */ +class RemoteContentService { + private readonly workers = new Map() + + /** Up to `max` jobs for a node, each locked for LOCK_MS. */ + async lease(nodeId: number, max: number): Promise { + const jobs: RemoteJob[] = [] + for (const queue of QUEUES) { + while (jobs.length < max) { + const token = `node-${nodeId}:${randomUUID()}` + const job = await this.worker(queue).getNextJob(token, { block: false }) + if (!job) break + try { + const task = await this.prepare(queue, job) + if (!task) { + await job.moveToCompleted(null, token, false) + continue + } + // What applying needs, kept with the job: the node's copy of the task isn't trusted. + if (task.kind === 'inspect') await job.updateData({ ...job.data, remote: { fileId: task.fileId, name: task.name } }) + jobs.push({ queue, jobId: job.id!, token, task }) + } catch (error: any) { + await job.moveToFailed(error, token, false).catch(() => {}) + } + } + } + return jobs + } + + /** Applies a node's result and completes the job. */ + async finish(queue: QueueName, jobId: string, token: string, result: InspectProbe | WalkProbe) { + const job = await this.job(queue, jobId) + let summary: unknown + if (queue === 'collection-queue') { + const remote = job.data.remote + if (!remote) throw new JobLost(`Job ${jobId} was not leased to a node`) + const task: InspectTask = { kind: 'inspect', fileId: Number(remote.fileId), name: String(remote.name), segments: [], sevenZip: null } + const content = await ContentService.apply(task, result as InspectProbe) + for (const line of describeContent(task.name, content)) logger.info(`[node] ${line}`) + summary = content.contentType + } else { + const outcome = await ArchiveWalkService.apply(Number(job.data.releaseId), result as WalkProbe) + logger.info( + `[node] Archive walk of release ${outcome.releaseId}: ${outcome.before} → ${outcome.entries} entries from ${outcome.segments} segments` + + (outcome.complete ? '' : ` (stopped: ${outcome.stopped})`) + + (outcome.renamed ? `, renamed to "${outcome.renamed}"` : '') + ) + summary = outcome + } + await this.locked(() => job.moveToCompleted(summary, token, false)) + } + + /** The probe failed on the node (e.g. a missing article): retried like a local failure. */ + async fail(queue: QueueName, jobId: string, token: string, error: string) { + const job = await this.job(queue, jobId) + await this.locked(() => job.moveToFailed(new Error(`node: ${error}`), token, false)) + } + + /** A stopping node gives its jobs back, without using up an attempt. */ + async release(queue: QueueName, jobId: string, token: string) { + const job = await this.job(queue, jobId) + await this.locked(() => job.moveToWait(token)) + } + + /** One article body for a relay node, on this process's NNTP pool. Null when the server doesn't have it. */ + async article(messageId: string): Promise { + const conn = await NntpService.acquire() + let failed = false + try { + return await NntpService.timed(NntpService.fetchBody(conn, messageId), `BODY ${messageId}`) + } catch (error: any) { + // 430 no such article: a fact about the article, the connection is fine. + if (/^4\d\d/.test(String(error.message))) return null + failed = true + throw error + } finally { + NntpService.release(conn, failed) + } + } + + /** Closes the lease workers' Redis connections (tests; a server just exits). */ + async close() { + await Promise.all([...this.workers.values()].map((worker) => worker.close())) + this.workers.clear() + } + + isQueue(name: unknown): name is QueueName { + return QUEUES.includes(name as QueueName) + } + + private async prepare(queue: QueueName, job: Job): Promise { + if (queue === 'collection-queue') return ContentService.prepare(Number(job.data.fileId)) + const prepared = await ArchiveWalkService.prepare(Number(job.data.releaseId)) + return 'task' in prepared ? prepared.task : null + } + + private async job(queue: QueueName, jobId: string): Promise { + const job = await Job.fromId(queue === 'collection-queue' ? QueueService.collectionQueue : QueueService.archiveWalkQueue, jobId) + if (!job) throw new JobLost(`Job ${jobId} is gone`) + return job + } + + /** BullMQ refuses a job whose lock another holder has now: the node's result came too late. */ + private async locked(action: () => Promise) { + try { + await action() + } catch (error: any) { + if (/lock|not in the active state|Missing key/i.test(String(error.message))) throw new JobLost(error.message) + throw error + } + } + + /** A worker that only hands out jobs (no processor, never runs). */ + private worker(queue: QueueName): Worker { + let worker = this.workers.get(queue) + if (!worker) { + worker = new Worker(queue, null, { connection: queueConfig.connection, lockDuration: LOCK_MS, autorun: false }) + this.workers.set(queue, worker) + } + return worker + } +} + +export default new RemoteContentService() diff --git a/app/services/SevenZipService.ts b/app/services/SevenZipService.ts index 8bb6421..762a11b 100644 --- a/app/services/SevenZipService.ts +++ b/app/services/SevenZipService.ts @@ -1,35 +1,10 @@ -import { execFile } from 'node:child_process' -import { mkdtemp, open, rm } from 'node:fs/promises' -import { tmpdir } from 'node:os' -import path from 'node:path' -import { promisify } from 'node:util' import db from '@adonisjs/lucid/services/db' -import logger from '@adonisjs/core/services/logger' -import NntpService from '#services/NntpService' -import { segmentsOf } from '#services/FileStore' -import { decodeSegment, type DecodedSegment } from '#lib/yenc/index' -import type { RarInfo } from '#lib/rar/index' -import { - checkHeader, - encodedHeaderEncrypted, - listingStart, - parseBsdtarListing, - parseSevenZipStart, - sevenZipVolume, - type SevenZipStart, -} from '#lib/sevenzip/index' - -const run = promisify(execFile) - -/** The tail read for a listing: header plus packed header; more is not a header. */ -const MAX_TAIL_BYTES = 16 * 1024 * 1024 +import { sevenZipVolume } from '#lib/sevenzip/index' /** - * Lists 7z archives, also split ones (.7z.001, .002, …), from the first - * segment of the first volume and the last segment(s) of the last one - * (CONCEPT.md D19). Those bytes are put at their place in a sparse file the - * size of the whole archive, which bsdtar (libarchive) lists: it only reads - * the start and the header at the end. + * The database side of 7z listings (CONCEPT.md D19): which volumes make up + * an archive. Listing itself (first segment, tail, bsdtar) is + * lib/sevenzip/list.ts, run by inspection here or on a node. */ class SevenZipService { /** @@ -56,152 +31,6 @@ class SevenZipService { const ordered = Array.from({ length: max }, (_, i) => byNumber.get(i + 1)) return max > 0 && ordered.every(Boolean) ? ordered : null } - - /** - * Lists the archive whose first volume starts with `firstData` (a decoded - * first segment). Null when it isn't a 7z start, when volumes are missing, - * or when the tail doesn't hold the header its start points to. - */ - async list(firstFile: any, firstData: Buffer): Promise { - const start = parseSevenZipStart(firstData) - if (!start) return null - const volumes = await this.volumes(firstFile) - if (!volumes) return null - if (start.headerSize === 0) return this.info([], false, false, volumes.length > 1) - - const tail = await this.readTail(volumes[volumes.length - 1], start) - if (!tail) return null - // Encrypted headers (-mhe): known from the header alone, no listing possible (most 7z uploads). - if ('headersEncrypted' in tail) return this.info([], true, true, volumes.length > 1) - - const dir = await mkdtemp(path.join(tmpdir(), '7z-')) - try { - const archive = path.join(dir, 'archive.7z') - const handle = await open(archive, 'w') - try { - await handle.write(firstData, 0, Math.min(firstData.length, start.archiveSize), 0) - await handle.write(tail.data, 0, tail.data.length, tail.from) - await handle.truncate(start.archiveSize) - } finally { - await handle.close() - } - return await this.bsdtar(archive, volumes.length > 1) - } finally { - await rm(dir, { recursive: true, force: true }) - } - } - - /** Lists the sparse archive; a trial extraction of one entry tells whether its data is encrypted. */ - private async bsdtar(archive: string, split: boolean): Promise { - let stdout = '' - try { - stdout = (await run('bsdtar', ['-tvf', archive], { timeout: 60_000, maxBuffer: 32 * 1024 * 1024 })).stdout - } catch (error: any) { - if (/header is encrypted/i.test(String(error.stderr))) return this.info([], true, true, split) - logger.warn(`bsdtar could not list a 7z archive: ${String(error.stderr || error.message).trim()}`) - return null - } - const entries = parseBsdtarListing(stdout) - const sample = entries.filter((e) => !e.directory).sort((a, b) => a.size - b.size)[0] - let encrypted = false - if (sample) { - try { - await run('bsdtar', ['-xOf', archive, sample.name], { timeout: 30_000, maxBuffer: 4 * 1024 * 1024, encoding: 'buffer' }) - } catch (error: any) { - // Without the data (it's all zeros here), anything else fails too: only this message counts. - encrypted = /content is encrypted/i.test(String(error.stderr)) - } - } - return this.info(entries, false, encrypted, split) - } - - private info(entries: { name: string; size: number; directory: boolean }[], headersEncrypted: boolean, encrypted: boolean, split: boolean): RarInfo { - return { - format: '7z', - volume: split, - firstVolume: true, - solid: false, - headersEncrypted, - truncated: false, - next: 0, - entries: entries.map((e) => ({ - name: e.name, - size: e.size, - packedSize: 0, - directory: e.directory, - encrypted: encrypted && !e.directory, - splitBefore: false, - splitAfter: false, - stored: false, - dataOffset: 0, - })), - } - } - - /** - * The archive's tail from the last volume: the header (checked against its - * CRC) and, for an encoded header, the packed stream before it. `from` is - * its position in the whole archive. - */ - private async readTail(last: any, start: SevenZipStart): Promise<{ from: number; data: Buffer } | { headersEncrypted: true } | null> { - const ids = new Map(segmentsOf(last).map((s) => [s.number, s])) - const numbers = [...ids.keys()] - if (numbers.length === 0) return null - const segments = new Map() - const segment = async (n: number) => { - if (!segments.has(n)) { - const id = ids.get(n)?.id - if (!id) throw new Error(`part ${n} of ${last.filename} is missing`) - segments.set(n, decodeSegment(await this.body(id), { lenientSize: true })) - } - return segments.get(n)! - } - - const lastNumber = numbers[numbers.length - 1] - const lastSegment = await segment(lastNumber) - const volumeSize = lastSegment.meta.begin.size - const volumeStart = start.archiveSize - volumeSize - const partBegin = (lastSegment.meta.part?.begin ?? 1) - 1 - const partSize = lastNumber > 1 ? partBegin / (lastNumber - 1) : volumeSize - if (volumeStart < 0 || !Number.isInteger(partSize) || partSize <= 0) return null - - // Bytes [from, to) of the whole archive, from the last volume's parts. - const read = async (from: number, to: number): Promise => { - if (from < volumeStart) return null - const a = from - volumeStart - const b = to - volumeStart - const pieces: Buffer[] = [] - for (let n = Math.floor(a / partSize) + 1; n <= Math.floor((b - 1) / partSize) + 1; n++) { - const s = await segment(n) - const offset = (s.meta.part?.begin ?? 1) - 1 - if (offset !== (n - 1) * partSize) return null - pieces.push(s.data.subarray(Math.max(0, a - offset), Math.max(0, Math.min(s.data.length, b - offset)))) - } - const data = Buffer.concat(pieces) - return data.length === b - a ? data : null - } - - const header = await read(start.headerStart, start.archiveSize) - if (!header || !checkHeader(start, header)) return null - if (encodedHeaderEncrypted(header)) return { headersEncrypted: true } - const from = Math.min(listingStart(start, header), start.headerStart) - if (start.archiveSize - from > MAX_TAIL_BYTES) return null - const data = await read(from, start.archiveSize) - return data ? { from, data } : null - } - - private async body(messageId: string): Promise { - const conn = await NntpService.acquire() - let failed = false - try { - return await NntpService.timed(NntpService.fetchBody(conn, messageId), `BODY ${messageId}`) - } catch (error: any) { - failed = !/^4\d\d/.test(String(error.message)) - throw error - } finally { - NntpService.release(conn, failed) - } - } } export default new SevenZipService() diff --git a/bin/console.ts b/bin/console.ts index 86bf4d2..6c9c434 100644 --- a/bin/console.ts +++ b/bin/console.ts @@ -18,6 +18,12 @@ const { Ignitor, prettyPrintError } = await import('@adonisjs/core') * URL to the application root. AdonisJS need it to resolve * paths to file and directories for scaffolding commands */ +/** + * The command being run, for start/env.ts: Ace takes the arguments out of + * process.argv before the environment is loaded. + */ +process.env.ACE_COMMAND ??= process.argv[2] ?? '' + const APP_ROOT = new URL('../', import.meta.url) /** diff --git a/commands/SeedNode.ts b/commands/SeedNode.ts index ffc3bcf..627eb8b 100644 --- a/commands/SeedNode.ts +++ b/commands/SeedNode.ts @@ -8,10 +8,18 @@ import { XoverClient, parseOverviews, type Overview } from '#lib/nntp/xover_clie import { runRange, type OverviewSource } from '#lib/seed/runner' import type { SeedBatch } from '#lib/seed/wire' import type { OverviewHeader } from '#lib/ingest/segment' +import { probeInspection, probeWalk, type ContentTask } from '#lib/content/probe' class LeaseLost extends Error {} class Stopping extends Error {} +interface InspectJob { + queue: string + jobId: string + token: string + task: ContentTask +} + interface Lease { id: number group: string @@ -45,11 +53,18 @@ export default class SeedNode extends BaseCommand { @flags.number({ description: - 'Ranges worked on at once, one NNTP connection each (default: SEED_CONNECTIONS or 1)', - default: Number(process.env.SEED_CONNECTIONS) || 1, + 'Ranges worked on at once, one NNTP connection each; 0 for an inspection-only node (default: SEED_CONNECTIONS or 1)', + default: process.env.SEED_CONNECTIONS ? Number(process.env.SEED_CONNECTIONS) : 1, }) declare connections: number + @flags.number({ + description: + 'Content inspections and archive walks worked on at once, articles through the seed API (default: SEED_INSPECTORS or 0)', + default: Number(process.env.SEED_INSPECTORS) || 0, + }) + declare inspectors: number + @flags.number({ description: 'Articles per XOVER', default: 20_000 }) declare chunk: number @@ -85,9 +100,13 @@ export default class SeedNode extends BaseCommand { const api = new SeedApi(base, token, (message) => this.logger.warning(message)) this.logger.info( - `Seeding node for ${base}: ${this.connections} range(s) at once, ${this.fetch === 'direct' ? 'own NNTP connections' : 'headers through the seed API'}.` + `Seeding node for ${base}: ${this.connections} range(s) at once, ${this.fetch === 'direct' ? 'own NNTP connections' : 'headers through the seed API'}` + + (this.inspectors > 0 ? `; ${this.inspectors} inspection(s) at once, articles through the seed API.` : '.') ) - await Promise.all(Array.from({ length: this.connections }, (_, i) => this.work(api, i + 1))) + await Promise.all([ + ...Array.from({ length: this.connections }, (_, i) => this.work(api, i + 1)), + ...Array.from({ length: this.inspectors }, (_, i) => this.inspect(api, i + 1)), + ]) this.logger.info('Seeding node stopped.') } @@ -137,6 +156,50 @@ export default class SeedNode extends BaseCommand { } } + /** + * Content inspection and archive walks (CONCEPT.md D41): lease a few jobs, + * probe each with articles through the seed API, post the results. The + * indexer applies them; a job left unanswered goes back after 10 minutes. + */ + private async inspect(api: SeedApi, slot: number) { + const log = (message: string) => this.logger.info(`[inspect ${slot}] ${message}`) + const fetchArticle = (id: string) => api.get(`/api/v1/seed/articles?id=${encodeURIComponent(id)}`, 4) + while (!this.stopping) { + let jobs: InspectJob[] = [] + try { + jobs = ((await api.post('/api/v1/seed/inspect/lease', { max: 2 })) as { jobs: InspectJob[] } | null)?.jobs ?? [] + } catch (error: any) { + if (error instanceof Unauthorized) throw error + log(`Lease failed: ${error.message}`) + } + if (jobs.length === 0) { + if (this.once) return + await this.pause(30_000) + continue + } + for (const job of jobs) { + const ref = { queue: job.queue, jobId: job.jobId, token: job.token } + if (this.stopping) { + await api.post('/api/v1/seed/inspect/release', ref).catch(() => {}) + continue + } + try { + const result = + job.task.kind === 'inspect' + ? await probeInspection(job.task, fetchArticle, (message) => log(message)) + : await probeWalk(job.task, fetchArticle) + await api.post('/api/v1/seed/inspect/done', { ...ref, result: JSON.stringify(result) }) + } catch (error: any) { + if (error instanceof Unauthorized) throw error + // The job's lock expired and it went to another node: nothing to report. + if (error instanceof LeaseLost) continue + log(`${job.task.kind === 'inspect' ? `"${job.task.name}"` : `walk of release ${job.task.releaseId}`} failed: ${error.message}`) + await api.post('/api/v1/seed/inspect/failed', { ...ref, error: error.message }).catch(() => {}) + } + } + } + } + private connect() { return XoverClient.connect({ host: nntpConfig.host, @@ -249,12 +312,12 @@ class SeedApi { return text ? JSON.parse(text) : null } - /** A binary GET (a relay chunk). */ - async get(path: string): Promise { - return (await this.request('GET', path)) ?? Buffer.alloc(0) + /** A binary GET (a relay chunk, an article). */ + async get(path: string, attempts = 14): Promise { + return (await this.request('GET', path, undefined, attempts)) ?? Buffer.alloc(0) } - private async request(method: string, path: string, payload?: Buffer): Promise { + private async request(method: string, path: string, payload?: Buffer, attempts = 14): Promise { for (let attempt = 1; ; attempt++) { let status = 0 let data = Buffer.alloc(0) @@ -283,7 +346,7 @@ class SeedApi { throw new Error(`Seed API ${path}: ${status} ${text()}`) // Network errors, 404 and 5xx: storing a batch twice is harmless, so try again. // ~20 minutes in all, longer than a lease lasts: a restart or maintenance of the API is waited out. - if (attempt >= 14) throw new Error(`Seed API ${path}: ${status || 'no response'} ${text()}`) + if (attempt >= attempts) throw new Error(`Seed API ${path}: ${status || 'no response'} ${text()}`) const wait = Math.min(120_000, 2000 * 2 ** (attempt - 1)) this.warn(`Seed API ${path}: ${status || text()}; retrying in ${wait / 1000} s`) await sleep(wait) diff --git a/commands/workers/Collection.ts b/commands/workers/Collection.ts index 9b42579..9a955e8 100644 --- a/commands/workers/Collection.ts +++ b/commands/workers/Collection.ts @@ -6,7 +6,7 @@ import env from '#start/env' import Shutdown from '#services/Shutdown' import QueueService from '#services/QueueService' import NntpService from '#services/NntpService' -import ContentService from '#services/ContentService' +import ContentService, { describeContent } from '#services/ContentService' import ArchiveWalkService from '#services/ArchiveWalkService' import File from '#models/file' @@ -33,31 +33,7 @@ export default class CollectionWorker extends BaseCommand { try { const content = await ContentService.inspect(fileId) - if (content?.par2) { - const { par2 } = content - this.logger.info(`Par2 "${file.filename}": release "${par2.releaseName}", ${par2.listed} files listed.`) - } else if (content?.archive) { - const { info, releaseRenamed, grouped } = content.archive - const flags = [info.headersEncrypted && 'encrypted headers', info.entries.some((e) => e.encrypted) && 'encrypted files'] - .filter(Boolean) - .join(', ') - this.logger.info( - `Archive "${file.filename}": ${info.format}, ${info.entries.length} entries${flags ? ` (${flags})` : ''}` + - (grouped ? ', grouped into a release' : '') + - (releaseRenamed ? `, release renamed to "${releaseRenamed}"` : '') - ) - } - if (content?.named) this.logger.info(`"${file.filename}" got its real name from par2.`) - if (content?.contentType === 'nzbe') this.logger.info(`"${file.filename}" is an encrypted NZB container.`) - if (content?.contentType === 'rclone') this.logger.info(`"${file.filename}" is rclone-encrypted.`) - if (content?.yencName) this.logger.info(`"${file.filename}" is "${content.yencName}" (yEnc name).`) - if (content?.nicovideo) this.logger.info(`"${file.filename}" → "${content.nicovideo}".`) - if (content?.sfvMoved) this.logger.info(`SFV "${file.filename}" moved ${content.sfvMoved} files into its release.`) - if (content?.contentType === 'yencbin') this.logger.info(`"${file.filename}" is a yEncBin encrypted container.`) - if (content?.nzb) { - const n = content.nzb - this.logger.info(`NZB "${file.filename}": ${n.listed} files (${n.known} known, ${n.added} added) → "${n.releaseName}".`) - } + for (const line of describeContent(file.filename, content)) this.logger.info(line) } catch (error: any) { this.logger.warning(`Content inspection of "${file.filename}" failed: ${error.message}`) } diff --git a/lib/content/probe.ts b/lib/content/probe.ts new file mode 100644 index 0000000..32c4388 --- /dev/null +++ b/lib/content/probe.ts @@ -0,0 +1,193 @@ +import { createHash } from 'node:crypto' +import { decodeSegment, YencCrcError, YencError, YencFile, type DecodedSegment } from '../yenc/index.js' +import { isPar2, parsePar2, type Par2Set } from '../par2/index.js' +import { parseRarHeaders, type RarEntry, type RarInfo } from '../rar/parse.js' +import { extractEntry } from '../rar/extract.js' +import { walkRar } from '../rar/walk.js' +import { SegmentReader, type BodyFetcher } from '../rar/segment_reader.js' +import { listSevenZip } from '../sevenzip/list.js' +import { sniffContentType, type ContentType } from './sniff.js' +import { NICOVIDEO_FILE, nicovideoInfo, zstdPrefix, type NicovideoInfo } from './nicovideo.js' + +/** + * Content inspection's download-and-parse half (CONCEPT.md D21, D41), + * without a database: what the indexer needs to know about a file, from its + * segments. The indexer prepares a task from the database, the probe runs + * here or on a node (articles through the seed API's relay), and the + * indexer applies the result. Tasks and results are plain JSON. + */ + +export interface TaskSegment { + number: number + id: string + /** Article size in bytes. */ + size?: number +} + +/** Inspect one file: its first segment, and more where its kind needs it. */ +export interface InspectTask { + kind: 'inspect' + fileId: number + /** real_name, else filename. */ + name: string + segments: TaskSegment[] + /** For a 7z listing: the archive's volume count and its last volume's segments; null when volumes are missing. */ + sevenZip: { volumes: number; last: TaskSegment[] } | null +} + +export interface InspectProbe { + /** The whole file's size, from the yEnc header. */ + size: number + contentType: ContentType | null + md5_16k: string | null + /** The file name in the yEnc header. */ + yencName: string + nicovideo?: NicovideoInfo | null + /** A whole small SFV (latin1) or NZB (utf8). */ + text?: string | null + par2?: Par2Set | null + archive?: RarInfo | null + /** NZBs read from inside an unsplit archive, and how many NZB entries it lists. */ + archiveNzbs?: { entryName: string; xml: string }[] + archiveNzbEntries?: number +} + +/** Walk a RAR set's headers for its complete listing (D40). */ +export interface WalkTask { + kind: 'walk' + releaseId: number + volumes: TaskSegment[][] + maxSegments: number +} + +export interface WalkProbe { + entries: Omit[] + complete: boolean + stopped?: string + headersEncrypted: boolean + segments: number +} + +export type ContentTask = InspectTask | WalkTask + +/** Small posted files (NZBs, SFVs) that are read completely. */ +const WHOLE_MAX_PARTS = 40 +const WHOLE_MAX_BYTES = 25 * 1024 * 1024 +/** Par2 index files are small; recovery volumes can be large and are not needed. */ +const PAR2_MAX_PARTS = 20 +const FIRST_CHUNK = 16 * 1024 + +/** MD5 of the first 16 KiB of a file, as par2 records it; undefined when the data is shorter than that but the file isn't. */ +export function chunkMd5(data: Buffer, fileSize: number): string | undefined { + if (data.length < Math.min(FIRST_CHUNK, fileSize)) return undefined + return createHash('md5').update(data.subarray(0, FIRST_CHUNK)).digest('hex') +} + +/** Decodes a body; some tools declare a size that leaves out a header (nzbe), the CRC still has to match. */ +export function decodeBody(body: Buffer): DecodedSegment { + try { + return decodeSegment(body) + } catch (error) { + if (!(error instanceof YencError) || error instanceof YencCrcError) throw error + return decodeSegment(body, { lenientSize: true }) + } +} + +/** All of a (small) file whose first segment is already decoded; null when it is too big. */ +export async function readWhole(first: Buffer, segments: TaskSegment[], fetch: BodyFetcher): Promise { + const bytes = segments.reduce((sum, p) => sum + (p.size ?? 0), 0) + if (segments.length > WHOLE_MAX_PARTS || bytes > WHOLE_MAX_BYTES) return null + if (segments.length === 1) return first + let assembled: YencFile | undefined + for (const part of segments) { + const segment = decodeSegment(await fetch(part.id), { lenientSize: true }) + assembled ??= YencFile.fromSegment(segment) + assembled.add(segment) + } + return assembled!.getBuffer() +} + +/** A par2 file whose first segment is already decoded, with its other parts if any; null when it is not par2. */ +export async function readPar2(first: Buffer, segments: TaskSegment[], fetch: BodyFetcher): Promise { + if (!isPar2(first)) return null + if (segments.length > PAR2_MAX_PARTS) return null + if (segments.length === 1) return parsePar2(first) + let assembled: YencFile | undefined + for (const part of segments) { + const segment = decodeSegment(await fetch(part.id)) + assembled ??= YencFile.fromSegment(segment) + assembled.add(segment) + } + return parsePar2(assembled!.getBuffer()) +} + +export function isNzbEntry(entry: Pick): boolean { + return /\.nzb$/i.test(entry.name) && !entry.encrypted && !entry.directory +} + +/** NZBs that can be read from a single, unencrypted archive. */ +function hasNzbEntries(info: RarInfo): boolean { + return !info.volume && !info.headersEncrypted && info.entries.some(isNzbEntry) +} + +export async function probeInspection( + task: InspectTask, + fetch: BodyFetcher, + warn: (message: string) => void = () => {} +): Promise { + const segment = decodeBody(await fetch(task.segments[0].id)) + const size = segment.meta.begin.size + const contentType = sniffContentType(segment.data) + const probe: InspectProbe = { + size, + contentType, + md5_16k: chunkMd5(segment.data, size) ?? null, + yencName: segment.meta.begin.name, + } + // An encrypted NZB, rclone crypt or yEncBin's container: nothing to read. + if (contentType === 'nzbe' || contentType === 'rclone' || contentType === 'yencbin') return probe + if (NICOVIDEO_FILE.test(task.name)) { + probe.nicovideo = nicovideoInfo(await zstdPrefix(segment.data)) + return probe + } + if (/\.sfv$/i.test(task.name)) { + probe.text = (await readWhole(segment.data, task.segments, fetch))?.toString('latin1') ?? null + return probe + } + if (contentType === 'nzb') { + probe.text = (await readWhole(segment.data, task.segments, fetch))?.toString('utf8') ?? null + return probe + } + probe.par2 = await readPar2(segment.data, task.segments, fetch) + if (probe.par2) return probe + if (contentType === '7z') { + probe.archive = task.sevenZip ? await listSevenZip(segment.data, task.sevenZip.volumes, task.sevenZip.last, fetch, warn) : null + return probe + } + const info = parseRarHeaders(segment.data) + probe.archive = info + if (info && hasNzbEntries(info)) { + probe.archiveNzbs = [] + const archive = await readWhole(segment.data, task.segments, fetch) + const archived = archive && parseRarHeaders(archive) + const entries = archived ? archived.entries.filter(isNzbEntry) : [] + probe.archiveNzbEntries = entries.length + for (const entry of entries) { + const xml = await extractEntry(archive!, entry, undefined, warn) + if (xml) probe.archiveNzbs.push({ entryName: entry.name, xml: xml.toString('utf8') }) + } + } + return probe +} + +export async function probeWalk(task: WalkTask, fetch: BodyFetcher): Promise { + const reader = new SegmentReader(task.volumes, task.maxSegments, fetch) + const walked = await walkRar(reader) + return { + entries: walked.entries.map(({ stored: _s, dataOffset: _d, ...entry }) => entry), + complete: walked.complete, + stopped: walked.stopped, + headersEncrypted: walked.headersEncrypted, + segments: reader.fetched, + } +} diff --git a/lib/rar/extract.ts b/lib/rar/extract.ts new file mode 100644 index 0000000..a83fbe2 --- /dev/null +++ b/lib/rar/extract.ts @@ -0,0 +1,48 @@ +import { execFile } from 'node:child_process' +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { promisify } from 'node:util' +import { storedEntryData, type RarEntry } from './parse.js' + +const run = promisify(execFile) + +/** Tools that print one archive entry to stdout, tried in order. */ +const EXTRACTORS: [string, (archive: string, entry: string) => string[]][] = [ + ['bsdtar', (archive, entry) => ['-xOf', archive, entry]], + ['unrar', (archive, entry) => ['p', '-inul', '-p-', '--', archive, entry]], +] + +/** + * The unpacked data of one entry of a complete (single-volume) archive: + * copied out when stored, else unpacked with bsdtar (libarchive, in the + * Docker image) or unrar. Null when that is not possible. + */ +export async function extractEntry( + archive: Buffer, + entry: RarEntry, + maxBytes = 64 * 1024 * 1024, + warn: (message: string) => void = () => {} +): Promise { + const stored = storedEntryData(archive, entry) + if (stored) return stored + if (entry.encrypted || entry.directory || entry.splitBefore || entry.splitAfter) return null + if (entry.size !== null && entry.size > maxBytes) return null + + const dir = await mkdtemp(path.join(tmpdir(), 'rar-')) + try { + const file = path.join(dir, 'archive.rar') + await writeFile(file, archive) + for (const [tool, args] of EXTRACTORS) { + try { + const { stdout } = await run(tool, args(file, entry.name), { encoding: 'buffer', maxBuffer: maxBytes, timeout: 60_000 }) + if (stdout.length > 0) return stdout + } catch (error: any) { + if (error.code !== 'ENOENT') warn(`${tool} could not extract "${entry.name}": ${error.message}`) + } + } + return null + } finally { + await rm(dir, { recursive: true, force: true }) + } +} diff --git a/lib/rar/index.ts b/lib/rar/index.ts index b9d3527..e45e870 100644 --- a/lib/rar/index.ts +++ b/lib/rar/index.ts @@ -4,3 +4,5 @@ */ export { parseRarHeaders, rarFormat, decodeRar4Name, storedEntryData, type RarInfo, type RarEntry } from './parse.js' export { walkRar, type VolumeReader, type RarWalk } from './walk.js' +export { SegmentReader, type BodyFetcher } from './segment_reader.js' +export { extractEntry } from './extract.js' diff --git a/lib/rar/segment_reader.ts b/lib/rar/segment_reader.ts new file mode 100644 index 0000000..e76522d --- /dev/null +++ b/lib/rar/segment_reader.ts @@ -0,0 +1,82 @@ +import { decodeSegment, YencCrcError, YencError, type DecodedSegment } from '../yenc/index.js' +import type { VolumeReader } from './walk.js' + +/** Downloads one article body (on a pooled connection, or through the seed API's relay). */ +export type BodyFetcher = (messageId: string) => Promise + +/** + * Reads volumes (each a list of segment message ids, in order) a segment at + * a time, finding the segment for an offset from the part size (yEnc + * posters use one size per set) and correcting from the `=ypart` offsets + * when a guess is off. Stops reading after `maxSegments` downloads. + */ +export class SegmentReader implements VolumeReader { + fetched = 0 + private partSize = 0 + private readonly sizes = new Map() + private readonly cache = new Map() + + constructor( + private readonly files: { id: string }[][], + private readonly maxSegments: number, + private readonly fetchBody: BodyFetcher + ) {} + + get volumes() { + return this.files.length + } + + size(v: number) { + return this.sizes.get(v) ?? null + } + + async read(v: number, offset: number, more: boolean): Promise { + const count = this.files[v].length + let index = this.partSize ? Math.min(count - 1, Math.floor(offset / this.partSize)) : 0 + for (let tries = 0; tries < 4; tries++) { + const segment = await this.segment(v, index) + if (!segment) return null + const end = segment.offset + segment.data.length + if (offset >= segment.offset && offset < end) { + const data = segment.data.subarray(offset - segment.offset) + if (!more || index + 1 >= count) return data + const next = await this.segment(v, index + 1) + return next ? Buffer.concat([data, next.data]) : null + } + const step = Math.floor((offset - segment.offset) / Math.max(1, segment.data.length)) + const moved = Math.max(0, Math.min(count - 1, index + (step === 0 ? (offset < segment.offset ? -1 : 1) : step))) + if (moved === index) return null + index = moved + } + return null + } + + private async segment(v: number, index: number): Promise { + const key = `${v}:${index}` + const cached = this.cache.get(key) + if (cached) return cached + const part = this.files[v][index] + if (!part?.id || this.fetched >= this.maxSegments) return null + this.fetched++ + let body: Buffer + try { + body = await this.fetchBody(part.id) + } catch { + // A missing article (or the relay failing) ends the walk with what was found. + return null + } + let segment: DecodedSegment + try { + segment = decodeSegment(body) + } catch (error) { + if (!(error instanceof YencError) || error instanceof YencCrcError) return null + segment = decodeSegment(body, { lenientSize: true }) + } + this.sizes.set(v, segment.meta.begin.size) + if (!this.partSize && segment.meta.part) this.partSize = segment.meta.part.end - segment.meta.part.begin + 1 + // Walks move forward: the last few segments are enough. + if (this.cache.size >= 4) this.cache.delete(this.cache.keys().next().value!) + this.cache.set(key, segment) + return segment + } +} diff --git a/lib/sevenzip/list.ts b/lib/sevenzip/list.ts new file mode 100644 index 0000000..d5e483f --- /dev/null +++ b/lib/sevenzip/list.ts @@ -0,0 +1,171 @@ +import { execFile } from 'node:child_process' +import { mkdtemp, open, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { promisify } from 'node:util' +import { decodeSegment, type DecodedSegment } from '../yenc/index.js' +import type { RarInfo } from '../rar/parse.js' +import type { BodyFetcher } from '../rar/segment_reader.js' +import { + checkHeader, + encodedHeaderEncrypted, + listingStart, + parseBsdtarListing, + parseSevenZipStart, + type SevenZipStart, +} from './index.js' + +const run = promisify(execFile) + +/** The tail read for a listing: header plus packed header; more is not a header. */ +const MAX_TAIL_BYTES = 16 * 1024 * 1024 + +/** + * Lists a 7z archive, also a split one (.7z.001, .002, …), from the first + * segment of the first volume (`firstData`, decoded) and the last segment(s) + * of the last one (CONCEPT.md D19). Those bytes are put at their place in a + * sparse file the size of the whole archive, which bsdtar (libarchive) + * lists: it only reads the start and the header at the end. Null when it + * isn't a 7z start, or the tail doesn't hold the header its start points to. + */ +export async function listSevenZip( + firstData: Buffer, + volumeCount: number, + lastVolume: { number: number; id: string }[], + fetch: BodyFetcher, + warn: (message: string) => void = () => {} +): Promise { + const start = parseSevenZipStart(firstData) + if (!start) return null + const split = volumeCount > 1 + if (start.headerSize === 0) return sevenZipInfo([], false, false, split) + + const tail = await readTail(lastVolume, start, fetch) + if (!tail) return null + // Encrypted headers (-mhe): known from the header alone, no listing possible (most 7z uploads). + if ('headersEncrypted' in tail) return sevenZipInfo([], true, true, split) + + const dir = await mkdtemp(path.join(tmpdir(), '7z-')) + try { + const archive = path.join(dir, 'archive.7z') + const handle = await open(archive, 'w') + try { + await handle.write(firstData, 0, Math.min(firstData.length, start.archiveSize), 0) + await handle.write(tail.data, 0, tail.data.length, tail.from) + await handle.truncate(start.archiveSize) + } finally { + await handle.close() + } + return await bsdtar(archive, split, warn) + } finally { + await rm(dir, { recursive: true, force: true }) + } +} + +/** Lists the sparse archive; a trial extraction of one entry tells whether its data is encrypted. */ +async function bsdtar(archive: string, split: boolean, warn: (message: string) => void): Promise { + let stdout = '' + try { + stdout = (await run('bsdtar', ['-tvf', archive], { timeout: 60_000, maxBuffer: 32 * 1024 * 1024 })).stdout + } catch (error: any) { + if (/header is encrypted/i.test(String(error.stderr))) return sevenZipInfo([], true, true, split) + warn(`bsdtar could not list a 7z archive: ${String(error.stderr || error.message).trim()}`) + return null + } + const entries = parseBsdtarListing(stdout) + const sample = entries.filter((e) => !e.directory).sort((a, b) => a.size - b.size)[0] + let encrypted = false + if (sample) { + try { + await run('bsdtar', ['-xOf', archive, sample.name], { timeout: 30_000, maxBuffer: 4 * 1024 * 1024, encoding: 'buffer' }) + } catch (error: any) { + // Without the data (it's all zeros here), anything else fails too: only this message counts. + encrypted = /content is encrypted/i.test(String(error.stderr)) + } + } + return sevenZipInfo(entries, false, encrypted, split) +} + +export function sevenZipInfo( + entries: { name: string; size: number; directory: boolean }[], + headersEncrypted: boolean, + encrypted: boolean, + split: boolean +): RarInfo { + return { + format: '7z', + volume: split, + firstVolume: true, + solid: false, + headersEncrypted, + truncated: false, + next: 0, + entries: entries.map((e) => ({ + name: e.name, + size: e.size, + packedSize: 0, + directory: e.directory, + encrypted: encrypted && !e.directory, + splitBefore: false, + splitAfter: false, + stored: false, + dataOffset: 0, + })), + } +} + +/** + * The archive's tail from the last volume: the header (checked against its + * CRC) and, for an encoded header, the packed stream before it. `from` is + * its position in the whole archive. + */ +async function readTail( + last: { number: number; id: string }[], + start: SevenZipStart, + fetch: BodyFetcher +): Promise<{ from: number; data: Buffer } | { headersEncrypted: true } | null> { + const ids = new Map(last.map((s) => [s.number, s])) + const numbers = [...ids.keys()] + if (numbers.length === 0) return null + const segments = new Map() + const segment = async (n: number) => { + if (!segments.has(n)) { + const id = ids.get(n)?.id + if (!id) throw new Error(`part ${n} of the last 7z volume is missing`) + segments.set(n, decodeSegment(await fetch(id), { lenientSize: true })) + } + return segments.get(n)! + } + + const lastNumber = numbers[numbers.length - 1] + const lastSegment = await segment(lastNumber) + const volumeSize = lastSegment.meta.begin.size + const volumeStart = start.archiveSize - volumeSize + const partBegin = (lastSegment.meta.part?.begin ?? 1) - 1 + const partSize = lastNumber > 1 ? partBegin / (lastNumber - 1) : volumeSize + if (volumeStart < 0 || !Number.isInteger(partSize) || partSize <= 0) return null + + // Bytes [from, to) of the whole archive, from the last volume's parts. + const read = async (from: number, to: number): Promise => { + if (from < volumeStart) return null + const a = from - volumeStart + const b = to - volumeStart + const pieces: Buffer[] = [] + for (let n = Math.floor(a / partSize) + 1; n <= Math.floor((b - 1) / partSize) + 1; n++) { + const s = await segment(n) + const offset = (s.meta.part?.begin ?? 1) - 1 + if (offset !== (n - 1) * partSize) return null + pieces.push(s.data.subarray(Math.max(0, a - offset), Math.max(0, Math.min(s.data.length, b - offset)))) + } + const data = Buffer.concat(pieces) + return data.length === b - a ? data : null + } + + const header = await read(start.headerStart, start.archiveSize) + if (!header || !checkHeader(start, header)) return null + if (encodedHeaderEncrypted(header)) return { headersEncrypted: true } + const from = Math.min(listingStart(start, header), start.headerStart) + if (start.archiveSize - from > MAX_TAIL_BYTES) return null + const data = await read(from, start.archiveSize) + return data ? { from, data } : null +} diff --git a/seed-node/README.md b/seed-node/README.md index 2c210d2..b31a95a 100644 --- a/seed-node/README.md +++ b/seed-node/README.md @@ -5,8 +5,14 @@ D38). It asks the seed API for a range over HTTPS, gets that range's headers (by default through the seed API, which fetches them from the news server and passes them on gzip-compressed), groups them into files, and posts the results back. It only makes outgoing connections: no port forwarding, -database, Redis or news server account. Only a token and -`SEED_CONNECTIONS` need setting. +database, Redis or news server account. Only the token needs setting, plus +how much work to take on. + +A node can also do content inspection and archive walks (CONCEPT.md D40, +D41) with `SEED_INSPECTORS`: it takes files from the indexer's inspection +queue, downloads their first segment(s) through the seed API, works out +what they are (par2 sets, RAR and 7z listings, NZBs, passwords) and posts +the findings back; the indexer only stores them. Headers through the seed API (`SEED_FETCH=relay`, the default) use the indexer's news server account: `SEED_NNTP_CONNECTIONS` in the indexer's @@ -22,8 +28,9 @@ the NNTP lines in `node.env`). server (port 563 for TLS). - Per connection: roughly 0.4 CPU core and 1–3 MB/s download (headers come gzip-compressed); a few hundred MB of RAM per node. Disk: about 1.5 GB for - the image. -- An NNTP account. Every connection a node opens counts against that + the image. Per inspector: up to a core while decoding, and a few MB/s + download (whole articles, ~700 KB each). +- Only with `SEED_FETCH=direct`: an NNTP account. Every connection a node opens counts against that account's connection limit, next to the ~20 the indexer itself uses if it's the same account. Too many gives `481`/`502` errors at login. Some providers also limit the number of IP addresses at once (Eweka: 2, as @@ -55,13 +62,12 @@ for the same name replaces the token; `--disable` revokes it. |---|---| | `SEED_URL` | `https://seed.exp.sforum.nl` (leave as is) | | `SEED_TOKEN` | the token from step 1 | -| `SEED_CONNECTIONS` | ranges worked on at once, one NNTP connection each. Start with 2–4. | -| `NNTP_HOST`, `NNTP_PORT`, `NNTP_SECURE` | the news server; over the internet use TLS: port `563`, `NNTP_SECURE=true` (for Eweka: `news.eweka.nl`) | -| `NNTP_USER`, `NNTP_PASSWORD` | the NNTP account | -| `NNTP_TLS_SERVERNAME` | only through a tunnel: the news server's real hostname, for the TLS certificate check | +| `SEED_CONNECTIONS` | ranges seeded at once. Start with 2–4; `0` for no seeding. | +| `SEED_INSPECTORS` | inspections and archive walks at once (default `0`). Try 2–4 on a machine with spare cores. | +| `SEED_FETCH=direct` and `NNTP_*` | optional: seed with this node's own NNTP account (see the commented lines). `NNTP_TLS_SERVERNAME` only through a tunnel: the news server's real hostname, for the TLS certificate check. | -Leave the block below "Required by the app's configuration, but unused" as -it is: the app checks these settings at startup, but a node never uses them. +Nothing else: the database, Redis and web server settings the indexer +needs are filled in for a node automatically. ## 4. Start it @@ -109,3 +115,4 @@ articles reached, and per node its leases and headers. | `Hostname/IP does not match certificate's altnames: Host: localhost` | TLS through a tunnel: set `NNTP_TLS_SERVERNAME` to the news server's hostname | | `Range … failed: …` | the range is given back and retried later, possibly by another node | | `Range … passed to another node; dropping it.` | this node was too slow to report (15 minutes); harmless | +| `[inspect n] "…" failed: …` | an inspection failed (often a missing article); the indexer retries it later | diff --git a/seed-node/node.env.example b/seed-node/node.env.example index b836bd8..b5298b3 100644 --- a/seed-node/node.env.example +++ b/seed-node/node.env.example @@ -1,34 +1,18 @@ -# --- Seed API: the orchestrator, and this node's token (`node ace seed:token ` on the indexer). +# The seed API (the orchestrator) and this node's token (`node ace seed:token ` on the indexer). SEED_URL=https://seed.exp.sforum.nl SEED_TOKEN=sn_replace_me -# Ranges worked on at once (about 0.4 CPU core each). -SEED_CONNECTIONS=4 -# relay: headers come through the seed API, no news server account needed (the default). -# direct: the node fetches with its own NNTP account; then fill in the NNTP_* lines below. -SEED_FETCH=relay -# --- Only for SEED_FETCH=direct: the NNTP account (counts against its connection and IP limits). -NNTP_HOST=localhost -NNTP_PORT=563 -NNTP_SECURE=true -NNTP_USER=unused -NNTP_PASSWORD=unused +# Ranges seeded at once (about 0.4 CPU core each). 0: no seeding. +SEED_CONNECTIONS=4 +# Content inspections and archive walks at once (CPU for decoding and parsing; articles come through the seed API). +SEED_INSPECTORS=0 + +# Optional: SEED_FETCH=direct seeds with this node's own NNTP account instead of through the seed API. +# SEED_FETCH=direct +# NNTP_HOST=news.eweka.nl +# NNTP_PORT=563 +# NNTP_SECURE=true +# NNTP_USER= +# NNTP_PASSWORD= # Through a tunnel or proxy: the hostname the server's TLS certificate is for. # NNTP_TLS_SERVERNAME=news.eweka.nl - -# --- Required by the app's configuration, but unused on a seeding node (no database, Redis or web server). -NODE_ENV=production -TZ=UTC -PORT=3333 -HOST=0.0.0.0 -LOG_LEVEL=info -APP_KEY=unused-on-seed-nodes-unused-on-seed-nodes -APP_URL=http://localhost:3333 -SESSION_DRIVER=memory -DB_CONNECTION=pg -DB_HOST=localhost -DB_PORT=5432 -DB_USER=unused -DB_DATABASE=unused -REDIS_HOST=localhost -REDIS_PORT=6379 diff --git a/start/env.ts b/start/env.ts index 8350d3d..c3033a2 100644 --- a/start/env.ts +++ b/start/env.ts @@ -11,6 +11,38 @@ import { Env } from '@adonisjs/core/env' +/** + * A seeding node (`node ace seed:node`, CONCEPT.md D38) has no database, + * Redis or web server, and in relay mode no NNTP account, but the app's + * configuration still reads those settings: they get placeholders here, so + * a node needs only SEED_URL and SEED_TOKEN. Values that are set win (e.g. + * the NNTP account of a node with SEED_FETCH=direct). + */ +if (process.env.ACE_COMMAND === 'seed:node') { + const placeholders: Record = { + NODE_ENV: 'production', + PORT: '3333', + HOST: '0.0.0.0', + LOG_LEVEL: 'info', + APP_KEY: 'unused-on-seed-nodes-unused-on-seed-nodes', + APP_URL: 'http://localhost:3333', + SESSION_DRIVER: 'memory', + DB_CONNECTION: 'pg', + DB_HOST: 'localhost', + DB_PORT: '5432', + DB_USER: 'unused', + DB_DATABASE: 'unused', + REDIS_HOST: 'localhost', + REDIS_PORT: '6379', + NNTP_HOST: 'localhost', + NNTP_PORT: '563', + NNTP_USER: 'unused', + NNTP_PASSWORD: 'unused', + NNTP_SECURE: 'true', + } + for (const [name, value] of Object.entries(placeholders)) process.env[name] ??= value +} + export default await Env.create(new URL('../', import.meta.url), { // Node NODE_ENV: Env.schema.enum(['development', 'production', 'test'] as const), diff --git a/start/routes.ts b/start/routes.ts index b0cd4b6..7eaa633 100644 --- a/start/routes.ts +++ b/start/routes.ts @@ -59,6 +59,12 @@ router router.post('ranges/:id/done', [SeedController, 'done']).where('id', router.matchers.number()) router.post('ranges/:id/release', [SeedController, 'release']).where('id', router.matchers.number()) router.get('status', [SeedController, 'status']) + // Content inspection and archive walks on nodes (CONCEPT.md D41). + router.post('inspect/lease', [SeedController, 'inspectLease']) + router.post('inspect/done', [SeedController, 'inspectDone']) + router.post('inspect/failed', [SeedController, 'inspectFailed']) + router.post('inspect/release', [SeedController, 'inspectRelease']) + router.get('articles', [SeedController, 'article']) }) .prefix('/api/v1/seed') .use(middleware.seedNode()) diff --git a/tests/functional/seed_api.spec.ts b/tests/functional/seed_api.spec.ts index 8161d48..de6fb9e 100644 --- a/tests/functional/seed_api.spec.ts +++ b/tests/functional/seed_api.spec.ts @@ -11,6 +11,9 @@ import { segmentFromHeader } from '#lib/ingest/segment' import { FileAggregator } from '#lib/aggregate/index' import { toFragment, toSeedFile, type SeedBatch } from '#lib/seed/wire' import type { IngestSegment } from '#lib/ingest/segment' +import FileStore from '#services/FileStore' +import RemoteContentService from '#services/RemoteContentService' +import { Job } from 'bullmq' function header(subject: string, n: number) { return { 'subject': subject, 'from': 'api seeder ', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', 'message-id': ``, ':bytes': '500000' } @@ -25,7 +28,10 @@ function batch(next: number): SeedBatch { test.group('Seed API', (group) => { group.each.setup(() => testUtils.db().withGlobalTransaction()) - group.teardown(() => QueueService.closeAll()) + group.teardown(async () => { + await RemoteContentService.close() + await QueueService.closeAll() + }) test('a node leases a range, posts a gzip batch, and finishes it', async ({ client, assert }) => { const token = await SeedOrchestrator.issueToken('test-node') @@ -128,6 +134,87 @@ test.group('Seed API', (group) => { } }) + test('a node inspects a file: lease, probe result, applied here (D41)', async ({ client, assert }) => { + await QueueService.collectionQueue.drain() + await QueueService.archiveWalkQueue.drain() + const token = await SeedOrchestrator.issueToken('inspect-node') + const auth = { authorization: `Bearer ${token}` } + const now = new Date() + const [release] = await db + .table('releases') + .insert({ release_key: 'i'.repeat(40), name: 'Xq8r2KdLm3Zw91PqA7zz', source: 'stem', groups: '[]', first_seen_at: now, last_seen_at: now }) + .returning('id') + const [file] = await FileStore.insert([ + { + filename: 'Xq8r2KdLm3Zw91PqA7zz.part01.rar', + file_key: 'inspect-node-file'.padEnd(40, '0'), + poster: 'x', + date: now.getTime(), + parts: 2, + groups: '[]', + segments: [ + { number: 1, id: 'one@test', size: 100 }, + { number: 2, id: 'two@test', size: 100 }, + ], + release_id: release.id, + created_at: now, + updated_at: now, + }, + ]) + await QueueService.collectionQueue.add('process-collection', { fileId: file.id }, { jobId: `node-test-${file.id}-${Date.now()}` }) + + const leased = await client.post('/api/v1/seed/inspect/lease').json({ max: 5 }).headers(auth) + leased.assertStatus(200) + const job = leased.body().jobs.find((j: any) => j.task.kind === 'inspect' && j.task.fileId === file.id) + assert.deepEqual(job.task.segments.map((s: any) => s.id), ['one@test', 'two@test']) + // Locked: nobody else gets it meanwhile. + ;(await client.post('/api/v1/seed/inspect/lease').json({ max: 5 }).headers(auth)).assertStatus(204) + + // What probeInspection found in the first segment; posted as a JSON string, empty strings and all. + const result = { + size: 15_728_640, + contentType: null, + md5_16k: null, + yencName: '', + archive: { + format: 'rar5', volume: true, firstVolume: true, solid: false, headersEncrypted: false, truncated: true, next: 120, + entries: [{ name: 'Real.Release.Name.2026.1080p-GRP/real.release.name.mkv', size: 5_000_000_000, packedSize: 15_700_000, directory: false, encrypted: true, splitBefore: false, splitAfter: true, stored: false, dataOffset: 100 }], + }, + } + const done = { queue: job.queue, jobId: job.jobId, token: job.token, result: JSON.stringify(result) } + ;(await client.post('/api/v1/seed/inspect/done').json(done).headers(auth)).assertStatus(200) + + const row = await db.from('releases').where('id', release.id).first() + assert.equal(row.name, 'Real.Release.Name.2026.1080p-GRP') + assert.isTrue(row.encrypted) + assert.lengthOf(row.contents, 1) + assert.isOk((await db.from('files').where('id', file.id).first()).archive) + assert.equal(await (await Job.fromId(QueueService.collectionQueue, job.jobId))!.getState(), 'completed') + // Too late (or twice): the lock is gone. + ;(await client.post('/api/v1/seed/inspect/done').json(done).headers(auth)).assertStatus(409) + }) + + test('a node gives an inspection job back, or reports it failed', async ({ client, assert }) => { + await QueueService.collectionQueue.drain() + await QueueService.archiveWalkQueue.drain() + const token = await SeedOrchestrator.issueToken('inspect-node-2') + const auth = { authorization: `Bearer ${token}` } + const now = new Date() + const [file] = await FileStore.insert([ + { filename: 'Zk3m9QpL2xW8rT5vN1bY.bin', file_key: 'inspect-node-file-2'.padEnd(40, '0'), poster: 'x', date: now.getTime(), parts: 1, groups: '[]', segments: [{ number: 1, id: 'x@test', size: 1 }], created_at: now, updated_at: now }, + ]) + await QueueService.collectionQueue.add('process-collection', { fileId: file.id }, { jobId: `node-test-${file.id}-${Date.now()}` }) + const lease = async () => (await client.post('/api/v1/seed/inspect/lease').json({ max: 1 }).headers(auth)).body().jobs[0] + + const first = await lease() + ;(await client.post('/api/v1/seed/inspect/release').json({ queue: first.queue, jobId: first.jobId, token: first.token }).headers(auth)).assertStatus(200) + const again = await lease() + assert.equal(again.jobId, first.jobId) + ;(await client.post('/api/v1/seed/inspect/failed').json({ queue: again.queue, jobId: again.jobId, token: again.token, error: '430 no such article' }).headers(auth)).assertStatus(200) + // Retried later (backoff), like a local failure. + assert.include(['delayed', 'waiting'], await (await Job.fromId(QueueService.collectionQueue, again.jobId))!.getState()) + }) + test('a range reported to its end but never finished is marked done, not leased', async ({ client, assert }) => { const token = await SeedOrchestrator.issueToken('late-node') const [range] = await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 101 }).returning('id') diff --git a/tests/unit/probe.spec.ts b/tests/unit/probe.spec.ts new file mode 100644 index 0000000..c8752a6 --- /dev/null +++ b/tests/unit/probe.spec.ts @@ -0,0 +1,78 @@ +import { test } from '@japa/runner' +import { crc32Hex } from '#lib/yenc/index' +import { probeInspection, type InspectTask } from '#lib/content/probe' + +/** One yEnc article of `file`: part `index` of `size` bytes (escaping '.' and whitespace too, so no dot-stuffing). */ +function yencArticle(name: string, file: Buffer, index: number, size: number) { + const data = file.subarray(index * size, (index + 1) * size) + const total = Math.ceil(file.length / size) + const out: number[] = [] + for (const byte of data) { + const c = (byte + 42) & 0xff + if ([0x00, 0x0a, 0x0d, 0x3d, 0x2e, 0x09, 0x20].includes(c)) out.push(0x3d, (c + 64) & 0xff) + else out.push(c) + } + const head = `=ybegin part=${index + 1} total=${total} line=128 size=${file.length} name=${name}\r\n=ypart begin=${index * size + 1} end=${index * size + data.length}\r\n` + const tail = `\r\n=yend size=${data.length} part=${index + 1} pcrc32=${crc32Hex(data)}\r\n` + return Buffer.from(head + Buffer.from(out).toString('latin1') + tail, 'latin1') +} + +/** A posted file as articles, and its inspection task. */ +function posted(name: string, file: Buffer, size = 400) { + const articles = new Map() + const segments = [] + for (let i = 0; i * size < file.length; i++) { + const id = `${name}-${i}@test` + articles.set(id, yencArticle(name, file, i, size)) + segments.push({ number: i + 1, id, size: size + 100 }) + } + const fetched: string[] = [] + const fetch = async (id: string) => { + fetched.push(id) + const body = articles.get(id) + if (!body) throw new Error(`430 no such article ${id}`) + return body + } + const task: InspectTask = { kind: 'inspect', fileId: 1, name, segments, sevenZip: { volumes: 1, last: segments } } + return { task, fetch, fetched } +} + +// `7z a -pSecret -mhe=on`: an encrypted header (tests/unit/sevenzip.spec.ts). +const HEADER_ENCRYPTED = Buffer.from( + 'N3q8ryccAARtHJcDoAAAAAAAAAA+AAAAAAAAAGqzNxflXzLifhRjmRlgXvMEO9zhcADrwDAjtLT4COwZQUsqMkFT0jGipxcTP5MNbocZW7iGLP5PCAuqfouDxAoI2Zw2AqMRF2m/XOVYRokWM4bHuH73xp1LIva5JpyAQSViuI66Jf5xxemaGnSvoC5/mHDHe6VZUxQwMmH3QuTR15+loSuNhbEU2feP1E2EdwIvHSYXdJ9jDqHe6XjW3gfh/2+NFwYQAQmAkAAHCwEAAiQG8QcBElMPIu8lRTeoM0YvSGcM+B6l/SMDAQEFXQAQAAABAAyAgoCmCgGDcQMzAAA=', + 'base64' +) + +test.group('content probe', () => { + test('a first segment gives size, fingerprint and the yEnc name', async ({ assert }) => { + const file = Buffer.alloc(20_000, 7) + const { task, fetch, fetched } = posted('Some.Name.bin', file, 17_000) + const probe = await probeInspection(task, fetch) + assert.include(probe, { size: 20_000, contentType: null, yencName: 'Some.Name.bin' }) + assert.match(probe.md5_16k!, /^[0-9a-f]{32}$/) + assert.deepEqual(fetched, ['Some.Name.bin-0@test']) + }) + + test('an NZB is read whole', async ({ assert }) => { + const xml = `\n${''.repeat(60)}` + const { task, fetch, fetched } = posted('x.nzb', Buffer.from(xml)) + const probe = await probeInspection(task, fetch) + assert.equal(probe.contentType, 'nzb') + assert.equal(probe.text, xml) + assert.lengthOf(fetched, task.segments.length + 1) // the first segment twice: once to sniff, once in the whole + }) + + test('a 7z archive with encrypted headers is known from its tail, without bsdtar', async ({ assert }) => { + const { task, fetch } = posted('a.7z', HEADER_ENCRYPTED, 100) + const probe = await probeInspection(task, fetch) + assert.equal(probe.contentType, '7z') + assert.include(probe.archive!, { format: '7z', headersEncrypted: true }) + assert.lengthOf(probe.archive!.entries, 0) + }) + + test('a missing article fails the probe (the job is retried)', async ({ assert }) => { + const { task, fetch } = posted('gone.bin', Buffer.alloc(10, 1)) + task.segments[0].id = 'missing@test' + await assert.rejects(() => probeInspection(task, fetch), /430/) + }) +}) diff --git a/tests/unit/rar.spec.ts b/tests/unit/rar.spec.ts index 92daa13..b6c6d8a 100644 --- a/tests/unit/rar.spec.ts +++ b/tests/unit/rar.spec.ts @@ -325,7 +325,7 @@ test.group('rar / archive walks', () => { }) test('finds segments from their yEnc offsets', async ({ assert }) => { - const { SegmentReader } = await import('#services/ArchiveWalkService') + const { SegmentReader } = await import('#lib/rar/index') const set = rar5Set() const size = 700 const articles = new Map() @@ -336,7 +336,7 @@ test.group('rar / archive walks', () => { articles.set(id, yencArticle(`x.part${v + 1}.rar`, volume, i, size)) segments.push({ number: i + 1, id, size: 0 }) } - return { id: v + 1, segments } + return segments }) const reader = new SegmentReader(files, 100, async (id) => articles.get(id)!) const walk = await walkRar(reader)