From b2cc75fdd006188881437423b3dac711d130df29 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Tue, 6 Oct 2026 20:03:43 +0200 Subject: [PATCH] Keep binaries until inspected; seeding writes binaries only for inspection Par2 naming needs the fingerprint inspection stores on a binary, and the GC deleted seeded binaries (6 h unfingerprinted, 48 h stale) long before the inspection queue reached them. binaries.awaiting_inspection protects them for up to 7 days; the GC deletes use partial indexes so they don't read past waiting rows. Seeding no longer creates binaries for files that aren't inspected (~85% of a.b.ath), which only churned the table. Co-Authored-By: Claude Opus 5.5 --- app/services/BinaryFileService.ts | 9 ++-- app/services/ContentService.ts | 2 + app/services/SeedService.ts | 43 ++++++++++++------- commands/GcScheduler.ts | 31 ++++++++----- config/gc.ts | 5 +++ ...8620765230_binaries_awaiting_inspection.ts | 20 +++++++++ .../1768620765240_binaries_gc_indexes.ts | 32 ++++++++++++++ database/schema.ts | 4 +- start/env.ts | 1 + tests/functional/binary_files.spec.ts | 18 ++++++++ tests/functional/seed.spec.ts | 19 ++++++-- 11 files changed, 149 insertions(+), 35 deletions(-) create mode 100644 database/migrations/1768620765230_binaries_awaiting_inspection.ts create mode 100644 database/migrations/1768620765240_binaries_gc_indexes.ts diff --git a/app/services/BinaryFileService.ts b/app/services/BinaryFileService.ts index 74b748d..13d4f15 100644 --- a/app/services/BinaryFileService.ts +++ b/app/services/BinaryFileService.ts @@ -99,6 +99,7 @@ class BinaryFileService { const id = insertedByKey.get(b.file_key) ?? byKey.get(b.file_key) if (id) fileIdOf.set(Number(b.id), id) } + const fresher = new Map(fresh.map((b) => [b.file_key, b])) result.saved = inserted.length result.existing = fileIdOf.size - inserted.length @@ -110,9 +111,12 @@ class BinaryFileService { `SELECT id FROM binaries WHERE id IN (SELECT unnest(?::bigint[])) ORDER BY file_key FOR UPDATE`, [linked.map((b) => b.id)] ) + // Files queued for inspection below keep their binary until it has its fingerprint (par2 naming, D16). + const queued = new Set(inserted.filter((row) => row.file_key && ContentService.isCandidate(String(fresher.get(row.file_key)?.name ?? ''))).map((row) => row.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))!)] + `UPDATE binaries b SET file_id = v.file_id, awaiting_inspection = CASE WHEN v.queued THEN true ELSE b.awaiting_inspection END + FROM unnest(?::bigint[], ?::bigint[], ?::boolean[]) AS v(id, file_id, queued) WHERE b.id = v.id`, + [linked.map((b) => b.id), linked.map((b) => fileIdOf.get(Number(b.id))!), linked.map((b) => queued.has(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). @@ -120,7 +124,6 @@ class BinaryFileService { } // 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 diff --git a/app/services/ContentService.ts b/app/services/ContentService.ts index 747ace2..05c8171 100644 --- a/app/services/ContentService.ts +++ b/app/services/ContentService.ts @@ -165,6 +165,8 @@ class ContentService { .update({ size: db.raw('coalesce(size, ?)', [probe.size]), md5_16k: db.raw('coalesce(md5_16k, ?)', [probe.md5_16k]), + // Inspected: the GC may delete it again by its usual rules (LOG.md 2026-10-06 (11)). + awaiting_inspection: null, }) } diff --git a/app/services/SeedService.ts b/app/services/SeedService.ts index 38c6a57..2733c67 100644 --- a/app/services/SeedService.ts +++ b/app/services/SeedService.ts @@ -62,27 +62,38 @@ class SeedService { 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 = files.map((file, i) => ({ - file_key: file.fileKey, - poster: file.poster, - subject: file.subject, - name: file.name, - total_parts: file.totalParts, - groups: rows[i].groups, - posted_at: postedOf(file) ?? null, - first_seen_at: now, - last_seen_at: now, - completed_at: now, - release_id: rows[i].release_id, - file_id: fileIds.get(file.fileKey) ?? null, - })) + // A completed binary, as worker:file would have left behind, only for files queued for content + // inspection: par2 naming (D16) needs the fingerprint inspection stores on it. For the others + // (~85% in a.b.ath) it was only GC churn: 1.6M rows an hour inserted and deleted (LOG.md 2026-10-06 (11)). + const insertedKeys = new Set(inserted.map((row) => row.file_key)) + const binaries = files.flatMap((file, i) => { + if (!ContentService.isCandidate(file.name)) return [] + return [{ + file_key: file.fileKey, + poster: file.poster, + subject: file.subject, + name: file.name, + total_parts: file.totalParts, + groups: rows[i].groups, + posted_at: postedOf(file) ?? null, + first_seen_at: now, + last_seen_at: now, + completed_at: now, + release_id: rows[i].release_id, + file_id: fileIds.get(file.fileKey) ?? null, + // New files are queued for inspection below; an existing file was queued when it was added. + awaiting_inspection: insertedKeys.has(file.fileKey) ? true : null, + }] + }) binaries.sort((a, b) => a.file_key.localeCompare(b.file_key)) for (let i = 0; i < binaries.length; i += insertLimits.rows) { await knex('binaries') .insert(binaries.slice(i, i + insertLimits.rows)) .onConflict('file_key') - .merge({ file_id: knex.raw('coalesce(binaries.file_id, excluded.file_id)') as any }) + .merge({ + file_id: knex.raw('coalesce(binaries.file_id, excluded.file_id)') as any, + awaiting_inspection: knex.raw('coalesce(binaries.awaiting_inspection, excluded.awaiting_inspection)') as any, + }) } // As worker:file does for a new file: an SFV posted earlier may list it (D34), then inspection. diff --git a/commands/GcScheduler.ts b/commands/GcScheduler.ts index 739b35d..e6d741f 100644 --- a/commands/GcScheduler.ts +++ b/commands/GcScheduler.ts @@ -178,19 +178,27 @@ export default class GcScheduler extends BaseCommand { * already live on in `files`; conflicted ones will never become a file. * Completed ones still waiting for worker:file are kept. */ + /** + * Binaries without new segments for staleAfterSeconds: incomplete ones are given up on, + * finished ones have their file. A finished one whose file waits for content inspection + * stays until it's inspected (its fingerprint names it from par2 sets, D16), at most + * awaitInspectionSeconds (LOG.md 2026-10-06 (11)). Three deletes, one per partial index, + * so none reads past millions of waiting rows. + */ private async deleteStaleBinaries(): Promise { - const query = db - .from('binaries') - .where('last_seen_at', '<', new Date(Date.now() - gcConfig.binaries.staleAfterSeconds * 1000)) - .where((q) => - q.whereNull('completed_at').orWhereNotNull('file_id').orWhereNotNull('conflicted_at') - ) - - if (this.dryRun) { - const [{ count }] = await query.count('* as count') - return Number(count) + const stale = new Date(Date.now() - gcConfig.binaries.staleAfterSeconds * 1000) + const waited = new Date(Date.now() - gcConfig.binaries.awaitInspectionSeconds * 1000) + const deletes = [ + db.from('binaries').where('last_seen_at', '<', stale).whereNotNull('file_id').whereRaw('awaiting_inspection IS NOT TRUE'), + db.from('binaries').where('last_seen_at', '<', stale).where((q) => q.whereNull('completed_at').orWhereNotNull('conflicted_at')), + db.from('binaries').where('last_seen_at', '<', waited).whereRaw('awaiting_inspection IS TRUE'), + ] + let total = 0 + for (const query of deletes) { + if (this.dryRun) total += Number((await query.count('* as count'))[0].count) + else total += Number(await query.delete()) } - return Number(await query.delete()) + return total } private async deleteOldDeferredFiles(): Promise { @@ -283,6 +291,7 @@ export default class GcScheduler extends BaseCommand { SELECT id FROM binaries WHERE last_seen_at < ? AND file_id IS NOT NULL AND completed_at IS NOT NULL AND size IS NULL AND md5_16k IS NULL AND conflicted_at IS NULL + AND awaiting_inspection IS NOT TRUE ORDER BY last_seen_at LIMIT ?)`, [cutoff, gcConfig.deleteBatch] ) diff --git a/config/gc.ts b/config/gc.ts index c339d5d..eb17ca3 100644 --- a/config/gc.ts +++ b/config/gc.ts @@ -46,6 +46,11 @@ export default { * later, so they go sooner than staleAfterSeconds. */ unfingerprintedAfterSeconds: env.get('GC_UNFINGERPRINTED_SECONDS', 6 * 60 * 60), + /** + * A finished binary whose file waits for content inspection is kept this long at + * most (it needs the fingerprint inspection stores for par2 naming, D16). + */ + awaitInspectionSeconds: env.get('GC_AWAIT_INSPECTION_SECONDS', 7 * 24 * 60 * 60), }, /** diff --git a/database/migrations/1768620765230_binaries_awaiting_inspection.ts b/database/migrations/1768620765230_binaries_awaiting_inspection.ts new file mode 100644 index 0000000..1ccc38c --- /dev/null +++ b/database/migrations/1768620765230_binaries_awaiting_inspection.ts @@ -0,0 +1,20 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * A completed binary whose file is queued for content inspection keeps its + * row until inspection has stored the file's fingerprint on it (LOG.md + * 2026-10-06 (11)): par2 naming (D16) matches par2 listings against binary + * fingerprints and attaches sets to binaries. With the inspection queue + * days behind, the GC deleted them first (after 6 hours unfingerprinted, + * 48 hours stale). TRUE while waiting; cleared by inspection. + */ +export default class extends BaseSchema { + async up() { + await this.db.rawQuery(`SET LOCAL lock_timeout = '2min'`) + await this.db.rawQuery(`ALTER TABLE binaries ADD COLUMN IF NOT EXISTS awaiting_inspection boolean`) + } + + async down() { + await this.db.rawQuery(`ALTER TABLE binaries DROP COLUMN IF EXISTS awaiting_inspection`) + } +} diff --git a/database/migrations/1768620765240_binaries_gc_indexes.ts b/database/migrations/1768620765240_binaries_gc_indexes.ts new file mode 100644 index 0000000..bae06f1 --- /dev/null +++ b/database/migrations/1768620765240_binaries_gc_indexes.ts @@ -0,0 +1,32 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Indexes for the GC's deletes of finished binaries, without the ones + * waiting for inspection (LOG.md 2026-10-06 (11)): those can be millions + * when inspection is behind, and the deletes must not read past them every + * sweep. Built CONCURRENTLY: binaries takes writes all the time. + */ +export default class extends BaseSchema { + static disableTransactions = true + + async up() { + await this.db.rawQuery( + `CREATE INDEX CONCURRENTLY IF NOT EXISTS binaries_done_last_seen_idx ON binaries (last_seen_at) + WHERE file_id IS NOT NULL AND awaiting_inspection IS NOT TRUE` + ) + await this.db.rawQuery( + `CREATE INDEX CONCURRENTLY IF NOT EXISTS binaries_awaiting_inspection_idx ON binaries (last_seen_at) + WHERE awaiting_inspection IS TRUE` + ) + await this.db.rawQuery( + `CREATE INDEX CONCURRENTLY IF NOT EXISTS binaries_open_last_seen_idx ON binaries (last_seen_at) + WHERE completed_at IS NULL OR conflicted_at IS NOT NULL` + ) + } + + async down() { + await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS binaries_open_last_seen_idx`) + await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS binaries_awaiting_inspection_idx`) + await this.db.rawQuery(`DROP INDEX CONCURRENTLY IF EXISTS binaries_done_last_seen_idx`) + } +} diff --git a/database/schema.ts b/database/schema.ts index c5ee683..cdcb623 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -33,8 +33,10 @@ export class AuthAccessTokenSchema extends BaseModel { } export class BinarySchema extends BaseModel { - static $columns = ['completedAt', 'conflictedAt', 'fileId', 'fileKey', 'firstSeenAt', 'groups', 'id', 'lastSeenAt', 'md516K', 'name', 'parentBinaryId', 'partCount', 'postedAt', 'poster', 'realName', 'releaseId', 'size', 'splitAt', 'subject', 'totalParts'] as const + static $columns = ['awaitingInspection', 'completedAt', 'conflictedAt', 'fileId', 'fileKey', 'firstSeenAt', 'groups', 'id', 'lastSeenAt', 'md516K', 'name', 'parentBinaryId', 'partCount', 'postedAt', 'poster', 'realName', 'releaseId', 'size', 'splitAt', 'subject', 'totalParts'] as const $columns = BinarySchema.$columns + @column() + declare awaitingInspection: boolean | null @column.dateTime() declare completedAt: DateTime | null @column.dateTime() diff --git a/start/env.ts b/start/env.ts index 5ebf0f7..9b45a99 100644 --- a/start/env.ts +++ b/start/env.ts @@ -105,6 +105,7 @@ export default await Env.create(new URL('../', import.meta.url), { ARCHIVE_WALK_CONCURRENCY: Env.schema.number.optional(), RELEASE_COMPLETE_INTERVAL_MS: Env.schema.number.optional(), FILE_KEYS_FILL_SECONDS: Env.schema.number.optional(), + GC_AWAIT_INSPECTION_SECONDS: Env.schema.number.optional(), ARCHIVE_WALK_MAX_SEGMENTS: Env.schema.number.optional(), DEFERRED_SETTLE_MS: Env.schema.number.optional(), DEFERRED_EXPAND_MAX_PARTS: Env.schema.number.optional(), diff --git a/tests/functional/binary_files.spec.ts b/tests/functional/binary_files.spec.ts index 105fe8d..797e4bf 100644 --- a/tests/functional/binary_files.spec.ts +++ b/tests/functional/binary_files.spec.ts @@ -6,6 +6,7 @@ import BinaryFileService from '#services/BinaryFileService' import QueueService from '#services/QueueService' import FileStore, { segmentsOf } from '#services/FileStore' import { fileKey } from '#lib/subject/index' +import ContentService from '#services/ContentService' function segment(name: string, number: number, total = 3): IngestSegment { const poster = 'batch-test@example.invalid' @@ -65,4 +66,21 @@ test.group('Binaries to files', (group) => { assert.include(again, { saved: 0, skipped: 2 }) assert.deepEqual(again.incomplete, [ids[2]]) }) + + test('a queued obfuscated file keeps its binary until inspection fingerprints it', async ({ assert }) => { + const name = 'Zk3m9QpL2xW8rT5vN1bYq.bin' + await IngestService.ingest([1, 2, 3].map((n) => segment(name, n))) + const b = await binary(name) + const result = await BinaryFileService.store([Number(b.id)]) + assert.include(result, { saved: 1, inspections: 1 }) + const waiting = await binary(name) + assert.isTrue(waiting.awaiting_inspection) + + // Inspection stores the fingerprint and clears the flag. + const task = { kind: 'inspect' as const, fileId: Number(waiting.file_id), name, segments: [], sevenZip: null } + await ContentService.apply(task, { size: 300, contentType: null, md5_16k: 'a'.repeat(32), yencName: name }) + const inspected = await binary(name) + assert.isNull(inspected.awaiting_inspection) + assert.equal(inspected.md5_16k, 'a'.repeat(32)) + }) }) diff --git a/tests/functional/seed.spec.ts b/tests/functional/seed.spec.ts index 934fb6d..5dcdb04 100644 --- a/tests/functional/seed.spec.ts +++ b/tests/functional/seed.spec.ts @@ -37,15 +37,26 @@ test.group('SeedService', (group) => { 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) + // A readable name isn't inspected: no binary (LOG.md 2026-10-06 (11)). + assert.isNull(await db.from('binaries').where('file_key', stored.file_key).first()) // The same files again (a crosspost, a second run): nothing new. const again = await SeedService.store(files) assert.include(again, { files: 0, existing: 2 }) }) + test('an obfuscated file keeps a binary until inspection has fingerprinted it', async ({ assert }) => { + const aggregator = new FileAggregator() + const subject = 'Xq8r2KdLm3Zw91PqA7zz [1/1] - "Xq8r2KdLm3Zw91PqA7zz.bin" yEnc (%d/2)' + const segments = [1, 2].map((n) => segmentFromHeader(header(subject.replace('%d', String(n)), n), 'alt.binaries.test', n)!) + const files = aggregator.add(segments).map(toSeedFile) + const result = await SeedService.store(files) + assert.include(result, { files: 1, inspections: 1 }) + const binary = await db.from('binaries').where('file_key', files[0].fileKey).first() + assert.isTrue(binary.awaiting_inspection) + assert.isNotNull(binary.file_id) + }) + test('stores a batch in several INSERTs', async ({ assert }) => { // Postgres takes 65,535 bind parameters per statement: ~5,400 rows of 12 columns. const rows = insertLimits.rows @@ -57,7 +68,7 @@ test.group('SeedService', (group) => { ) const files = aggregator.add(segments).map(toSeedFile) assert.include(await SeedService.store(files), { files: 120 }) - const [{ count }] = await db.from('binaries').whereLike('name', 'dense.%').whereNotNull('file_id').count('* as count') + const [{ count }] = await db.from('files').whereLike('filename', 'dense.%').count('* as count') assert.equal(Number(count), 120) } finally { insertLimits.rows = rows