From f1773744c08b8fe711259a76eb7789d873c8dd7d Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 5 Oct 2026 15:57:51 +0200 Subject: [PATCH] Partition files by age (id) New rows went into free space scattered through the 9.6 GB heap on the HDD (left behind by inspection's updates): one random HDD read per insert, and seed batches took a median 13 s to store. The existing table becomes the first partition as it is (a CHECK scan, no rewrite; indexes, its foreign key and the triggers are taken over by the partitioned parent); new partitions of 4M ids go in the same tablespace and stay in memory while they fill. - Every partition has a unique file_key index; FileStore.insert looks keys up in all partitions first (a key can't be unique across partitions). - FilePartitions.ensure() (gc:scheduler) keeps two partitions ahead. - Stats count files from the partitions; seed:node waits ~20 minutes for the API before giving a range up. Co-Authored-By: Claude Opus 5.5 --- app/services/FilePartitions.ts | 39 ++++++ app/services/FileStore.ts | 31 ++++- app/services/SearchService.ts | 3 +- commands/GcScheduler.ts | 4 + commands/SeedNode.ts | 3 +- .../1768620765150_partition_files.ts | 77 ++++++++++++ database/schema.ts | 111 ++++++++++++++++++ tests/functional/file_partitions.spec.ts | 50 ++++++++ 8 files changed, 310 insertions(+), 8 deletions(-) create mode 100644 app/services/FilePartitions.ts create mode 100644 database/migrations/1768620765150_partition_files.ts create mode 100644 tests/functional/file_partitions.spec.ts diff --git a/app/services/FilePartitions.ts b/app/services/FilePartitions.ts new file mode 100644 index 0000000..83e7ece --- /dev/null +++ b/app/services/FilePartitions.ts @@ -0,0 +1,39 @@ +import db from '@adonisjs/lucid/services/db' +import logger from '@adonisjs/core/services/logger' + +/** Ids per partition (as the migration that partitioned files: …150). */ +export const FILE_PARTITION_SIZE = 4_000_000 +/** Ids kept covered beyond the newest file: two partitions. */ +const AHEAD = 2 * FILE_PARTITION_SIZE + +/** + * `files` is partitioned by id (CONCEPT.md D39). An insert with an id no + * partition covers fails, so gc:scheduler calls ensure() every sweep to keep + * two partitions ready ahead of the newest id. Each new partition gets its + * own unique index on file_key (FileStore.insert checks the older ones), + * in the parent's tablespace (the HDD on homelab). + */ +class FilePartitions { + async ensure(): Promise { + const created: string[] = [] + while (true) { + const { rows } = await db.rawQuery( + `SELECT (SELECT coalesce(max(id), 0) FROM files) AS max_id, + -- ''{0,1}, not ''?: knex reads ? as a binding. + max((regexp_match(pg_get_expr(c.relpartbound, c.oid), 'TO \\(''{0,1}(\\d+)''{0,1}\\)'))[1]::bigint) AS upper + FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid + WHERE i.inhparent = 'files'::regclass` + ) + const maxId = Number(rows[0].max_id) + const upper = Number(rows[0].upper) + if (upper - maxId >= AHEAD) return created + const name = `files_p${upper}` + await db.rawQuery(`CREATE TABLE ${name} PARTITION OF files FOR VALUES FROM (${upper}) TO (${upper + FILE_PARTITION_SIZE})`) + await db.rawQuery(`ALTER TABLE ${name} ADD CONSTRAINT ${name}_file_key_unique UNIQUE (file_key)`) + logger.info(`files: partition ${name} created (ids ${upper}–${upper + FILE_PARTITION_SIZE - 1}).`) + created.push(name) + } + } +} + +export default new FilePartitions() diff --git a/app/services/FileStore.ts b/app/services/FileStore.ts index c64136e..1d83a38 100644 --- a/app/services/FileStore.ts +++ b/app/services/FileStore.ts @@ -39,15 +39,34 @@ export function loadedSegments(stored: Buffer | null): Segment[] { class FileStore { /** * Inserts files, skipping those whose `file_key` exists. Returns the rows - * that were inserted (id and filename). + * that were inserted (id and filename). `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. */ async insert(files: NewFile[], client: Pick = db): Promise<{ id: number; filename: string }[]> { if (files.length === 0) return [] - const rows = files.map(({ segments, ...file }) => { - const sorted = [...segments].sort((a, b) => a.number - b.number) - return { ...file, segments: packSegments(sorted), first_message_id: sorted[0]?.id ?? null } - }) - const inserted = await client.knexQuery().table('files').insert(rows).onConflict('file_key').ignore().returning(['id', 'filename']) + const keys = [...new Set(files.map((file) => file.file_key).filter((key): key is string => typeof key === 'string'))] + const known = new Set() + 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') + for (const row of found) known.add(row.file_key) + } + const seen = new Set() + const rows = files + .filter((file) => { + const key = file.file_key as string | null | undefined + if (!key) return true + if (known.has(key) || seen.has(key)) return false + seen.add(key) + return true + }) + .map(({ segments, ...file }) => { + const sorted = [...segments].sort((a, b) => a.number - b.number) + return { ...file, segments: packSegments(sorted), first_message_id: sorted[0]?.id ?? null } + }) + if (rows.length === 0) return [] + const inserted = await client.knexQuery().table('files').insert(rows).onConflict().ignore().returning(['id', 'filename']) return inserted.map((row: any) => ({ id: Number(row.id), filename: row.filename })) } } diff --git a/app/services/SearchService.ts b/app/services/SearchService.ts index c4359db..4396c43 100644 --- a/app/services/SearchService.ts +++ b/app/services/SearchService.ts @@ -390,7 +390,8 @@ class SearchService { SELECT r.name, r.size, r.file_count, r.encrypted, coalesce(r.source, 'unknown') AS source FROM releases r TABLESAMPLE SYSTEM (0.5) WHERE ${VISIBLE_SQL}) SELECT (SELECT reltuples FROM pg_class WHERE relname = 'releases_category_idx') AS releases, - (SELECT reltuples FROM pg_class WHERE relname = 'files') AS files, + -- files is partitioned (D39): its partitions hold the counts. + (SELECT sum(c.reltuples) FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid WHERE i.inhparent = 'files'::regclass) AS files, (SELECT count(*) FROM sample) AS sampled, (SELECT count(*) FROM sample WHERE is_readable_name(name)) AS sampled_readable, -- Indexed data only: a spot listing's size is what the spot announces. diff --git a/commands/GcScheduler.ts b/commands/GcScheduler.ts index 1d9577c..bd4b082 100644 --- a/commands/GcScheduler.ts +++ b/commands/GcScheduler.ts @@ -8,6 +8,7 @@ import RedisService from '#services/RedisService' import IndexMaintenance from '#services/IndexMaintenance' import ReleaseMergeService from '#services/ReleaseMergeService' import StagingService from '#services/StagingService' +import FilePartitions from '#services/FilePartitions' import Shutdown from '#services/Shutdown' const CLEAN_BATCH = 1000 @@ -61,6 +62,9 @@ export default class GcScheduler extends BaseCommand { } } + // Inserts fail without a partition for their id (D39). + if (!this.dryRun) await FilePartitions.ensure() + const trimmed = await this.trimBodyQueue() if (trimmed) this.logger.info(`Body queue: ${this.verb('drop', 'dropped')} ${trimmed} oldest waiting jobs.`) diff --git a/commands/SeedNode.ts b/commands/SeedNode.ts index ce11e8c..ffc3bcf 100644 --- a/commands/SeedNode.ts +++ b/commands/SeedNode.ts @@ -282,7 +282,8 @@ class SeedApi { if (status >= 400 && status < 500 && status !== 404) throw new Error(`Seed API ${path}: ${status} ${text()}`) // Network errors, 404 and 5xx: storing a batch twice is harmless, so try again. - if (attempt >= 8) throw new Error(`Seed API ${path}: ${status || 'no response'} ${text()}`) + // ~20 minutes in all, longer than a lease lasts: a restart or maintenance of the API is waited out. + if (attempt >= 14) throw new Error(`Seed API ${path}: ${status || 'no response'} ${text()}`) const wait = Math.min(120_000, 2000 * 2 ** (attempt - 1)) this.warn(`Seed API ${path}: ${status || text()}; retrying in ${wait / 1000} s`) await sleep(wait) diff --git a/database/migrations/1768620765150_partition_files.ts b/database/migrations/1768620765150_partition_files.ts new file mode 100644 index 0000000..a5350d9 --- /dev/null +++ b/database/migrations/1768620765150_partition_files.ts @@ -0,0 +1,77 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** Ids per partition: 4M files is a few days of indexing, ~2 GB of heap (FilePartitions uses the same). */ +const FILE_PARTITION_SIZE = 4_000_000 + +/** + * `files` partitioned by id, i.e. by age (CONCEPT.md D39). New rows went into + * free space scattered through the 9.6 GB heap on the HDD (left by + * inspection's updates), one random read per insert; now they go into a + * small recent partition that stays in memory. The existing table becomes + * the first partition as it is: only a CHECK scan, no rewrite. + * + * Postgres can't enforce a unique file_key across partitions: every + * partition has its own unique index, and FileStore.insert looks keys up in + * the older ones. Run with the writers stopped (a few minutes). + */ +export default class extends BaseSchema { + static disableTransactions = true + + async up() { + const q = (sql: string) => this.db.rawQuery(sql) + const { rows } = await q(`SELECT coalesce(max(id), 0) AS max FROM files`) + // The first new partition starts at the next million. + const boundary = (Math.floor(Number(rows[0].max) / 1_000_000) + 1) * 1_000_000 + // The HDD tablespace on homelab (D32), the default elsewhere; new partitions inherit the parent's. + const { rows: ts } = await q(`SELECT t.spcname FROM pg_class c JOIN pg_tablespace t ON t.oid = c.reltablespace WHERE c.oid = 'files'::regclass`) + const tablespace = ts[0]?.spcname ? ` TABLESPACE ${ts[0].spcname}` : '' + + // The old table, renamed; its constraint and index names get a files_p0 prefix. + await q(`ALTER TABLE files ADD CONSTRAINT files_p0_bound CHECK (id < ${boundary}) NOT VALID`) + await q(`ALTER TABLE files VALIDATE CONSTRAINT files_p0_bound`) + await q(`ALTER TABLE files RENAME TO files_p0`) + await q(`ALTER TABLE files_p0 RENAME CONSTRAINT files_pkey TO files_p0_pkey`) + await q(`ALTER TABLE files_p0 RENAME CONSTRAINT files_file_key_unique TO files_p0_file_key_unique`) + await q(`ALTER TABLE files_p0 RENAME CONSTRAINT files_release_id_foreign TO files_p0_release_id_foreign`) + for (const index of ['files_content_type_index', 'files_first_message_id_idx', 'files_name_idx', 'files_release_id_index']) { + await q(`ALTER INDEX ${index} RENAME TO ${index.replace(/^files_/, 'files_p0_')}`) + } + // Recreated on the parent, which clones them to every partition. + for (const trigger of ['files_release_extension', 'files_release_search', 'files_release_totals', 'files_zz_release_complete']) { + await q(`DROP TRIGGER ${trigger} ON files_p0`) + } + + // The parent: same columns and defaults (the id sequence), partitioned by id, in the old table's tablespace. + await q(`CREATE TABLE files (LIKE files_p0 INCLUDING DEFAULTS) PARTITION BY RANGE (id)${tablespace}`) + await q(`ALTER SEQUENCE files_id_seq OWNED BY files.id`) + await q(`ALTER TABLE files ADD CONSTRAINT files_pkey PRIMARY KEY (id)`) + await q(`ALTER TABLE files ADD CONSTRAINT files_release_id_foreign FOREIGN KEY (release_id) REFERENCES releases(id) ON DELETE SET NULL`) + await q(`CREATE INDEX files_content_type_index ON files (content_type)`) + await q(`CREATE INDEX files_first_message_id_idx ON files (first_message_id)`) + await q(`CREATE INDEX files_name_idx ON files ((coalesce(real_name, filename::text)))`) + await q(`CREATE INDEX files_release_id_index ON files (release_id)`) + // Matching indexes and the foreign key of files_p0 are taken over, not rebuilt; the CHECK spares the scan. + await q(`ALTER TABLE files ATTACH PARTITION files_p0 FOR VALUES FROM (MINVALUE) TO (${boundary})`) + await q(`ALTER TABLE files_p0 DROP CONSTRAINT files_p0_bound`) + + 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()`) + + // binaries.file_id followed the rename to files_p0; point it at the parent (no scan needed). + await q(`ALTER TABLE binaries DROP CONSTRAINT binaries_file_id_foreign`) + await q(`ALTER TABLE binaries ADD CONSTRAINT binaries_file_id_foreign FOREIGN KEY (file_id) REFERENCES files(id) ON DELETE SET NULL NOT VALID`) + + // The next two partitions; FilePartitions.ensure() (gc:scheduler) keeps two ahead from now on. + for (const start of [boundary, boundary + FILE_PARTITION_SIZE]) { + const name = `files_p${start}` + await q(`CREATE TABLE ${name} PARTITION OF files FOR VALUES FROM (${start}) TO (${start + FILE_PARTITION_SIZE})`) + await q(`ALTER TABLE ${name} ADD CONSTRAINT ${name}_file_key_unique UNIQUE (file_key)`) + } + } + + async down() { + throw new Error('Not reversible: merging partitions back means rewriting the files table.') + } +} diff --git a/database/schema.ts b/database/schema.ts index 48d635a..2def622 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -209,6 +209,117 @@ export class FileSchema extends BaseModel { declare updatedAt: DateTime | null } +export class FilesP0Schema extends BaseModel { + static $columns = ['archive', 'contentType', 'createdAt', 'date', 'fileKey', 'filename', 'firstMessageId', 'groups', 'id', 'parts', 'poster', 'realName', 'releaseId', 'segments', 'size', 'updatedAt'] as const + $columns = FilesP0Schema.$columns + @column() + declare archive: any | null + @column() + declare contentType: string | null + @column.dateTime({ autoCreate: true }) + declare createdAt: DateTime | null + @column() + declare date: bigint | number + @column() + declare fileKey: string | null + @column() + declare filename: string + @column() + declare firstMessageId: string | null + @column() + declare groups: any | null + @column({ isPrimary: true }) + declare id: number + @column() + declare parts: number + @column() + declare poster: string + @column() + declare realName: string | null + @column() + declare releaseId: bigint | number | null + @column() + declare segments: Buffer + @column() + declare size: bigint | number | null + @column.dateTime({ autoCreate: true, autoUpdate: true }) + declare updatedAt: DateTime | null +} + +export class FilesP1000000Schema extends BaseModel { + static $columns = ['archive', 'contentType', 'createdAt', 'date', 'fileKey', 'filename', 'firstMessageId', 'groups', 'id', 'parts', 'poster', 'realName', 'releaseId', 'segments', 'size', 'updatedAt'] as const + $columns = FilesP1000000Schema.$columns + @column() + declare archive: any | null + @column() + declare contentType: string | null + @column.dateTime({ autoCreate: true }) + declare createdAt: DateTime | null + @column() + declare date: bigint | number + @column() + declare fileKey: string | null + @column() + declare filename: string + @column() + declare firstMessageId: string | null + @column() + declare groups: any | null + @column({ isPrimary: true }) + declare id: number + @column() + declare parts: number + @column() + declare poster: string + @column() + declare realName: string | null + @column() + declare releaseId: bigint | number | null + @column() + declare segments: Buffer + @column() + declare size: bigint | number | null + @column.dateTime({ autoCreate: true, autoUpdate: true }) + declare updatedAt: DateTime | null +} + +export class FilesP5000000Schema extends BaseModel { + static $columns = ['archive', 'contentType', 'createdAt', 'date', 'fileKey', 'filename', 'firstMessageId', 'groups', 'id', 'parts', 'poster', 'realName', 'releaseId', 'segments', 'size', 'updatedAt'] as const + $columns = FilesP5000000Schema.$columns + @column() + declare archive: any | null + @column() + declare contentType: string | null + @column.dateTime({ autoCreate: true }) + declare createdAt: DateTime | null + @column() + declare date: bigint | number + @column() + declare fileKey: string | null + @column() + declare filename: string + @column() + declare firstMessageId: string | null + @column() + declare groups: any | null + @column({ isPrimary: true }) + declare id: number + @column() + declare parts: number + @column() + declare poster: string + @column() + declare realName: string | null + @column() + declare releaseId: bigint | number | null + @column() + declare segments: Buffer + @column() + declare size: bigint | number | null + @column.dateTime({ autoCreate: true, autoUpdate: true }) + declare updatedAt: DateTime | null +} + export class GroupSchema extends BaseModel { static $columns = ['active', 'backfillNextId', 'backfillUntilId', 'createdAt', 'fetchBodies', 'id', 'lastIndexedId', 'lastQueuedId', 'name', 'updatedAt'] as const $columns = GroupSchema.$columns diff --git a/tests/functional/file_partitions.spec.ts b/tests/functional/file_partitions.spec.ts new file mode 100644 index 0000000..2110093 --- /dev/null +++ b/tests/functional/file_partitions.spec.ts @@ -0,0 +1,50 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import FileStore from '#services/FileStore' +import FilePartitions, { FILE_PARTITION_SIZE } from '#services/FilePartitions' + +const file = (fileKey: string, name: string) => { + const now = new Date() + return { file_key: fileKey, filename: name, poster: 'p', date: now.getTime(), parts: 1, size: 1, groups: '[]', segments: [], created_at: now, updated_at: now } +} + +test.group('files partitions', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + // Sequences ignore rollbacks: put files_id_seq back after each test. + group.each.setup(async () => { + const { rows } = await db.rawQuery(`SELECT last_value, is_called FROM files_id_seq`) + return async () => { + await db.rawQuery(`SELECT setval('files_id_seq', ?, ?)`, [rows[0].last_value, rows[0].is_called]) + } + }) + + test('a file_key already in an older partition is not inserted again', async ({ assert }) => { + await db.rawQuery(`SELECT setval('files_id_seq', 500000, false)`) + const [first] = await FileStore.insert([file('k'.repeat(40), 'old.bin')]) + // The next ids fall in the second partition, whose own unique index doesn't know the key. + await db.rawQuery(`SELECT setval('files_id_seq', 1000000, false)`) + const again = await FileStore.insert([file('k'.repeat(40), 'again.bin'), file('n'.repeat(40), 'new.bin'), file('n'.repeat(40), 'twice.bin')]) + assert.deepEqual(again.map((f) => f.filename), ['new.bin']) + assert.isAtLeast(again[0].id, 1000000) + assert.equal(await db.from('files').where('file_key', 'k'.repeat(40)).count('* as n').then((r) => Number(r[0].n)), 1) + assert.isBelow(first.id, 1000000) + }) + + test('partitions are kept ready ahead of the newest id', async ({ assert }) => { + const upper = async () => { + const { rows } = await db.rawQuery( + `SELECT max((regexp_match(pg_get_expr(c.relpartbound, c.oid), 'TO \\(''{0,1}(\\d+)''{0,1}\\)'))[1]::bigint) AS upper + FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid WHERE i.inhparent = 'files'::regclass` + ) + return Number(rows[0].upper) + } + const before = await upper() + assert.deepEqual(await FilePartitions.ensure(), []) + // A file near the end of the covered ids: one more partition is needed. + await db.rawQuery(`SELECT setval('files_id_seq', ?, false)`, [before - FILE_PARTITION_SIZE - 10]) + await FileStore.insert([file('e'.repeat(40), 'late.bin')]) + assert.deepEqual(await FilePartitions.ensure(), [`files_p${before}`]) + assert.equal(await upper(), before + FILE_PARTITION_SIZE) + }) +})