Voyager 3 schema: the slim catalog (migration 6000, the reset) and blocks (6100)
Test and deploy / test (push) Failing after 50s
Test and deploy / edge (push) Successful in 1m53s
Test and deploy / deploy (push) Skipped

releases rebuilt without its text copies: poster_id, group_set_id,
bytea keys, source smallint, edge, settled_at as the finish. posters by
id, group_sets. blocks as the lease unit, ranges.block_id. fragments,
range_blocks, release_final, the settler and the fragment merger gone;
release_chunks, rule_counts and rule_samples keyed by block. Cold tables
on CHUNK_TABLESPACE. Migration 6000 truncates the catalog and refuses
without VOYAGER3_RESET=1.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Daan MeijerandClaude Fable 5.1 committed 2026-10-25 13:44:54 +01:00
1 parent 1a448f37f9
commit 346493b6c3
17 files changed
+527 -545

No files matched your search

+49
View File
@@ -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
-12
View File
@@ -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)
-153
View File
@@ -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<MergeResult> {
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<string, typeof rows>()
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<EdgeSegment>; 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<string, { files: AggregatedFile<EdgeSegment>[]; 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<string, string>()
const byStart = new Map<string, string>()
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()
+1 -1
View File
@@ -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)
-31
View File
@@ -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<number> {
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()
+2 -2
View File
@@ -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)`,
+6 -7
View File
@@ -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,
-49
View File
@@ -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<number> {
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()
+43 -60
View File
@@ -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.')
@@ -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)')
}
}
@@ -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`)
}
}
+36
View File
@@ -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<number, Source>
/** 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'
}
@@ -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 <p@x>', 'date': date, 'message-id': `<f${i}@x>`, ':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<string, unknown> = {}) => {
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)
})
})
-65
View File
@@ -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 <p@x>', '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)
})
})
+129 -31
View File
@@ -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<string, unknown> = {}) => {
@@ -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<string, unknown>) => {
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<string, unknown> = {}) => {
@@ -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: '<jc1@x>', file: 'a', status: 'waiting' },
{ release_id: dead, kind: 'lone', message_id: '<jc2@x>', file: 'b', status: 'waiting' },
]).returning('id')
const ids = await db
.table('probe_candidates')
.multiInsert([
{ release_id: dead, kind: 'lone', message_id: '<jc1@x>', file: 'a', status: 'waiting' },
{ release_id: dead, kind: 'lone', message_id: '<jc2@x>', 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')
-39
View File
@@ -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: '<a@x>', file: 'a' },
{ release_id: ids.fresh, kind: 'lone', message_id: '<b@x>', 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'])
})
})
+19
View File
@@ -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')
})
})