From 2178847513081500a0ce58634a26e0fa3d0afbbd Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Sun, 4 Oct 2026 02:53:11 +0200 Subject: [PATCH] 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 --- app/services/IngestService.ts | 19 +++++++++++++++--- commands/GcScheduler.ts | 35 ++++++++++++++++++++++----------- tests/functional/ingest.spec.ts | 20 +++++++++++++++++++ 3 files changed, 60 insertions(+), 14 deletions(-) diff --git a/app/services/IngestService.ts b/app/services/IngestService.ts index d8e17aa..77941ba 100644 --- a/app/services/IngestService.ts +++ b/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() + 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() } diff --git a/commands/GcScheduler.ts b/commands/GcScheduler.ts index 55a263e..1d9577c 100644 --- a/commands/GcScheduler.ts +++ b/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 { 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 } diff --git a/tests/functional/ingest.spec.ts b/tests/functional/ingest.spec.ts index c7fccf3..9c76198 100644 --- a/tests/functional/ingest.spec.ts +++ b/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 { 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)