Content type and archive listing move from files to file_details
Every probe rewrote its file's row on the HDD (content_type, archive): a new row version on a random page and an entry in each of the five indexes. They now live in file_details on the NVMe; files rows are inserted once. files:move-details copies the existing values after the deploy; a later migration drops the old columns. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
1520a8dc35
commit
fad1716d3a
12 files changed
+138
-22
No files matched your search
@@ -1,5 +1,6 @@
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import logger from '@adonisjs/core/services/logger'
|
||||
import FileStore from '#services/FileStore'
|
||||
import IngestService from '#services/IngestService'
|
||||
import type { RarInfo } from '#lib/rar/index'
|
||||
import { fileStem, isGenericName, rarReleaseIdentity } from '#lib/release/index'
|
||||
@@ -24,7 +25,7 @@ class ArchiveService {
|
||||
async record(file: any, binary: any, info: RarInfo): Promise<ArchiveResult> {
|
||||
const { next: _next, ...listing } = info
|
||||
const stored = { ...listing, entries: info.entries.slice(0, MAX_ENTRIES).map(({ stored: _s, dataOffset: _d, ...entry }) => entry) }
|
||||
await db.from('files').where('id', file.id).update({ archive: JSON.stringify(stored) })
|
||||
await FileStore.setDetails(Number(file.id), { archive: stored })
|
||||
|
||||
let releaseId = file.release_id ?? binary?.release_id ?? null
|
||||
let grouped = false
|
||||
|
||||
@@ -24,7 +24,7 @@ import { sevenZipVolume } from '#lib/sevenzip/index'
|
||||
import SevenZipService from '#services/SevenZipService'
|
||||
import SfvService from '#services/SfvService'
|
||||
import ArchiveWalkService from '#services/ArchiveWalkService'
|
||||
import { segmentsOf } from '#services/FileStore'
|
||||
import FileStore, { segmentsOf } from '#services/FileStore'
|
||||
import GarbageFilter from '#services/GarbageFilter'
|
||||
|
||||
/**
|
||||
@@ -136,7 +136,7 @@ class ContentService {
|
||||
if (await GarbageFilter.skipProbe(file)) return null
|
||||
if ((sevenZipVolume(String(file.real_name ?? file.filename)) ?? 0) > 1) {
|
||||
const volumes = await SevenZipService.volumes(file)
|
||||
if (!volumes || volumes[0].archive) return null
|
||||
if (!volumes || (await FileStore.details(Number(volumes[0].id))).archive) return null
|
||||
return this.prepare(Number(volumes[0].id))
|
||||
}
|
||||
const segments = segmentsOf(file)
|
||||
@@ -185,7 +185,7 @@ class ContentService {
|
||||
if (!file) return result
|
||||
const binary = await db.from('binaries').where('file_id', task.fileId).first()
|
||||
const { contentType } = probe
|
||||
await db.from('files').where('id', file.id).update({ content_type: contentType })
|
||||
await FileStore.setDetails(Number(file.id), { content_type: contentType })
|
||||
if (contentType === 'missing') {
|
||||
// Nothing to read; the binary needn't wait for inspection any more.
|
||||
if (binary) await db.from('binaries').where('id', binary.id).update({ awaiting_inspection: null })
|
||||
|
||||
@@ -37,10 +37,35 @@ export function loadedSegments(stored: Buffer | null): Segment[] {
|
||||
return stored ? unpackSegments(stored) : []
|
||||
}
|
||||
|
||||
/** What inspection learns about a file, kept beside `files` (LOG.md 2026-10-08 (14)). */
|
||||
export interface FileDetails {
|
||||
content_type?: string | null
|
||||
archive?: unknown
|
||||
}
|
||||
|
||||
const DETAIL_COLUMNS = ['content_type', 'archive'] as const
|
||||
|
||||
/** Rows per INSERT, well below Postgres' 65,535 bind parameters (tests lower it). */
|
||||
export const insertLimits = { rows: 2000 }
|
||||
|
||||
class FileStore {
|
||||
/** Sets some of a file's details, leaving the others as they are. */
|
||||
async setDetails(fileId: number, details: FileDetails, client: Pick<QueryClientContract, 'rawQuery'> = db) {
|
||||
const columns = DETAIL_COLUMNS.filter((c) => details[c] !== undefined)
|
||||
if (columns.length === 0) return
|
||||
const values = columns.map((c) => (c === 'archive' && details[c] !== null ? JSON.stringify(details[c]) : details[c]))
|
||||
await client.rawQuery(
|
||||
`INSERT INTO file_details (file_id, ${columns.join(', ')}) VALUES (?, ${columns.map(() => '?').join(', ')})
|
||||
ON CONFLICT (file_id) DO UPDATE SET ${columns.map((c) => `${c} = excluded.${c}`).join(', ')}`,
|
||||
[fileId, ...(values as any[])]
|
||||
)
|
||||
}
|
||||
|
||||
/** A file's details (empty when it has none). */
|
||||
async details(fileId: number): Promise<FileDetails> {
|
||||
return (await db.from('file_details').where('file_id', fileId).select('content_type', 'archive').first()) ?? {}
|
||||
}
|
||||
|
||||
/**
|
||||
* Inserts files, skipping those whose `file_key` exists. Returns the rows
|
||||
* that were inserted (id, filename, file_key). `files` is partitioned by id
|
||||
@@ -88,12 +113,24 @@ class FileStore {
|
||||
return { ...file, segments: packSegments(sorted), first_message_id: sorted[0]?.id ?? null }
|
||||
})
|
||||
if (rows.length === 0) return []
|
||||
// Details given with a new file (e.g. a known content type) go to file_details.
|
||||
const details = rows.map((row: Record<string, unknown>) => {
|
||||
const d: FileDetails = {}
|
||||
for (const c of DETAIL_COLUMNS) if (c in row) (d as any)[c] = row[c]
|
||||
for (const c of DETAIL_COLUMNS) delete row[c]
|
||||
return d
|
||||
})
|
||||
// A statement takes at most 65,535 parameters: ~5,400 rows of 12 columns.
|
||||
const inserted: { id: number; filename: string; file_key: string | null }[] = []
|
||||
for (let i = 0; i < rows.length; i += insertLimits.rows) {
|
||||
const chunk = await client.knexQuery().table('files').insert(rows.slice(i, i + insertLimits.rows)).onConflict().ignore().returning(['id', 'filename', 'file_key'])
|
||||
for (const row of chunk) inserted.push({ id: Number(row.id), filename: row.filename, file_key: row.file_key ?? null })
|
||||
}
|
||||
const detailsByKey = new Map(rows.map((row: Record<string, unknown>, i: number) => [row.file_key ?? row.filename, details[i]]))
|
||||
for (const row of inserted) {
|
||||
const d = detailsByKey.get(row.file_key ?? row.filename)
|
||||
if (d && Object.keys(d).length > 0) await this.setDetails(row.id, d, client)
|
||||
}
|
||||
return inserted
|
||||
}
|
||||
}
|
||||
|
||||
@@ -196,8 +196,8 @@ class GarbageFilter {
|
||||
async seed(percent: number): Promise<number> {
|
||||
const { rows } = await db.rawQuery(
|
||||
`SELECT DISTINCT ON (f.release_id) f.release_id, f.first_message_id, f.groups
|
||||
FROM files f TABLESAMPLE SYSTEM (?)
|
||||
WHERE f.content_type IS NOT NULL AND f.release_id IS NOT NULL AND NOT is_readable_name(f.filename)`,
|
||||
FROM files f TABLESAMPLE SYSTEM (?) JOIN file_details d ON d.file_id = f.id
|
||||
WHERE d.content_type IS NOT NULL AND f.release_id IS NOT NULL AND NOT is_readable_name(f.filename)`,
|
||||
[percent]
|
||||
)
|
||||
let added = 0
|
||||
|
||||
@@ -70,8 +70,8 @@ const VISIBLE_SQL = `(r.file_count > 0 OR r.source = 'spot')`
|
||||
|
||||
/** Incomplete, and every file is a par2 file: nothing to download but repair data. */
|
||||
const PAR2_ONLY_SQL = `(r.file_count > 0 AND NOT r.complete AND NOT EXISTS (
|
||||
SELECT 1 FROM files f WHERE f.release_id = r.id
|
||||
AND f.content_type IS DISTINCT FROM 'par2' AND f.filename !~* '\\.par2$'))`
|
||||
SELECT 1 FROM files f LEFT JOIN file_details d ON d.file_id = f.id WHERE f.release_id = r.id
|
||||
AND d.content_type IS DISTINCT FROM 'par2' AND f.filename !~* '\\.par2$'))`
|
||||
|
||||
/** A search word as an ILIKE pattern; dots, underscores, dashes and spaces are interchangeable. */
|
||||
function pattern(word: string): string {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
import logger from '@adonisjs/core/services/logger'
|
||||
import FileStore from '#services/FileStore'
|
||||
import { parseSfv, sfvReleaseName } from '#lib/content/index'
|
||||
|
||||
/** Releases made by name rules alone; an SFV's listing outranks them. */
|
||||
@@ -30,7 +31,7 @@ class SfvService {
|
||||
.insert(entries.map((e) => ({ release_id: releaseId, name: e.name, crc: e.crc, poster: sfvFile.poster, posted_at: postedAt })))
|
||||
.onConflict(['release_id', 'name'])
|
||||
.ignore()
|
||||
await db.from('files').where('id', sfvFile.id).update({ content_type: 'sfv' })
|
||||
await FileStore.setDetails(Number(sfvFile.id), { content_type: 'sfv' })
|
||||
|
||||
const candidates = await db
|
||||
.from('files as f')
|
||||
|
||||
@@ -48,20 +48,20 @@ export default class ContentBackfill extends BaseCommand {
|
||||
try {
|
||||
while (!this.limit || queued < this.limit) {
|
||||
// Files never inspected: no fingerprint on their binary and no archive listing.
|
||||
const query = db.from('files as f').where('f.id', '>', lastId)
|
||||
const query = db.from('files as f').leftJoin('file_details as d', 'd.file_id', 'f.id').where('f.id', '>', lastId)
|
||||
// Files from NZBs (spots) have no binary.
|
||||
if (this.sevenZip || this.archives || this.sfv || this.nicovideo) query.leftJoin('binaries as b', 'b.file_id', 'f.id')
|
||||
else query.join('binaries as b', 'b.file_id', 'f.id')
|
||||
if (this.nicovideo) {
|
||||
query.whereNull('f.content_type').whereRaw(`f.filename ~* '^\\d{9,11}_nicovideo_jp_watch_sm\\d+\\.tar\\.zst$'`)
|
||||
query.whereNull('d.content_type').whereRaw(`f.filename ~* '^\\d{9,11}_nicovideo_jp_watch_sm\\d+\\.tar\\.zst$'`)
|
||||
} else if (this.sfv) {
|
||||
query.whereNull('f.content_type').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.sfv$'`)
|
||||
query.whereNull('d.content_type').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.sfv$'`)
|
||||
} else if (this.archives) {
|
||||
query.whereNull('f.content_type').whereNull('f.archive').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.(rar|7z|7z\\.0*1|bin)$'`)
|
||||
query.whereNull('d.content_type').whereNull('d.archive').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.(rar|7z|7z\\.0*1|bin)$'`)
|
||||
} else if (this.sevenZip) {
|
||||
query.whereNull('f.archive').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.7z(\\.0*1){0,1}$'`)
|
||||
} else if (this.archiveNzbs) query.whereRaw(`f.archive::text ~* '\\.nzb"'`)
|
||||
else if (!this.all) query.whereNull('b.md5_16k').whereNull('f.archive')
|
||||
query.whereNull('d.archive').whereRaw(`coalesce(f.real_name, f.filename) ~* '\\.7z(\\.0*1){0,1}$'`)
|
||||
} else if (this.archiveNzbs) query.whereRaw(`d.archive::text ~* '\\.nzb"'`)
|
||||
else if (!this.all) query.whereNull('b.md5_16k').whereNull('d.archive')
|
||||
const files = await query
|
||||
.orderBy('f.id')
|
||||
.limit(BATCH)
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
import { BaseCommand } from '@adonisjs/core/ace'
|
||||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||||
import db from '@adonisjs/lucid/services/db'
|
||||
|
||||
/**
|
||||
* One-off (LOG.md 2026-10-08 (14)): copies the content types and archive
|
||||
* listings still on `files` into file_details, a partition at a time.
|
||||
* Values written since the deploy (by the new code, into file_details) win:
|
||||
* a file that has details already is skipped.
|
||||
*/
|
||||
export default class FilesMoveDetails extends BaseCommand {
|
||||
public static commandName = 'files:move-details'
|
||||
public static description = "Copies files' content types and archive listings into file_details."
|
||||
|
||||
public static options: CommandOptions = {
|
||||
startApp: true,
|
||||
}
|
||||
|
||||
public async run() {
|
||||
const { rows: partitions } = await db.rawQuery(
|
||||
`SELECT c.relname FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid WHERE i.inhparent = 'files'::regclass ORDER BY c.relname`
|
||||
)
|
||||
let total = 0
|
||||
for (const { relname } of partitions) {
|
||||
const started = Date.now()
|
||||
const result = await db.rawQuery(
|
||||
`INSERT INTO file_details (file_id, content_type, archive)
|
||||
SELECT id, content_type, archive FROM ${relname} WHERE content_type IS NOT NULL OR archive IS NOT NULL
|
||||
ON CONFLICT (file_id) DO NOTHING`
|
||||
)
|
||||
total += result.rowCount ?? 0
|
||||
this.logger.info(`${relname}: ${result.rowCount} copied in ${Math.round((Date.now() - started) / 1000)} s.`)
|
||||
}
|
||||
this.logger.info(`${total} files' details copied.`)
|
||||
}
|
||||
}
|
||||
@@ -70,7 +70,7 @@ export default class ReleaseRebuild extends BaseCommand {
|
||||
AND NOT (v.release_id IS NULL
|
||||
AND b.release_id IN (SELECT id FROM releases WHERE source IN ('par2', 'rar')))
|
||||
-- A posted NZB decided this grouping (NzbImportService); name rules don't override it.
|
||||
AND NOT EXISTS (SELECT 1 FROM files n WHERE n.release_id = b.release_id AND n.content_type = 'nzb')`,
|
||||
AND NOT EXISTS (SELECT 1 FROM files n JOIN file_details nd ON nd.file_id = n.id WHERE n.release_id = b.release_id AND nd.content_type = 'nzb')`,
|
||||
[ids, releaseIdsPerBinary] as any
|
||||
)
|
||||
})
|
||||
@@ -85,12 +85,13 @@ export default class ReleaseRebuild extends BaseCommand {
|
||||
const rows = await db
|
||||
.from('files as f')
|
||||
.join('binaries as b', 'b.file_id', 'f.id')
|
||||
.whereNotNull('f.archive')
|
||||
.join('file_details as d', 'd.file_id', 'f.id')
|
||||
.whereNotNull('d.archive')
|
||||
.whereNull('b.release_id')
|
||||
.where('f.id', '>', rarCursor)
|
||||
.orderBy('f.id')
|
||||
.limit(BATCH)
|
||||
.select('f.id as file_id', 'f.archive', 'b.id as binary_id', 'b.poster', 'b.groups', 'b.posted_at')
|
||||
.select('f.id as file_id', 'd.archive', 'b.id as binary_id', 'b.poster', 'b.groups', 'b.posted_at')
|
||||
if (rows.length === 0) break
|
||||
rarCursor = Number(rows.at(-1).file_id)
|
||||
const segments = rows.map((r) => ({
|
||||
@@ -147,7 +148,8 @@ export default class ReleaseRebuild extends BaseCommand {
|
||||
await db.transaction(async (trx) => {
|
||||
await trx.rawQuery('DELETE FROM release_files')
|
||||
await trx.rawQuery(`UPDATE binaries SET real_name = NULL WHERE real_name IS NOT NULL`)
|
||||
await trx.rawQuery(`UPDATE files SET real_name = NULL, archive = NULL WHERE real_name IS NOT NULL OR archive IS NOT NULL`)
|
||||
await trx.rawQuery(`UPDATE files SET real_name = NULL WHERE real_name IS NOT NULL`)
|
||||
await trx.rawQuery(`UPDATE file_details SET archive = NULL WHERE archive IS NOT NULL`)
|
||||
await trx.rawQuery(`UPDATE releases SET contents = NULL, encrypted = NULL WHERE contents IS NOT NULL OR encrypted IS NOT NULL`)
|
||||
// Binaries whose name rules find no release were grouped from contents (or wrongly by size).
|
||||
await trx.rawQuery(`
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
import { BaseSchema } from '@adonisjs/lucid/schema'
|
||||
|
||||
/**
|
||||
* What inspection learns about a file (content type, archive listing) lives
|
||||
* in file_details, on the NVMe, instead of in `files` (LOG.md 2026-10-08 (14)).
|
||||
* `files` sits on the HDD and was updated after its insert by every probe:
|
||||
* a new row version on a random page and an entry in each of its indexes
|
||||
* (3.7M updates against 49.5M inserts in a day, 27% HOT). Its rows are now
|
||||
* only inserted (and moved between releases or renamed, rarely).
|
||||
* The existing values are copied by `files:move-details` after this deploy;
|
||||
* a later migration drops the columns.
|
||||
*/
|
||||
export default class extends BaseSchema {
|
||||
async up() {
|
||||
// The foreign key locks `files` briefly: take the lock first, so it can't deadlock with writers.
|
||||
await this.db.rawQuery(`SET LOCAL lock_timeout = '60s'`)
|
||||
await this.db.rawQuery(`LOCK TABLE files IN SHARE ROW EXCLUSIVE MODE`)
|
||||
await this.db.rawQuery(`
|
||||
CREATE TABLE IF NOT EXISTS file_details (
|
||||
file_id bigint PRIMARY KEY REFERENCES files (id) ON DELETE CASCADE,
|
||||
content_type varchar(16),
|
||||
archive jsonb)`)
|
||||
}
|
||||
|
||||
async down() {
|
||||
await this.db.rawQuery(`DROP TABLE IF EXISTS file_details`)
|
||||
}
|
||||
}
|
||||
@@ -187,6 +187,17 @@ export class FetchRangeSchema extends BaseModel {
|
||||
declare updatedAt: DateTime
|
||||
}
|
||||
|
||||
export class FileDetailSchema extends BaseModel {
|
||||
static $columns = ['archive', 'contentType', 'fileId'] as const
|
||||
$columns = FileDetailSchema.$columns
|
||||
@column()
|
||||
declare archive: any | null
|
||||
@column()
|
||||
declare contentType: string | null
|
||||
@column({ isPrimary: true })
|
||||
declare fileId: bigint | number
|
||||
}
|
||||
|
||||
export class FileKeySchema extends BaseModel {
|
||||
static $columns = ['keyHash'] as const
|
||||
$columns = FileKeySchema.$columns
|
||||
|
||||
@@ -212,7 +212,7 @@ test.group('Seed API', (group) => {
|
||||
assert.equal(row.name, 'Real.Release.Name.2026.1080p-GRP')
|
||||
assert.isTrue(row.encrypted)
|
||||
assert.lengthOf(row.contents, 1)
|
||||
assert.isOk((await db.from('files').where('id', file.id).first()).archive)
|
||||
assert.isOk((await db.from('file_details').where('file_id', file.id).first())?.archive)
|
||||
assert.equal(await (await Job.fromId(QueueService.collectionQueue, job.jobId))!.getState(), 'completed')
|
||||
// Too late (or twice): the lock is gone.
|
||||
;(await client.post('/api/v1/seed/inspect/done').json(done).headers(auth)).assertStatus(409)
|
||||
@@ -252,7 +252,7 @@ test.group('Seed API', (group) => {
|
||||
const job = (await client.post('/api/v1/seed/inspect/lease').json({ max: 1 }).headers(auth)).body().jobs[0]
|
||||
const error = 'Seed API /api/v1/seed/articles?id=gone%40test: 410 {"error":"No such article"}'
|
||||
;(await client.post('/api/v1/seed/inspect/failed').json({ queue: job.queue, jobId: job.jobId, token: job.token, error }).headers(auth)).assertStatus(200)
|
||||
assert.equal((await db.from('files').where('id', file.id).first()).content_type, 'missing')
|
||||
assert.equal((await db.from('file_details').where('file_id', file.id).first())?.content_type, 'missing')
|
||||
assert.equal(await (await Job.fromId(QueueService.collectionQueue, job.jobId))!.getState(), 'completed')
|
||||
})
|
||||
|
||||
|
||||
Reference in new issue
Block a user