file_keys: fill in the background, use it once complete
Test and deploy / test (push) Successful in 1m37s
Test and deploy / deploy (push) Successful in 2m23s

Copying all existing keys in the migration read the files rows from the
HDD for over half an hour and held up the deploy. gc:scheduler now copies
them in slices (FILE_KEYS_FILL_SECONDS per sweep) and FileStore checks
every partition until the fill is complete.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Daan MeijerandClaude Opus 5.5 committed 2026-10-06 14:54:12 +02:00
1 parent 693e6c3244
commit f0ff8e200f
8 files changed
+163 -18

No files matched your search

+60
View File
@@ -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<boolean> {
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()
+9 -2
View File
@@ -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<string>()
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))`,
+17
View File
@@ -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()
+3
View File
@@ -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),
+17 -16
View File
@@ -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`)
}
}
+15
View File
@@ -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
+1
View File
@@ -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(),
+41
View File
@@ -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)
})
})