diff --git a/app/services/BinaryFileService.ts b/app/services/BinaryFileService.ts new file mode 100644 index 0000000..74b748d --- /dev/null +++ b/app/services/BinaryFileService.ts @@ -0,0 +1,141 @@ +import db from '@adonisjs/lucid/services/db' +import FileStore from '#services/FileStore' +import SfvService from '#services/SfvService' +import QueueService from '#services/QueueService' +import ContentService from '#services/ContentService' + +export interface BinaryFilesResult { + /** New files. */ + saved: number + /** Binaries whose file already existed (same first segment or file_key). */ + existing: number + /** Conflicted, gone or already linked: nothing to do. */ + skipped: number + /** Not all parts stored yet (the job is retried for these). */ + incomplete: number[] + inspections: number +} + +/** + * Turns completed binaries into files (worker:file), a batch at a time + * (LOG.md 2026-10-06 (7)): one query each for the binaries, their parts and + * files that exist already, one INSERT for the new files (so the files + * statement trigger updates each release once), and one UPDATE and DELETE + * for the binaries' links and copied parts. It was one job of ~8 queries per + * binary, ~700 a minute. + */ +class BinaryFileService { + async store(binaryIds: number[]): Promise { + const result: BinaryFilesResult = { saved: 0, existing: 0, skipped: 0, incomplete: [], inspections: 0 } + const ids = [...new Set(binaryIds)] + const binaries = await db + .from('binaries') + .whereIn('id', ids) + .whereNull('conflicted_at') + .whereNull('file_id') + .select('id', 'file_key', 'name', 'real_name', 'poster', 'posted_at', 'groups', 'release_id', 'total_parts') + result.skipped = ids.length - binaries.length + if (binaries.length === 0) return result + + const partsOf = new Map() + const parts = await db + .from('parts') + .whereIn('binary_id', binaries.map((b) => b.id)) + .orderBy([{ column: 'binary_id' }, { column: 'number' }]) + .select('binary_id', 'number', 'message_id', 'bytes') + for (const part of parts) { + const list = partsOf.get(Number(part.binary_id)) ?? [] + list.push({ number: Number(part.number), id: part.message_id, size: Number(part.bytes) }) + partsOf.set(Number(part.binary_id), list) + } + const complete = binaries.filter((b) => { + const ok = (partsOf.get(Number(b.id))?.length ?? 0) >= b.total_parts + if (!ok) result.incomplete.push(Number(b.id)) + return ok + }) + if (complete.length === 0) return result + + // Already known as a file, e.g. added from a posted NZB before its own articles completed (id only: the row holds the segment list). + const firsts = complete.map((b) => partsOf.get(Number(b.id))![0].id) + const byFirst = new Map() + for (const row of await db.from('files').whereIn('first_message_id', firsts).select('id', 'first_message_id')) { + byFirst.set(row.first_message_id, Number(row.id)) + } + const fileIdOf = new Map() + const fresh = complete.filter((b) => { + const existing = byFirst.get(partsOf.get(Number(b.id))![0].id) + if (existing) fileIdOf.set(Number(b.id), existing) + return !existing + }) + + const now = new Date() + // file_key is unique: a second completion of the same upload inserts nothing. + const inserted = await FileStore.insert( + fresh.map((b) => { + const segments = partsOf.get(Number(b.id))! + return { + file_key: b.file_key, + filename: b.name, + poster: b.poster, + date: b.posted_at ? new Date(b.posted_at).getTime() : now.getTime(), + parts: segments.length, + segments, + groups: JSON.stringify(b.groups), + release_id: b.release_id, + real_name: b.real_name, + size: segments.reduce((sum, s) => sum + s.size, 0), + created_at: now, + updated_at: now, + } + }) + ) + const insertedByKey = new Map(inserted.map((row) => [row.file_key, row.id])) + const missing = fresh.filter((b) => !insertedByKey.has(b.file_key)).map((b) => b.file_key) + const byKey = new Map() + if (missing.length > 0) { + for (const row of await db.from('files').whereIn('file_key', missing).select('id', 'file_key')) byKey.set(row.file_key, Number(row.id)) + } + for (const b of fresh) { + const id = insertedByKey.get(b.file_key) ?? byKey.get(b.file_key) + if (id) fileIdOf.set(Number(b.id), id) + } + result.saved = inserted.length + result.existing = fileIdOf.size - inserted.length + + // Links, locked in file_key order like ingest's upsert, so the two can't deadlock. + const linked = complete.filter((b) => fileIdOf.has(Number(b.id))).sort((a, b) => a.file_key.localeCompare(b.file_key)) + if (linked.length > 0) { + await db.transaction(async (trx) => { + await trx.rawQuery( + `SELECT id FROM binaries WHERE id IN (SELECT unnest(?::bigint[])) ORDER BY file_key FOR UPDATE`, + [linked.map((b) => b.id)] + ) + await trx.rawQuery( + `UPDATE binaries b SET file_id = v.file_id FROM unnest(?::bigint[], ?::bigint[]) AS v(id, file_id) WHERE b.id = v.id`, + [linked.map((b) => b.id), linked.map((b) => fileIdOf.get(Number(b.id))!)] + ) + }) + // Header-derived parts are in the file's segment list now; body-derived ones (with a CRC) stay for split checks (D20). + await db.rawQuery(`DELETE FROM parts WHERE binary_id = ANY(?::bigint[]) AND pcrc32 IS NULL`, [linked.map((b) => b.id)]) + } + + // As for one file before: an SFV posted earlier may list them (D34), then inspection. + const fresher = new Map(fresh.map((b) => [b.file_key, b])) + const added = inserted.flatMap((row) => { + const b = row.file_key ? fresher.get(row.file_key) : undefined + return b + ? [{ id: row.id, name: b.real_name ?? b.name, posted: String(b.name), poster: b.poster, date: b.posted_at ? new Date(b.posted_at).getTime() : null, release_id: b.release_id }] + : [] + }) + if (added.length > 0) await SfvService.claimAll(added.map(({ posted: _p, ...file }) => file)) + // By the posted name, as before: inspection skips later volumes and readable media names anyway. + const jobs = added + .filter((file) => ContentService.isCandidate(file.posted)) + .map((file) => ({ name: 'process-collection', data: { fileId: file.id }, opts: { jobId: `file-${file.id}` } })) + if (jobs.length > 0) await QueueService.collectionQueue.addBulk(jobs) + result.inspections = jobs.length + return result + } +} + +export default new BinaryFileService() diff --git a/app/services/FileStore.ts b/app/services/FileStore.ts index b9bdeb8..5ec0373 100644 --- a/app/services/FileStore.ts +++ b/app/services/FileStore.ts @@ -42,17 +42,29 @@ export const insertLimits = { rows: 2000 } class FileStore { /** * Inserts files, skipping those whose `file_key` exists. Returns the rows - * that were inserted (id and filename). `files` is partitioned by id + * 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: - * 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. + * 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 + * 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 }[]> { + async insert( + files: NewFile[], + client: Pick = db + ): Promise<{ id: number; filename: string; file_key: string | null }[]> { 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() 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') + const chunk = keys.slice(i, i + 5000) + 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))`, + [chunk] + ) + const maybe = rows.map((row: { key: string }) => row.key) + if (maybe.length === 0) continue + const found = await client.knexQuery().from('files').whereIn('file_key', maybe).select('file_key') for (const row of found) known.add(row.file_key) } const seen = new Set() @@ -70,10 +82,10 @@ class FileStore { }) if (rows.length === 0) return [] // A statement takes at most 65,535 parameters: ~5,400 rows of 12 columns. - const inserted: { id: number; filename: string }[] = [] + 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']) - for (const row of chunk) inserted.push({ id: Number(row.id), filename: row.filename }) + 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 }) } return inserted } diff --git a/app/services/IngestService.ts b/app/services/IngestService.ts index 8f65053..a76d4c4 100644 --- a/app/services/IngestService.ts +++ b/app/services/IngestService.ts @@ -85,6 +85,9 @@ async function upsertKeyed( } } } +/** Completed binaries per worker:file job. */ +const FILE_BATCH = 100 + /** Size-only matching below this is too ambiguous (nfo, sfv, small volumes). */ export const PAR2_SIZE_ONLY_MIN = 1024 * 1024 /** A par2 set and its files are posted together; don't match across days. */ @@ -187,8 +190,29 @@ class IngestService { const newParts = partRows.filter((row) => !(done.has(row.binary_id) && row.pcrc32 === null)) newParts.sort((a, b) => a.binary_id - b.binary_id || a.number - b.number) + // Parts and the binaries' stored-part counts in one statement, so the count can't miss a + // part (LOG.md 2026-10-06 (7)); claimCompleted compares it instead of counting parts. NULL + // (binaries from before part_count) is counted once: the statement's own rows aren't in + // its snapshot, so they're added on top. Concurrent batches can lock binaries in a + // different order: a deadlock rolls the whole statement back, and it is retried. for (const chunk of chunks(newParts, PART_CHUNK)) { - await knex('parts').insert(chunk).onConflict(['binary_id', 'number']).ignore() + const insert = knex('parts').insert(chunk).onConflict(['binary_id', 'number']).ignore().returning('binary_id').toSQL() + for (let attempt = 1; ; attempt++) { + try { + await knex.raw( + `WITH added AS (${insert.sql}), + counted AS (SELECT binary_id, count(*)::int AS n FROM added GROUP BY binary_id) + UPDATE binaries b SET part_count = CASE + WHEN b.part_count IS NULL THEN (SELECT count(*) FROM parts p WHERE p.binary_id = b.id) + c.n + ELSE b.part_count + c.n END + FROM counted c WHERE b.id = c.binary_id`, + insert.bindings as any[] + ) + break + } catch (error: any) { + if (error.code !== '40P01' || attempt >= 5) throw error + } + } } if (weakRows.length > 0) await this.flagConflicts(db.connection(), weakRows) @@ -366,7 +390,7 @@ class IngestService { const result = await db.rawQuery( `UPDATE binaries b SET completed_at = now() WHERE b.id = ANY(?) AND b.completed_at IS NULL AND b.conflicted_at IS NULL - AND (SELECT count(*) FROM parts p WHERE p.binary_id = b.id) >= b.total_parts + AND coalesce(b.part_count, (SELECT count(*) FROM parts p WHERE p.binary_id = b.id)) >= b.total_parts RETURNING b.id, b.file_key`, [binaryIds] ) @@ -378,14 +402,14 @@ class IngestService { async enqueue(completed: CompletedBinary[]) { if (completed.length === 0) return - // jobId = file_key: BullMQ ignores an add() for an id that already exists. - await QueueService.fileQueue.addBulk( - completed.map(({ id, fileKey }) => ({ - name: 'process-file', - data: { binaryId: id }, - opts: { jobId: fileKey }, - })) - ) + // A job per FILE_BATCH binaries: worker:file stores them with a few set-based queries (LOG.md 2026-10-06 (7)). + // jobId from the first file_key: BullMQ ignores an add() for an id that already exists. + const jobs = [] + for (let i = 0; i < completed.length; i += FILE_BATCH) { + const batch = completed.slice(i, i + FILE_BATCH) + jobs.push({ name: 'process-files', data: { binaryIds: batch.map((b) => b.id) }, opts: { jobId: `${batch[0].fileKey}-${batch.length}` } }) + } + await QueueService.fileQueue.addBulk(jobs) } } diff --git a/app/services/SeedOrchestrator.ts b/app/services/SeedOrchestrator.ts index beafbf8..16923cd 100644 --- a/app/services/SeedOrchestrator.ts +++ b/app/services/SeedOrchestrator.ts @@ -1,5 +1,6 @@ import { createHash, randomBytes } from 'node:crypto' import db from '@adonisjs/lucid/services/db' +import { cut, uncovered } from '#lib/seed/plan' import NntpService from '#services/NntpService' import SeedService from '#services/SeedService' import BackfillService from '#services/BackfillService' @@ -77,7 +78,7 @@ class SeedOrchestrator { WHERE id = ( SELECT id FROM seed_ranges WHERE status = 'queued' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)) - ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED) + ORDER BY priority DESC, id LIMIT 1 FOR UPDATE SKIP LOCKED) RETURNING id, group_name, next_id, end_id`, [nodeId, LEASE_TTL_SECONDS] ) @@ -143,9 +144,12 @@ class SeedOrchestrator { * server's oldest article. A group's backfill stops where its seed plan * begins. Ranges are interleaved across groups, newest first. */ - async plan(groups: string[], options: { days?: number; size: number }): Promise<{ group: string; from: number; to: number; ranges: number }[]> { + async plan( + groups: string[], + options: { days?: number; size: number; priority?: number } + ): Promise<{ group: string; from: number; to: number; ranges: number; gaps: number }[]> { const perGroup: { group: string; ranges: [number, number][] }[] = [] - const summary: { group: string; from: number; to: number; ranges: number }[] = [] + const summary: { group: string; from: number; to: number; ranges: number; gaps: number }[] = [] const conn = await NntpService.acquire() let failed = false try { @@ -159,20 +163,21 @@ class SeedOrchestrator { const date = header ? Date.parse(header.date) : Number.NaN return Number.isNaN(date) ? null : date } - // Top: where live indexing began, or the backfill's position, or below earlier seed ranges. + // Top: where live indexing began, or as far down as the backfill has come. let top = Number(await articleAtDate(first, last, new Date(group.created_at).getTime(), dateOf)) + if (group.backfill_next_id !== null) top = Math.min(top, Number(group.backfill_next_id)) const backfilling = group.backfill_next_id !== null && group.backfill_until_id !== null && Number(group.backfill_next_id) > Number(group.backfill_until_id) - if (backfilling) top = Math.min(top, Number(group.backfill_next_id)) - const seeded = await db.from('seed_ranges').where('group_name', name).min('start_id as low').first() - if (seeded?.low !== null && seeded?.low !== undefined) top = Math.min(top, Number(seeded.low)) const bottom = options.days ? Number(await articleAtDate(first, last, Date.now() - options.days * 24 * 60 * 60 * 1000, dateOf)) : Number(first) - const ranges: [number, number][] = [] - for (let end = top - 1; end >= bottom; end -= options.size) ranges.push([Math.max(bottom, end - options.size + 1), end]) + // Everything below the top that no earlier range covers (in any state). Planning below + // the lowest earlier range left a 13-month gap in a.b.ath (LOG.md 2026-10-06 (7)). + const planned = await db.from('seed_ranges').where('group_name', name).select('start_id', 'end_id') + const gaps = uncovered(bottom, top - 1, planned.map((r) => [Number(r.start_id), Number(r.end_id)] as [number, number])) + const ranges = cut(gaps, options.size) perGroup.push({ group: name, ranges }) - summary.push({ group: name, from: bottom, to: top - 1, ranges: ranges.length }) + summary.push({ group: name, from: bottom, to: top - 1, ranges: ranges.length, gaps: gaps.length }) // The seed plan takes over the rest of this group's backfill. if (backfilling && ranges.length > 0) { await db.from('groups').where('id', group.id).update({ backfill_until_id: group.backfill_next_id }) @@ -190,7 +195,7 @@ class SeedOrchestrator { for (let i = 0; perGroup.some((g) => i < g.ranges.length); i++) { for (const g of perGroup) { const range = g.ranges[i] - if (range) rows.push({ group_name: g.group, start_id: range[0], end_id: range[1], next_id: range[0] }) + if (range) rows.push({ group_name: g.group, start_id: range[0], end_id: range[1], next_id: range[0], priority: options.priority ?? 0 }) } } for (let i = 0; i < rows.length; i += 5000) await db.table('seed_ranges').multiInsert(rows.slice(i, i + 5000)) diff --git a/commands/SeedPlan.ts b/commands/SeedPlan.ts index 706cdc0..cccd1a6 100644 --- a/commands/SeedPlan.ts +++ b/commands/SeedPlan.ts @@ -4,9 +4,10 @@ import NntpService from '#services/NntpService' import SeedOrchestrator from '#services/SeedOrchestrator' /** - * Plans seed ranges (CONCEPT.md D38) for groups, from below what live - * indexing, the backfill and earlier plans cover down to `--days` back (or - * the server's oldest article). A group's backfill stops where its plan + * Plans seed ranges (CONCEPT.md D38) for groups: every article from below + * what live indexing and the backfill cover down to `--days` back (or the + * server's oldest article) that no earlier range covers. Running it again + * plans only what's missing. A group's backfill stops where its plan * starts. Ranges are interleaved across groups, newest first. * * node ace seed:plan alt.binaries.ath alt.binaries.mom --days 365 @@ -28,11 +29,16 @@ export default class SeedPlan extends BaseCommand { @flags.number({ description: 'Articles per range', default: 2_000_000 }) declare size: number + @flags.number({ description: 'Leased before ranges of a lower priority (default 0)', default: 0 }) + declare priority: number + public async run() { try { - const summary = await SeedOrchestrator.plan(this.groups, { days: this.days, size: this.size }) + const summary = await SeedOrchestrator.plan(this.groups, { days: this.days, size: this.size, priority: this.priority }) for (const s of summary) { - this.logger.info(`${s.group}: ${s.ranges.toLocaleString('en')} ranges, articles ${s.from.toLocaleString('en')}–${s.to.toLocaleString('en')}.`) + this.logger.info( + `${s.group}: ${s.ranges.toLocaleString('en')} new ranges in ${s.gaps} gap(s) not yet planned, articles ${s.from.toLocaleString('en')}–${s.to.toLocaleString('en')}.` + ) } } finally { await NntpService.shutdown() diff --git a/commands/workers/File.ts b/commands/workers/File.ts index 33bfb47..931309c 100644 --- a/commands/workers/File.ts +++ b/commands/workers/File.ts @@ -1,14 +1,11 @@ import { BaseCommand } from '@adonisjs/core/ace' import { CommandOptions } from '@adonisjs/core/types/ace' import { Worker } from 'bullmq' -import db from '@adonisjs/lucid/services/db' import env from '#start/env' -import FileStore from '#services/FileStore' -import SfvService from '#services/SfvService' +import BinaryFileService from '#services/BinaryFileService' import queueConfig from '#config/queue' import Shutdown from '#services/Shutdown' import QueueService from '#services/QueueService' -import ContentService from '#services/ContentService' export default class FileWorker extends BaseCommand { public static commandName = 'worker:file' @@ -21,83 +18,21 @@ export default class FileWorker extends BaseCommand { public async run() { this.logger.info('Starting file worker...') - const collectionQueue = QueueService.collectionQueue - const worker = new Worker('file-queue', async (job) => { - const { binaryId } = job.data - - const binary = await db.from('binaries').where('id', binaryId).first() - if (!binary) { - this.logger.warning(`Binary ${binaryId} no longer exists (garbage collected?).`) - return + // A batch of completed binaries ('process-files'), or one from before batching ('process-file'). + const ids: number[] = (job.data.binaryIds ?? [job.data.binaryId]).map(Number) + const result = await BinaryFileService.store(ids) + if (result.saved || result.existing) { + this.logger.info( + `Files: ${result.saved} saved, ${result.existing} existed already` + + (result.skipped ? `, ${result.skipped} skipped` : '') + + (result.inspections ? `, ${result.inspections} queued for inspection` : '') + + ` (${ids.length} binaries).` + ) } - - if (binary.conflicted_at) { - this.logger.warning(`Binary ${binaryId} ("${binary.name}") has conflicting parts, skipping.`) - return - } - - const parts = await db - .from('parts') - .where('binary_id', binaryId) - .orderBy('number') - .select('number', 'message_id', 'bytes') - if (parts.length < binary.total_parts) { - throw new Error(`Binary ${binaryId} has ${parts.length}/${binary.total_parts} parts.`) - } - - const segments = parts.map((part) => ({ number: Number(part.number), id: part.message_id, size: Number(part.bytes) })) - - // Already known as a file, e.g. added from a posted NZB before its own articles completed. - const existing = await db.from('files').where('first_message_id', segments[0]?.id ?? '').first() - if (existing) { - await db.from('binaries').where('id', binaryId).update({ file_id: existing.id }) - await dropCopiedParts(binary) - this.logger.debug(`File "${binary.name}" already exists as ID ${existing.id} (same first segment).`) - return - } - - // file_key is unique: a second completion of the same upload inserts nothing. - const now = new Date() - const [inserted] = await FileStore.insert([ - { - file_key: binary.file_key, - filename: binary.name, - poster: binary.poster, - date: binary.posted_at ? new Date(binary.posted_at).getTime() : now.getTime(), - parts: parts.length, - segments, - groups: JSON.stringify(binary.groups), - release_id: binary.release_id, - real_name: binary.real_name, - size: parts.reduce((sum, part) => sum + Number(part.bytes), 0), - created_at: now, - updated_at: now, - }, - ]) - - const fileId = - inserted?.id ?? (await db.from('files').where('file_key', binary.file_key).first())?.id - await db.from('binaries').where('id', binaryId).update({ file_id: fileId }) - await dropCopiedParts(binary) - - if (!inserted) { - this.logger.debug(`File "${binary.name}" (${binary.file_key}) already exists as ID ${fileId}.`) - return - } - - this.logger.info(`Saved file "${binary.name}" (${parts.length} parts) as ID ${fileId}.`) - // An SFV posted earlier may already list it (D34). - await SfvService.claim({ - id: Number(fileId), - name: binary.real_name ?? binary.name, - poster: binary.poster, - date: binary.posted_at ? new Date(binary.posted_at).getTime() : null, - release_id: binary.release_id, - }) - // Files inspection would skip anyway (later volumes, readable media names) don't take a queue slot. - if (ContentService.isCandidate(binary.name)) { - await collectionQueue.add('process-collection', { fileId }, { jobId: `file-${fileId}` }) + // The rest is stored; the job's retry picks these up (the others are linked now and skipped). + if (result.incomplete.length > 0) { + throw new Error(`${result.incomplete.length} of ${ids.length} binaries don't have all their parts yet.`) } }, { connection: queueConfig.connection, @@ -117,20 +52,3 @@ export default class FileWorker extends BaseCommand { await QueueService.closeAll() } } - -/** - * The file holds every message id now. Binaries identified by their subject - * (no size: header-derived) don't need their parts any more; body-derived - * ones keep them, so articles of an equal-sized next volume are recognised - * as a conflict and split off (CONCEPT.md D20). LOG.md 2026-09-28 (17). - */ -/** - * The file holds the segment list now. Header-derived parts (no per-part - * CRC) aren't needed any more; body-derived ones are kept for split-volume - * verification (D20). Going by the CRC, not the binary's size: content - * inspection gives header-derived binaries a size too, which kept ~20M - * parts around (LOG.md 2026-10-02). - */ -async function dropCopiedParts(binary: { id: number }) { - await db.from('parts').where('binary_id', binary.id).whereNull('pcrc32').delete() -} diff --git a/database/migrations/1768620765200_seed_range_priority.ts b/database/migrations/1768620765200_seed_range_priority.ts new file mode 100644 index 0000000..3706593 --- /dev/null +++ b/database/migrations/1768620765200_seed_range_priority.ts @@ -0,0 +1,14 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** Seed ranges of a higher priority are leased first (`seed:plan --priority`), e.g. a gap in recent history. */ +export default class extends BaseSchema { + async up() { + this.schema.alterTable('seed_ranges', (table) => { + table.smallint('priority').notNullable().defaultTo(0) + }) + } + + async down() { + this.schema.alterTable('seed_ranges', (table) => table.dropColumn('priority')) + } +} diff --git a/database/migrations/1768620765210_binary_part_count.ts b/database/migrations/1768620765210_binary_part_count.ts new file mode 100644 index 0000000..6bfdf10 --- /dev/null +++ b/database/migrations/1768620765210_binary_part_count.ts @@ -0,0 +1,21 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Stored parts per binary, kept by IngestService as it inserts parts + * (LOG.md 2026-10-06 (7)). claimCompleted counted every binary's parts in + * every batch (15% of Postgres' active time); a binary of thousands of + * parts that gets a few per batch was counted thousands of times. NULL: + * not counted yet (binaries from before), counted exactly on the next new + * part. Only parts of binaries without a file are deleted, so it stays exact + * until a binary completes. + */ +export default class extends BaseSchema { + async up() { + await this.db.rawQuery(`SET LOCAL lock_timeout = '2min'`) + await this.db.rawQuery(`ALTER TABLE binaries ADD COLUMN IF NOT EXISTS part_count integer`) + } + + async down() { + await this.db.rawQuery(`ALTER TABLE binaries DROP COLUMN IF EXISTS part_count`) + } +} diff --git a/database/migrations/1768620765220_file_keys.ts b/database/migrations/1768620765220_file_keys.ts new file mode 100644 index 0000000..154a7c8 --- /dev/null +++ b/database/migrations/1768620765220_file_keys.ts @@ -0,0 +1,68 @@ +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. + * + * 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). + */ +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)`) + 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; + INSERT INTO file_keys (key_hash) + SELECT DISTINCT hashtextextended(file_key, 0) FROM new_rows WHERE file_key IS NOT NULL + 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`) + } +} diff --git a/database/schema.ts b/database/schema.ts index 26d22f6..3dee504 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -33,7 +33,7 @@ export class AuthAccessTokenSchema extends BaseModel { } export class BinarySchema extends BaseModel { - static $columns = ['completedAt', 'conflictedAt', 'fileId', 'fileKey', 'firstSeenAt', 'groups', 'id', 'lastSeenAt', 'md516K', 'name', 'parentBinaryId', 'postedAt', 'poster', 'realName', 'releaseId', 'size', 'splitAt', 'subject', 'totalParts'] as const + static $columns = ['completedAt', 'conflictedAt', 'fileId', 'fileKey', 'firstSeenAt', 'groups', 'id', 'lastSeenAt', 'md516K', 'name', 'parentBinaryId', 'partCount', 'postedAt', 'poster', 'realName', 'releaseId', 'size', 'splitAt', 'subject', 'totalParts'] as const $columns = BinarySchema.$columns @column.dateTime() declare completedAt: DateTime | null @@ -57,6 +57,8 @@ export class BinarySchema extends BaseModel { declare name: string @column() declare parentBinaryId: bigint | number | null + @column() + declare partCount: number | null @column.dateTime() declare postedAt: DateTime | null @column() @@ -172,6 +174,13 @@ export class FetchRangeSchema extends BaseModel { declare updatedAt: DateTime } +export class FileKeySchema extends BaseModel { + static $columns = ['keyHash'] as const + $columns = FileKeySchema.$columns + @column({ isPrimary: true }) + declare keyHash: bigint | number +} + 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 @@ -486,7 +495,7 @@ export class SeedNodeSchema extends BaseModel { } export class SeedRangeSchema extends BaseModel { - static $columns = ['attempts', 'createdAt', 'doneAt', 'endId', 'error', 'files', 'fragments', 'groupName', 'headers', 'heartbeatAt', 'id', 'leasedAt', 'nextId', 'nodeId', 'startId', 'status'] as const + static $columns = ['attempts', 'createdAt', 'doneAt', 'endId', 'error', 'files', 'fragments', 'groupName', 'headers', 'heartbeatAt', 'id', 'leasedAt', 'nextId', 'nodeId', 'priority', 'startId', 'status'] as const $columns = SeedRangeSchema.$columns @column() declare attempts: number @@ -517,6 +526,8 @@ export class SeedRangeSchema extends BaseModel { @column() declare nodeId: number | null @column() + declare priority: number + @column() declare startId: bigint | number @column() declare status: string diff --git a/lib/seed/plan.ts b/lib/seed/plan.ts new file mode 100644 index 0000000..6916453 --- /dev/null +++ b/lib/seed/plan.ts @@ -0,0 +1,27 @@ +/** + * Seed planning arithmetic (CONCEPT.md D38), independent of the database. + */ + +/** The parts of `from`..`to` (inclusive) that no range in `covered` (inclusive pairs) overlaps, oldest first. */ +export function uncovered(from: number, to: number, covered: [number, number][]): [number, number][] { + const gaps: [number, number][] = [] + let next = from + for (const [start, end] of [...covered].sort((a, b) => a[0] - b[0])) { + if (end < next) continue + if (start > to) break + if (start > next) gaps.push([next, Math.min(start - 1, to)]) + next = Math.max(next, end + 1) + if (next > to) return gaps + } + if (next <= to) gaps.push([next, to]) + return gaps +} + +/** `gaps` cut into ranges of at most `size` articles, newest first (from the top of the newest gap down). */ +export function cut(gaps: [number, number][], size: number): [number, number][] { + const ranges: [number, number][] = [] + for (const [from, to] of [...gaps].sort((a, b) => b[1] - a[1])) { + for (let end = to; end >= from; end -= size) ranges.push([Math.max(from, end - size + 1), end]) + } + return ranges +} diff --git a/tests/functional/binary_files.spec.ts b/tests/functional/binary_files.spec.ts new file mode 100644 index 0000000..105fe8d --- /dev/null +++ b/tests/functional/binary_files.spec.ts @@ -0,0 +1,68 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import IngestService, { type IngestSegment } from '#services/IngestService' +import BinaryFileService from '#services/BinaryFileService' +import QueueService from '#services/QueueService' +import FileStore, { segmentsOf } from '#services/FileStore' +import { fileKey } from '#lib/subject/index' + +function segment(name: string, number: number, total = 3): IngestSegment { + const poster = 'batch-test@example.invalid' + const subject = `"${name}" yEnc` + return { fileKey: fileKey(poster, subject, total), poster, subject, name, totalParts: total, number, messageId: `${name}-${number}@test`, bytes: 100, groups: ['alt.binaries.test'] } +} + +const binary = (name: string, total = 3) => db.from('binaries').where('file_key', segment(name, 1, total).fileKey).first() + +test.group('Binaries to files', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + group.teardown(() => QueueService.closeAll()) + + test('ingest counts stored parts, once each, and claims a binary by that count', async ({ assert }) => { + await IngestService.ingest([segment('a.bin', 1), segment('a.bin', 2)]) + // The same article again (a crosspost): not counted twice. + await IngestService.ingest([segment('a.bin', 2)]) + assert.equal((await binary('a.bin')).part_count, 2) + assert.isNull((await binary('a.bin')).completed_at) + const completed = await IngestService.ingest([segment('a.bin', 3)]) + assert.equal((await binary('a.bin')).part_count, 3) + assert.deepEqual(completed.map((c) => c.fileKey), [segment('a.bin', 1).fileKey]) + }) + + test('a binary from before the count is counted exactly on its next part', async ({ assert }) => { + await IngestService.ingest([segment('b.bin', 1), segment('b.bin', 2)]) + await db.from('binaries').where('id', (await binary('b.bin')).id).update({ part_count: null }) + const completed = await IngestService.ingest([segment('b.bin', 3)]) + assert.equal((await binary('b.bin')).part_count, 3) + assert.lengthOf(completed, 1) + }) + + test('stores a batch: new files, existing ones linked, incomplete ones left for the retry', async ({ assert }) => { + await IngestService.ingest([1, 2, 3].flatMap((n) => [segment('new.bin', n), segment('known.bin', n)])) + await IngestService.ingest([segment('half.bin', 1)]) + // known.bin is a file already (e.g. from an NZB), found by its first segment. + const now = new Date() + const [known] = await FileStore.insert([ + { filename: 'known.bin', poster: 'x', date: now.getTime(), parts: 3, groups: '[]', segments: [1, 2, 3].map((n) => ({ number: n, id: `known.bin-${n}@test`, size: 100 })), created_at: now, updated_at: now }, + ]) + const ids = [await binary('new.bin'), await binary('known.bin'), await binary('half.bin')].map((b) => Number(b.id)) + + const result = await BinaryFileService.store(ids) + assert.include(result, { saved: 1, existing: 1 }) + assert.deepEqual(result.incomplete, [ids[2]]) + + const stored = await db.from('files').where('file_key', segment('new.bin', 1).fileKey).first() + assert.deepEqual(segmentsOf(stored).map((s) => s.number), [1, 2, 3]) + assert.equal(Number((await binary('new.bin')).file_id), Number(stored.id)) + assert.equal(Number((await binary('known.bin')).file_id), known.id) + // Header parts are in the file's segment list now. + const [{ n }] = await db.from('parts').whereIn('binary_id', [ids[0], ids[1]]).count('* as n') + assert.equal(Number(n), 0) + + // The retry: done ones are skipped, the incomplete one again incomplete. + const again = await BinaryFileService.store(ids) + assert.include(again, { saved: 0, skipped: 2 }) + assert.deepEqual(again.incomplete, [ids[2]]) + }) +}) diff --git a/tests/functional/seed_api.spec.ts b/tests/functional/seed_api.spec.ts index de6fb9e..ff6fd00 100644 --- a/tests/functional/seed_api.spec.ts +++ b/tests/functional/seed_api.spec.ts @@ -134,6 +134,16 @@ test.group('Seed API', (group) => { } }) + test('ranges of a higher priority are leased first', async ({ client, assert }) => { + const token = await SeedOrchestrator.issueToken('priority-node') + await db.table('seed_ranges').insert([ + { group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 }, + { group_name: 'alt.binaries.test', start_id: 101, end_id: 200, next_id: 101, priority: 1 }, + ]) + const leased = await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${token}` }) + assert.include(leased.body().range, { start: 101, end: 200 }) + }) + test('a node inspects a file: lease, probe result, applied here (D41)', async ({ client, assert }) => { await QueueService.collectionQueue.drain() await QueueService.archiveWalkQueue.drain() diff --git a/tests/unit/seed_plan.spec.ts b/tests/unit/seed_plan.spec.ts new file mode 100644 index 0000000..1b9536f --- /dev/null +++ b/tests/unit/seed_plan.spec.ts @@ -0,0 +1,16 @@ +import { test } from '@japa/runner' +import { cut, uncovered } from '#lib/seed/plan' + +test.group('seed plan', () => { + test('plans only what earlier ranges leave uncovered', ({ assert }) => { + // a.b.ath on 2026-10-06: test ranges around 49.8B and a plan below them; the backfill ends at 58.4B. + assert.deepEqual(uncovered(1, 100, [[40, 50], [10, 20], [45, 60]]), [[1, 9], [21, 39], [61, 100]]) + assert.deepEqual(uncovered(1, 100, []), [[1, 100]]) + assert.deepEqual(uncovered(1, 100, [[0, 200]]), []) + assert.deepEqual(uncovered(10, 20, [[1, 5], [25, 30]]), [[10, 20]]) + }) + + test('cuts gaps into ranges, newest first', ({ assert }) => { + assert.deepEqual(cut([[1, 9], [61, 100]], 25), [[76, 100], [61, 75], [1, 9]]) + }) +})