From be2112f4ad392dcada65c9f4e182d09a1d93bbb0 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Tue, 6 Oct 2026 13:48:59 +0200 Subject: [PATCH] Par2 size matching 100x faster; release bookkeeping once per statement matchPar2Releases checked every same-size release_files row's set and date (18 s per batch of body-derived binaries); now an index on release_files (release_id, size) drops the repeated sizes first: 0.16 s. File inserts updated their release in four row triggers; one statement trigger now updates each release once per INSERT. Completeness is noted in release_dirty and recomputed by gc:scheduler every 30 minutes. Co-Authored-By: Claude Opus 5.5 --- app/services/IngestService.ts | 27 +++-- app/services/ReleaseCompleteness.ts | 35 ++++++ commands/GcScheduler.ts | 17 +++ config/gc.ts | 3 + ...8620765190_release_totals_per_statement.ts | 107 ++++++++++++++++++ database/schema.ts | 7 ++ start/env.ts | 1 + tests/functional/release_complete.spec.ts | 14 +++ tests/functional/web_api.spec.ts | 8 +- 9 files changed, 208 insertions(+), 11 deletions(-) create mode 100644 app/services/ReleaseCompleteness.ts create mode 100644 database/migrations/1768620765190_release_totals_per_statement.ts diff --git a/app/services/IngestService.ts b/app/services/IngestService.ts index 2eef3b7..8f65053 100644 --- a/app/services/IngestService.ts +++ b/app/services/IngestService.ts @@ -272,18 +272,27 @@ class IngestService { */ async matchPar2Releases(binaryIds: number[]) { if (binaryIds.length === 0) return + // Listed-once first (index-only on release_files (release_id, size): most candidate + // sizes are a set's equal-sized volumes), the release's date only for what's left; + // OFFSET 0 keeps that order. Was: every size match's date and count, 18 s per batch. await db.rawQuery( `UPDATE binaries b SET release_id = m.release_id FROM ( - SELECT b2.id, min(rf.release_id) AS release_id - FROM binaries b2 - JOIN release_files rf ON rf.size = b2.size - JOIN releases r ON r.id = rf.release_id - WHERE b2.id = ANY(?) AND b2.release_id IS NULL AND b2.md5_16k IS NULL AND b2.size >= ? - AND abs(extract(epoch FROM (r.posted_at - b2.posted_at))) < ? - AND (SELECT count(*) FROM release_files x WHERE x.release_id = rf.release_id AND x.size = rf.size) = 1 - GROUP BY b2.id - HAVING count(DISTINCT rf.release_id) = 1 + SELECT c.id, min(c.release_id) AS release_id + FROM ( + SELECT b2.id, b2.posted_at, rf.release_id + FROM binaries b2 + JOIN release_files rf ON rf.size = b2.size + WHERE b2.id = ANY(?) AND b2.release_id IS NULL AND b2.md5_16k IS NULL AND b2.size >= ? + AND NOT EXISTS ( + SELECT 1 FROM release_files x + WHERE x.release_id = rf.release_id AND x.size = rf.size AND x.par2_file_id <> rf.par2_file_id) + OFFSET 0 + ) c + JOIN releases r ON r.id = c.release_id + WHERE abs(extract(epoch FROM (r.posted_at - c.posted_at))) < ? + GROUP BY c.id + HAVING count(DISTINCT c.release_id) = 1 ) m WHERE b.id = m.id`, [binaryIds, PAR2_SIZE_ONLY_MIN, PAR2_MATCH_WINDOW_SECONDS] diff --git a/app/services/ReleaseCompleteness.ts b/app/services/ReleaseCompleteness.ts new file mode 100644 index 0000000..56fd44a --- /dev/null +++ b/app/services/ReleaseCompleteness.ts @@ -0,0 +1,35 @@ +import db from '@adonisjs/lucid/services/db' + +/** Noted releases taken per transaction. */ +const BATCH = 2000 + +/** + * Release completeness (D24) for releases whose files or par2 listing + * changed: triggers only note them in release_dirty (a release can be noted + * thousands of times between sweeps), and gc:scheduler recomputes each + * noted release once every RELEASE_COMPLETE_INTERVAL_MS (LOG.md 2026-10-06 (6)). + */ +class ReleaseCompleteness { + /** Recomputes all noted releases; returns how many distinct releases were checked. */ + async sweep(): Promise { + let checked = 0 + for (;;) { + const done = await db.transaction(async (trx) => { + // Taken and recomputed in one transaction: a failure leaves the notes for the next sweep. + const { rows } = await trx.rawQuery( + `DELETE FROM release_dirty WHERE ctid = ANY(ARRAY(SELECT ctid FROM release_dirty LIMIT ?)) RETURNING release_id`, + [BATCH] + ) + const ids: number[] = [...new Set(rows.map((row: { release_id: string }) => Number(row.release_id)))] + if (ids.length > 0) { + await trx.rawQuery(`SELECT recompute_release_complete(id) FROM unnest(?::bigint[]) AS id`, [ids]) + } + checked += ids.length + return rows.length < BATCH + }) + if (done) return checked + } + } +} + +export default new ReleaseCompleteness() diff --git a/commands/GcScheduler.ts b/commands/GcScheduler.ts index bd4b082..eef6a5d 100644 --- a/commands/GcScheduler.ts +++ b/commands/GcScheduler.ts @@ -9,6 +9,7 @@ import IndexMaintenance from '#services/IndexMaintenance' import ReleaseMergeService from '#services/ReleaseMergeService' import StagingService from '#services/StagingService' import FilePartitions from '#services/FilePartitions' +import ReleaseCompleteness from '#services/ReleaseCompleteness' import Shutdown from '#services/Shutdown' const CLEAN_BATCH = 1000 @@ -30,6 +31,9 @@ export default class GcScheduler extends BaseCommand { @flags.boolean({ description: 'Report what would be removed without changing anything' }) declare dryRun: boolean + /** When release completeness was last recomputed (it runs less often than the sweep). */ + private completenessAt = 0 + public async run() { this.logger.info( `GC started (interval ${gcConfig.intervalMs / 1000}s${this.dryRun ? ', dry run' : ''}).` @@ -42,6 +46,11 @@ export default class GcScheduler extends BaseCommand { } catch (err: any) { this.logger.error(`GC sweep failed: ${err.message}`) } + try { + await this.releaseCompleteness() + } catch (err: any) { + this.logger.error(`Release completeness failed: ${err.message}`) + } if (!this.once) await Shutdown.sleep(gcConfig.intervalMs) } while (!this.once && !Shutdown.requested) } finally { @@ -108,6 +117,14 @@ export default class GcScheduler extends BaseCommand { } /** Rebuilds bloated indexes, at most every gcConfig.reindexIntervalMs (CONCEPT.md D33). */ + private async releaseCompleteness() { + if (this.dryRun || (!this.once && Date.now() - this.completenessAt < gcConfig.releaseCompleteIntervalMs)) return + this.completenessAt = Date.now() + const started = Date.now() + const checked = await ReleaseCompleteness.sweep() + if (checked > 0) this.logger.info(`Release completeness: ${checked} changed releases checked in ${Math.round((Date.now() - started) / 1000)} s.`) + } + private async rebuildIndexes() { const key = 'gc:reindex:last' const last = Number((await RedisService.client.get(key)) ?? 0) diff --git a/config/gc.ts b/config/gc.ts index e247bcb..de3bfa5 100644 --- a/config/gc.ts +++ b/config/gc.ts @@ -7,6 +7,9 @@ export default { /** How often `gc:scheduler` sweeps. */ intervalMs: env.get('GC_INTERVAL_MS', 3 * 60 * 1000), + /** How often gc:scheduler recomputes the completeness of releases that changed (LOG.md 2026-10-06 (6)). */ + releaseCompleteIntervalMs: env.get('RELEASE_COMPLETE_INTERVAL_MS', 30 * 60 * 1000), + jobs: { /** Completed jobs are only kept for inspection; keep them briefly. */ completedAgeSeconds: env.get('GC_COMPLETED_AGE_SECONDS', 60 * 60), diff --git a/database/migrations/1768620765190_release_totals_per_statement.ts b/database/migrations/1768620765190_release_totals_per_statement.ts new file mode 100644 index 0000000..62a7a73 --- /dev/null +++ b/database/migrations/1768620765190_release_totals_per_statement.ts @@ -0,0 +1,107 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Cheaper release bookkeeping for file inserts (LOG.md 2026-10-06 (6)). + * + * Every inserted file fired four row triggers, each updating its release: + * file_count/size, search_files, extensions and completeness, so a seed + * batch of thousands of files rewrote its releases thousands of times (a + * quarter of Postgres' active time on homelab). Now one statement-level + * trigger updates each release once per INSERT, with all four. + * + * Completeness (D24), the expensive part, is no longer computed in triggers: + * refresh_release_complete() only notes the release in release_dirty, and + * gc:scheduler recomputes the noted releases every 30 minutes + * (recompute_release_complete). Counts, sizes, search text and extensions + * stay current: code decides on them right away (weak releases, deleting + * empty releases). + * + * Also an index on release_files (release_id, size) for the par2 size + * match's "listed once in its set" check (IngestService.matchPar2Releases). + */ +export default class extends BaseSchema { + async up() { + const q = (sql: string) => this.db.rawQuery(sql) + // Swapping triggers locks files and its partitions: give up rather than queue every writer behind it. + await q(`SET LOCAL lock_timeout = '2min'`) + // Built CONCURRENTLY by hand on homelab (2026-10-06); IF NOT EXISTS keeps that one. + await q(`CREATE INDEX IF NOT EXISTS release_files_release_size_idx ON release_files (release_id, size)`) + + // Append-only (no key): writers never wait on each other or on the sweep. + await q(`CREATE TABLE IF NOT EXISTS release_dirty (release_id bigint NOT NULL)`) + await q(` + CREATE OR REPLACE FUNCTION recompute_release_complete(rid bigint) RETURNS void LANGUAGE sql AS $$ + UPDATE releases r SET complete = c.complete + FROM (SELECT release_complete(rid) AS complete) c + WHERE r.id = rid AND r.complete IS DISTINCT FROM c.complete + $$`) + // Every trigger that called this (files, release_files, binaries.size, releases.file_total) now just notes the release. + await q(` + CREATE OR REPLACE FUNCTION refresh_release_complete(rid bigint) RETURNS void LANGUAGE sql AS $$ + INSERT INTO release_dirty (release_id) SELECT rid WHERE rid IS NOT NULL + $$`) + + await q(` + CREATE OR REPLACE FUNCTION files_release_insert_statement() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + WITH added AS ( + SELECT release_id, count(*) AS n, coalesce(sum(size), 0) AS bytes, + array_agg(DISTINCT e) FILTER (WHERE length(e) BETWEEN 2 AND 4) AS exts, + array_agg(DISTINCT w) FILTER (WHERE w <> '') AS words + FROM ( + SELECT release_id, size, name_extension(nm) AS e, + CASE WHEN is_readable_name(nm) THEN search_normalize(name_stem(nm)) END AS w + FROM (SELECT release_id, size, coalesce(real_name, filename) AS nm FROM new_rows WHERE release_id IS NOT NULL) f + ) x + GROUP BY release_id + ) + UPDATE releases r SET + file_count = r.file_count + a.n, + size = r.size + a.bytes, + extensions = CASE + WHEN a.exts IS NULL OR coalesce(r.extensions, '{}') @> a.exts THEN r.extensions + ELSE array(SELECT DISTINCT x FROM unnest(coalesce(r.extensions, '{}') || a.exts) x ORDER BY x) END, + search_files = CASE + WHEN a.words IS NULL OR length(coalesce(r.search_files, '')) >= 4000 THEN r.search_files + ELSE nullif(left(concat_ws(' ', r.search_files, ( + SELECT string_agg(w, ' ') FROM unnest(a.words) w + WHERE position(w IN coalesce(r.search_text, search_normalize(r.name))) = 0)), 4000), '') END + FROM added a + WHERE r.id = a.release_id; + + INSERT INTO release_dirty (release_id) SELECT DISTINCT release_id FROM new_rows WHERE release_id IS NOT NULL; + RETURN NULL; + END $$`) + + // The row triggers keep UPDATE and DELETE (renames, moves between releases); INSERT is the statement trigger. + for (const name of ['files_release_extension', 'files_release_search', 'files_release_totals', 'files_zz_release_complete']) { + await q(`DROP TRIGGER IF EXISTS ${name} ON files`) + } + await q(`CREATE TRIGGER files_release_extension AFTER UPDATE OF release_id, filename, real_name ON files FOR EACH ROW EXECUTE FUNCTION files_release_extension()`) + await q(`CREATE TRIGGER files_release_search AFTER UPDATE OF release_id, filename, real_name ON files FOR EACH ROW EXECUTE FUNCTION files_release_search()`) + await q(`CREATE TRIGGER files_release_totals AFTER DELETE OR UPDATE OF release_id, size ON files FOR EACH ROW EXECUTE FUNCTION files_release_totals()`) + await q(`CREATE TRIGGER files_zz_release_complete AFTER DELETE OR UPDATE OF release_id, size, real_name ON files FOR EACH ROW EXECUTE FUNCTION files_release_complete()`) + await q(`CREATE TRIGGER files_release_insert AFTER INSERT ON files REFERENCING NEW TABLE AS new_rows FOR EACH STATEMENT EXECUTE FUNCTION files_release_insert_statement()`) + } + + async down() { + const q = (sql: string) => this.db.rawQuery(sql) + await q(`DROP TRIGGER IF EXISTS files_release_insert ON files`) + for (const name of ['files_release_extension', 'files_release_search', 'files_release_totals', 'files_zz_release_complete']) { + await q(`DROP TRIGGER IF EXISTS ${name} ON files`) + } + await q(`CREATE TRIGGER files_release_extension AFTER INSERT OR UPDATE OF release_id, filename, real_name ON files FOR EACH ROW EXECUTE FUNCTION files_release_extension()`) + await q(`CREATE TRIGGER files_release_search AFTER INSERT OR UPDATE OF release_id, filename, real_name ON files FOR EACH ROW EXECUTE FUNCTION files_release_search()`) + await q(`CREATE TRIGGER files_release_totals AFTER INSERT OR DELETE OR UPDATE OF release_id, size ON files FOR EACH ROW EXECUTE FUNCTION files_release_totals()`) + await q(`CREATE TRIGGER files_zz_release_complete AFTER INSERT OR DELETE OR UPDATE OF release_id, size, real_name ON files FOR EACH ROW EXECUTE FUNCTION files_release_complete()`) + await q(`DROP FUNCTION IF EXISTS files_release_insert_statement()`) + await q(` + CREATE OR REPLACE FUNCTION refresh_release_complete(rid bigint) RETURNS void LANGUAGE sql AS $$ + UPDATE releases r SET complete = c.complete + FROM (SELECT release_complete(rid) AS complete) c + WHERE r.id = rid AND r.complete IS DISTINCT FROM c.complete + $$`) + await q(`DROP FUNCTION IF EXISTS recompute_release_complete(bigint)`) + await q(`DROP TABLE IF EXISTS release_dirty`) + } +} diff --git a/database/schema.ts b/database/schema.ts index f196bfb..26d22f6 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -391,6 +391,13 @@ export class PartSchema extends BaseModel { declare pcrc32: string | null } +export class ReleaseDirtySchema extends BaseModel { + static $columns = ['releaseId'] as const + $columns = ReleaseDirtySchema.$columns + @column() + declare releaseId: bigint | number +} + export class ReleaseFileSchema extends BaseModel { static $columns = ['md5', 'md516K', 'name', 'par2FileId', 'releaseId', 'setId', 'size'] as const $columns = ReleaseFileSchema.$columns diff --git a/start/env.ts b/start/env.ts index c3033a2..7332df4 100644 --- a/start/env.ts +++ b/start/env.ts @@ -102,6 +102,7 @@ export default await Env.create(new URL('../', import.meta.url), { BODY_CONCURRENCY: Env.schema.number.optional(), COLLECTION_CONCURRENCY: Env.schema.number.optional(), ARCHIVE_WALK_CONCURRENCY: Env.schema.number.optional(), + RELEASE_COMPLETE_INTERVAL_MS: 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/release_complete.spec.ts b/tests/functional/release_complete.spec.ts index 6500265..eb2a882 100644 --- a/tests/functional/release_complete.spec.ts +++ b/tests/functional/release_complete.spec.ts @@ -1,6 +1,7 @@ import { test } from '@japa/runner' import db from '@adonisjs/lucid/services/db' import testUtils from '@adonisjs/core/services/test_utils' +import ReleaseCompleteness from '#services/ReleaseCompleteness' /** A release with `n` files (each with a binary of size 100000 + i), named f.bin. */ async function release(n: number) { @@ -31,12 +32,24 @@ async function listing(releaseId: number, count: number, offset = 0) { SELECT ?::bigint, md5(i::text), 'f' || i || '.bin', 100000 + i, md5(i::text), 'set' FROM generate_series(1 + ?::int, ?::int + ?::int) i`, [releaseId, offset, count, offset] ) + // Triggers only note the release; gc:scheduler's sweep recomputes it. + await ReleaseCompleteness.sweep() return Boolean((await db.from('releases').where('id', releaseId).first()).complete) } test.group('release completeness', (group) => { group.each.setup(() => testUtils.db().withGlobalTransaction()) + test('file inserts keep counts, size, extensions and search text current, once per statement', async ({ assert }) => { + const id = await release(3) + const row = await db.from('releases').where('id', id).first() + assert.equal(Number(row.file_count), 3) + assert.deepEqual(row.extensions, ['bin']) + // Noted for the completeness sweep once, not per file. + const [{ count }] = await db.from('release_dirty').where('release_id', id).count('* as count') + assert.equal(Number(count), 1) + }) + test('complete when every listed file of 64 KiB or more is there, by size or name', async ({ assert }) => { const id = await release(3) assert.isTrue(await listing(id, 3)) @@ -47,6 +60,7 @@ test.group('release completeness', (group) => { test('a par2 listing of thousands of files is checked in one go', async ({ assert }) => { const id = await release(3000) + await ReleaseCompleteness.sweep() const started = Date.now() assert.isTrue(await listing(id, 3000)) // Was cubic: minutes for a few thousand rows. diff --git a/tests/functional/web_api.spec.ts b/tests/functional/web_api.spec.ts index 65f823a..d4995a7 100644 --- a/tests/functional/web_api.spec.ts +++ b/tests/functional/web_api.spec.ts @@ -1,4 +1,5 @@ import { test } from '@japa/runner' +import ReleaseCompleteness from '#services/ReleaseCompleteness' import db from '@adonisjs/lucid/services/db' import FileStore from '#services/FileStore' import testUtils from '@adonisjs/core/services/test_utils' @@ -89,9 +90,12 @@ test.group('Web API', (group) => { response.assertBodyContains({ sources: [] }) }) - test('completeness follows files and par2 lists through triggers', async ({ assert }) => { + test('completeness follows files and par2 lists (triggers note releases, the sweep recomputes)', async ({ assert }) => { const now = new Date() - const complete = async (id: number) => (await db.from('releases').where('id', id).first()).complete + const complete = async (id: number) => { + await ReleaseCompleteness.sweep() + return (await db.from('releases').where('id', id).first()).complete + } const file = (releaseId: number, name: string) => ({ filename: name, poster: 'p', date: now.getTime(), parts: 1, size: 100_000, release_id: releaseId, groups: '[]', segments: [], created_at: now, updated_at: now,