diff --git a/REBUILD.md b/REBUILD.md index 694a94e..cd8fd02 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -472,6 +472,55 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md verdicts, file hashes. Resets are paused (2026-10-12); this one waits for the user's go-ahead, and for the new NVMe. +### Schema (built 2026-10-25; migrations 6000 and 6100) + +Migration 6000 *is* the reset: it truncates every table the indexing +recomputes and refuses to run on a catalog with releases unless +`VOYAGER3_RESET=1`, so merging `voyager-3` into `rebuild` is the go-ahead. +Survives: `groups`, `ranges` (planned again from their start, with their +priorities, `capture`, `est_date`, `stored_at`, `plain_chunks` and measured +`posted_low/high`), the corpus, `probe_cache`, `spots` (unlinked from +releases; the inspector matches them again), `nntp_shares`, `nodes`, +`users`, the rule sets. + +- `releases`: created anew, 8-byte columns first: `id`, `posted_at`, + `last_posted_at`, `first_article`, `last_article`, `segments`, `bytes`, + `settled_at` (the finish; null while a tail release is provisional), + `inspected_at`, `first_seen_at`; `group_id`, `group_set_id`, `poster_id`, + `file_total`, `files`, `chunks`, `category`; `source smallint` + (`lib/release/source.ts`: 1 subject, 2 post, 3 stem, 4 spot, 5 nzb, 6 + album); `readable`, `has_par2`, `complete`, `passworded`, `sidecar_only`, + `edge` (open at its block's end); `release_key` and `identity_key` as + 20-byte `bytea`; `name`, `inner_name`. Gone: `poster`, `groups`, + `last_seen_at`. Indexes: pkey, release_key unique, (identity_key, + group_id), name and inner-name trigram, newest, (poster_id, newest), + passworded, `(group_id, identity_key) WHERE edge` for the block merger, + `id WHERE settled_at IS NULL` for the provisional ones. The category + trigger reads the groups from `group_sets`; the judge's dirty trigger + watches `poster_id` instead of `poster`. +- `posters(id serial, poster unique)` with its trigram index; + `group_sets(id serial, groups jsonb unique)`, a sorted array of names. +- `blocks` (6100): `group_id`, `start_id`, `end_id`, `next_id` (the resume + point), `posted_low/high`, `status`, `priority`, `node_id`, `attempts`, + `rule_version`, `headers`, `releases`, `leased_at`, `heartbeat_at`, + `done_at`, `error`; indexes todo (priority, id) and leased. + `ranges.block_id` points at the block; the ranges' own lease columns stay + until the leases move (then a migration drops them). +- `release_chunks`, `rule_counts`, `rule_samples`: `range_id` → `block_id`. +- Gone: `fragments`, `range_blocks`, `release_final()` (final = finished: + `settled_at IS NOT NULL`), the settler, the fragment merger, + `releases_category_newest_idx`. +- `release_inspections` and `probe_candidates` on `CHUNK_TABLESPACE` (the + HDD) with their indexes. Note LOG.md 2026-10-19 (81): the judge's fake + rule read all of `release_inspections` every round, 17 min on the HDD; + the judge must read it by release or keep what it needs on the NVMe. + +Next: the batch store (write-once releases, posters and group sets by id, +bytea keys, blocks), then the window and albums in the Rust edge, block +leases and the block merger, the corpus replay harness. Until the batch +store is done the 38 functional tests that write the old columns fail on +`voyager-3`. + ## Decisions a. Chunk payload: a write-once `bytea` table in Postgres, partitioned by diff --git a/app/services/batch_store.ts b/app/services/batch_store.ts index ef6c179..a6c9c8d 100644 --- a/app/services/batch_store.ts +++ b/app/services/batch_store.ts @@ -86,18 +86,6 @@ class BatchStore { }) } - for (let i = 0; i < batch.fragments.length; i += 2000) { - const rows = batch.fragments.slice(i, i + 2000).map((f) => ({ - range_id: rangeId, - file_key: f.fileKey, - // Not Math.min(...): a big file's fragment has more segments than the call stack takes arguments. - article_number: f.segments.reduce((min, s) => (s[3] < min ? s[3] : min), Infinity), - segments: f.segments.length, - fragment: JSON.stringify(f), - })) - await trx.insertQuery().table('fragments').multiInsert(rows) - } - const moved = await trx .from('ranges') .where('id', rangeId) diff --git a/app/services/fragment_merger.ts b/app/services/fragment_merger.ts deleted file mode 100644 index 0d86cd9..0000000 --- a/app/services/fragment_merger.ts +++ /dev/null @@ -1,153 +0,0 @@ -import db from '@adonisjs/lucid/services/db' -import BatchStore from '#services/batch_store' -import ChunkPartitions from '#services/chunk_partitions' -import RuleSets from '#services/rule_sets' -import { RuleSet } from '#lib/rules/index' -import { candidatesOf, chunkOf, mergeFragments, releaseKeyOf, uploadsOf, type Fragment, type ProbeCandidate, type ReleaseChunk } from '#lib/edge/index' -import type { AggregatedFile } from '#lib/files/index' -import type { EdgeSegment } from '#lib/edge/index' - -export interface MergeResult { - files: number - completed: number - incomplete: number - waiting: number - /** This round reached the end of the file keys (the next one starts over). */ - wrapped: boolean -} - -/** File keys per round. */ -const BATCH = 20_000 - -/** - * Files cut by a range's edge (REBUILD.md, fragments), put together once - * the ranges around them are done: a complete file is judged by the rules - * like any other and written as a release chunk (with probe candidates); a - * file still missing parts while every neighbouring range is done is - * incomplete, and dropped (counted on its ranges). Others wait. - */ -class FragmentMerger { - /** Where the next round goes on: waiting files must not take every round's places. */ - private cursor = '' - - async run(): Promise { - const result: MergeResult = { files: 0, completed: 0, incomplete: 0, waiting: 0, wrapped: false } - const { rows: keys } = await db.rawQuery( - `SELECT DISTINCT f.file_key FROM fragments f JOIN ranges r ON r.id = f.range_id - WHERE r.status = 'done' AND f.file_key > ? ORDER BY f.file_key LIMIT ?`, - [this.cursor, BATCH] - ) - // At the end of the keys: the next round starts over. - result.wrapped = keys.length < BATCH - this.cursor = result.wrapped ? '' : keys.at(-1).file_key - if (keys.length === 0) return result - const rules = new RuleSet(await RuleSets.current()) - const rows = await db - .from('fragments as f') - .join('ranges as r', 'r.id', 'f.range_id') - .whereIn('f.file_key', keys.map((k: { file_key: string }) => k.file_key)) - .select('f.id', 'f.file_key', 'f.range_id', 'f.fragment', 'r.group_id', 'r.start_id', 'r.end_id', 'r.status') - const byKey = new Map() - for (const row of rows) { - const list = byKey.get(row.file_key) - if (list) list.push(row) - else byKey.set(row.file_key, [row]) - } - - const completed: { file: AggregatedFile; rangeId: number; ids: number[] }[] = [] - const incomplete: { rangeId: number; ids: number[]; segments: number }[] = [] - const neighbours = await this.neighbourStatus([...byKey.values()]) - for (const list of byKey.values()) { - result.files++ - const file = mergeFragments(list.map((r) => r.fragment as Fragment)) - const ids = list.map((r) => Number(r.id)) - const rangeId = Math.max(...list.map((r) => Number(r.range_id))) - if (file) { - completed.push({ file, rangeId, ids }) - continue - } - if (list.every((r) => r.status === 'done') && neighbours(list)) { - incomplete.push({ rangeId, ids, segments: list.reduce((n, r) => n + (r.fragment as Fragment).segments.length, 0) }) - } else { - result.waiting++ - } - } - - // Completed files, grouped into their releases, judged and written like a batch's. - const byRelease = new Map[]; rangeId: number }>() - for (const { file, rangeId } of completed) { - const key = releaseKeyOf(file.segments[0]) - const entry = byRelease.get(key) - if (entry) entry.files.push(file) - else byRelease.set(key, { files: [file], rangeId }) - } - const releases: { rangeId: number; chunk: ReleaseChunk; candidates: ProbeCandidate[] }[] = [] - for (const [key, { files, rangeId }] of byRelease) { - const verdict = rules.judge(key, files.map((f) => ({ name: f.segments[0].file.name, totalParts: f.segments[0].totalParts, groups: f.segments[0].file.groups, firstMessageId: f.segments[0].messageId, poster: f.segments[0].file.poster }))) - if (verdict.rule !== null) continue - uploadsOf(files).forEach((list, upload) => releases.push({ rangeId, chunk: chunkOf(key, list, upload), candidates: candidatesOf(key, list, upload) })) - } - await ChunkPartitions.ensure() - await db.transaction(async (trx) => { - // In key order (batches lock their group; the merger only ever waits on one batch at a time). - releases.sort((a, b) => a.chunk.key.localeCompare(b.chunk.key) || a.chunk.upload - b.chunk.upload) - for (const { rangeId, chunk, candidates } of releases) await BatchStore.writeReleases(trx, rangeId, 0, [chunk], candidates) - const done = [...completed.flatMap((c) => c.ids), ...incomplete.flatMap((c) => c.ids)] - for (let i = 0; i < done.length; i += 5000) await trx.from('fragments').whereIn('id', done.slice(i, i + 5000)).delete() - for (const { rangeId, segments } of incomplete) { - await trx.rawQuery( - `INSERT INTO rule_counts (range_id, rule, key, articles, files) VALUES (?, 'incomplete', 'edge', ?, 1) - ON CONFLICT (range_id, rule, key) DO UPDATE SET articles = rule_counts.articles + excluded.articles, files = rule_counts.files + 1`, - [rangeId, segments] - ) - } - }) - result.completed = completed.length - result.incomplete = incomplete.length - return result - } - - /** - * For each file's fragments: whether the ranges right before and after - * their ranges (where a missing part could be) are done, read in one - * query for the whole round (per file it was three, ~2,000 files a - * minute). A neighbour not planned yet counts as not done (newer articles - * come later), except before the group's oldest article. - */ - private async neighbourStatus(lists: { group_id: number; start_id: number | string; end_id: number | string }[][]) { - const span = (list: (typeof lists)[number]) => ({ - group: Number(list[0].group_id), - first: Math.min(...list.map((r) => Number(r.start_id))), - last: Math.max(...list.map((r) => Number(r.end_id))), - }) - const spans = lists.map(span) - const groups = [...new Set(spans.map((s) => s.group))] - // The range holding the article before and the one after, each by the (group_id, start_id) index: - // "end_id = ? OR start_id = ?" read every range of the group per file (5.5 s a round; LOG.md 2026-10-09). - const { rows } = await db.rawQuery( - `SELECT s.group_id, s.before_end, s.after_start, b.status AS before, a.status AS after - FROM unnest(?::int[], ?::bigint[], ?::bigint[]) AS s(group_id, before_end, after_start) - LEFT JOIN LATERAL (SELECT r.status FROM ranges r WHERE r.group_id = s.group_id AND r.start_id <= s.before_end AND r.end_id >= s.before_end - ORDER BY r.start_id DESC LIMIT 1) b ON true - LEFT JOIN LATERAL (SELECT r.status FROM ranges r WHERE r.group_id = s.group_id AND r.start_id <= s.after_start AND r.end_id >= s.after_start - ORDER BY r.start_id DESC LIMIT 1) a ON true`, - [spans.map((s) => s.group), spans.map((s) => s.first - 1), spans.map((s) => s.last + 1)] - ) - const byEnd = new Map() - const byStart = new Map() - for (const r of rows) { - if (r.before !== null) byEnd.set(`${r.group_id}:${r.before_end}`, r.before) - if (r.after !== null) byStart.set(`${r.group_id}:${r.after_start}`, r.after) - } - const lows = new Map((await db.from('groups').whereIn('id', groups).select('id', 'low')).map((g) => [Number(g.id), g.low === null ? null : Number(g.low)])) - return (list: (typeof lists)[number]) => { - const { group, first, last } = span(list) - const before = byEnd.get(`${group}:${first - 1}`) - const low = lows.get(group) - const beforeDone = before !== undefined ? before === 'done' : low === null || low === undefined || first <= low - return beforeDone && byStart.get(`${group}:${last + 1}`) === 'done' - } - } -} - -export default new FragmentMerger() diff --git a/app/services/par2_pruner.ts b/app/services/par2_pruner.ts index e440553..fabda38 100644 --- a/app/services/par2_pruner.ts +++ b/app/services/par2_pruner.ts @@ -75,7 +75,7 @@ export class Par2Pruner { WHERE o.set_id = b.set_id AND v.release_id IS NULL) THEN 'judged' WHEN NOT EXISTS (SELECT 1 FROM par2_sets s WHERE s.set_id = b.set_id AND s.created_at > ?) AND NOT EXISTS (SELECT 1 FROM par2_sets s JOIN releases r ON r.id = s.release_id WHERE s.set_id = b.set_id AND ( - release_final(r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at, r.complete) IS NOT TRUE + r.settled_at IS NULL OR EXISTS (SELECT 1 FROM recent x WHERE x.group_id = r.group_id AND x.posted_low <= coalesce(r.last_posted_at, r.posted_at) + interval '6 hours' AND x.posted_high >= r.posted_at - interval '6 hours'))) AND NOT EXISTS (SELECT 1 FROM par2_files f JOIN unseen u ON u.hash16k = f.hash16k AND u.size = f.length WHERE f.set_id = b.set_id) diff --git a/app/services/range_blocks.ts b/app/services/range_blocks.ts deleted file mode 100644 index bc16831..0000000 --- a/app/services/range_blocks.ts +++ /dev/null @@ -1,31 +0,0 @@ -import db from '@adonisjs/lucid/services/db' - -/** - * range_blocks (migration 3900): per group, each run of consecutive done ranges with its posting time span, read by - * release_final(). A few thousand rows from ~650k ranges, rebuilt whole in one statement each time; a run breaks at a - * range that isn't done or doesn't start where the one before ended. - */ -class RangeBlocks { - async rebuild(): Promise { - return db.transaction(async (trx) => { - await trx.rawQuery(`DELETE FROM range_blocks`) - const result = await trx.rawQuery( - `WITH r AS ( - SELECT group_id, start_id, end_id, status = 'done' AS done, posted_low, posted_high, - min(start_id) OVER (PARTITION BY group_id) AS group_first, - lag(status = 'done') OVER w AS done_before, lag(end_id) OVER w AS end_before - FROM ranges WINDOW w AS (PARTITION BY group_id ORDER BY start_id)), - runs AS ( - SELECT *, sum(CASE WHEN done IS DISTINCT FROM done_before OR end_before IS DISTINCT FROM start_id - 1 THEN 1 ELSE 0 END) - OVER (PARTITION BY group_id ORDER BY start_id) AS run - FROM r) - INSERT INTO range_blocks (group_id, start_id, end_id, posted_low, posted_high, at_group_start) - SELECT group_id, min(start_id), max(end_id), min(posted_low), max(posted_high), min(start_id) = min(group_first) - FROM runs WHERE done GROUP BY group_id, run` - ) - return result.rowCount ?? 0 - }) - } -} - -export default new RangeBlocks() diff --git a/app/services/release_judge.ts b/app/services/release_judge.ts index 67ec2f8..f0b23e3 100644 --- a/app/services/release_judge.ts +++ b/app/services/release_judge.ts @@ -102,10 +102,10 @@ export const RULES: Rule[] = [ { name: 'unusable', says: 'random names, no name inside, no par2 or inner-name merge, no spot, a day after settling, and final', - // Final (release_final): next to an unread range its par2 or the rest of its files may still come (LOG.md 2026-10-16 (67)). + // Finished (settled_at): a provisional tail release's par2 or the rest of its files may still come. sql: (scoped) => `SELECT r.id FROM releases r ${only(scoped, 'r.id')} WHERE r.settled_at < now() - interval '1 day' AND NOT r.readable AND r.inner_name IS NULL - AND release_final(r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at, r.complete) + AND r.settled_at IS NOT NULL AND NOT EXISTS (SELECT 1 FROM release_merges m WHERE m.into_id = r.id) AND NOT EXISTS (SELECT 1 FROM release_spots s WHERE s.release_id = r.id) AND NOT EXISTS (SELECT 1 FROM par2_sets p WHERE p.release_id = r.id)`, diff --git a/app/services/search_service.ts b/app/services/search_service.ts index d493212..9deb74d 100644 --- a/app/services/search_service.ts +++ b/app/services/search_service.ts @@ -51,7 +51,7 @@ export interface ReleaseSummary { size: number readable: boolean complete: boolean | null - /** The complete verdict is definite: done ranges cover the release's posting times and 6 hours either side (release_final). */ + /** The complete verdict is definite: the release is finished (voyager 3: the window closed on it; a tail release is provisional until then). */ final: boolean category: number settled: boolean @@ -98,8 +98,8 @@ class SearchService { } if (params.readable) where.push('(r.readable OR r.inner_name IS NOT NULL)') if (params.complete) where.push('r.complete') - // Incomplete only when final: a release next to unread ranges may yet get its missing files (release_final). - if (params.notIncomplete) where.push('(r.complete IS NOT FALSE OR NOT release_final(r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at, r.complete))') + // Incomplete only when final: a provisional tail release may yet get its missing files. + if (params.notIncomplete) where.push('(r.complete IS NOT FALSE OR r.settled_at IS NULL)') if (params.unlocked) where.push('r.passworded IS NOT TRUE') if (params.source) { where.push('r.source = ?') @@ -166,7 +166,7 @@ class SearchService { db.rawQuery( `SELECT r.id, r.name, r.source, r.poster, r.groups, r.posted_at, r.file_total, r.files, r.bytes, r.readable, r.complete, r.category, r.settled_at, r.passworded, r.inner_name, EXISTS (SELECT 1 FROM release_passwords p WHERE p.release_id = r.id) AS password_known, - release_final(r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at, r.complete) AS final + r.settled_at IS NOT NULL AS final FROM ${source.from} WHERE ${source.where} ORDER BY ${order} LIMIT ? OFFSET ?`, [...source.bindings, ...bindings, perPage, offset] ), @@ -205,10 +205,9 @@ class SearchService { async release(id: number) { const r = await db.from('releases').where('id', id).first() if (!r) return null - const [files, { rows: inspections }, { rows: finality }, passwords] = await Promise.all([ + const [files, { rows: inspections }, passwords] = await Promise.all([ NzbService.files(id), db.rawQuery(`SELECT kind, result, created_at FROM release_inspections WHERE release_id = ? ORDER BY candidate_id`, [id]), - db.rawQuery(`SELECT release_final(?, ?, ?, ?, ?, ?) AS final`, [r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at, r.complete]), NzbService.passwords(id), ]) return { @@ -224,7 +223,7 @@ class SearchService { segments: Number(r.segments), readable: r.readable, complete: r.complete, - final: finality[0].final, + final: r.settled_at !== null, hasPar2: r.has_par2, category: r.category, firstSeenAt: r.first_seen_at, diff --git a/app/services/settler.ts b/app/services/settler.ts deleted file mode 100644 index 38fcf79..0000000 --- a/app/services/settler.ts +++ /dev/null @@ -1,49 +0,0 @@ -import db from '@adonisjs/lucid/services/db' - -/** Releases unchanged for this long are settled (ranges are read in order; the merger adds within minutes). */ -const QUIET_HOURS = 6 -/** - * History: a release whose last post is QUIET_HOURS old gets nothing new from the tail, only from the - * neighbouring ranges' fragments, which the merger adds within minutes: settled once unchanged this long. - * Waiting QUIET_HOURS after it was read held every history release back from phase 2 for six hours after - * each reset (the hourly sweeps' resets kept the prober idle; LOG.md 2026-10-12 (13)). - */ -const QUIET_HISTORY_MINUTES = 30 -const BATCH = 50_000 - -/** - * Settles releases (REBUILD.md, part 8): once nothing has been added for - * QUIET_HOURS (QUIET_HISTORY_MINUTES once its last post is that old), its completeness is decided: complete when every file the - * [i/N] counter announced was found, unknown without a counter (phase 2 - * decides those by their par2 sets: app/services/par2_completeness.ts). - * Phase 2 picks its work from settled releases. - */ -class Settler { - /** Up to `batches` batches of BATCH (all there are without). */ - async run(batches = Infinity): Promise { - let settled = 0 - for (let batch = 0; batch < batches; batch++) { - // Their probe candidates become claimable (Prober.claim), the unreadable ones first. - const { rows } = await db.rawQuery( - `WITH settled AS ( - UPDATE releases SET settled_at = now(), complete = CASE WHEN file_total IS NULL THEN complete ELSE files >= file_total END - WHERE id IN (SELECT id FROM releases WHERE settled_at IS NULL AND last_seen_at < now() - make_interval(mins => ?) - AND (last_seen_at < now() - make_interval(hours => ?) OR coalesce(last_posted_at, posted_at) < now() - make_interval(hours => ?)) LIMIT ?) - RETURNING id, readable, files, bytes, poster, sidecar_only), - marked AS (UPDATE probe_candidates pc SET ready = true, unreadable = NOT s.readable FROM settled s WHERE pc.release_id = s.id), - -- A small one of sidecars only (the batch store's sidecar_only) may be an upload's .nfo or .nzb that came apart - -- from it: the merger's sidecar fold looks at them. - queued AS (INSERT INTO fold_candidates (release_id) SELECT id FROM settled WHERE sidecar_only AND files <= 3 AND bytes < 1000000 AND poster IS NOT NULL - ON CONFLICT DO NOTHING) - SELECT count(*)::int AS n FROM settled`, - [QUIET_HISTORY_MINUTES, QUIET_HOURS, QUIET_HOURS, BATCH] - ) - const n = rows[0].n - settled += n - if (n < BATCH) return settled - } - return settled - } -} - -export default new Settler() diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index e8ecace..cce4e78 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -3,9 +3,6 @@ import type { CommandOptions } from '@adonisjs/core/types/ace' import { setTimeout as sleep } from 'node:timers/promises' import { Prefetcher } from '#services/prefetcher' import Planner from '#services/planner' -import FragmentMerger from '#services/fragment_merger' -import Settler from '#services/settler' -import RangeBlocks from '#services/range_blocks' import Leases from '#services/leases' import NntpPool from '#services/nntp_pool' import { NntpShare } from '#services/nntp_shares' @@ -13,38 +10,38 @@ import coordinatorConfig from '#config/coordinator' const TICK_MS = 2000 const PLAN_EVERY_MS = 60 * 60 * 1000 -const MERGE_EVERY_MS = 60 * 1000 -const MERGE_BUDGET_MS = 30 * 1000 const STOP_WAIT_MS = 15 * 1000 -/** - * Settling in rounds of at most SETTLE_BATCHES batches every few minutes: once an hour, ~3M releases rewrote their - * rows and index entries in one go (2.65M in 30 min), the NVMe saturated and the batch stores queued on their locks - * (LOG.md 2026-10-16 (67)). - */ -const SETTLE_EVERY_MS = 5 * 60 * 1000 -const SETTLE_BATCHES = 4 -/** range_blocks (release_final) after ranges finish. */ -const BLOCKS_EVERY_MS = 5 * 60 * 1000 +/** Leases whose worker went quiet after reading everything are finished every few minutes. */ +const ABANDONED_EVERY_MS = 5 * 60 * 1000 /** - * The coordinator's background work (REBUILD.md, part 1): the prefetcher, - * merging fragments of finished ranges every minute, and every hour - * planning ranges for new articles and settling quiet releases. The API (leases, stream, - * batches) runs in the web server. + * The coordinator's background work (REBUILD.md, part 1): the prefetcher, the tail's ranges every few minutes, + * new groups' history every hour, and abandoned leases. The API (leases, stream, batches) runs in the web server. + * Voyager 3: the settler and the fragment merger are gone (the window finishes a release once; the block merger + * joins edge releases). */ export default class CoordinatorRun extends BaseCommand { static commandName = 'coordinator:run' static description = 'Prefetches ranges into the spool and plans new ones' static options: CommandOptions = { startApp: true, staysAlive: true } - @flags.string({ description: 'This process\'s share of the prefetching, "i/n" (ranges with id % n = i - 1); only 1/n plans, merges and settles' }) + @flags.string({ + description: + 'This process\'s share of the prefetching, "i/n" (ranges with id % n = i - 1); only 1/n plans, merges and settles', + }) declare shard?: string async run() { const [index, count] = (this.shard ?? '1/1').split('/').map(Number) - if (!(count >= 1 && index >= 1 && index <= count)) throw new Error(`--shard: "i/n" with 1 <= i <= n, not "${this.shard}"`) + if (!(count >= 1 && index >= 1 && index <= count)) + throw new Error(`--shard: "i/n" with 1 <= i <= n, not "${this.shard}"`) const prefetcher = new Prefetcher({ index: index - 1, count }) - const share = new NntpShare(count > 1 ? `coordinator ${index}/${count}` : 'coordinator', coordinatorConfig.nntpShares.coordinatorPriority, coordinatorConfig.nntpMinConnections, coordinatorConfig.nntpConnections) + const share = new NntpShare( + count > 1 ? `coordinator ${index}/${count}` : 'coordinator', + coordinatorConfig.nntpShares.coordinatorPriority, + coordinatorConfig.nntpMinConnections, + coordinatorConfig.nntpConnections + ) const main = index === 1 let stopping = false let stopped: () => void @@ -56,7 +53,11 @@ export default class CoordinatorRun extends BaseCommand { prefetcher.stop() await done }) - this.logger.info(count > 1 ? `Coordinator started (shard ${index}/${count}${main ? '' : ', prefetching only'}).` : 'Coordinator started.') + this.logger.info( + count > 1 + ? `Coordinator started (shard ${index}/${count}${main ? '' : ', prefetching only'}).` + : 'Coordinator started.' + ) // The pool's size: its share of the account's connections, the prober the rest (app/services/nntp_shares.ts). // Before the first connection: until then it's NNTP_CONNECTIONS. const first = await NntpPool.share(share) @@ -69,11 +70,16 @@ export default class CoordinatorRun extends BaseCommand { } catch (error: any) { this.logger.error(`NNTP share failed: ${error.message}`) } - for (let waited = 0; waited < coordinatorConfig.nntpShares.everyMs && !stopping; waited += TICK_MS) await sleep(TICK_MS) + for ( + let waited = 0; + waited < coordinatorConfig.nntpShares.everyMs && !stopping; + waited += TICK_MS + ) + await sleep(TICK_MS) } })() - // Two loops: the prefetcher's ticks don't wait for merging or planning (a merge round - // of up to MERGE_BUDGET_MS left the connections idle once their targets were fetched). + // The prefetcher's ticks in a loop of their own: planning doesn't hold them (a merge round of up to 30 s once + // left the connections idle once their targets were fetched). const prefetching = (async () => { while (!stopping) { try { @@ -84,24 +90,24 @@ export default class CoordinatorRun extends BaseCommand { await sleep(TICK_MS) } })() - // range_blocks in a loop of its own: a settling or merging round doesn't hold release_final back. - const blocks = (async () => { + const abandoned = (async () => { while (main && !stopping) { try { - const abandoned = await Leases.finishAbandoned() - if (abandoned > 0) this.logger.info(`Finished ${abandoned} ranges read to their end whose worker went quiet.`) - await RangeBlocks.rebuild() + const finished = await Leases.finishAbandoned() + if (finished > 0) + this.logger.info( + `Finished ${finished} ranges read to their end whose worker went quiet.` + ) } catch (error: any) { - this.logger.error(`Range blocks failed: ${error.message}`) + this.logger.error(`Finishing abandoned leases failed: ${error.message}`) } - for (let waited = 0; waited < BLOCKS_EVERY_MS && !stopping; waited += TICK_MS) await sleep(TICK_MS) + for (let waited = 0; waited < ABANDONED_EVERY_MS && !stopping; waited += TICK_MS) + await sleep(TICK_MS) } })() // History for new groups hourly; the tail every few minutes. let plannedAt = 0 let tailedAt = 0 - let mergedAt = 0 - let settledAt = 0 while (!stopping) { try { if (main && Date.now() - plannedAt >= PLAN_EVERY_MS) { @@ -109,45 +115,22 @@ export default class CoordinatorRun extends BaseCommand { const { groups, ranges } = await Planner.planAll() if (ranges > 0) this.logger.info(`Planned ${ranges} ranges over ${groups} groups.`) } - if (main && Date.now() - settledAt >= SETTLE_EVERY_MS) { - settledAt = Date.now() - const settled = await Settler.run(SETTLE_BATCHES) - if (settled > 0) this.logger.info(`Settled ${settled} releases.`) - } if (main && Date.now() - tailedAt >= coordinatorConfig.tail.everySeconds * 1000) { tailedAt = Date.now() const { groups, ranges } = await Planner.planTail() if (ranges > 0) this.logger.info(`Tail: ${ranges} ranges for ${groups} groups.`) } - if (main && Date.now() - mergedAt >= MERGE_EVERY_MS) { - mergedAt = Date.now() - // Rounds until a whole pass over the file keys decides nothing more, or MERGE_BUDGET_MS. - // (Going on while only waiting files were left kept Postgres busy all the time.) - const totals = { completed: 0, incomplete: 0, waiting: 0 } - let decidedThisPass = 0 - while (!stopping && Date.now() - mergedAt < MERGE_BUDGET_MS) { - const merged = await FragmentMerger.run() - totals.completed += merged.completed - totals.incomplete += merged.incomplete - totals.waiting = merged.waiting - decidedThisPass += merged.completed + merged.incomplete - if (merged.wrapped) { - if (decidedThisPass === 0) break - decidedThisPass = 0 - } - } - if (totals.completed || totals.incomplete) this.logger.info(`Fragments: ${totals.completed} files completed, ${totals.incomplete} incomplete, ${totals.waiting} waiting.`) - } } catch (error: any) { this.logger.error(`Coordinator round failed: ${error.message}`) } await sleep(TICK_MS) } - await Promise.all([prefetching, blocks, sharing]) + await Promise.all([prefetching, abandoned, sharing]) // The chunks in flight, for at most STOP_WAIT_MS (a connection the server holds back can take minutes); then closed. const waitUntil = Date.now() + STOP_WAIT_MS while (prefetcher.active > 0 && Date.now() < waitUntil) await sleep(200) - if (prefetcher.active > 0) this.logger.info(`Closing ${prefetcher.active} connections still fetching.`) + if (prefetcher.active > 0) + this.logger.info(`Closing ${prefetcher.active} connections still fetching.`) await NntpPool.close() await share.leave().catch(() => {}) this.logger.info('Coordinator stopped.') diff --git a/database/migrations/1770000006000_voyager3_catalog.ts b/database/migrations/1770000006000_voyager3_catalog.ts new file mode 100644 index 0000000..98f65a7 --- /dev/null +++ b/database/migrations/1770000006000_voyager3_catalog.ts @@ -0,0 +1,181 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' +import env from '#start/env' + +/** + * Voyager 3's catalog (REBUILD.md, "Voyager 3: window indexing", 4 and 7), built from a reset, not converted in + * place. This migration IS the reset: it empties every table the indexing recomputes (releases, chunks, + * inspections, candidates, par2 data, file hashes, verdicts, merges, rule counts and samples; what survives: + * groups, ranges, the corpus, probe_cache, spots, nntp_shares, nodes, users), so it refuses to run on a catalog + * with releases unless VOYAGER3_RESET=1 (the user's go-ahead; resets are paused since 2026-10-12). Then: + * - posters get an id; group_sets holds each distinct group combination once (~2,600); releases is created anew + * without its text copies (poster_id, group_set_id, the keys as 20-byte bytea, source a smallint), 8-byte + * columns first, with `edge` (still open at its block's end, for the block merger) and settled_at as the finish + * (null: a tail release still provisional). The category and judge triggers follow the new columns; + * releases_category_newest_idx (5.3 GB, used once) goes. + * - fragments and range_blocks go with the fragment merger and the settler (the window finishes a release once); + * release_final() goes: a release is final once finished. + * - release_chunks, rule_counts and rule_samples are keyed by the block that wrote them (block_id; migration 6100). + * - release_inspections and probe_candidates move to CHUNK_TABLESPACE (the HDD) when set: written once, read per + * release, not by search. + */ +export default class extends BaseSchema { + async up() { + this.defer(async (db) => { + const { rows } = await db.rawQuery(`SELECT count(*)::int AS n FROM releases`) + if (rows[0].n > 0 && process.env.VOYAGER3_RESET !== '1') { + throw new Error( + `The voyager 3 catalog migration empties the catalog (${rows[0].n} releases): run it with VOYAGER3_RESET=1 after the go-ahead` + ) + } + }) + this.schema.raw(` + TRUNCATE releases, release_chunks, release_inspections, probe_candidates, file_hashes, par2_files, par2_sets, + release_verdicts, release_merges, release_spots, release_passwords, rule_counts, rule_samples, judge_dirty, + judge_full, fold_candidates, posters, worker_cursors RESTART IDENTITY`) + // Spots matched to releases that are gone: the spot inspector matches them again. + this.schema.raw(`UPDATE spots SET release_id = NULL WHERE release_id IS NOT NULL`) + this.schema.raw(`UPDATE spot_files SET release_id = NULL WHERE release_id IS NOT NULL`) + + this.schema.raw(`DROP TABLE IF EXISTS fragments`) + this.schema.raw(`DROP TABLE IF EXISTS range_blocks`) + this.schema.raw( + `DROP FUNCTION IF EXISTS release_final(integer, bigint, bigint, timestamptz, timestamptz, boolean)` + ) + this.schema.raw( + `DROP FUNCTION IF EXISTS release_final(integer, bigint, bigint, timestamptz, timestamptz)` + ) + + // Posters by id (a release's poster_id); searched by trigram as before (poster_search). + this.schema.raw(`DROP TABLE IF EXISTS posters`) + this.schema.raw(` + CREATE TABLE posters ( + id serial PRIMARY KEY, + poster text NOT NULL UNIQUE + )`) + this.schema.raw(`CREATE INDEX posters_trgm ON posters USING gin (poster gin_trgm_ops)`) + // Each distinct combination of groups a release was posted to, once: a sorted jsonb array of names. + this.schema.raw(` + CREATE TABLE group_sets ( + id serial PRIMARY KEY, + groups jsonb NOT NULL UNIQUE + )`) + + this.schema.raw(`DROP TABLE IF EXISTS releases CASCADE`) + this.schema.raw(` + CREATE TABLE releases ( + id bigserial PRIMARY KEY, + -- The upload's posting window (first and last article's Date). + posted_at timestamptz, + last_posted_at timestamptz, + first_article bigint, + last_article bigint, + segments bigint NOT NULL DEFAULT 0, + bytes bigint NOT NULL DEFAULT 0, + -- The finish: the window closed on it (or the block ended); null while a tail release is provisional. + settled_at timestamptz, + inspected_at timestamptz, + first_seen_at timestamptz NOT NULL DEFAULT now(), + -- The owner group (null: a spot's release), the set of groups it was posted to, its poster. + group_id integer, + group_set_id integer NOT NULL, + poster_id integer, + file_total integer, + files integer NOT NULL DEFAULT 0, + chunks integer NOT NULL DEFAULT 0, + category integer, + -- lib/release/source.ts: 1 subject, 2 post, 3 stem, 4 spot, 5 nzb, 6 album. + source smallint NOT NULL, + readable boolean NOT NULL DEFAULT false, + has_par2 boolean NOT NULL DEFAULT false, + complete boolean, + passworded boolean, + sidecar_only boolean, + -- Still open when its block ended: the block merger joins it with the neighbour block's part by identity. + edge boolean NOT NULL DEFAULT false, + -- SHA-1s (lib/release/identity.ts): the identity (poster, name, group), and the release (identity and upload). + release_key bytea NOT NULL, + identity_key bytea NOT NULL, + name text NOT NULL, + inner_name varchar(255) + )`) + this.schema.raw(`CREATE UNIQUE INDEX releases_release_key_unique ON releases (release_key)`) + this.schema.raw(`CREATE INDEX releases_identity_idx ON releases (identity_key, group_id)`) + this.schema.raw( + `CREATE INDEX releases_name_trgm ON releases USING gin (name gin_trgm_ops) WITH (fastupdate = on, gin_pending_list_limit = 65536)` + ) + this.schema.raw( + `CREATE INDEX releases_inner_name_trgm ON releases USING gin (inner_name gin_trgm_ops) WHERE inner_name IS NOT NULL` + ) + this.schema.raw( + `CREATE INDEX releases_newest_idx ON releases (posted_at DESC NULLS LAST, id DESC)` + ) + this.schema.raw( + `CREATE INDEX releases_poster_newest_idx ON releases (poster_id, posted_at DESC NULLS LAST, id DESC)` + ) + this.schema.raw(`CREATE INDEX releases_passworded_idx ON releases (id) WHERE passworded`) + this.schema.raw( + `CREATE INDEX releases_edge_idx ON releases (group_id, identity_key) WHERE edge` + ) + this.schema.raw( + `CREATE INDEX releases_provisional_idx ON releases (id) WHERE settled_at IS NULL` + ) + + // The category from the name, the inner name, or the groups (migration 5200), the groups now in group_sets. + this.schema.raw(` + CREATE OR REPLACE FUNCTION releases_set_category() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + NEW.category := release_category_for(NEW.name, NEW.inner_name, (SELECT groups FROM group_sets WHERE id = NEW.group_set_id), + CASE WHEN TG_OP = 'UPDATE' THEN OLD.category ELSE NEW.category END); + RETURN NEW; + END $$`) + this.schema.raw(` + CREATE TRIGGER releases_category BEFORE INSERT OR UPDATE OF name, inner_name, group_set_id ON releases + FOR EACH ROW EXECUTE FUNCTION releases_set_category()`) + // The judge's dirty marks (migration 4900), on the columns its rules read. + this.schema.raw(` + CREATE TRIGGER judge_dirty_insert AFTER INSERT ON releases REFERENCING NEW TABLE AS new_rows + FOR EACH STATEMENT EXECUTE FUNCTION judge_dirty_releases()`) + this.schema.raw(` + CREATE TRIGGER judge_dirty_delete AFTER DELETE ON releases REFERENCING OLD TABLE AS old_rows + FOR EACH STATEMENT EXECUTE FUNCTION judge_dirty_releases()`) + this.schema.raw(` + CREATE TRIGGER judge_dirty_update AFTER UPDATE ON releases FOR EACH ROW + WHEN (old.name IS DISTINCT FROM new.name OR old.poster_id IS DISTINCT FROM new.poster_id OR old.source IS DISTINCT FROM new.source + OR old.file_total IS DISTINCT FROM new.file_total OR old.passworded IS DISTINCT FROM new.passworded + OR old.settled_at IS DISTINCT FROM new.settled_at OR old.readable IS DISTINCT FROM new.readable + OR old.inner_name IS DISTINCT FROM new.inner_name OR old.group_id IS DISTINCT FROM new.group_id + OR old.first_article IS DISTINCT FROM new.first_article OR old.last_article IS DISTINCT FROM new.last_article + OR old.posted_at IS DISTINCT FROM new.posted_at OR old.last_posted_at IS DISTINCT FROM new.last_posted_at + OR old.complete IS DISTINCT FROM new.complete) + EXECUTE FUNCTION judge_dirty_releases()`) + + // Chunks, rule counts and samples by the block that wrote them. + this.schema.raw(`ALTER TABLE release_chunks RENAME COLUMN range_id TO block_id`) + this.schema.raw( + `ALTER INDEX IF EXISTS release_chunks_range_idx RENAME TO release_chunks_block_idx` + ) + this.schema.raw(`ALTER TABLE rule_counts RENAME COLUMN range_id TO block_id`) + this.schema.raw(`ALTER TABLE rule_samples RENAME COLUMN range_id TO block_id`) + this.schema.raw( + `ALTER INDEX IF EXISTS rule_samples_range_id_index RENAME TO rule_samples_block_id_index` + ) + + // Cold tables to the HDD (CHUNK_TABLESPACE), with their indexes; empty now, so at once. + const hdd = env.get('CHUNK_TABLESPACE') + if (hdd) { + for (const table of ['release_inspections', 'probe_candidates']) { + this.schema.raw(`ALTER TABLE ${table} SET TABLESPACE ${hdd}`) + this.schema.raw(` + DO $$ DECLARE i record; BEGIN + FOR i IN SELECT indexname FROM pg_indexes WHERE schemaname = 'public' AND tablename = '${table}' LOOP + EXECUTE format('ALTER INDEX %I SET TABLESPACE %I', i.indexname, '${hdd}'); + END LOOP; + END $$`) + } + } + } + + async down() { + throw new Error('The voyager 3 catalog migration cannot be undone (it emptied the catalog)') + } +} diff --git a/database/migrations/1770000006100_blocks.ts b/database/migrations/1770000006100_blocks.ts new file mode 100644 index 0000000..80f24c9 --- /dev/null +++ b/database/migrations/1770000006100_blocks.ts @@ -0,0 +1,61 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * Blocks (REBUILD.md, "Voyager 3: window indexing", 1): what a worker leases. A block is a run of consecutive + * ranges of one group covering at least 30 days of posting time (planned from ranges.est_date and posted_low/high), + * read in article order through the 24 h window; blocks of a group run in parallel. Ranges stay the unit the + * prefetcher fetches (the spool, the corpus) and point at their block; the lease, resume point, heartbeat and + * counts move here. Part of the reset (migration 6000): every range is read again, from the corpus where stored. + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + CREATE TABLE blocks ( + id bigserial PRIMARY KEY, + group_id integer NOT NULL REFERENCES groups (id) ON DELETE CASCADE, + -- The first range's start and the last range's end. + start_id bigint NOT NULL, + end_id bigint NOT NULL, + -- The resume point: the first article not stored yet (batches are refused out of order, as ranges' were). + next_id bigint NOT NULL, + -- The posting time the block covers: estimated when planned, the worker's measured span once done. + posted_low timestamptz, + posted_high timestamptz, + -- planned, leased, done, failed. + status varchar(8) NOT NULL DEFAULT 'planned', + -- As ranges': the tail before everything, then the history by era. + priority integer NOT NULL DEFAULT 0, + node_id integer, + attempts integer NOT NULL DEFAULT 0, + rule_version integer, + headers bigint NOT NULL DEFAULT 0, + releases integer NOT NULL DEFAULT 0, + leased_at timestamptz, + heartbeat_at timestamptz, + done_at timestamptz, + error text, + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (group_id, start_id) + )`) + this.schema.raw( + `CREATE INDEX blocks_todo_idx ON blocks (priority, id) WHERE status IN ('planned', 'leased')` + ) + this.schema.raw(`CREATE INDEX blocks_leased_idx ON blocks (leased_at) WHERE status = 'leased'`) + this.schema.raw( + `ALTER TABLE ranges ADD COLUMN IF NOT EXISTS block_id bigint REFERENCES blocks (id) ON DELETE SET NULL` + ) + this.schema.raw(`CREATE INDEX IF NOT EXISTS ranges_block_idx ON ranges (block_id)`) + // The reset: every range planned again from its start (the corpus serves the stored ones). Priorities, capture, + // est_date, stored_at, plain_chunks and the measured posted_low/high stay. + this.schema.raw(` + UPDATE ranges SET status = 'planned', next_id = start_id, node_id = NULL, attempts = 0, spooled_chunks = 0, + spooled_at = NULL, leased_at = NULL, heartbeat_at = NULL, done_at = NULL, headers = 0, error = NULL, + rule_version = NULL + WHERE status <> 'planned' OR next_id <> start_id OR spooled_chunks <> 0 OR node_id IS NOT NULL`) + } + + async down() { + this.schema.raw(`ALTER TABLE ranges DROP COLUMN IF EXISTS block_id`) + this.schema.raw(`DROP TABLE IF EXISTS blocks`) + } +} diff --git a/lib/release/source.ts b/lib/release/source.ts new file mode 100644 index 0000000..f8d11fb --- /dev/null +++ b/lib/release/source.ts @@ -0,0 +1,36 @@ +/** + * How a release got its name, as releases.source stores it (a smallint; voyager 3, migration 6000). The names are + * what the edge, the spot inspector and the API use; the numbers never change meaning once written. + */ +export const SOURCES = { + /** A name the subject announces ("name" yEnc (k/N)). */ + subject: 1, + /** The post's files, named after the first file (no release name in the subject). */ + post: 2, + /** The files' shared stem. */ + stem: 3, + /** A Spotnet spot's files that matched no release. */ + spot: 4, + /** An NZB's files. */ + nzb: 5, + /** One poster's files that differ only by a number sequence, or share a prefix (voyager 3). */ + album: 6, +} as const + +export type Source = keyof typeof SOURCES + +const NAMES = Object.fromEntries( + Object.entries(SOURCES).map(([name, code]) => [code, name]) +) as Record + +/** The code of a source name; throws for an unknown one (a typo would write a wrong number for good). */ +export function sourceCode(name: string): number { + const code = SOURCES[name as Source] + if (code === undefined) throw new Error(`Unknown release source "${name}"`) + return code +} + +/** The name of a stored code; "unknown" for one not listed (never written, but a row might be read by an older build). */ +export function sourceName(code: number | string | null | undefined): Source | 'unknown' { + return NAMES[Number(code)] ?? 'unknown' +} diff --git a/tests/functional/final_completeness.spec.ts b/tests/functional/final_completeness.spec.ts deleted file mode 100644 index d586045..0000000 --- a/tests/functional/final_completeness.spec.ts +++ /dev/null @@ -1,95 +0,0 @@ -import { test } from '@japa/runner' -import db from '@adonisjs/lucid/services/db' -import testUtils from '@adonisjs/core/services/test_utils' -import BatchStore from '#services/batch_store' -import RangeBlocks from '#services/range_blocks' -import SearchService from '#services/search_service' -import { RangeIndexer, postedSpan } from '#lib/edge/index' -import { DEFAULT_RULES } from '#lib/rules/index' -import type { Overview } from '#lib/nntp/index' - -const GROUP = 'alt.binaries.final' -const at = (iso: string) => new Date(iso) - -test.group('Final completeness', (group) => { - group.each.setup(() => testUtils.db().withGlobalTransaction()) - - test("a batch's posting times, without the earliest and latest 1%, become the range's span", async ({ assert }) => { - assert.deepEqual(postedSpan([]), [null, null]) - const dates = Array.from({ length: 200 }, (_, i) => 1_000_000 + i) - dates.push(5, 9_999_999_999) - // 202 dates: 2 trimmed at each end, so the two wild ones go. - assert.deepEqual(postedSpan(dates), [1_000_001, 1_000_198]) - - const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id') - const [node] = await db.table('nodes').insert({ name: 'final-node' }).returning('id') - const [range] = await db - .table('ranges') - .insert({ group_id: g.id, start_id: 1, end_id: 1000, next_id: 1, status: 'leased', node_id: node.id, rule_version: 1 }) - .returning('id') - const overviews = Array.from({ length: 10 }, (_, i) => { - const date = new Date(Date.parse('2026-01-01T10:00:00Z') + i * 60_000).toUTCString().replace('GMT', '+0000') - return [1 + i, { 'subject': `[1/1] - "Some.Thing.2026.rar" yEnc (${i + 1}/10)`, 'from': 'p ', 'date': date, 'message-id': ``, ':bytes': '1000', 'xref': `x ${GROUP}:${1 + i}` }] as Overview - }) - const ix = new RangeIndexer({ group: GROUP, start: 1, end: 1000, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } } }) - ix.add(overviews, 11) - await BatchStore.store(Number(range.id), Number(node.id), ix.take()) - const stored = await db.from('ranges').where('id', range.id).firstOrFail() - assert.equal(new Date(stored.posted_low).toISOString(), '2026-01-01T10:00:00.000Z') - assert.equal(new Date(stored.posted_high).toISOString(), '2026-01-01T10:09:00.000Z') - }) - - test('a release is final once one block of done ranges covers its posting times and 6 hours either side', async ({ assert }) => { - const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id') - const range = (start: number, status: string, low: string | null, high: string | null) => - db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: start + 99_999, next_id: start, status, posted_low: low && at(low), posted_high: high && at(high) }) - await range(1, 'done', '2026-01-01T00:00:00Z', '2026-01-01T12:00:00Z') - await range(100_001, 'done', '2026-01-01T12:00:00Z', '2026-01-01T23:00:00Z') - await range(200_001, 'planned', null, null) - await range(300_001, 'done', '2026-01-02T10:00:00Z', '2026-01-02T20:00:00Z') - assert.equal(await RangeBlocks.rebuild(), 2) - - let n = 0 - const release = async (first: number, posted: string, extra: Record = {}) => { - n++ - const key = `fc${n}`.padEnd(40, '0') - const [row] = await db - .table('releases') - .insert({ release_key: key, identity_key: key, name: `Final.Test.${n}`, source: 'stem', groups: '[]', group_id: g.id, files: 3, file_total: 5, complete: false, first_article: first, last_article: first + 10, posted_at: at(posted), last_posted_at: at(posted), readable: true, settled_at: new Date(), ...extra }) - .returning('id') - return Number(row.id) - } - const final = async (id: number) => { - const r = await db.from('releases').where('id', id).firstOrFail() - const { rows } = await db.rawQuery(`SELECT release_final(?, ?, ?, ?, ?, ?) AS final`, [r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at, r.complete]) - return rows[0].final as boolean - } - - // Mid-block: final either way. - const middle = await release(90_000, '2026-01-01T11:00:00Z') - // In the first range: nothing older to wait for, although 6 hours before it isn't covered. - const first = await release(1_000, '2026-01-01T02:00:00Z') - // Near the block's end, the next range unread: its missing files may be there. - const edge = await release(195_000, '2026-01-01T22:00:00Z') - // In the island after the gap: the gap is before it. - const island = await release(310_000, '2026-01-02T12:00:00Z') - assert.isTrue(await final(middle)) - assert.isTrue(await final(first)) - assert.isFalse(await final(edge)) - // Complete there: every file found, nothing to wait for. - assert.isTrue(await final(await release(196_000, '2026-01-01T22:30:00Z', { files: 5, complete: true, name: 'Complete.Edge' }))) - assert.isFalse(await final(island)) - - // Search: an incomplete release that isn't final isn't left out as incomplete; the API says which is final. - const shown = await SearchService.search({ q: 'Final.Test', notIncomplete: true }) - assert.sameMembers(shown.releases.map((r) => r.id), [edge, island]) - assert.isTrue(shown.releases.every((r) => r.final === false)) - - // The gap read: one block, the edge release final (the island's 12:00 + 6 h is inside 20:00). - await db.from('ranges').where('start_id', 200_001).update({ status: 'done', posted_low: at('2026-01-01T23:00:00Z'), posted_high: at('2026-01-02T10:00:00Z') }) - assert.equal(await RangeBlocks.rebuild(), 1) - assert.isTrue(await final(edge)) - assert.isTrue(await final(island)) - assert.equal((await SearchService.search({ q: 'Final.Test', notIncomplete: true })).total, 0) - }) -}) diff --git a/tests/functional/fragment_merger.spec.ts b/tests/functional/fragment_merger.spec.ts deleted file mode 100644 index 9bc2e20..0000000 --- a/tests/functional/fragment_merger.spec.ts +++ /dev/null @@ -1,65 +0,0 @@ -import { test } from '@japa/runner' -import db from '@adonisjs/lucid/services/db' -import testUtils from '@adonisjs/core/services/test_utils' -import BatchStore from '#services/batch_store' -import FragmentMerger from '#services/fragment_merger' -import { RangeIndexer } from '#lib/edge/index' -import { decodeChunk } from '#lib/chunks/index' -import { DEFAULT_RULES } from '#lib/rules/index' -import type { Overview } from '#lib/nntp/index' - -const GROUP = 'alt.binaries.test' -const rules = { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } } - -function parts(name: string, total: number, numbers: number[], at: number): Overview[] { - return numbers.map((n, i) => [at + i, { - 'subject': `Big.Upload.2020 [1/1] - "${name}" yEnc (${n}/${total})`, 'from': 'p ', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', - 'message-id': `<${name}-${n}@test>`, ':bytes': '700000', 'xref': `news.example ${GROUP}:${at + i}`, - }] as Overview) -} - -/** Indexes one range's overviews and stores them as the range's only batch; the range ends done. */ -async function indexRange(groupId: number, nodeId: number, start: number, end: number, overviews: Overview[]) { - const [r] = await db.table('ranges').insert({ group_id: groupId, start_id: start, end_id: end, next_id: start, status: 'leased', node_id: nodeId }).returning('id') - const indexer = new RangeIndexer({ group: GROUP, start, end, rules, edgeArticles: 100 }) - indexer.add(overviews, end + 1) - indexer.finish() - await BatchStore.store(Number(r.id), nodeId, indexer.take()) - await db.from('ranges').where('id', r.id).update({ status: 'done' }) - return Number(r.id) -} - -test.group('FragmentMerger', (group) => { - group.each.setup(() => testUtils.db().withGlobalTransaction()) - - test('a file cut by a range edge is completed from both sides once both ranges are done', async ({ assert }) => { - const [g] = await db.table('groups').insert({ name: GROUP, priority: 1, low: 1 }).returning('id') - const [n] = await db.table('nodes').insert({ name: 'merge-node' }).returning('id') - // Parts 1–2 at the end of the first range (still open there), 3–4 at the start of the next. - await indexRange(g.id, n.id, 1, 1000, parts('big.part1.rar', 4, [1, 2], 990)) - assert.equal((await db.from('fragments').count('* as c'))[0].c, '1') - assert.include(await FragmentMerger.run(), { completed: 0, incomplete: 0, waiting: 1 }) - - await indexRange(g.id, n.id, 1001, 2000, parts('big.part1.rar', 4, [3, 4], 1001)) - assert.include(await FragmentMerger.run(), { completed: 1, incomplete: 0 }) - assert.equal((await db.from('fragments').count('* as c'))[0].c, '0') - const release = await db.from('releases').first() - assert.include(release, { files: 1, chunks: 1 }) - const chunk = await db.from('release_chunks').where('release_id', release.id).first() - assert.deepEqual(decodeChunk(chunk.payload)[0].segments.map((s) => s[0]), [1, 2, 3, 4]) - assert.deepEqual((await db.from('probe_candidates')).map((c) => c.kind), ['archive']) - }) - - test('a file still missing parts when all its neighbours are done is dropped', async ({ assert }) => { - const [g] = await db.table('groups').insert({ name: GROUP, priority: 1, low: 1 }).returning('id') - const [n] = await db.table('nodes').insert({ name: 'merge-node' }).returning('id') - await indexRange(g.id, n.id, 1, 1000, parts('gap.part1.rar', 5, [1, 2], 990)) - await indexRange(g.id, n.id, 1001, 2000, parts('gap.part1.rar', 5, [4, 5], 1001)) - // Its later neighbour isn't there yet: the missing part could still come. - assert.include(await FragmentMerger.run(), { completed: 0, incomplete: 0, waiting: 1 }) - await indexRange(g.id, n.id, 2001, 3000, []) - assert.include(await FragmentMerger.run(), { completed: 0, incomplete: 1 }) - assert.equal((await db.from('fragments').count('* as c'))[0].c, '0') - assert.lengthOf(await db.from('releases'), 0) - }) -}) diff --git a/tests/functional/release_judge.spec.ts b/tests/functional/release_judge.spec.ts index de02bd1..ba22011 100644 --- a/tests/functional/release_judge.spec.ts +++ b/tests/functional/release_judge.spec.ts @@ -3,12 +3,13 @@ import db from '@adonisjs/lucid/services/db' import testUtils from '@adonisjs/core/services/test_utils' import { ReleaseJudge } from '#services/release_judge' import SearchService from '#services/search_service' -import RangeBlocks from '#services/range_blocks' test.group('ReleaseJudge', (group) => { group.each.setup(() => testUtils.db().withGlobalTransaction()) - test('judges dead, fake, repeated, locked and unusable releases, and takes a verdict back', async ({ assert }) => { + test('judges dead, fake, repeated, locked and unusable releases, and takes a verdict back', async ({ + assert, + }) => { let n = 0 const settled = new Date(Date.now() - 3 * 86400_000) const release = async (name: string, extra: Record = {}) => { @@ -16,19 +17,40 @@ test.group('ReleaseJudge', (group) => { const key = `jj${String(n).padStart(6, '0')}`.padEnd(40, '0') const [row] = await db .table('releases') - .insert({ release_key: key, identity_key: key, name, source: 'stem', groups: '[]', group_id: 1, files: 3, posted_at: settled, readable: true, settled_at: settled, ...extra }) + .insert({ + release_key: key, + identity_key: key, + name, + source: 'stem', + groups: '[]', + group_id: 1, + files: 3, + posted_at: settled, + readable: true, + settled_at: settled, + ...extra, + }) .returning('id') return Number(row.id) } let candidate = 9000 const probe = async (releaseId: number, status: string) => { candidate++ - await db.table('probe_candidates').insert({ id: candidate, release_id: releaseId, kind: 'lone', message_id: `<${candidate}@x>`, file: 'x', status }) + await db.table('probe_candidates').insert({ + id: candidate, + release_id: releaseId, + kind: 'lone', + message_id: `<${candidate}@x>`, + file: 'x', + status, + }) return candidate } const inspect = async (releaseId: number, kind: string, result: Record) => { const id = await probe(releaseId, 'done') - await db.table('release_inspections').insert({ release_id: releaseId, candidate_id: id, kind, result: JSON.stringify(result) }) + await db + .table('release_inspections') + .insert({ release_id: releaseId, candidate_id: id, kind, result: JSON.stringify(result) }) } const dead = await release('Gone.Movie.2020.1080p') @@ -39,7 +61,11 @@ test.group('ReleaseJudge', (group) => { await probe(halfDead, 'done') const fakeArchive = await release('Das.A-Team.S02E23.Das.Spiel.ist.aus') - await inspect(fakeArchive, 'archive', { type: 'rar4', names: ['Das.A-Team.S02E23.Das.Spiel.ist.aus.exe'], entries: 1 }) + await inspect(fakeArchive, 'archive', { + type: 'rar4', + names: ['Das.A-Team.S02E23.Das.Spiel.ist.aus.exe'], + entries: 1, + }) const fakeFile = await release('Some.Film.2021.1080p.WEB-DL') // Only "H.264" and "WEB-DL" say it's media here (a "?" in the pattern never matched). const fakeCodec = await release('Some.Show.H.264.WEB-DL') @@ -54,34 +80,71 @@ test.group('ReleaseJudge', (group) => { for (let i = 0; i < 20; i++) { const id = await release(`kantodo${i}`) repeatedIds.push(id) - await db.table('file_hashes').insert({ candidate_id: 20000 + i, release_id: id, hash16k: hash, size: 1073741824 }) + await db + .table('file_hashes') + .insert({ candidate_id: 20000 + i, release_id: id, hash16k: hash, size: 1073741824 }) } - const scraped = await release('Jamie Lee Curtis Nude Celebrity Pictures', { poster: '@_CelebMovie_@', source: 'subject' }) - const celeborn = await release('Some.Album.2010', { poster: 'celeborn@not.here (Celeborn)', source: 'subject' }) + const scraped = await release('Jamie Lee Curtis Nude Celebrity Pictures', { + poster: '@_CelebMovie_@', + source: 'subject', + }) + const celeborn = await release('Some.Album.2010', { + poster: 'celeborn@not.here (Celeborn)', + source: 'subject', + }) // Scraped posters are found through posters (the batch store adds each release's). - await db.table('posters').multiInsert([{ poster: '@_CelebMovie_@' }, { poster: 'celeborn@not.here (Celeborn)' }]) + await db + .table('posters') + .multiInsert([{ poster: '@_CelebMovie_@' }, { poster: 'celeborn@not.here (Celeborn)' }]) const locked = await release('Locked.Thing', { passworded: true }) const unlocked = await release('Known.Password', { passworded: true }) - await db.table('release_passwords').insert({ release_id: unlocked, password: 'pw', source: 'spot-text' }) + await db + .table('release_passwords') + .insert({ release_id: unlocked, password: 'pw', source: 'spot-text' }) // A block of done ranges around the unusable ones (release_final); one beside an unread range isn't judged yet. - await db.table('groups').insert({ id: 1, name: 'alt.binaries.judged', priority: 1 }).onConflict('id').ignore() + await db + .table('groups') + .insert({ id: 1, name: 'alt.binaries.judged', priority: 1 }) + .onConflict('id') + .ignore() await db.table('ranges').multiInsert([ - { group_id: 1, start_id: 1, end_id: 1000, next_id: 1001, status: 'done', posted_low: new Date(settled.getTime() - 86400_000), posted_high: new Date(settled.getTime() + 86400_000) }, + { + group_id: 1, + start_id: 1, + end_id: 1000, + next_id: 1001, + status: 'done', + posted_low: new Date(settled.getTime() - 86400_000), + posted_high: new Date(settled.getTime() + 86400_000), + }, { group_id: 1, start_id: 1001, end_id: 2000, next_id: 1001, status: 'planned' }, ]) - await RangeBlocks.rebuild() const placed = { first_article: 100, last_article: 110 } const unusable = await release('ab12CD34ef56GH78ij90', { readable: false, ...placed }) - const pending = await release('pp12CD34ef56GH78ij90', { readable: false, first_article: 990, last_article: 1000, posted_at: new Date(settled.getTime() + 86400_000 - 3600_000) }) - const fresh = await release('zz12CD34ef56GH78ij90', { readable: false, settled_at: new Date(), ...placed }) + const pending = await release('pp12CD34ef56GH78ij90', { + readable: false, + first_article: 990, + last_article: 1000, + posted_at: new Date(settled.getTime() + 86400_000 - 3600_000), + }) + const fresh = await release('zz12CD34ef56GH78ij90', { + readable: false, + settled_at: new Date(), + ...placed, + }) const spotted = await release('yy12CD34ef56GH78ij90', { readable: false, ...placed }) await db.table('release_spots').insert({ release_id: spotted, spot_id: 's@x', files: 1 }) const round = await new ReleaseJudge().full() - const verdicts = new Map((await db.from('release_verdicts').select('release_id', 'rule')).map((v) => [Number(v.release_id), v.rule])) + const verdicts = new Map( + (await db.from('release_verdicts').select('release_id', 'rule')).map((v) => [ + Number(v.release_id), + v.rule, + ]) + ) assert.equal(verdicts.get(dead), 'dead') assert.isUndefined(verdicts.get(halfDead)) assert.equal(verdicts.get(fakeArchive), 'fake') @@ -106,7 +169,9 @@ test.group('ReleaseJudge', (group) => { assert.equal((await SearchService.search({ q: 'Locked', judged: true })).total, 1) // A password found: the next round takes the verdict back. - await db.table('release_passwords').insert({ release_id: locked, password: 'found', source: 'nzb' }) + await db + .table('release_passwords') + .insert({ release_id: locked, password: 'found', source: 'nzb' }) const again = await new ReleaseJudge().full() assert.equal(again.rules.locked.removed, 1) // Verdicts that stay are not written again. @@ -115,7 +180,9 @@ test.group('ReleaseJudge', (group) => { assert.isNull(await db.from('release_verdicts').where('release_id', locked).first()) }) - test('judges the releases whose facts changed, and those a hash of theirs tipped over', async ({ assert }) => { + test('judges the releases whose facts changed, and those a hash of theirs tipped over', async ({ + assert, + }) => { let n = 0 const settled = new Date(Date.now() - 3 * 86400_000) const release = async (name: string, extra: Record = {}) => { @@ -123,12 +190,24 @@ test.group('ReleaseJudge', (group) => { const key = `jc${String(n).padStart(6, '0')}`.padEnd(40, '0') const [row] = await db .table('releases') - .insert({ release_key: key, identity_key: key, name, source: 'stem', groups: '[]', files: 3, posted_at: settled, readable: true, settled_at: settled, ...extra }) + .insert({ + release_key: key, + identity_key: key, + name, + source: 'stem', + groups: '[]', + files: 3, + posted_at: settled, + readable: true, + settled_at: settled, + ...extra, + }) .returning('id') return Number(row.id) } const judge = new ReleaseJudge() - const verdict = async (id: number) => (await db.from('release_verdicts').where('release_id', id).first())?.rule + const verdict = async (id: number) => + (await db.from('release_verdicts').where('release_id', id).first())?.rule const dirty = async () => Number((await db.from('judge_dirty').count('* as n'))[0].n) // New releases are noted; a passworded one is locked, until a password turns up. @@ -138,20 +217,31 @@ test.group('ReleaseJudge', (group) => { assert.equal(await verdict(locked), 'locked') assert.equal(round.rules.locked.added, 1) assert.equal(await dirty(), 0) - await db.table('release_passwords').insert({ release_id: locked, password: 'pw', source: 'nzb' }) + await db + .table('release_passwords') + .insert({ release_id: locked, password: 'pw', source: 'nzb' }) round = await judge.changed() assert.isUndefined(await verdict(locked)) assert.equal(round.rules.locked.removed, 1) // Probes: a release whose probed articles are all missing is dead once the prober's statuses come in. const dead = await release('Gone.Movie.2020.1080p') - const ids = await db.table('probe_candidates').multiInsert([ - { release_id: dead, kind: 'lone', message_id: '', file: 'a', status: 'waiting' }, - { release_id: dead, kind: 'lone', message_id: '', file: 'b', status: 'waiting' }, - ]).returning('id') + const ids = await db + .table('probe_candidates') + .multiInsert([ + { release_id: dead, kind: 'lone', message_id: '', file: 'a', status: 'waiting' }, + { release_id: dead, kind: 'lone', message_id: '', file: 'b', status: 'waiting' }, + ]) + .returning('id') await judge.changed() assert.isUndefined(await verdict(dead)) - await db.from('probe_candidates').whereIn('id', ids.map((r: { id: string }) => r.id)).update({ status: 'missing' }) + await db + .from('probe_candidates') + .whereIn( + 'id', + ids.map((r: { id: string }) => r.id) + ) + .update({ status: 'missing' }) await judge.changed() assert.equal(await verdict(dead), 'dead') @@ -159,13 +249,19 @@ test.group('ReleaseJudge', (group) => { // comes first), though only the 20th changed. const hash = 'b'.repeat(32) const holders: number[] = [] - for (let i = 0; i < 19; i++) holders.push(await release(`holder${i}`, i === 0 ? { passworded: true } : {})) - for (const [i, id] of holders.entries()) await db.table('file_hashes').insert({ candidate_id: 30000 + i, release_id: id, hash16k: hash, size: 5000 }) + for (let i = 0; i < 19; i++) + holders.push(await release(`holder${i}`, i === 0 ? { passworded: true } : {})) + for (const [i, id] of holders.entries()) + await db + .table('file_hashes') + .insert({ candidate_id: 30000 + i, release_id: id, hash16k: hash, size: 5000 }) await judge.changed() assert.equal(await verdict(holders[0]), 'locked') assert.isUndefined(await verdict(holders[1])) const twentieth = await release('holder19') - await db.table('file_hashes').insert({ candidate_id: 30019, release_id: twentieth, hash16k: hash, size: 5000 }) + await db + .table('file_hashes') + .insert({ candidate_id: 30019, release_id: twentieth, hash16k: hash, size: 5000 }) round = await judge.changed() assert.equal(round.releases, 20) const rules = new Set() @@ -173,7 +269,9 @@ test.group('ReleaseJudge', (group) => { assert.deepEqual([...rules], ['repeated']) // A 21st: the file was in 20 releases without it, so only it is judged. const twentyFirst = await release('holder20') - await db.table('file_hashes').insert({ candidate_id: 30020, release_id: twentyFirst, hash16k: hash, size: 5000 }) + await db + .table('file_hashes') + .insert({ candidate_id: 30020, release_id: twentyFirst, hash16k: hash, size: 5000 }) round = await judge.changed() assert.equal(round.releases, 1) assert.equal(await verdict(twentyFirst), 'repeated') diff --git a/tests/functional/settler.spec.ts b/tests/functional/settler.spec.ts deleted file mode 100644 index 21611ef..0000000 --- a/tests/functional/settler.spec.ts +++ /dev/null @@ -1,39 +0,0 @@ -import { test } from '@japa/runner' -import db from '@adonisjs/lucid/services/db' -import testUtils from '@adonisjs/core/services/test_utils' -import Settler from '#services/settler' - -test.group('Settler', (group) => { - group.each.setup(() => testUtils.db().withGlobalTransaction()) - - test('settles quiet releases: complete by the [i/N] counter, unknown without one', async ({ assert }) => { - const old = new Date(Date.now() - 7 * 3600 * 1000) - const row = (key: string, files: number, fileTotal: number | null, lastSeen: Date) => ({ - release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'subject', groups: '[]', files, file_total: fileTotal, last_seen_at: lastSeen, - }) - await db.table('releases').multiInsert([row('a', 10, 10, old), row('b', 8, 10, old), row('c', 3, null, old), row('d', 10, 10, new Date())]) - assert.equal(await Settler.run(), 3) - const byName = Object.fromEntries((await db.from('releases').select('name', 'complete', 'settled_at')).map((r) => [r.name, [r.complete, r.settled_at !== null]])) - assert.deepEqual(byName, { a: [true, true], b: [false, true], c: [null, true], d: [null, false] }) - }) - - test('settles history releases (posted long ago) once unchanged for half an hour', async ({ assert }) => { - const posted = new Date('2015-01-01T00:00:00Z') - const row = (key: string, lastSeen: Date, lastPosted: Date) => ({ - release_key: key.padEnd(40, '0'), identity_key: key.padEnd(40, '0'), name: key, source: 'subject', groups: '[]', files: 1, last_seen_at: lastSeen, posted_at: lastPosted, last_posted_at: lastPosted, - }) - const minutesAgo = (m: number) => new Date(Date.now() - m * 60_000) - await db.table('releases').multiInsert([row('old', minutesAgo(31), posted), row('fresh', minutesAgo(5), posted), row('tail', minutesAgo(31), minutesAgo(40))]) - const ids = Object.fromEntries((await db.from('releases').select('id', 'name')).map((r) => [r.name, r.id])) - await db.table('probe_candidates').multiInsert([ - { release_id: ids.old, kind: 'lone', message_id: '', file: 'a' }, - { release_id: ids.fresh, kind: 'lone', message_id: '', file: 'b' }, - ]) - assert.equal(await Settler.run(), 1) - // The settled release's candidates are claimable now, as unreadable (readable defaults to false). - const ready = await db.from('probe_candidates').orderBy('message_id').select('ready', 'unreadable') - assert.deepEqual(ready.map((r) => [r.ready, r.unreadable]), [[true, true], [false, false]]) - const settled = (await db.from('releases').whereNotNull('settled_at').select('name')).map((r) => r.name) - assert.deepEqual(settled, ['old']) - }) -}) diff --git a/tests/unit/release/source.spec.ts b/tests/unit/release/source.spec.ts new file mode 100644 index 0000000..d9c7644 --- /dev/null +++ b/tests/unit/release/source.spec.ts @@ -0,0 +1,19 @@ +import { test } from '@japa/runner' +import { SOURCES, sourceCode, sourceName } from '#lib/release/source' + +test.group('release source codes', () => { + test('names and codes map both ways; the codes are fixed', ({ assert }) => { + assert.deepEqual(SOURCES, { subject: 1, post: 2, stem: 3, spot: 4, nzb: 5, album: 6 }) + for (const [name, code] of Object.entries(SOURCES)) { + assert.equal(sourceCode(name), code) + assert.equal(sourceName(code), name) + assert.equal(sourceName(String(code)), name) + } + }) + + test('an unknown name throws, an unknown code reads as unknown', ({ assert }) => { + assert.throws(() => sourceCode('subjekt'), /Unknown release source/) + assert.equal(sourceName(99), 'unknown') + assert.equal(sourceName(null), 'unknown') + }) +})