Content inspection and archive walks on seeding nodes (relay)

Inspection is split into prepare (database), probe (lib/content/probe:
downloads, decoding and parsing, no database) and apply (database).
worker:collection runs all three; seed:node with SEED_INSPECTORS leases
jobs from the same queues through the seed API (BullMQ manual leases),
fetches articles through the seed API and posts the probes back.

A node now needs only SEED_URL and SEED_TOKEN: the settings the app
requires but a node never uses get placeholders for seed:node.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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<Buffer | null> {
-  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

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<WalkOutcome> {
-    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<WalkOutcome> {
+    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<string, unknown> = {
         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<WalkOutcome> {
+    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<number, number>()
-  private readonly cache = new Map<string, DecodedSegment>()
-
-  constructor(
-    private readonly files: any[],
-    private readonly maxSegments: number,
-    private readonly fetchBody: (messageId: string) => Promise<Buffer> = 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<Buffer | null> {
-    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<DecodedSegment | null> {
-    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<Buffer> {
   const conn = await NntpService.acquire(true)

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<Buffer | null> {
-    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<NzbImportResult[]> {
-    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<ContentResult | null> {
+  async prepare(fileId: number): Promise<InspectTask | null> {
     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<ContentResult> {
+    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<ContentResult | null> {
+    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<Buffer> {
+  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()

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<Par2Set | null> {
-    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<Par2Result> {
     const result = await db.transaction(async (trx) => {
       // Exact: size and first-16-KiB MD5 of a listed file. Only these may pull

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<QueueName, Worker>()
+
+  /** Up to `max` jobs for a node, each locked for LOCK_MS. */
+  async lease(nodeId: number, max: number): Promise<RemoteJob[]> {
+    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<Buffer | null> {
+    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<ContentTask | null> {
+    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<Job> {
+    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<unknown>) {
+    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()

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<RarInfo | null> {
-    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<RarInfo | null> {
-    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<number, DecodedSegment>()
-    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<Buffer | null> => {
-      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<Buffer> {
-    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()