From f4ded062dfab65f3e2f19f909aab589c08a1c3a7 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 5 Oct 2026 23:32:52 +0200 Subject: [PATCH] Complete archive listings by walking RAR headers, at low priority Inspection lists only the entries whose headers are in the first segment of a first volume. A walk hops from header to header, fetching one segment per entry and volume, in its own queue (one job per collection worker) on NNTP connections inspection isn't waiting for. content:walk-archives queues releases inspected earlier (CONCEPT D40). Co-Authored-By: Claude Opus 5.5 --- app/services/ArchiveService.ts | 3 +- app/services/ArchiveWalkService.ts | 213 ++++++++++++++++++ app/services/ContentService.ts | 3 + app/services/NntpService.ts | 16 +- app/services/QueueService.ts | 6 +- app/services/SevenZipService.ts | 1 + commands/ArchiveWalk.ts | 70 ++++++ commands/workers/Collection.ts | 21 +- .../1768620765170_release_contents_walked.ts | 14 ++ database/schema.ts | 4 +- lib/content/archive_names.ts | 15 ++ lib/content/index.ts | 2 +- lib/rar/index.ts | 1 + lib/rar/parse.ts | 31 ++- lib/rar/walk.ts | 77 +++++++ start/env.ts | 2 + tests/unit/rar.spec.ts | 135 ++++++++++- 17 files changed, 597 insertions(+), 17 deletions(-) create mode 100644 app/services/ArchiveWalkService.ts create mode 100644 commands/ArchiveWalk.ts create mode 100644 database/migrations/1768620765170_release_contents_walked.ts create mode 100644 lib/rar/walk.ts diff --git a/app/services/ArchiveService.ts b/app/services/ArchiveService.ts index 04a5726..613e30d 100644 --- a/app/services/ArchiveService.ts +++ b/app/services/ArchiveService.ts @@ -27,7 +27,8 @@ export interface ArchiveResult { */ class ArchiveService { async record(file: any, binary: any, info: RarInfo): Promise { - const stored = { ...info, entries: info.entries.slice(0, MAX_ENTRIES).map(({ stored: _s, dataOffset: _d, ...entry }) => entry) } + const { next: _next, ...listing } = info + const stored = { ...listing, entries: info.entries.slice(0, MAX_ENTRIES).map(({ stored: _s, dataOffset: _d, ...entry }) => entry) } await db.from('files').where('id', file.id).update({ archive: JSON.stringify(stored) }) let releaseId = file.release_id ?? binary?.release_id ?? null diff --git a/app/services/ArchiveWalkService.ts b/app/services/ArchiveWalkService.ts new file mode 100644 index 0000000..29becf2 --- /dev/null +++ b/app/services/ArchiveWalkService.ts @@ -0,0 +1,213 @@ +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' + +/** Stored per release, as for a first-volume listing (ArchiveService). */ +const MAX_ENTRIES = 500 + +/** Listed sizes below this share of the release's size: the first segment listed only part of it. */ +const COVERED = 0.7 + +/** Inspection queues a walk this long after the first volume, when the release has had time to complete. */ +const WALK_DELAY_MS = 30 * 60 * 1000 + +export interface WalkOutcome { + releaseId: number + /** Nothing downloaded: why. */ + skipped?: 'missing' | 'covered' | 'volumes' + entries?: number + before?: number + complete?: boolean + stopped?: string + segments?: number + renamed?: string +} + +/** + * Complete archive listings by walking RAR headers (CONCEPT.md D40). The + * first segment of a first volume lists the entries that start in it: for + * a set of many small files (an audiobook's MP3s, a music album) that is + * the first one or two. The walk reads one segment per further entry and + * volume, at low priority: its own queue, one job at a time per collection + * worker, and NNTP connections only when inspection isn't waiting for one. + */ +class ArchiveWalkService { + /** Segments one walk may download (~0.7 MB each). */ + readonly maxSegments = env.get('ARCHIVE_WALK_MAX_SEGMENTS', 400) + + /** A listing that leaves most of the release unaccounted for. */ + worthWalking(entries: number, listedBytes: number, releaseSize: number): boolean { + if (entries === 0 || entries >= MAX_ENTRIES) return false + return listedBytes < COVERED * releaseSize + } + + /** After inspection listed a first RAR volume whose listing ends inside the set. */ + async queueAfterInspection(releaseId: number | null, info: RarInfo) { + if (!releaseId || info.format === '7z' || info.firstVolume === false) return + if (!info.truncated || info.headersEncrypted || info.entries.length === 0) return + 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' } + 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' }) + + const volumes = await this.volumes(releaseId) + if (!volumes) return this.done(releaseId, { releaseId, skipped: 'volumes' }) + + const reader = new SegmentReader(volumes, this.maxSegments) + const walked = await walkRar(reader) + const outcome: WalkOutcome = { + releaseId, + before: before.length, + entries: walked.entries.length, + complete: walked.complete, + stopped: walked.stopped, + segments: reader.fetched, + } + if (walked.entries.length > before.length) { + const contents = walked.entries.slice(0, MAX_ENTRIES).map(({ stored: _s, dataOffset: _d, ...entry }) => entry) + 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 }) + // 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 + outcome.renamed = better + } + await db.from('releases').where('id', releaseId).update(update) + } + return this.done(releaseId, outcome) + } + + private async done(releaseId: number, outcome: WalkOutcome) { + await db.from('releases').where('id', releaseId).update({ contents_walked_at: new Date() }) + return outcome + } + + /** + * The release's RAR volumes in order, from volume 1 up to the first gap. + * Null without a first volume (e.g. volumes with random names). + */ + private async volumes(releaseId: number): Promise { + const files = await db.from('files').where('release_id', releaseId).select('id', 'filename', 'real_name', 'segments') + const sets = new Map>() + for (const file of files) { + const volume = rarVolume(String(file.real_name ?? file.filename)) + if (!volume || segmentsOf(file).length === 0) continue + const set = sets.get(volume.base) ?? new Map() + if (!set.has(volume.number)) set.set(volume.number, file) + sets.set(volume.base, set) + } + // A release holds one set, now and then a second (a sample, subs): the biggest. + const set = [...sets.values()].sort((a, b) => b.size - a.size)[0] + if (!set?.has(1)) return null + const ordered: any[] = [] + for (let n = 1; set.has(n); n++) ordered.push(set.get(n)) + return ordered + } +} + +/** + * 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) + let failed = false + try { + return await NntpService.timed(NntpService.fetchBody(conn, messageId), 'BODY') + } catch (error) { + failed = true + throw error + } finally { + NntpService.release(conn, failed) + } +} + +export default new ArchiveWalkService() diff --git a/app/services/ContentService.ts b/app/services/ContentService.ts index 8ae264b..496b415 100644 --- a/app/services/ContentService.ts +++ b/app/services/ContentService.ts @@ -25,6 +25,7 @@ import { looksLikeFilename } from '#lib/subject/index' import { sevenZipVolume } from '#lib/sevenzip/index' import SevenZipService from '#services/SevenZipService' import SfvService from '#services/SfvService' +import ArchiveWalkService from '#services/ArchiveWalkService' import { segmentsOf } from '#services/FileStore' const FIRST_CHUNK = 16 * 1024 @@ -259,6 +260,8 @@ class ContentService { 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) } } diff --git a/app/services/NntpService.ts b/app/services/NntpService.ts index 20caeb3..a3413df 100644 --- a/app/services/NntpService.ts +++ b/app/services/NntpService.ts @@ -7,6 +7,8 @@ class NntpService { private allConnections: Set; private idleConnections: any[]; private waiters: ((conn: any) => void)[]; + /** Background work (archive walks): served only when no one else waits. */ + private lowWaiters: ((conn: any) => void)[] = []; private createdCount: number; constructor(poolSize = 10) { @@ -34,7 +36,8 @@ class NntpService { return conn; } - public async acquire() { + /** A connection from the pool; with `low`, after every waiting normal caller. */ + public async acquire(low = false) { if (this.idleConnections.length > 0) { return this.idleConnections.pop(); } @@ -45,7 +48,7 @@ class NntpService { return this._createConnection(); } - return new Promise(resolve => this.waiters.push(resolve)); + return new Promise(resolve => (low ? this.lowWaiters : this.waiters).push(resolve)); } /** @@ -58,7 +61,7 @@ class NntpService { this.discard(conn); return; } - const resolve = this.waiters.shift(); + const resolve = this.nextWaiter(); if (resolve) { resolve(conn); } else { @@ -73,13 +76,17 @@ class NntpService { this._destroy(conn); // Someone is waiting for a slot: open a fresh connection for them. - const resolve = this.waiters.shift(); + const resolve = this.nextWaiter(); if (resolve) { this.createdCount++; this._createConnection().then(resolve, () => this.waiters.unshift(resolve)); } } + private nextWaiter() { + return this.waiters.shift() ?? this.lowWaiters.shift(); + } + private _destroy(conn: any) { try { conn.sock?.destroy(); @@ -126,6 +133,7 @@ class NntpService { this.allConnections.clear(); this.idleConnections.length = 0; this.waiters.length = 0; + this.lowWaiters.length = 0; this.createdCount = 0; } } diff --git a/app/services/QueueService.ts b/app/services/QueueService.ts index 2adf00c..2f2547d 100644 --- a/app/services/QueueService.ts +++ b/app/services/QueueService.ts @@ -41,6 +41,10 @@ class QueueService { get collectionQueue() { return this.queue('collection-queue') } + /** Complete archive listings, at low priority (ArchiveWalkService). */ + get archiveWalkQueue() { + return this.queue('archive-walk-queue') + } /** Queues no longer used by the pipeline, kept here so GC can empty them. */ get legacy(): Queue[] { @@ -48,7 +52,7 @@ class QueueService { } get all(): Queue[] { - return [this.nntpFetchQueue, this.fileQueue, this.bodyQueue, this.collectionQueue] + return [this.nntpFetchQueue, this.fileQueue, this.bodyQueue, this.collectionQueue, this.archiveWalkQueue] } /** Closes the queues that were opened. */ diff --git a/app/services/SevenZipService.ts b/app/services/SevenZipService.ts index 15bc0cb..9c6cc61 100644 --- a/app/services/SevenZipService.ts +++ b/app/services/SevenZipService.ts @@ -120,6 +120,7 @@ class SevenZipService { solid: false, headersEncrypted, truncated: false, + next: 0, entries: entries.map((e) => ({ name: e.name, size: e.size, diff --git a/commands/ArchiveWalk.ts b/commands/ArchiveWalk.ts new file mode 100644 index 0000000..bddcd7b --- /dev/null +++ b/commands/ArchiveWalk.ts @@ -0,0 +1,70 @@ +import { BaseCommand, args, flags } from '@adonisjs/core/ace' +import { CommandOptions } from '@adonisjs/core/types/ace' +import db from '@adonisjs/lucid/services/db' +import QueueService from '#services/QueueService' +import NntpService from '#services/NntpService' +import ArchiveWalkService from '#services/ArchiveWalkService' + +const BATCH = 5000 + +/** + * Complete archive listings for releases inspected before archive walks + * existed (CONCEPT.md D40): queues those whose listing leaves most of the + * release unaccounted for. worker:collection walks them, one at a time and + * behind inspection. With release ids, walks those right here instead. + */ +export default class ArchiveWalk extends BaseCommand { + public static commandName = 'content:walk-archives' + public static description = 'Queues releases with partial archive listings for a (low priority) header walk.' + + public static options: CommandOptions = { + startApp: true, + } + + @args.spread({ description: 'Release ids to walk now (instead of queueing)', required: false }) + declare releaseIds?: string[] + + @flags.number({ description: 'Queue at most this many releases' }) + declare limit?: number + + @flags.boolean({ description: 'Also releases that were walked before' }) + declare again: boolean + + public async run() { + try { + if (this.releaseIds?.length) { + for (const id of this.releaseIds) { + this.logger.info(JSON.stringify(await ArchiveWalkService.walk(Number(id)))) + } + return + } + let lastId = 0 + let queued = 0 + while (!this.limit || queued < this.limit) { + const query = db + .from('releases as r') + .where('r.id', '>', lastId) + .whereNotNull('r.contents') + .orderBy('r.id') + .limit(BATCH) + .select('r.id', 'r.size', 'r.contents_walked_at', db.raw(`( + SELECT count(*) FROM jsonb_array_elements(r.contents)) AS n`), db.raw(`( + SELECT coalesce(sum((e->>'size')::bigint), 0) FROM jsonb_array_elements(r.contents) e) AS listed`)) + const rows = await query + if (rows.length === 0) break + lastId = Number(rows.at(-1).id) + const jobs = rows + .filter((r) => this.again || !r.contents_walked_at) + .filter((r) => ArchiveWalkService.worthWalking(Number(r.n), Number(r.listed), Number(r.size))) + .slice(0, this.limit ? this.limit - queued : undefined) + .map((r) => ({ name: 'walk', data: { releaseId: Number(r.id) }, opts: { jobId: `walk-${r.id}-${Date.now()}` } })) + if (jobs.length > 0) await QueueService.archiveWalkQueue.addBulk(jobs) + queued += jobs.length + } + this.logger.info(`Queued ${queued} releases for an archive walk.`) + } finally { + await QueueService.closeAll() + await NntpService.shutdown() + } + } +} diff --git a/commands/workers/Collection.ts b/commands/workers/Collection.ts index 08c96a8..9b42579 100644 --- a/commands/workers/Collection.ts +++ b/commands/workers/Collection.ts @@ -7,6 +7,7 @@ import Shutdown from '#services/Shutdown' import QueueService from '#services/QueueService' import NntpService from '#services/NntpService' import ContentService from '#services/ContentService' +import ArchiveWalkService from '#services/ArchiveWalkService' import File from '#models/file' export default class CollectionWorker extends BaseCommand { @@ -70,10 +71,28 @@ export default class CollectionWorker extends BaseCommand { this.logger.error(`Collection job ${job?.id} failed: ${err.message}`) }) + // Complete archive listings (D40): background work beside inspection, on the same NNTP pool. + const walkConcurrency = env.get('ARCHIVE_WALK_CONCURRENCY', 1) + const walker = walkConcurrency > 0 + ? new Worker('archive-walk-queue', async (job) => { + const o = await ArchiveWalkService.walk(Number(job.data.releaseId)) + if (o.skipped) return o + this.logger.info( + `Archive walk of release ${o.releaseId}: ${o.before} → ${o.entries} entries from ${o.segments} segments` + + (o.complete ? '' : ` (stopped: ${o.stopped})`) + + (o.renamed ? `, renamed to "${o.renamed}"` : '') + ) + return o + }, { connection: queueConfig.connection, concurrency: walkConcurrency }) + : null + walker?.on('failed', (job, err) => { + this.logger.error(`Archive walk job ${job?.id} failed: ${err.message}`) + }) + this.logger.info('Collection worker started and listening for jobs.') await Shutdown.wait() this.logger.info('Shutting down: finishing active jobs...') - await worker.close() + await Promise.all([worker.close(), walker?.close()]) await NntpService.shutdown() await QueueService.closeAll() } diff --git a/database/migrations/1768620765170_release_contents_walked.ts b/database/migrations/1768620765170_release_contents_walked.ts new file mode 100644 index 0000000..8d2d86c --- /dev/null +++ b/database/migrations/1768620765170_release_contents_walked.ts @@ -0,0 +1,14 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** When an archive walk (CONCEPT.md D40) looked at a release, so it isn't walked again. */ +export default class extends BaseSchema { + async up() { + this.schema.alterTable('releases', (table) => { + table.timestamp('contents_walked_at', { useTz: true }) + }) + } + + async down() { + this.schema.alterTable('releases', (table) => table.dropColumn('contents_walked_at')) + } +} diff --git a/database/schema.ts b/database/schema.ts index e16a281..f196bfb 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -411,7 +411,7 @@ export class ReleaseFileSchema extends BaseModel { } export class ReleaseSchema extends BaseModel { - static $columns = ['category', 'complete', 'container', 'contents', 'encrypted', 'extensions', 'fileCount', 'fileTotal', 'firstSeenAt', 'groups', 'id', 'lastSeenAt', 'name', 'password', 'passwordSource', 'postedAt', 'poster', 'releaseKey', 'searchFiles', 'searchText', 'size', 'source'] as const + static $columns = ['category', 'complete', 'container', 'contents', 'contentsWalkedAt', 'encrypted', 'extensions', 'fileCount', 'fileTotal', 'firstSeenAt', 'groups', 'id', 'lastSeenAt', 'name', 'password', 'passwordSource', 'postedAt', 'poster', 'releaseKey', 'searchFiles', 'searchText', 'size', 'source'] as const $columns = ReleaseSchema.$columns @column() declare category: number | null @@ -421,6 +421,8 @@ export class ReleaseSchema extends BaseModel { declare container: string | null @column() declare contents: any | null + @column.dateTime() + declare contentsWalkedAt: DateTime | null @column() declare encrypted: boolean | null @column() diff --git a/lib/content/archive_names.ts b/lib/content/archive_names.ts index 2727ef9..dfec2be 100644 --- a/lib/content/archive_names.ts +++ b/lib/content/archive_names.ts @@ -23,3 +23,18 @@ export function isLaterVolume(name: string): boolean { if (part) return Number(part[1]) > 1 return /\.r\d{2,3}$/i.test(name) || /\.vol\d+[+-]\d+\.par2$/i.test(name) } + +/** + * The set and 1-based number of a RAR volume: `x.part03.rar` is volume 3 of + * `x`, and in the old naming `x.rar` is 1, `x.r00` 2, `x.r01` 3. Null for + * other names. + */ +export function rarVolume(name: string): { base: string; number: number } | null { + const part = /^(.*)\.part(\d+)\.rar$/i.exec(name) + if (part) return { base: part[1].toLowerCase(), number: Number(part[2]) } + const old = /^(.*)\.r(\d{2,3})$/i.exec(name) + if (old) return { base: old[1].toLowerCase(), number: Number(old[2]) + 2 } + const first = /^(.*)\.rar$/i.exec(name) + if (first) return { base: first[1].toLowerCase(), number: 1 } + return null +} diff --git a/lib/content/index.ts b/lib/content/index.ts index adbb5e7..090dfd7 100644 --- a/lib/content/index.ts +++ b/lib/content/index.ts @@ -2,6 +2,6 @@ * Content recognition by magic bytes, independent of AdonisJS. */ export { sniffContentType, type ContentType } from './sniff.js' -export { isArchiveStart, isLaterVolume } from './archive_names.js' +export { isArchiveStart, isLaterVolume, rarVolume } from './archive_names.js' export { parseSfv, sfvReleaseName, type SfvEntry } from './sfv.js' export { NICOVIDEO_FILE, NICOVIDEO_RELEASE, nicovideoInfo, nicovideoReleaseName, zstdPrefix, tarEntries, type NicovideoInfo } from './nicovideo.js' diff --git a/lib/rar/index.ts b/lib/rar/index.ts index 5faa2e6..b9d3527 100644 --- a/lib/rar/index.ts +++ b/lib/rar/index.ts @@ -3,3 +3,4 @@ * contents from the first segment of a first volume. See CONCEPT.md (D19). */ export { parseRarHeaders, rarFormat, decodeRar4Name, storedEntryData, type RarInfo, type RarEntry } from './parse.js' +export { walkRar, type VolumeReader, type RarWalk } from './walk.js' diff --git a/lib/rar/parse.ts b/lib/rar/parse.ts index 48f56d6..915e6bc 100644 --- a/lib/rar/parse.ts +++ b/lib/rar/parse.ts @@ -31,6 +31,12 @@ export interface RarInfo { entries: RarEntry[] /** The buffer ended before the end-of-archive header (normal for one segment). */ truncated: boolean + /** + * Offset of the first block that was not read: where the next header + * starts (possibly beyond the buffer, after an entry's data), or the + * header the buffer cut off. After the end-of-archive header, its end. + */ + next: number } const RAR4_MARKER = Buffer.from([0x52, 0x61, 0x72, 0x21, 0x1a, 0x07, 0x00]) @@ -45,11 +51,16 @@ export function rarFormat(data: Buffer): 'rar4' | 'rar5' | null { /** * Lists a RAR archive from its headers only, so the first segment of a first * volume is enough. Returns null when the data is not a RAR archive. + * + * With `continueAs`, `data` starts at a header in the middle of a volume (no + * signature), e.g. where a previous parse said the next header is. */ -export function parseRarHeaders(data: Buffer): RarInfo | null { +export function parseRarHeaders(data: Buffer, continueAs?: 'rar4' | 'rar5'): RarInfo | null { + if (continueAs === 'rar4') return parseRar4(data, 0) + if (continueAs === 'rar5') return parseRar5(data, 0) const format = rarFormat(data) - if (format === 'rar4') return parseRar4(data) - if (format === 'rar5') return parseRar5(data) + if (format === 'rar4') return parseRar4(data, RAR4_MARKER.length) + if (format === 'rar5') return parseRar5(data, RAR5_MARKER.length) return null } @@ -60,7 +71,7 @@ const R4_FILE = 0x74 const R4_END = 0x7b const R4_LONG_BLOCK = 0x8000 -function parseRar4(data: Buffer): RarInfo { +function parseRar4(data: Buffer, start: number): RarInfo { const info: RarInfo = { format: 'rar4', volume: false, @@ -69,8 +80,9 @@ function parseRar4(data: Buffer): RarInfo { headersEncrypted: false, entries: [], truncated: true, + next: start, } - let p = RAR4_MARKER.length + let p = start while (p + 7 <= data.length) { const headCrc = data.readUInt16LE(p) @@ -81,6 +93,7 @@ function parseRar4(data: Buffer): RarInfo { if (type === R4_END) { info.truncated = false + p += size break } @@ -130,6 +143,7 @@ function parseRar4(data: Buffer): RarInfo { if (info.volume && !info.firstVolume && info.entries.length > 0 && !info.entries[0].splitBefore) { info.firstVolume = true } + info.next = p return info } @@ -205,7 +219,7 @@ function vint(data: Buffer, offset: number): { value: number; next: number } | n return null } -function parseRar5(data: Buffer): RarInfo { +function parseRar5(data: Buffer, start: number): RarInfo { const info: RarInfo = { format: 'rar5', volume: false, @@ -214,8 +228,9 @@ function parseRar5(data: Buffer): RarInfo { headersEncrypted: false, entries: [], truncated: true, + next: start, } - let p = RAR5_MARKER.length + let p = start while (p + 5 <= data.length) { const headCrc = data.readUInt32LE(p) @@ -246,6 +261,7 @@ function parseRar5(data: Buffer): RarInfo { if (type.value === R5_END) { info.truncated = false + p = headerEnd break } if (type.value === R5_ENCRYPTION) { @@ -270,6 +286,7 @@ function parseRar5(data: Buffer): RarInfo { if (next <= p) break p = next } + info.next = p return info } diff --git a/lib/rar/walk.ts b/lib/rar/walk.ts new file mode 100644 index 0000000..d5d17a7 --- /dev/null +++ b/lib/rar/walk.ts @@ -0,0 +1,77 @@ +import { parseRarHeaders, rarFormat, type RarEntry, type RarInfo } from './parse.js' + +/** Reads the volumes of a RAR set, a segment at a time. */ +export interface VolumeReader { + readonly volumes: number + /** + * Bytes of volume `v` from `offset` to the end of the segment that holds + * it; with `more`, the following segment too (for a header cut off by the + * segment end). Null when that can't be read (missing article, budget). + */ + read(v: number, offset: number, more: boolean): Promise + /** Size of volume `v`, known once something of it was read. */ + size(v: number): number | null +} + +export interface RarWalk { + format: 'rar4' | 'rar5' | null + /** One entry per name, from its first header. */ + entries: RarEntry[] + /** Every volume was walked to its end. */ + complete: boolean + headersEncrypted: boolean + /** Why the walk ended early. */ + stopped?: 'unreadable' | 'invalid' | 'encrypted' +} + +/** + * Lists a RAR set completely by hopping from header to header (CONCEPT.md + * D40). Each header says how much data follows it, so the next header's + * offset is known and only the segment holding it is read: one segment per + * entry and one per volume, instead of the whole set. Split files repeat + * their header at the start of every volume they continue in. + */ +export async function walkRar(reader: VolumeReader): Promise { + const seen = new Map() + const result = (stopped?: RarWalk['stopped']): RarWalk => ({ + format, + entries: [...seen.values()], + complete: !stopped, + headersEncrypted: stopped === 'encrypted', + stopped, + }) + let format: 'rar4' | 'rar5' | null = null + let v = 0 + let offset = 0 + + while (v < reader.volumes) { + let info: RarInfo | null = null + let reached = -1 + for (const more of [false, true]) { + const data = await reader.read(v, offset, more) + if (!data) return result('unreadable') + if (offset === 0) format = rarFormat(data) ?? format + info = offset === 0 ? parseRarHeaders(data) : format && parseRarHeaders(data, format) + if (!info) return result('invalid') + // Stopped inside the data: a header the segment end cut off (read on), or a bad one. + if (!info.truncated || info.next >= data.length || info.headersEncrypted) break + // Read on and got further: the next header is cut off at the end of the following segment. + if (more && info.next > reached) break + if (more) return result('invalid') + reached = info.next + } + if (info!.headersEncrypted) return result('encrypted') + for (const entry of info!.entries) if (!seen.has(entry.name)) seen.set(entry.name, entry) + + const next = offset + info!.next + const size = reader.size(v) + // The end-of-archive header closes every volume; data running to the end does too. + if (!info!.truncated || size === null || next >= size) { + v++ + offset = 0 + } else { + offset = next + } + } + return result() +} diff --git a/start/env.ts b/start/env.ts index bfc4852..8350d3d 100644 --- a/start/env.ts +++ b/start/env.ts @@ -69,6 +69,8 @@ export default await Env.create(new URL('../', import.meta.url), { FETCH_CONCURRENCY: Env.schema.number.optional(), BODY_CONCURRENCY: Env.schema.number.optional(), COLLECTION_CONCURRENCY: Env.schema.number.optional(), + ARCHIVE_WALK_CONCURRENCY: Env.schema.number.optional(), + ARCHIVE_WALK_MAX_SEGMENTS: Env.schema.number.optional(), DEFERRED_SETTLE_MS: Env.schema.number.optional(), DEFERRED_EXPAND_MAX_PARTS: Env.schema.number.optional(), DEFERRED_CONCURRENCY: Env.schema.number.optional(), diff --git a/tests/unit/rar.spec.ts b/tests/unit/rar.spec.ts index 442fc5d..1260cfb 100644 --- a/tests/unit/rar.spec.ts +++ b/tests/unit/rar.spec.ts @@ -1,6 +1,7 @@ import { test } from '@japa/runner' import { crc32 } from 'node:zlib' -import { decodeRar4Name, parseRarHeaders, rarFormat, storedEntryData, type RarEntry } from '#lib/rar/index' +import { decodeRar4Name, parseRarHeaders, rarFormat, storedEntryData, walkRar, type RarEntry, type VolumeReader } from '#lib/rar/index' +import { crc32Hex } from '#lib/yenc/index' /** Entries without the extraction fields, for comparing listings. */ const listing = (entries: RarEntry[]) => entries.map(({ stored, dataOffset, ...rest }) => rest) @@ -228,3 +229,135 @@ test.group('rar / looksRandom capitals', () => { assert.isFalse(looksRandom('TheQuickBrownFoxJumps')) }) }) + +// --- Archive walks (D40) -------------------------------------------------------- + +/** Volumes split into segments of `segmentSize` bytes; counts the segments read. */ +function memoryReader(volumes: Buffer[], segmentSize: number) { + const read = new Set() + const reader: VolumeReader & { segments: Set } = { + segments: read, + volumes: volumes.length, + size: (v) => volumes[v].length, + async read(v, offset, more) { + const index = Math.floor(offset / segmentSize) + if (offset >= volumes[v].length) return null + read.add(`${v}:${index}`) + if (more) read.add(`${v}:${index + 1}`) + return volumes[v].subarray(offset, (index + (more ? 2 : 1)) * segmentSize) + }, + } + return reader +} + +function rar5Set() { + const main = (n: number) => rar5Block(1, 0, n === 0 ? vint(0x0001) : Buffer.concat([vint(0x0003), vint(n)])) + const end = rar5Block(5, 0, vint(0x0001)) + const SPLIT_BEFORE = 0x0008 + const SPLIT_AFTER = 0x0010 + return [ + Buffer.concat([RAR5, main(0), rar5File('Book/01.mp3', 3000, { packed: 3000 }), rar5File('Book/02.mp3', 7000, { packed: 5000, headerFlags: SPLIT_AFTER }), end]), + Buffer.concat([ + RAR5, + main(1), + rar5File('Book/02.mp3', 7000, { packed: 2000, headerFlags: SPLIT_BEFORE }), + rar5File('Book/03.mp3', 1000, { packed: 1000 }), + rar5File('Book/04.mp3', 4500, { packed: 4000, headerFlags: SPLIT_AFTER }), + end, + ]), + Buffer.concat([RAR5, main(2), rar5File('Book/04.mp3', 4500, { packed: 500, headerFlags: SPLIT_BEFORE }), rar5File('Book/05.mp3', 100, { packed: 100 }), rar5Block(5, 0, vint(0))]), + ] +} + +test.group('rar / archive walks', () => { + test('lists every entry of a RAR5 set, reading a segment per header', async ({ assert }) => { + // 700-byte segments: headers get cut off by segment ends. + const reader = memoryReader(rar5Set(), 700) + const walk = await walkRar(reader) + assert.isTrue(walk.complete) + assert.equal(walk.format, 'rar5') + assert.deepEqual(walk.entries.map((e) => e.name), ['Book/01.mp3', 'Book/02.mp3', 'Book/03.mp3', 'Book/04.mp3', 'Book/05.mp3']) + assert.equal(walk.entries[1].size, 7000) + const total = rar5Set().reduce((sum, v) => sum + Math.ceil(v.length / 700), 0) + assert.isBelow(reader.segments.size, total / 2) + }) + + test('lists a RAR4 set across volumes', async ({ assert }) => { + const end = rar4Block(0x7b, 0, Buffer.alloc(0)) + const volumes = [ + Buffer.concat([RAR4, rar4Main(0x0001 | 0x0100), rar4File(Buffer.from('a.flac'), { packed: 900, unpacked: 900 }), rar4File(Buffer.from('b.flac'), { packed: 2000, unpacked: 2500, flags: 0x0002 }), end]), + Buffer.concat([RAR4, rar4Main(0x0001), rar4File(Buffer.from('b.flac'), { packed: 500, unpacked: 2500, flags: 0x0001 }), rar4File(Buffer.from('c.flac'), { packed: 10, unpacked: 10 }), end]), + ] + const walk = await walkRar(memoryReader(volumes, 512)) + assert.isTrue(walk.complete) + assert.deepEqual(walk.entries.map((e) => e.name), ['a.flac', 'b.flac', 'c.flac']) + }) + + test('stops at a missing segment or a broken header, keeping what it found', async ({ assert }) => { + const set = rar5Set() + const missing = memoryReader(set, 700) + const read = missing.read.bind(missing) + missing.read = async (v, offset, more) => (v === 1 ? null : read(v, offset, more)) + const partial = await walkRar(missing) + assert.equal(partial.stopped, 'unreadable') + assert.deepEqual(partial.entries.map((e) => e.name), ['Book/01.mp3', 'Book/02.mp3']) + + const broken = rar5Set() + // The header of 03.mp3 in volume 2, after 02.mp3's continuation and its data. + const at = broken[1].indexOf(Buffer.from('Book/03.mp3')) - 4 + broken[1][at] ^= 0xff + const invalid = await walkRar(memoryReader(broken, 700)) + assert.equal(invalid.stopped, 'invalid') + assert.lengthOf(invalid.entries, 2) + }) + + test('finds segments from their yEnc offsets', async ({ assert }) => { + const { SegmentReader } = await import('#services/ArchiveWalkService') + const set = rar5Set() + const size = 700 + const articles = new Map() + const files = set.map((volume, v) => { + const segments = [] + for (let i = 0; i * size < volume.length; i++) { + const id = `v${v}s${i}@test` + 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 } + }) + const reader = new SegmentReader(files, 100, async (id) => articles.get(id)!) + const walk = await walkRar(reader) + assert.isTrue(walk.complete) + assert.lengthOf(walk.entries, 5) + assert.isBelow(reader.fetched, articles.size / 2) + + const capped = new SegmentReader(files, 3, async (id) => articles.get(id)!) + const stopped = await walkRar(capped) + assert.equal(stopped.stopped, 'unreadable') + assert.equal(capped.fetched, 3) + }) +}) + +/** One yEnc article of `file`: part `index` of `size` bytes. */ +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 + // Escape the critical characters, and '.' and whitespace so no line needs dot-stuffing. + if ([0x00, 0x0a, 0x0d, 0x3d, 0x2e, 0x09, 0x20].includes(c)) out.push(0x3d, (c + 64) & 0xff) + else out.push(c) + } + const lines: string[] = [] + const encoded = Buffer.from(out) + for (let i = 0; i < encoded.length; i += 128) { + let end = Math.min(encoded.length, i + 128) + if (encoded[end - 1] === 0x3d && end < encoded.length) end++ + lines.push(encoded.subarray(i, end).toString('latin1')) + i = end - 128 + } + 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 + lines.join('\r\n') + tail, 'latin1') +}