Stop refilling parts of binaries that are already a file

Crossposts, reposts and backfill ranges keep bringing articles of files
that worker:file already created. Ingest inserted their header parts again,
and GC's copied-parts pass (which searched all binaries for parts on every
batch) removed only 150k-450k per sweep: parts grew to 68M rows, 56M of
them belonging to finished files, and the 23 GB table paused the backfill.

- Ingest skips header parts (no per-part CRC) for binaries with a file_id.
- GC drops copied parts with a cursor over binaries, 2000 per statement.

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

app/services/IngestService.ts

@@ -260,6 +260,17 @@ class IngestService {
         for (const [key, id] of ids) idsByKey.set(key, id)
       }
 
+      // A binary that is a file already has its segment list in `files`; its
+      // header parts were dropped then. Crossposts in other groups, reposts
+      // and backfill ranges keep bringing the same articles, and inserting
+      // them again grew `parts` to 56M rows nobody reads (LOG.md 2026-10-04).
+      const done = new Set<number>()
+      const ids = [...idsByKey.values()]
+      for (const chunk of chunks(ids, BINARY_CHUNK)) {
+        const { rows } = await knex.raw('SELECT id FROM binaries WHERE id = ANY(?) AND file_id IS NOT NULL', [chunk])
+        for (const row of rows) done.add(Number(row.id))
+      }
+
       const partRows = segments
         .map((segment) => ({
           binary_id: idsByKey.get(segment.fileKey)!,
@@ -272,10 +283,12 @@ class IngestService {
           file_crc32: segment.fileCrc32 ?? null,
           md5_16k: segment.md5_16k ?? null,
         }))
-      const weakRows = partRows.filter((_, i) => segments[i].weakIdentity)
+      const weakRows = partRows.filter((row, i) => segments[i].weakIdentity && !(done.has(row.binary_id) && row.pcrc32 === null))
+      // Body-derived parts (with a per-part CRC) are kept, as worker:file does.
+      const newParts = partRows.filter((row) => !(done.has(row.binary_id) && row.pcrc32 === null))
 
-      partRows.sort((a, b) => a.binary_id - b.binary_id || a.number - b.number)
-      for (const chunk of chunks(partRows, PART_CHUNK)) {
+      newParts.sort((a, b) => a.binary_id - b.binary_id || a.number - b.number)
+      for (const chunk of chunks(newParts, PART_CHUNK)) {
         await knex('parts').insert(chunk).onConflict(['binary_id', 'number']).ignore()
       }
 

commands/GcScheduler.ts

@@ -12,6 +12,8 @@ import Shutdown from '#services/Shutdown'
 
 const CLEAN_BATCH = 1000
 const SCAN_COUNT = 1000
+/** Binaries per statement when dropping copied parts (a few hundred parts each at most). */
+const COPIED_PARTS_BATCH = 2000
 
 export default class GcScheduler extends BaseCommand {
   public static commandName = 'gc:scheduler'
@@ -173,21 +175,32 @@ export default class GcScheduler extends BaseCommand {
   private singleCursor = 0
   private releaseCursor = 0
 
-  /** Parts of header-derived binaries that are a file already (worker:file now drops them itself). */
+  /**
+   * Header parts of binaries that are a file already. worker:file drops them
+   * when it creates the file, and ingest no longer adds them afterwards; this
+   * catches what slipped through. A cursor over binaries (as for single
+   * articles) visits each one once per pass, instead of searching all of
+   * them for parts on every batch (LOG.md 2026-10-04).
+   */
+  private copiedCursor = 0
   private async dropCopiedParts(budget: number): Promise<number> {
     let total = 0
     while (Date.now() < budget) {
-      const result = await db.rawQuery(
-        // Header-derived parts (no per-part CRC) of binaries that are a file now.
-        `DELETE FROM parts WHERE pcrc32 IS NULL AND binary_id IN (
-           SELECT b.id FROM binaries b
-           WHERE b.file_id IS NOT NULL
-             AND EXISTS (SELECT 1 FROM parts p WHERE p.binary_id = b.id AND p.pcrc32 IS NULL)
-           LIMIT ?)`,
-        [500]
+      const { rows } = await db.rawQuery(
+        `WITH batch AS (
+           SELECT id FROM binaries WHERE id > ? AND file_id IS NOT NULL ORDER BY id LIMIT ?
+         ), gone AS (
+           DELETE FROM parts p USING batch WHERE p.binary_id = batch.id AND p.pcrc32 IS NULL
+           RETURNING 1)
+         SELECT (SELECT max(id) FROM batch) AS last, (SELECT count(*) FROM gone) AS deleted`,
+        [this.copiedCursor, COPIED_PARTS_BATCH]
       )
-      total += result.rowCount ?? 0
-      if (!result.rowCount) break
+      total += Number(rows[0].deleted)
+      if (rows[0].last === null) {
+        this.copiedCursor = 0
+        break
+      }
+      this.copiedCursor = Number(rows[0].last)
     }
     return total
   }

tests/functional/ingest.spec.ts

@@ -4,6 +4,7 @@ import testUtils from '@adonisjs/core/services/test_utils'
 import IngestService, { type IngestSegment } from '#services/IngestService'
 import QueueService from '#services/QueueService'
 import { fileKey } from '#lib/subject/index'
+import FileStore from '#services/FileStore'
 
 function segment(overrides: Partial<IngestSegment>): IngestSegment {
   const poster = 'ingest-test@example.invalid'
@@ -51,6 +52,25 @@ test.group('IngestService', (group) => {
     assert.isAbove(+after.last_seen_at, Date.now() - 60_000)
   })
 
+  test('a binary that is a file already takes no new header parts', async ({ assert }) => {
+    await IngestService.ingest([segment({ number: 1 })])
+    const binary = await binaryFor(segment({}).fileKey)
+    const now = new Date()
+    const [file] = await FileStore.insert([
+      { filename: 'test-file.bin', poster: 'x', date: now.getTime(), parts: 3, size: 300, groups: '[]', segments: [], created_at: now, updated_at: now },
+    ])
+    await db.from('binaries').where('id', binary.id).update({ file_id: file.id })
+    await db.from('parts').where('binary_id', binary.id).delete()
+
+    // The same articles again (a crosspost, a backfill range): nothing to store.
+    await IngestService.ingest([segment({ number: 2, messageId: 'part2@test' }), segment({ number: 3, messageId: 'part3@test' })])
+    assert.equal(await db.from('parts').where('binary_id', binary.id).count('* as n').then((r) => Number(r[0].n)), 0)
+
+    // Body-derived parts (with a per-part CRC) are still kept.
+    await IngestService.ingest([segment({ number: 2, messageId: 'part2@test', pcrc32: 'abcd1234' })])
+    assert.equal(await db.from('parts').where('binary_id', binary.id).count('* as n').then((r) => Number(r[0].n)), 1)
+  })
+
   test('ingesting the same segments twice is a no-op', async ({ assert }) => {
     const segments = [segment({ number: 1 }), segment({ number: 2, messageId: 'part2@test' })]
     await IngestService.ingest(segments)