Seed plans fill every uncovered gap, with priorities; cheaper ingest and file storing

- seed:plan plans every part of a group's history no range covers (it
  started below the lowest known range and left a.b.ath's 13 most recent
  months out); seed_ranges.priority orders leases (seed:plan --priority).
- binaries.part_count, kept by the parts insert in the same statement:
  claimCompleted compares it instead of counting parts every batch.
- file_keys: 64-bit hashes of all file keys; FileStore.insert asks it
  before probing every files partition.
- worker:file stores 100 completed binaries per job with set-based queries.

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

app/services/BinaryFileService.ts

@@ -0,0 +1,141 @@
+import db from '@adonisjs/lucid/services/db'
+import FileStore from '#services/FileStore'
+import SfvService from '#services/SfvService'
+import QueueService from '#services/QueueService'
+import ContentService from '#services/ContentService'
+
+export interface BinaryFilesResult {
+  /** New files. */
+  saved: number
+  /** Binaries whose file already existed (same first segment or file_key). */
+  existing: number
+  /** Conflicted, gone or already linked: nothing to do. */
+  skipped: number
+  /** Not all parts stored yet (the job is retried for these). */
+  incomplete: number[]
+  inspections: number
+}
+
+/**
+ * Turns completed binaries into files (worker:file), a batch at a time
+ * (LOG.md 2026-10-06 (7)): one query each for the binaries, their parts and
+ * files that exist already, one INSERT for the new files (so the files
+ * statement trigger updates each release once), and one UPDATE and DELETE
+ * for the binaries' links and copied parts. It was one job of ~8 queries per
+ * binary, ~700 a minute.
+ */
+class BinaryFileService {
+  async store(binaryIds: number[]): Promise<BinaryFilesResult> {
+    const result: BinaryFilesResult = { saved: 0, existing: 0, skipped: 0, incomplete: [], inspections: 0 }
+    const ids = [...new Set(binaryIds)]
+    const binaries = await db
+      .from('binaries')
+      .whereIn('id', ids)
+      .whereNull('conflicted_at')
+      .whereNull('file_id')
+      .select('id', 'file_key', 'name', 'real_name', 'poster', 'posted_at', 'groups', 'release_id', 'total_parts')
+    result.skipped = ids.length - binaries.length
+    if (binaries.length === 0) return result
+
+    const partsOf = new Map<number, { number: number; id: string; size: number }[]>()
+    const parts = await db
+      .from('parts')
+      .whereIn('binary_id', binaries.map((b) => b.id))
+      .orderBy([{ column: 'binary_id' }, { column: 'number' }])
+      .select('binary_id', 'number', 'message_id', 'bytes')
+    for (const part of parts) {
+      const list = partsOf.get(Number(part.binary_id)) ?? []
+      list.push({ number: Number(part.number), id: part.message_id, size: Number(part.bytes) })
+      partsOf.set(Number(part.binary_id), list)
+    }
+    const complete = binaries.filter((b) => {
+      const ok = (partsOf.get(Number(b.id))?.length ?? 0) >= b.total_parts
+      if (!ok) result.incomplete.push(Number(b.id))
+      return ok
+    })
+    if (complete.length === 0) return result
+
+    // Already known as a file, e.g. added from a posted NZB before its own articles completed (id only: the row holds the segment list).
+    const firsts = complete.map((b) => partsOf.get(Number(b.id))![0].id)
+    const byFirst = new Map<string, number>()
+    for (const row of await db.from('files').whereIn('first_message_id', firsts).select('id', 'first_message_id')) {
+      byFirst.set(row.first_message_id, Number(row.id))
+    }
+    const fileIdOf = new Map<number, number>()
+    const fresh = complete.filter((b) => {
+      const existing = byFirst.get(partsOf.get(Number(b.id))![0].id)
+      if (existing) fileIdOf.set(Number(b.id), existing)
+      return !existing
+    })
+
+    const now = new Date()
+    // file_key is unique: a second completion of the same upload inserts nothing.
+    const inserted = await FileStore.insert(
+      fresh.map((b) => {
+        const segments = partsOf.get(Number(b.id))!
+        return {
+          file_key: b.file_key,
+          filename: b.name,
+          poster: b.poster,
+          date: b.posted_at ? new Date(b.posted_at).getTime() : now.getTime(),
+          parts: segments.length,
+          segments,
+          groups: JSON.stringify(b.groups),
+          release_id: b.release_id,
+          real_name: b.real_name,
+          size: segments.reduce((sum, s) => sum + s.size, 0),
+          created_at: now,
+          updated_at: now,
+        }
+      })
+    )
+    const insertedByKey = new Map(inserted.map((row) => [row.file_key, row.id]))
+    const missing = fresh.filter((b) => !insertedByKey.has(b.file_key)).map((b) => b.file_key)
+    const byKey = new Map<string, number>()
+    if (missing.length > 0) {
+      for (const row of await db.from('files').whereIn('file_key', missing).select('id', 'file_key')) byKey.set(row.file_key, Number(row.id))
+    }
+    for (const b of fresh) {
+      const id = insertedByKey.get(b.file_key) ?? byKey.get(b.file_key)
+      if (id) fileIdOf.set(Number(b.id), id)
+    }
+    result.saved = inserted.length
+    result.existing = fileIdOf.size - inserted.length
+
+    // Links, locked in file_key order like ingest's upsert, so the two can't deadlock.
+    const linked = complete.filter((b) => fileIdOf.has(Number(b.id))).sort((a, b) => a.file_key.localeCompare(b.file_key))
+    if (linked.length > 0) {
+      await db.transaction(async (trx) => {
+        await trx.rawQuery(
+          `SELECT id FROM binaries WHERE id IN (SELECT unnest(?::bigint[])) ORDER BY file_key FOR UPDATE`,
+          [linked.map((b) => b.id)]
+        )
+        await trx.rawQuery(
+          `UPDATE binaries b SET file_id = v.file_id FROM unnest(?::bigint[], ?::bigint[]) AS v(id, file_id) WHERE b.id = v.id`,
+          [linked.map((b) => b.id), linked.map((b) => fileIdOf.get(Number(b.id))!)]
+        )
+      })
+      // Header-derived parts are in the file's segment list now; body-derived ones (with a CRC) stay for split checks (D20).
+      await db.rawQuery(`DELETE FROM parts WHERE binary_id = ANY(?::bigint[]) AND pcrc32 IS NULL`, [linked.map((b) => b.id)])
+    }
+
+    // As for one file before: an SFV posted earlier may list them (D34), then inspection.
+    const fresher = new Map(fresh.map((b) => [b.file_key, b]))
+    const added = inserted.flatMap((row) => {
+      const b = row.file_key ? fresher.get(row.file_key) : undefined
+      return b
+        ? [{ id: row.id, name: b.real_name ?? b.name, posted: String(b.name), poster: b.poster, date: b.posted_at ? new Date(b.posted_at).getTime() : null, release_id: b.release_id }]
+        : []
+    })
+    if (added.length > 0) await SfvService.claimAll(added.map(({ posted: _p, ...file }) => file))
+    // By the posted name, as before: inspection skips later volumes and readable media names anyway.
+    const jobs = added
+      .filter((file) => ContentService.isCandidate(file.posted))
+      .map((file) => ({ name: 'process-collection', data: { fileId: file.id }, opts: { jobId: `file-${file.id}` } }))
+    if (jobs.length > 0) await QueueService.collectionQueue.addBulk(jobs)
+    result.inspections = jobs.length
+    return result
+  }
+}
+
+export default new BinaryFileService()

app/services/FileStore.ts

@@ -42,17 +42,29 @@ export const insertLimits = { rows: 2000 }
 class FileStore {
   /**
    * Inserts files, skipping those whose `file_key` exists. Returns the rows
-   * that were inserted (id and filename). `files` is partitioned by id
+   * that were inserted (id, filename, file_key). `files` is partitioned by id
    * (CONCEPT.md D39) and each partition has its own unique file_key index:
-   * keys already in any partition are looked up first; two writers adding
-   * the same new key at once meet in the newest partition's index.
+   * file_keys (a 64-bit hash of every key, filled by the insert trigger) says
+   * which keys may exist, and only those are looked up in the partitions; two
+   * writers adding the same new key at once meet in the newest partition's index.
    */
-  async insert(files: NewFile[], client: Pick<QueryClientContract, 'knexQuery'> = db): Promise<{ id: number; filename: string }[]> {
+  async insert(
+    files: NewFile[],
+    client: Pick<QueryClientContract, 'knexQuery' | 'rawQuery'> = db
+  ): Promise<{ id: number; filename: string; file_key: string | null }[]> {
     if (files.length === 0) return []
     const keys = [...new Set(files.map((file) => file.file_key).filter((key): key is string => typeof key === 'string'))]
     const known = new Set<string>()
     for (let i = 0; i < keys.length; i += 5000) {
-      const found = await client.knexQuery().from('files').whereIn('file_key', keys.slice(i, i + 5000)).select('file_key')
+      const chunk = keys.slice(i, i + 5000)
+      const { rows } = await client.rawQuery(
+        `SELECT k.key FROM unnest(?::text[]) AS k(key)
+         WHERE EXISTS (SELECT 1 FROM file_keys f WHERE f.key_hash = hashtextextended(k.key, 0))`,
+        [chunk]
+      )
+      const maybe = rows.map((row: { key: string }) => row.key)
+      if (maybe.length === 0) continue
+      const found = await client.knexQuery().from('files').whereIn('file_key', maybe).select('file_key')
       for (const row of found) known.add(row.file_key)
     }
     const seen = new Set<string>()
@@ -70,10 +82,10 @@ class FileStore {
       })
     if (rows.length === 0) return []
     // A statement takes at most 65,535 parameters: ~5,400 rows of 12 columns.
-    const inserted: { id: number; filename: string }[] = []
+    const inserted: { id: number; filename: string; file_key: string | null }[] = []
     for (let i = 0; i < rows.length; i += insertLimits.rows) {
-      const chunk = await client.knexQuery().table('files').insert(rows.slice(i, i + insertLimits.rows)).onConflict().ignore().returning(['id', 'filename'])
-      for (const row of chunk) inserted.push({ id: Number(row.id), filename: row.filename })
+      const chunk = await client.knexQuery().table('files').insert(rows.slice(i, i + insertLimits.rows)).onConflict().ignore().returning(['id', 'filename', 'file_key'])
+      for (const row of chunk) inserted.push({ id: Number(row.id), filename: row.filename, file_key: row.file_key ?? null })
     }
     return inserted
   }

app/services/IngestService.ts

@@ -85,6 +85,9 @@ async function upsertKeyed(
     }
   }
 }
+/** Completed binaries per worker:file job. */
+const FILE_BATCH = 100
+
 /** Size-only matching below this is too ambiguous (nfo, sfv, small volumes). */
 export const PAR2_SIZE_ONLY_MIN = 1024 * 1024
 /** A par2 set and its files are posted together; don't match across days. */
@@ -187,8 +190,29 @@ class IngestService {
       const newParts = partRows.filter((row) => !(done.has(row.binary_id) && row.pcrc32 === null))
 
       newParts.sort((a, b) => a.binary_id - b.binary_id || a.number - b.number)
+      // Parts and the binaries' stored-part counts in one statement, so the count can't miss a
+      // part (LOG.md 2026-10-06 (7)); claimCompleted compares it instead of counting parts. NULL
+      // (binaries from before part_count) is counted once: the statement's own rows aren't in
+      // its snapshot, so they're added on top. Concurrent batches can lock binaries in a
+      // different order: a deadlock rolls the whole statement back, and it is retried.
       for (const chunk of chunks(newParts, PART_CHUNK)) {
-        await knex('parts').insert(chunk).onConflict(['binary_id', 'number']).ignore()
+        const insert = knex('parts').insert(chunk).onConflict(['binary_id', 'number']).ignore().returning('binary_id').toSQL()
+        for (let attempt = 1; ; attempt++) {
+          try {
+            await knex.raw(
+              `WITH added AS (${insert.sql}),
+               counted AS (SELECT binary_id, count(*)::int AS n FROM added GROUP BY binary_id)
+               UPDATE binaries b SET part_count = CASE
+                 WHEN b.part_count IS NULL THEN (SELECT count(*) FROM parts p WHERE p.binary_id = b.id) + c.n
+                 ELSE b.part_count + c.n END
+               FROM counted c WHERE b.id = c.binary_id`,
+              insert.bindings as any[]
+            )
+            break
+          } catch (error: any) {
+            if (error.code !== '40P01' || attempt >= 5) throw error
+          }
+        }
       }
 
       if (weakRows.length > 0) await this.flagConflicts(db.connection(), weakRows)
@@ -366,7 +390,7 @@ class IngestService {
     const result = await db.rawQuery(
       `UPDATE binaries b SET completed_at = now()
        WHERE b.id = ANY(?) AND b.completed_at IS NULL AND b.conflicted_at IS NULL
-         AND (SELECT count(*) FROM parts p WHERE p.binary_id = b.id) >= b.total_parts
+         AND coalesce(b.part_count, (SELECT count(*) FROM parts p WHERE p.binary_id = b.id)) >= b.total_parts
        RETURNING b.id, b.file_key`,
       [binaryIds]
     )
@@ -378,14 +402,14 @@ class IngestService {
 
   async enqueue(completed: CompletedBinary[]) {
     if (completed.length === 0) return
-    // jobId = file_key: BullMQ ignores an add() for an id that already exists.
-    await QueueService.fileQueue.addBulk(
-      completed.map(({ id, fileKey }) => ({
-        name: 'process-file',
-        data: { binaryId: id },
-        opts: { jobId: fileKey },
-      }))
-    )
+    // A job per FILE_BATCH binaries: worker:file stores them with a few set-based queries (LOG.md 2026-10-06 (7)).
+    // jobId from the first file_key: BullMQ ignores an add() for an id that already exists.
+    const jobs = []
+    for (let i = 0; i < completed.length; i += FILE_BATCH) {
+      const batch = completed.slice(i, i + FILE_BATCH)
+      jobs.push({ name: 'process-files', data: { binaryIds: batch.map((b) => b.id) }, opts: { jobId: `${batch[0].fileKey}-${batch.length}` } })
+    }
+    await QueueService.fileQueue.addBulk(jobs)
   }
 }
 

app/services/SeedOrchestrator.ts

@@ -1,5 +1,6 @@
 import { createHash, randomBytes } from 'node:crypto'
 import db from '@adonisjs/lucid/services/db'
+import { cut, uncovered } from '#lib/seed/plan'
 import NntpService from '#services/NntpService'
 import SeedService from '#services/SeedService'
 import BackfillService from '#services/BackfillService'
@@ -77,7 +78,7 @@ class SeedOrchestrator {
        WHERE id = (
          SELECT id FROM seed_ranges
          WHERE status = 'queued' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?))
-         ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED)
+         ORDER BY priority DESC, id LIMIT 1 FOR UPDATE SKIP LOCKED)
        RETURNING id, group_name, next_id, end_id`,
       [nodeId, LEASE_TTL_SECONDS]
     )
@@ -143,9 +144,12 @@ class SeedOrchestrator {
    * server's oldest article. A group's backfill stops where its seed plan
    * begins. Ranges are interleaved across groups, newest first.
    */
-  async plan(groups: string[], options: { days?: number; size: number }): Promise<{ group: string; from: number; to: number; ranges: number }[]> {
+  async plan(
+    groups: string[],
+    options: { days?: number; size: number; priority?: number }
+  ): Promise<{ group: string; from: number; to: number; ranges: number; gaps: number }[]> {
     const perGroup: { group: string; ranges: [number, number][] }[] = []
-    const summary: { group: string; from: number; to: number; ranges: number }[] = []
+    const summary: { group: string; from: number; to: number; ranges: number; gaps: number }[] = []
     const conn = await NntpService.acquire()
     let failed = false
     try {
@@ -159,20 +163,21 @@ class SeedOrchestrator {
           const date = header ? Date.parse(header.date) : Number.NaN
           return Number.isNaN(date) ? null : date
         }
-        // Top: where live indexing began, or the backfill's position, or below earlier seed ranges.
+        // Top: where live indexing began, or as far down as the backfill has come.
         let top = Number(await articleAtDate(first, last, new Date(group.created_at).getTime(), dateOf))
+        if (group.backfill_next_id !== null) top = Math.min(top, Number(group.backfill_next_id))
         const backfilling = group.backfill_next_id !== null && group.backfill_until_id !== null && Number(group.backfill_next_id) > Number(group.backfill_until_id)
-        if (backfilling) top = Math.min(top, Number(group.backfill_next_id))
-        const seeded = await db.from('seed_ranges').where('group_name', name).min('start_id as low').first()
-        if (seeded?.low !== null && seeded?.low !== undefined) top = Math.min(top, Number(seeded.low))
         const bottom = options.days
           ? Number(await articleAtDate(first, last, Date.now() - options.days * 24 * 60 * 60 * 1000, dateOf))
           : Number(first)
 
-        const ranges: [number, number][] = []
-        for (let end = top - 1; end >= bottom; end -= options.size) ranges.push([Math.max(bottom, end - options.size + 1), end])
+        // Everything below the top that no earlier range covers (in any state). Planning below
+        // the lowest earlier range left a 13-month gap in a.b.ath (LOG.md 2026-10-06 (7)).
+        const planned = await db.from('seed_ranges').where('group_name', name).select('start_id', 'end_id')
+        const gaps = uncovered(bottom, top - 1, planned.map((r) => [Number(r.start_id), Number(r.end_id)] as [number, number]))
+        const ranges = cut(gaps, options.size)
         perGroup.push({ group: name, ranges })
-        summary.push({ group: name, from: bottom, to: top - 1, ranges: ranges.length })
+        summary.push({ group: name, from: bottom, to: top - 1, ranges: ranges.length, gaps: gaps.length })
         // The seed plan takes over the rest of this group's backfill.
         if (backfilling && ranges.length > 0) {
           await db.from('groups').where('id', group.id).update({ backfill_until_id: group.backfill_next_id })
@@ -190,7 +195,7 @@ class SeedOrchestrator {
     for (let i = 0; perGroup.some((g) => i < g.ranges.length); i++) {
       for (const g of perGroup) {
         const range = g.ranges[i]
-        if (range) rows.push({ group_name: g.group, start_id: range[0], end_id: range[1], next_id: range[0] })
+        if (range) rows.push({ group_name: g.group, start_id: range[0], end_id: range[1], next_id: range[0], priority: options.priority ?? 0 })
       }
     }
     for (let i = 0; i < rows.length; i += 5000) await db.table('seed_ranges').multiInsert(rows.slice(i, i + 5000))