diff --git a/app/services/FileKeys.ts b/app/services/FileKeys.ts new file mode 100644 index 0000000..86b1a84 --- /dev/null +++ b/app/services/FileKeys.ts @@ -0,0 +1,60 @@ +import db from '@adonisjs/lucid/services/db' + +/** files ids per slice: a few seconds of reads from the HDD. */ +const SLICE = 100_000 + +/** + * The file_keys filter (LOG.md 2026-10-06 (7)): a 64-bit hash of every + * file_key. The insert trigger adds new keys; `fill` copies the keys of + * files from before, in id order and a slice at a time, and marks the + * filter complete when it reaches the files that existed when it started + * (everything after that has been added by the trigger, which was live + * first). FileStore uses the filter only once it is complete. + */ +class FileKeys { + private completeSince = 0 + + /** Whether every file's key is in file_keys; cached once true. */ + async complete(): Promise { + if (this.completeSince) return true + const row = await db.from('file_keys_fill').where('id', 1).first() + if (row?.complete) this.completeSince = Date.now() + return Boolean(row?.complete) + } + + /** Forgets the cached state (tests). */ + forget() { + this.completeSince = 0 + } + + /** Copies keys until `budgetMs` has passed or the fill is done; returns the files read and whether it's complete. */ + async fill(budgetMs: number): Promise<{ read: number; complete: boolean; nextId: number; untilId: number }> { + const started = Date.now() + let state = await db.from('file_keys_fill').where('id', 1).first() + if (!state) return { read: 0, complete: false, nextId: 0, untilId: 0 } + if (state.until_id === null) { + // Everything inserted from here on goes through the trigger (live since the migration). + const [{ max }] = await db.from('files').max('id as max') + await db.from('file_keys_fill').where('id', 1).update({ until_id: Number(max ?? 0), updated_at: new Date() }) + state = await db.from('file_keys_fill').where('id', 1).first() + } + let next = Number(state.next_id) + const until = Number(state.until_id) + let read = 0 + while (next <= until && Date.now() - started < budgetMs) { + const to = Math.min(until, next + SLICE - 1) + const result = await db.rawQuery( + `INSERT INTO file_keys (key_hash) + SELECT DISTINCT hashtextextended(file_key, 0) FROM files WHERE id BETWEEN ? AND ? AND file_key IS NOT NULL + ON CONFLICT DO NOTHING`, + [next, to] + ) + read += Number(result.rowCount ?? 0) + next = to + 1 + await db.from('file_keys_fill').where('id', 1).update({ next_id: next, complete: next > until, updated_at: new Date() }) + } + return { read, complete: next > until, nextId: next, untilId: until } + } +} + +export default new FileKeys() diff --git a/app/services/FileStore.ts b/app/services/FileStore.ts index 5ec0373..6455593 100644 --- a/app/services/FileStore.ts +++ b/app/services/FileStore.ts @@ -1,4 +1,5 @@ import db from '@adonisjs/lucid/services/db' +import FileKeys from '#services/FileKeys' import type { QueryClientContract } from '@adonisjs/lucid/types/database' import { packSegments, unpackSegments, type Segment } from '#lib/nzb/segments' @@ -44,8 +45,8 @@ class FileStore { * Inserts files, skipping those whose `file_key` exists. Returns the rows * that were inserted (id, filename, file_key). `files` is partitioned by id * (CONCEPT.md D39) and each partition has its own unique file_key index: - * file_keys (a 64-bit hash of every key, filled by the insert trigger) says - * which keys may exist, and only those are looked up in the partitions; two + * once file_keys (a 64-bit hash of every key) is complete, it says which keys + * may exist, and only those are looked up in the partitions; two * writers adding the same new key at once meet in the newest partition's index. */ async insert( @@ -55,8 +56,14 @@ class FileStore { if (files.length === 0) return [] const keys = [...new Set(files.map((file) => file.file_key).filter((key): key is string => typeof key === 'string'))] const known = new Set() + const filtered = await FileKeys.complete() for (let i = 0; i < keys.length; i += 5000) { const chunk = keys.slice(i, i + 5000) + if (!filtered) { + const found = await client.knexQuery().from('files').whereIn('file_key', chunk).select('file_key') + for (const row of found) known.add(row.file_key) + continue + } const { rows } = await client.rawQuery( `SELECT k.key FROM unnest(?::text[]) AS k(key) WHERE EXISTS (SELECT 1 FROM file_keys f WHERE f.key_hash = hashtextextended(k.key, 0))`, diff --git a/commands/GcScheduler.ts b/commands/GcScheduler.ts index eef6a5d..739b35d 100644 --- a/commands/GcScheduler.ts +++ b/commands/GcScheduler.ts @@ -10,6 +10,7 @@ import ReleaseMergeService from '#services/ReleaseMergeService' import StagingService from '#services/StagingService' import FilePartitions from '#services/FilePartitions' import ReleaseCompleteness from '#services/ReleaseCompleteness' +import FileKeys from '#services/FileKeys' import Shutdown from '#services/Shutdown' const CLEAN_BATCH = 1000 @@ -51,6 +52,11 @@ export default class GcScheduler extends BaseCommand { } catch (err: any) { this.logger.error(`Release completeness failed: ${err.message}`) } + try { + await this.fillFileKeys() + } catch (err: any) { + this.logger.error(`file_keys fill failed: ${err.message}`) + } if (!this.once) await Shutdown.sleep(gcConfig.intervalMs) } while (!this.once && !Shutdown.requested) } finally { @@ -117,6 +123,17 @@ export default class GcScheduler extends BaseCommand { } /** Rebuilds bloated indexes, at most every gcConfig.reindexIntervalMs (CONCEPT.md D33). */ + /** The keys of files from before file_keys, a slice of time per sweep, until done. */ + private async fillFileKeys() { + if (this.dryRun || (await FileKeys.complete())) return + const started = Date.now() + const { read, complete, nextId, untilId } = await FileKeys.fill(gcConfig.fileKeysFillMs) + this.logger.info( + `file_keys: ${read.toLocaleString('en')} keys copied in ${Math.round((Date.now() - started) / 1000)} s, ` + + (complete ? 'complete: file inserts use the filter now.' : `at file ${nextId.toLocaleString('en')} of ${untilId.toLocaleString('en')}.`) + ) + } + private async releaseCompleteness() { if (this.dryRun || (!this.once && Date.now() - this.completenessAt < gcConfig.releaseCompleteIntervalMs)) return this.completenessAt = Date.now() diff --git a/config/gc.ts b/config/gc.ts index de3bfa5..c339d5d 100644 --- a/config/gc.ts +++ b/config/gc.ts @@ -10,6 +10,9 @@ export default { /** 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), + /** Time per sweep for copying old file keys into file_keys until it's complete (LOG.md 2026-10-06 (7)). */ + fileKeysFillMs: env.get('FILE_KEYS_FILL_SECONDS', 90) * 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/1768620765220_file_keys.ts b/database/migrations/1768620765220_file_keys.ts index 154a7c8..052316d 100644 --- a/database/migrations/1768620765220_file_keys.ts +++ b/database/migrations/1768620765220_file_keys.ts @@ -4,20 +4,28 @@ import { BaseSchema } from '@adonisjs/lucid/schema' * A 64-bit hash of every file_key, in one index (LOG.md 2026-10-06 (7)). * file_key is unique per files partition only (D39), so FileStore.insert * looked every key up in every partition: 7 partitions and growing, 35,000 - * random index reads for a batch of 5,000 files. Now it asks file_keys - * first, and only keys whose hash is there (a duplicate, or rarely a - * collision) are looked up in files. + * random index reads for a batch of 5,000 files. With file_keys complete it + * asks file_keys first, and only keys whose hash is there (a duplicate, or + * rarely a collision) are looked up in files. * - * Filled by the files insert statement trigger; no transaction, so that the - * trigger is live before the existing keys are copied and no file inserted - * meanwhile is missed (a missed key would let a duplicate file in). + * New keys come from the files insert statement trigger. The keys from + * before are copied in the background (FileKeys.fill, gc:scheduler): + * copying them here took over half an hour of HDD reads and held up the + * deploy. Until file_keys_fill.complete, FileStore checks every partition. */ export default class extends BaseSchema { - static disableTransactions = true - async up() { const q = (sql: string) => this.db.rawQuery(sql) await q(`CREATE TABLE IF NOT EXISTS file_keys (key_hash bigint PRIMARY KEY)`) + // One row: the fill's position in files.id, the id it must reach, and whether it got there. + await q(` + CREATE TABLE IF NOT EXISTS file_keys_fill ( + id smallint PRIMARY KEY DEFAULT 1 CHECK (id = 1), + next_id bigint NOT NULL DEFAULT 0, + until_id bigint, + complete boolean NOT NULL DEFAULT false, + updated_at timestamptz NOT NULL DEFAULT now())`) + await q(`INSERT INTO file_keys_fill (id) VALUES (1) ON CONFLICT DO NOTHING`) await q(` CREATE OR REPLACE FUNCTION files_release_insert_statement() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN @@ -52,17 +60,10 @@ export default class extends BaseSchema { ON CONFLICT DO NOTHING; RETURN NULL; END $$`) - // The keys from before; read from the file_key indexes (NVMe), not the files heap (HDD). - await q(`SET enable_seqscan = off`) - await q(` - INSERT INTO file_keys (key_hash) - SELECT DISTINCT hashtextextended(file_key, 0) FROM files WHERE file_key IS NOT NULL - ON CONFLICT DO NOTHING`) - await q(`RESET enable_seqscan`) - await q(`ANALYZE file_keys`) } async down() { + await this.db.rawQuery(`DROP TABLE IF EXISTS file_keys_fill`) await this.db.rawQuery(`DROP TABLE IF EXISTS file_keys`) } } diff --git a/database/schema.ts b/database/schema.ts index 3dee504..c5ee683 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -181,6 +181,21 @@ export class FileKeySchema extends BaseModel { declare keyHash: bigint | number } +export class FileKeysFillSchema extends BaseModel { + static $columns = ['complete', 'id', 'nextId', 'untilId', 'updatedAt'] as const + $columns = FileKeysFillSchema.$columns + @column() + declare complete: boolean + @column({ isPrimary: true }) + declare id: number + @column() + declare nextId: bigint | number + @column() + declare untilId: bigint | number | null + @column.dateTime({ autoCreate: true, autoUpdate: true }) + declare updatedAt: DateTime +} + export class FileSchema extends BaseModel { static $columns = ['archive', 'contentType', 'createdAt', 'date', 'fileKey', 'filename', 'firstMessageId', 'groups', 'id', 'parts', 'poster', 'realName', 'releaseId', 'segments', 'size', 'updatedAt'] as const $columns = FileSchema.$columns diff --git a/start/env.ts b/start/env.ts index 7332df4..a2b5a12 100644 --- a/start/env.ts +++ b/start/env.ts @@ -103,6 +103,7 @@ export default await Env.create(new URL('../', import.meta.url), { COLLECTION_CONCURRENCY: Env.schema.number.optional(), ARCHIVE_WALK_CONCURRENCY: Env.schema.number.optional(), RELEASE_COMPLETE_INTERVAL_MS: Env.schema.number.optional(), + FILE_KEYS_FILL_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/file_keys.spec.ts b/tests/functional/file_keys.spec.ts new file mode 100644 index 0000000..cf5d88a --- /dev/null +++ b/tests/functional/file_keys.spec.ts @@ -0,0 +1,41 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import FileKeys from '#services/FileKeys' +import FileStore from '#services/FileStore' + +const file = (key: string, name: string) => { + const now = new Date() + return { file_key: key, filename: name, poster: 'x', date: now.getTime(), parts: 1, groups: '[]', segments: [], created_at: now, updated_at: now } +} + +test.group('file_keys', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + group.each.teardown(() => FileKeys.forget()) + + test('the fill copies keys from before, then file inserts use the filter', async ({ assert }) => { + // A file from before file_keys: in files, not in file_keys. + const [old] = await FileStore.insert([file('fk-old'.padEnd(40, '0'), 'old.bin')]) + await db.rawQuery(`DELETE FROM file_keys WHERE key_hash = hashtextextended(?, 0)`, ['fk-old'.padEnd(40, '0')]) + await db.from('file_keys_fill').where('id', 1).update({ next_id: 0, until_id: null, complete: false }) + FileKeys.forget() + + // Not complete: every partition is checked, so the old key is found anyway. + assert.isFalse(await FileKeys.complete()) + assert.lengthOf(await FileStore.insert([file('fk-old'.padEnd(40, '0'), 'again.bin')]), 0) + + const filled = await FileKeys.fill(60_000) + assert.isTrue(filled.complete) + assert.isAtLeast(filled.untilId, old.id) + const [{ n }] = (await db.rawQuery(`SELECT count(*) AS n FROM file_keys WHERE key_hash = hashtextextended(?, 0)`, ['fk-old'.padEnd(40, '0')])).rows + assert.equal(Number(n), 1) + + // Complete: the filter says the old key may exist, the partitions confirm it; a new key goes in. + assert.isTrue(await FileKeys.complete()) + assert.lengthOf(await FileStore.insert([file('fk-old'.padEnd(40, '0'), 'again.bin')]), 0) + assert.lengthOf(await FileStore.insert([file('fk-new'.padEnd(40, '0'), 'new.bin')]), 1) + // And the trigger put the new one in file_keys. + const [{ m }] = (await db.rawQuery(`SELECT count(*) AS m FROM file_keys WHERE key_hash = hashtextextended(?, 0)`, ['fk-new'.padEnd(40, '0')])).rows + assert.equal(Number(m), 1) + }) +})