Edge aggregation: seed:range indexes history one row per file

A seeding node fetches a range over one NNTP connection, parses headers
with the live rules, groups them into files in memory (lib/aggregate) and
stores per complete file its release, the file and a completed binary
(SeedService), instead of one parts row per article. What doesn't complete
within the window goes through the regular ingest path as fragments.
`--dry-run` only fetches and aggregates, to measure a node.

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

app/services/SeedService.ts

@@ -0,0 +1,107 @@
+import db from '@adonisjs/lucid/services/db'
+import type { AggregatedFile } from '#lib/aggregate/index'
+import IngestService, { type IngestSegment } from '#services/IngestService'
+import FileStore from '#services/FileStore'
+import SfvService from '#services/SfvService'
+import ContentService from '#services/ContentService'
+import QueueService from '#services/QueueService'
+
+export interface SeedStoreResult {
+  files: number
+  /** Files that existed already (same file_key, e.g. from live indexing or a crosspost). */
+  existing: number
+  releases: number
+  inspections: number
+}
+
+/**
+ * Stores files an edge node aggregated from headers (CONCEPT.md D37): their
+ * releases, the files with their segment lists, and one completed binary per
+ * file, as worker:file would have left it, so par2 naming, inspection and GC
+ * treat them like any other file. A batch is a few statements, not one per
+ * article.
+ */
+class SeedService {
+  async store(files: AggregatedFile<IngestSegment>[]): Promise<SeedStoreResult> {
+    const result: SeedStoreResult = { files: 0, existing: 0, releases: 0, inspections: 0 }
+    if (files.length === 0) return result
+    const knex = db.connection().getWriteClient()
+    const now = new Date()
+
+    // A file's identity and release come from its first segment, its date from the earliest.
+    const firstOf = (file: AggregatedFile<IngestSegment>) => file.segments[0]
+    const postedOf = (file: AggregatedFile<IngestSegment>) => {
+      const dates = file.segments.map((s) => s.postedAt?.getTime()).filter((t): t is number => Number.isFinite(t))
+      return dates.length > 0 ? new Date(Math.min(...dates)) : undefined
+    }
+    const releaseIds = await IngestService.upsertReleases(
+      knex,
+      files.map((file) => ({ ...firstOf(file), postedAt: postedOf(file) })),
+      now
+    )
+    result.releases = releaseIds.size
+
+    const rows = files.map((file) => {
+      const first = firstOf(file)
+      const postedAt = postedOf(file)
+      return {
+        file_key: file.fileKey,
+        filename: first.name,
+        poster: first.poster,
+        date: (postedAt ?? now).getTime(),
+        parts: file.segments.length,
+        segments: file.segments.map((s) => ({ number: s.number, id: s.messageId, size: s.bytes })),
+        groups: JSON.stringify(first.groups),
+        release_id: first.release ? (releaseIds.get(first.release.key) ?? null) : null,
+        size: file.segments.reduce((sum, s) => sum + s.bytes, 0),
+        created_at: now,
+        updated_at: now,
+      }
+    })
+    const inserted = await FileStore.insert(rows)
+    result.files = inserted.length
+    result.existing = rows.length - inserted.length
+
+    const fileIds = new Map<string, number>()
+    for (const row of await db.from('files').whereIn('file_key', rows.map((r) => r.file_key)).select('id', 'file_key')) {
+      fileIds.set(row.file_key, Number(row.id))
+    }
+    // The completed binary worker:file would have left behind; an existing one just learns its file.
+    const binaries = rows.map((row, i) => ({
+      file_key: row.file_key,
+      poster: row.poster,
+      subject: firstOf(files[i]).subject,
+      name: row.filename,
+      total_parts: firstOf(files[i]).totalParts,
+      groups: row.groups,
+      posted_at: postedOf(files[i]) ?? null,
+      first_seen_at: now,
+      last_seen_at: now,
+      completed_at: now,
+      release_id: row.release_id,
+      file_id: fileIds.get(row.file_key) ?? null,
+    }))
+    binaries.sort((a, b) => a.file_key.localeCompare(b.file_key))
+    await knex('binaries')
+      .insert(binaries)
+      .onConflict('file_key')
+      .merge({ file_id: knex.raw('coalesce(binaries.file_id, excluded.file_id)') as any })
+
+    // As worker:file does for a new file: an SFV posted earlier may list it (D34), then inspection.
+    const rowById = new Map(rows.map((row) => [fileIds.get(row.file_key), row]))
+    const candidates: { name: string; data: { fileId: number }; opts: { jobId: string } }[] = []
+    for (const file of inserted) {
+      const row = rowById.get(file.id)
+      if (!row) continue
+      await SfvService.claim({ id: file.id, name: row.filename, poster: row.poster, date: row.date, release_id: row.release_id })
+      if (ContentService.isCandidate(row.filename)) {
+        candidates.push({ name: 'process-collection', data: { fileId: file.id }, opts: { jobId: `file-${file.id}` } })
+      }
+    }
+    if (candidates.length > 0) await QueueService.collectionQueue.addBulk(candidates)
+    result.inspections = candidates.length
+    return result
+  }
+}
+
+export default new SeedService()

commands/SeedRange.ts

@@ -0,0 +1,155 @@
+import { BaseCommand, args, flags } from '@adonisjs/core/ace'
+import { CommandOptions } from '@adonisjs/core/types/ace'
+import NntpService from '#services/NntpService'
+import IngestService, { segmentFromHeader, type IngestSegment } from '#services/IngestService'
+import StagingService from '#services/StagingService'
+import SeedService from '#services/SeedService'
+import QueueService from '#services/QueueService'
+import { isDeferred } from '#services/DeferredService'
+import { FileAggregator } from '#lib/aggregate/index'
+
+/**
+ * Edge aggregation (CONCEPT.md D37): indexes a range of a group's history
+ * over one NNTP connection, grouping headers into files in memory and
+ * storing one row per file. Files cut by the range edges, or that never
+ * complete, go through the regular ingest path as fragments. Header-only,
+ * like the backfill: obfuscated subjects are counted, not downloaded.
+ *
+ *   node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000
+ *   node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000 --dry-run
+ */
+export default class SeedRange extends BaseCommand {
+  public static commandName = 'seed:range'
+  public static description = "Indexes a range of a group's headers by aggregating them into files on this node."
+
+  public static options: CommandOptions = {
+    startApp: true,
+  }
+
+  @args.string({ description: 'Group' })
+  declare group: string
+
+  @flags.string({ description: 'First article number' })
+  declare start: string
+
+  @flags.number({ description: 'Articles to index', default: 1_000_000 })
+  declare count: number
+
+  @flags.number({ description: 'Articles per XOVER', default: 20_000 })
+  declare chunk: number
+
+  @flags.number({ description: 'Batches without a new part before an open file is given up as fragments', default: 5 })
+  declare idle: number
+
+  @flags.boolean({ description: 'Fetch and aggregate only; store nothing' })
+  declare dryRun: boolean
+
+  public async run() {
+    const first = Number(this.start)
+    const last = first + this.count - 1
+    if (!Number.isSafeInteger(first) || first < 1) throw new Error(`--start must be an article number, got ${this.start}`)
+    const aggregator = new FileAggregator<IngestSegment>()
+    const stats = { headers: 0, segments: 0, obfuscated: 0, deferred: 0, complete: 0, files: 0, existing: 0, fragments: 0, inspections: 0 }
+    const time = { fetch: 0, parse: 0, store: 0 }
+    const cpuStart = process.cpuUsage()
+    const started = Date.now()
+
+    const conn = await NntpService.acquire()
+    let failed = false
+    try {
+      await NntpService.timed(conn.group(this.group), `GROUP ${this.group}`)
+      const fetch = async (from: number) => {
+        const to = Math.min(from + this.chunk - 1, last)
+        const t = Date.now()
+        const reply = await NntpService.timed<any>(conn.xover(from, to), `XOVER ${this.group} ${from}-${to}`)
+        time.fetch += Date.now() - t
+        return reply.overviews.filter(([n, h]: [number | null, Record<string, string>]) => n !== null && h?.['message-id'])
+      }
+
+      // One XOVER in flight while the previous batch is parsed and stored.
+      let from = first
+      const prefetch = (at: number) => {
+        const promise = fetch(at)
+        // Awaited after the current batch is stored; until then a failure must not count as unhandled.
+        promise.catch(() => {})
+        return promise
+      }
+      let next: Promise<[number, Record<string, string>][]> | null = prefetch(from)
+      while (next) {
+        const overviews = await next
+        from += this.chunk
+        next = from <= last ? prefetch(from) : null
+
+        let t = Date.now()
+        const segments: IngestSegment[] = []
+        for (const [articleNumber, header] of overviews) {
+          const segment = segmentFromHeader(header, this.group, articleNumber)
+          if (segment) segments.push(segment)
+          else if (isDeferred(header)) stats.deferred++
+          else stats.obfuscated++
+        }
+        stats.headers += overviews.length
+        stats.segments += segments.length
+        const complete = aggregator.add(segments)
+        const fragments = aggregator.evict(this.idle)
+        stats.complete += complete.length
+        stats.fragments += fragments.length
+        time.parse += Date.now() - t
+
+        t = Date.now()
+        if (!this.dryRun) await this.store(complete, fragments, stats)
+        time.store += Date.now() - t
+        this.progress(stats, started)
+      }
+
+      const rest = aggregator.evict(0)
+      stats.fragments += rest.length
+      if (!this.dryRun) await this.store([], rest, stats)
+    } catch (error) {
+      failed = true
+      throw error
+    } finally {
+      NntpService.release(conn, failed)
+      await NntpService.shutdown()
+      await QueueService.closeAll()
+    }
+
+    const seconds = (Date.now() - started) / 1000
+    const cpu = process.cpuUsage(cpuStart)
+    this.logger.info(
+      `${this.group} ${first}-${last}: ${stats.headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s ` +
+        `(${Math.round(stats.headers / seconds).toLocaleString('en')}/s). ` +
+        `${stats.segments.toLocaleString('en')} readable, ${stats.obfuscated.toLocaleString('en')} obfuscated, ${stats.deferred.toLocaleString('en')} deferred. ` +
+        `${stats.complete.toLocaleString('en')} complete files (${stats.files.toLocaleString('en')} new, ${stats.existing.toLocaleString('en')} known), ` +
+        `${stats.fragments.toLocaleString('en')} fragment segments, ${stats.inspections.toLocaleString('en')} queued for inspection.`
+    )
+    this.logger.info(
+      `Time: fetch ${(time.fetch / 1000).toFixed(0)} s (overlapped), parse+aggregate ${(time.parse / 1000).toFixed(0)} s, ` +
+        `store ${(time.store / 1000).toFixed(0)} s; node CPU ${((cpu.user + cpu.system) / 1e6).toFixed(0)} s.`
+    )
+  }
+
+  private async store(complete: Parameters<typeof SeedService.store>[0], fragments: IngestSegment[], stats: Record<string, number>) {
+    const stored = await SeedService.store(complete)
+    stats.files += stored.files
+    stats.existing += stored.existing
+    stats.inspections += stored.inspections
+    if (fragments.length === 0) return
+    // The regular path: staging (D35) for lone segments, binaries and parts for the rest.
+    const { ingest } = StagingService.enabled(this.group)
+      ? await StagingService.filter(fragments, { counters: StagingService.countersEnabled(this.group) })
+      : { ingest: fragments }
+    await IngestService.ingest(ingest)
+  }
+
+  private lastReport = 0
+  private progress(stats: Record<string, number>, started: number) {
+    if (Date.now() - this.lastReport < 30_000) return
+    this.lastReport = Date.now()
+    const seconds = (Date.now() - started) / 1000
+    this.logger.info(
+      `${stats.headers.toLocaleString('en')} headers (${Math.round(stats.headers / seconds).toLocaleString('en')}/s), ` +
+        `${stats.complete.toLocaleString('en')} files, ${stats.fragments.toLocaleString('en')} fragment segments.`
+    )
+  }
+}

lib/aggregate/index.ts

@@ -0,0 +1,86 @@
+/**
+ * Edge aggregation (CONCEPT.md D37): groups header segments into files in
+ * memory, so a seeding node writes one row per file instead of one per
+ * article. A file is done once every part number from 1 to its total has
+ * an article; what never completes within the node's window (an upload cut
+ * by the range edges, a random subject per article) leaves as fragments
+ * for the regular ingest path, where it can meet its other parts.
+ */
+
+/** What the aggregator needs of a segment (IngestSegment has all of it). */
+export interface AggregatableSegment {
+  fileKey: string
+  number: number
+  totalParts: number
+  messageId: string
+  bytes: number
+}
+
+export interface AggregatedFile<S extends AggregatableSegment> {
+  fileKey: string
+  /** One segment per part number, in part order; the first article seen wins (a later one is a repost). */
+  segments: S[]
+}
+
+interface OpenFile<S> {
+  parts: Map<number, S>
+  totalParts: number
+  /** Batch in which a segment last arrived. */
+  touched: number
+}
+
+export class FileAggregator<S extends AggregatableSegment> {
+  private open = new Map<string, OpenFile<S>>()
+  private batch = 0
+  /** Segments with a part number outside 1..total: never part of a complete file. */
+  private stray: S[] = []
+
+  /** Adds one batch of segments; returns the files they completed. */
+  add(segments: S[]): AggregatedFile<S>[] {
+    this.batch++
+    const touched = new Set<string>()
+    for (const segment of segments) {
+      if (segment.number < 1 || segment.number > segment.totalParts) {
+        this.stray.push(segment)
+        continue
+      }
+      let file = this.open.get(segment.fileKey)
+      if (!file) {
+        file = { parts: new Map(), totalParts: segment.totalParts, touched: this.batch }
+        this.open.set(segment.fileKey, file)
+      }
+      if (!file.parts.has(segment.number)) file.parts.set(segment.number, segment)
+      file.touched = this.batch
+      touched.add(segment.fileKey)
+    }
+
+    const complete: AggregatedFile<S>[] = []
+    for (const key of touched) {
+      const file = this.open.get(key)!
+      if (file.parts.size < file.totalParts) continue
+      this.open.delete(key)
+      complete.push({ fileKey: key, segments: [...file.parts.values()].sort((a, b) => a.number - b.number) })
+    }
+    return complete
+  }
+
+  /**
+   * Gives up on files that got no segment in the last `idleBatches` batches
+   * (all open files when 0) and returns their segments, plus stray ones.
+   */
+  evict(idleBatches = 0): S[] {
+    const out = this.stray
+    this.stray = []
+    for (const [key, file] of this.open) {
+      if (idleBatches > 0 && this.batch - file.touched < idleBatches) continue
+      this.open.delete(key)
+      out.push(...file.parts.values())
+    }
+    return out
+  }
+
+  /** Files still waiting for parts. */
+  get openFiles(): number {
+    return this.open.size
+  }
+}

tests/functional/seed.spec.ts

@@ -0,0 +1,47 @@
+import { test } from '@japa/runner'
+import db from '@adonisjs/lucid/services/db'
+import testUtils from '@adonisjs/core/services/test_utils'
+import { segmentFromHeader, type IngestSegment } from '#services/IngestService'
+import SeedService from '#services/SeedService'
+import QueueService from '#services/QueueService'
+import { FileAggregator } from '#lib/aggregate/index'
+import { segmentsOf } from '#services/FileStore'
+
+function header(subject: string, n: number) {
+  return {
+    'subject': subject,
+    'from': 'seeder <seed@example.invalid>',
+    'date': 'Mon, 05 Oct 2026 10:00:00 +0000',
+    'message-id': `<seed-${n}-${subject.length}@test>`,
+    ':bytes': '700000',
+  }
+}
+
+test.group('SeedService', (group) => {
+  group.each.setup(() => testUtils.db().withGlobalTransaction())
+  group.teardown(() => QueueService.closeAll())
+
+  test('stores aggregated files with their release and a completed binary', async ({ assert }) => {
+    const aggregator = new FileAggregator<IngestSegment>()
+    const subjects = [1, 2].map((i) => `Seed.Test.S01E01.1080p - [1/2] - "Seed.Test.S01E01.1080p.part${i}.mkv" yEnc (%d/3)`)
+    const segments = subjects.flatMap((s) => [1, 2, 3].map((n) => segmentFromHeader(header(s.replace('%d', String(n)), n), 'alt.binaries.test', n)!))
+    const files = aggregator.add(segments)
+    assert.lengthOf(files, 2)
+
+    const result = await SeedService.store(files)
+    assert.include(result, { files: 2, existing: 0, releases: 1 })
+
+    const stored = await db.from('files').where('filename', 'Seed.Test.S01E01.1080p.part1.mkv').first()
+    assert.equal(stored.parts, 3)
+    assert.deepEqual(segmentsOf(stored).map((s) => s.number), [1, 2, 3])
+    const release = await db.from('releases').where('id', stored.release_id).first()
+    assert.equal(release.name, segments[0].release!.name) // named as live indexing would
+    const binary = await db.from('binaries').where('file_key', stored.file_key).first()
+    assert.equal(Number(binary.file_id), Number(stored.id))
+    assert.isNotNull(binary.completed_at)
+
+    // The same files again (a crosspost, a second run): nothing new.
+    const again = await SeedService.store(files)
+    assert.include(again, { files: 0, existing: 2 })
+  })
+})

tests/unit/aggregate.spec.ts

@@ -0,0 +1,39 @@
+import { test } from '@japa/runner'
+import { FileAggregator, type AggregatableSegment } from '#lib/aggregate/index'
+
+const seg = (fileKey: string, number: number, totalParts: number, messageId = `${fileKey}-${number}@test`): AggregatableSegment => ({
+  fileKey,
+  number,
+  totalParts,
+  messageId,
+  bytes: 100,
+})
+
+test.group('aggregate / FileAggregator', () => {
+  test('completes a file once every part arrived, across batches and out of order', ({ assert }) => {
+    const aggregator = new FileAggregator()
+    assert.deepEqual(aggregator.add([seg('a', 2, 3), seg('b', 1, 1)]).map((f) => f.fileKey), ['b'])
+    assert.equal(aggregator.openFiles, 1)
+    const [done] = aggregator.add([seg('a', 3, 3), seg('a', 1, 3)])
+    assert.equal(done.fileKey, 'a')
+    assert.deepEqual(done.segments.map((s) => s.number), [1, 2, 3])
+    assert.equal(aggregator.openFiles, 0)
+  })
+
+  test('keeps the first article of a part number (a repost) and drops impossible numbers', ({ assert }) => {
+    const aggregator = new FileAggregator()
+    const [done] = aggregator.add([seg('a', 1, 2, 'first@test'), seg('a', 1, 2, 'repost@test'), seg('a', 5, 2), seg('a', 2, 2)])
+    assert.deepEqual(done.segments.map((s) => s.messageId), ['first@test', 'a-2@test'])
+    assert.deepEqual(aggregator.evict().map((s) => s.number), [5])
+  })
+
+  test('gives up idle files as fragments, all of them at the end', ({ assert }) => {
+    const aggregator = new FileAggregator()
+    aggregator.add([seg('old', 1, 3)])
+    aggregator.add([seg('new', 1, 3)])
+    aggregator.add([seg('new', 2, 3)])
+    assert.deepEqual(aggregator.evict(2).map((s) => s.fileKey), ['old'])
+    assert.deepEqual(aggregator.evict(0).map((s) => s.number), [1, 2])
+    assert.equal(aggregator.openFiles, 0)
+  })
+})