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 <noreply@anthropic.com>

app/services/ArchiveService.ts

@@ -27,7 +27,8 @@ export interface ArchiveResult {
  */
 class ArchiveService {
   async record(file: any, binary: any, info: RarInfo): Promise<ArchiveResult> {
-    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

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<WalkOutcome> {
+    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<string, unknown> = {
+        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<any[] | null> {
+    const files = await db.from('files').where('release_id', releaseId).select('id', 'filename', 'real_name', 'segments')
+    const sets = new Map<string, Map<number, any>>()
+    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<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)
+  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()

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)
       }
     }

app/services/NntpService.ts

@@ -7,6 +7,8 @@ class NntpService {
   private allConnections: Set<any>;
   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;
   }
 }

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. */

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,