From 8c367440880d30923aa17567b08b0cb5f04ae9e3 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 5 Oct 2026 12:02:01 +0200 Subject: [PATCH] Seed orchestrator: remote nodes lease ranges and post results over HTTPS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - seed_nodes (hashed bearer tokens) and seed_ranges (planned ranges with lease, resume point and counts); migration …140. - SeedOrchestrator: lease (oldest plan first, FOR UPDATE SKIP LOCKED; a lease without a batch for 15 min passes to the next node and resumes at next_id), batch (stores files and fragments, moves next_id), finish, release, plan (from below live indexing / the backfill / earlier plans down to N days back; takes over the group's backfill), status. - Seed API under /api/v1/seed (lease, ranges/:id/batches|done|release, status) behind a node-token middleware; JSON bodies up to 64 MB after gzip. - seed:node runs a node: no database or Redis, outgoing HTTPS only, N connections, gzip batches, retries for network errors and 5xx, drops a range on 409, gives its ranges back on SIGTERM. seed-node/ has a compose file that builds the image from this repository. - seed:token, seed:plan, seed:status for the operator. Header parsing moved to lib/ingest and the range loop to lib/seed/runner, shared by seed:range and seed:node. Co-Authored-By: Claude Opus 5.5 --- .adonisjs/server/controllers.ts | 1 + app/controllers/seed_controller.ts | 65 ++++++ app/middleware/seed_node_middleware.ts | 26 +++ app/services/IngestService.ts | 107 +--------- app/services/SeedOrchestrator.ts | 188 +++++++++++++++++ app/services/SeedService.ts | 88 ++++---- commands/SeedNode.ts | 194 ++++++++++++++++++ commands/SeedPlan.ts | 41 ++++ commands/SeedRange.ts | 143 ++++--------- commands/SeedStatus.ts | 25 +++ commands/SeedToken.ts | 37 ++++ config/bodyparser.ts | 5 + .../1768620765140_seed_orchestrator.ts | 47 +++++ database/schema.ts | 54 +++++ lib/ingest/segment.ts | 105 ++++++++++ lib/seed/runner.ts | 78 +++++++ lib/seed/wire.ts | 73 +++++++ seed-node/README.md | 26 +++ seed-node/compose.yaml | 18 ++ seed-node/node.env.example | 29 +++ start/kernel.ts | 1 + start/routes.ts | 13 ++ tests/functional/seed.spec.ts | 3 +- tests/functional/seed_api.spec.ts | 77 +++++++ 24 files changed, 1198 insertions(+), 246 deletions(-) create mode 100644 app/controllers/seed_controller.ts create mode 100644 app/middleware/seed_node_middleware.ts create mode 100644 app/services/SeedOrchestrator.ts create mode 100644 commands/SeedNode.ts create mode 100644 commands/SeedPlan.ts create mode 100644 commands/SeedStatus.ts create mode 100644 commands/SeedToken.ts create mode 100644 database/migrations/1768620765140_seed_orchestrator.ts create mode 100644 lib/ingest/segment.ts create mode 100644 lib/seed/runner.ts create mode 100644 lib/seed/wire.ts create mode 100644 seed-node/README.md create mode 100644 seed-node/compose.yaml create mode 100644 seed-node/node.env.example create mode 100644 tests/functional/seed_api.spec.ts diff --git a/.adonisjs/server/controllers.ts b/.adonisjs/server/controllers.ts index ed3a28e..34f4bc9 100644 --- a/.adonisjs/server/controllers.ts +++ b/.adonisjs/server/controllers.ts @@ -9,4 +9,5 @@ export const controllers = { Newznab: () => import('#controllers/newznab_controller'), Profile: () => import('#controllers/profile_controller'), Releases: () => import('#controllers/releases_controller'), + Seed: () => import('#controllers/seed_controller'), } diff --git a/app/controllers/seed_controller.ts b/app/controllers/seed_controller.ts new file mode 100644 index 0000000..9b991a6 --- /dev/null +++ b/app/controllers/seed_controller.ts @@ -0,0 +1,65 @@ +import type { HttpContext } from '@adonisjs/core/http' +import SeedOrchestrator, { LeaseLost } from '#services/SeedOrchestrator' +import type { SeedBatch } from '#lib/seed/wire' + +/** + * The seed API (CONCEPT.md D38). Nodes make outgoing HTTPS requests only: + * lease a range, post batches of what they found, then finish or release + * the range. Bodies may be gzip-compressed (Content-Encoding: gzip). + */ +export default class SeedController { + /** POST /api/v1/seed/lease → 200 { range } or 204 when there's nothing to do. */ + async lease({ seedNode, response }: HttpContext) { + const range = await SeedOrchestrator.lease(seedNode.id) + return range ? { range } : response.noContent() + } + + /** POST /api/v1/seed/ranges/:id/batches with a SeedBatch. */ + async batch({ seedNode, params, request, response }: HttpContext) { + const batch = request.body() as SeedBatch + if (!Number.isSafeInteger(batch?.next) || !Array.isArray(batch.files) || !Array.isArray(batch.fragments)) { + return response.badRequest({ error: 'Expected { next, headers, files[], fragments[] }' }) + } + return this.leased(response, async () => { + const stored = await SeedOrchestrator.batch(seedNode.id, Number(params.id), { + ...batch, + headers: Number(batch.headers) || 0, + obfuscated: Number(batch.obfuscated) || 0, + deferred: Number(batch.deferred) || 0, + }) + return { stored } + }) + } + + /** POST /api/v1/seed/ranges/:id/done */ + async done({ seedNode, params, response }: HttpContext) { + return this.leased(response, async () => { + await SeedOrchestrator.finish(seedNode.id, Number(params.id)) + return { ok: true } + }) + } + + /** POST /api/v1/seed/ranges/:id/release { error? } */ + async release({ seedNode, params, request, response }: HttpContext) { + const error = request.input('error') + return this.leased(response, async () => { + await SeedOrchestrator.release(seedNode.id, Number(params.id), typeof error === 'string' ? error : undefined) + return { ok: true } + }) + } + + /** GET /api/v1/seed/status */ + async status() { + return SeedOrchestrator.status() + } + + /** 409 when the range was given to another node meanwhile; the node should drop it. */ + private async leased(response: HttpContext['response'], action: () => Promise) { + try { + return await action() + } catch (error) { + if (error instanceof LeaseLost) return response.conflict({ error: error.message }) + throw error + } + } +} diff --git a/app/middleware/seed_node_middleware.ts b/app/middleware/seed_node_middleware.ts new file mode 100644 index 0000000..b8865d0 --- /dev/null +++ b/app/middleware/seed_node_middleware.ts @@ -0,0 +1,26 @@ +import type { HttpContext } from '@adonisjs/core/http' +import type { NextFn } from '@adonisjs/core/types/http' +import SeedOrchestrator from '#services/SeedOrchestrator' + +declare module '@adonisjs/core/http' { + interface HttpContext { + /** The seeding node that made this request (CONCEPT.md D38). */ + seedNode: { id: number; name: string } + } +} + +/** + * Seed API requests carry a node's bearer token (`node ace seed:token + * `); anything else gets 401. The seed API is the only part of the app + * reachable from the internet (traefik routes only /api/v1/seed there). + */ +export default class SeedNodeMiddleware { + async handle(ctx: HttpContext, next: NextFn) { + const header = ctx.request.header('authorization') ?? '' + const token = header.startsWith('Bearer ') ? header.slice(7).trim() : '' + const node = token ? await SeedOrchestrator.nodeForToken(token) : null + if (!node) return ctx.response.unauthorized({ error: 'Unknown or disabled seed node token' }) + ctx.seedNode = node + return next() + } +} diff --git a/app/services/IngestService.ts b/app/services/IngestService.ts index 77941ba..2eef3b7 100644 --- a/app/services/IngestService.ts +++ b/app/services/IngestService.ts @@ -4,43 +4,10 @@ import type { QueryClientContract } from '@adonisjs/lucid/types/database' import type { Knex } from 'knex' import QueueService from '#services/QueueService' import NameService from '#services/NameService' -import { parseXrefGroups } from '#lib/nntp/xref' -import { fileKey, parseSubject, weakFileKey } from '#lib/subject/index' -import { releaseIdentity, type ReleaseIdentity } from '#lib/release/index' -import { inferTotalParts, type YencMeta } from '#lib/yenc/index' +import type { IngestSegment } from '#lib/ingest/segment' -/** One segment, with the identity of the file it belongs to already resolved. */ -export interface IngestSegment { - fileKey: string - poster: string - /** Subject without segment counter. */ - subject: string - name: string - totalParts: number - number: number - /** Without angle brackets. */ - messageId: string - bytes: number - articleNumber?: bigint | number - postedAt?: Date - groups: string[] - /** - * The identity comes from the yEnc name only (obfuscated posts). Such names - * can be shared by several equal-sized files of one post, so a second - * article for the same part number marks the binary as conflicted. - */ - weakIdentity?: boolean - /** Release (download) this file belongs to, when it can be determined. */ - release?: ReleaseIdentity | null - /** Exact file size from =ybegin (body-derived segments only). */ - size?: number - /** yEnc CRCs (body-derived segments only), to verify split volumes. */ - pcrc32?: string - partSize?: number - fileCrc32?: string - /** Part 1 only: MD5 of the first 16 KiB of the file, as par2 records it. */ - md5_16k?: string -} +// Header parsing is pure and lives in lib, so seeding nodes need no database (CONCEPT.md D37). +export { segmentFromHeader, segmentFromYenc, type IngestSegment, type OverviewHeader } from '#lib/ingest/segment' export interface CompletedBinary { id: number @@ -134,74 +101,6 @@ function validDate(date?: Date) { return date && !Number.isNaN(date.getTime()) ? date : null } -/** An XOVER overview entry as returned by nntp-js. */ -export type OverviewHeader = Record - -function common(header: OverviewHeader, groupName: string, articleNumber?: bigint | number) { - const xrefGroups = parseXrefGroups(header.xref) - return { - poster: header.from ?? '', - messageId: (header['message-id'] ?? '').replace(/^<|>$/g, ''), - bytes: Number.parseInt(header[':bytes'] ?? '0', 10) || 0, - articleNumber, - postedAt: header.date ? new Date(header.date) : undefined, - groups: xrefGroups.length > 0 ? xrefGroups : [groupName], - } -} - -/** Builds a segment from the subject line; null when the subject is obfuscated. */ -export function segmentFromHeader( - header: OverviewHeader, - groupName: string, - articleNumber?: bigint | number -): IngestSegment | null { - const parsed = header.subject ? parseSubject(header.subject) : null - if (!parsed || !header['message-id']) return null - const base = common(header, groupName, articleNumber) - return { - ...base, - fileKey: fileKey(base.poster, parsed.base, parsed.totalSegments), - subject: parsed.base, - name: parsed.name, - totalParts: parsed.totalSegments, - number: parsed.segment, - release: releaseIdentity({ poster: base.poster, subject: parsed.base, name: parsed.name, weak: false }), - } -} - -/** - * Builds a segment for an obfuscated post from the yEnc header in its body. - * Subject and poster can be random per article, so the identity comes from - * the yEnc header only (see weakFileKey). - */ -export function segmentFromYenc( - header: OverviewHeader, - groupName: string, - meta: Pick & Partial>, - articleNumber?: bigint | number -): IngestSegment | null { - const base = common(header, groupName, articleNumber) - const totalParts = inferTotalParts(meta) - const number = meta.begin.part ?? 1 - if (!base.messageId || number < 1 || number > totalParts) return null - const subject = `"${meta.begin.name}" yEnc` - return { - ...base, - fileKey: weakFileKey(meta.begin.name, meta.begin.size, totalParts), - subject, - name: meta.begin.name, - totalParts, - number, - weakIdentity: true, - release: releaseIdentity({ poster: base.poster, subject, name: meta.begin.name, weak: true }), - size: meta.begin.size, - // Single-part posts only carry crc32, which is then also the part's CRC. - pcrc32: meta.end?.pcrc32 ?? (meta.part ? undefined : meta.end?.crc32), - partSize: meta.part ? meta.part.end - meta.part.begin + 1 : meta.begin.size, - fileCrc32: meta.end?.crc32, - } -} - /** * Stores segments in `binaries`/`parts` and queues every binary that became * complete exactly once. Idempotent: ingesting the same segments again is a diff --git a/app/services/SeedOrchestrator.ts b/app/services/SeedOrchestrator.ts new file mode 100644 index 0000000..b0da532 --- /dev/null +++ b/app/services/SeedOrchestrator.ts @@ -0,0 +1,188 @@ +import { createHash, randomBytes } from 'node:crypto' +import db from '@adonisjs/lucid/services/db' +import NntpService from '#services/NntpService' +import SeedService from '#services/SeedService' +import { articleAtDate } from '#lib/nntp/date_search' +import { normalizeSeedFile, type SeedBatch } from '#lib/seed/wire' + +/** A lease without a batch for this long is given to another node; the range resumes at next_id. */ +export const LEASE_TTL_SECONDS = 15 * 60 +/** Leases per range before it counts as failed (a range that keeps breaking nodes). */ +const MAX_ATTEMPTS = 5 +/** Articles per XOVER probe when looking for an article by date. */ +const PROBE = 50 + +export interface SeedLease { + id: number + group: string + /** Resume point: first article not reported yet. */ + start: number + end: number +} + +export class LeaseLost extends Error {} + +const hash = (token: string) => createHash('sha256').update(token).digest('hex') + +/** + * The seed orchestrator (CONCEPT.md D38): seeding nodes lease article + * ranges over HTTPS, report what they aggregated in batches (D37), and + * finish or give back the range. Nodes only make outgoing requests; a lease + * that goes quiet for LEASE_TTL_SECONDS passes to the next node that asks. + */ +class SeedOrchestrator { + /** Creates a node, or gives an existing one a new token. Returns the token (stored hashed only). */ + async issueToken(name: string): Promise { + const token = `sn_${randomBytes(24).toString('base64url')}` + await db + .table('seed_nodes') + .insert({ name, token_hash: hash(token), disabled: false }) + .onConflict('name') + .merge({ token_hash: hash(token), disabled: false }) + return token + } + + async nodeForToken(token: string): Promise<{ id: number; name: string } | null> { + const node = await db.from('seed_nodes').where('token_hash', hash(token)).where('disabled', false).first() + if (!node) return null + if (!node.last_seen_at || Date.now() - new Date(node.last_seen_at).getTime() > 60_000) { + await db.from('seed_nodes').where('id', node.id).update({ last_seen_at: new Date() }) + } + return { id: Number(node.id), name: node.name } + } + + /** The next range for this node, oldest plan first; a quiet lease is taken over. */ + async lease(nodeId: number): Promise { + // Ranges that broke node after node stop being handed out. + await db.rawQuery( + `UPDATE seed_ranges SET status = 'failed', error = coalesce(error, 'lease expired too often') + WHERE status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?) AND attempts >= ?`, + [LEASE_TTL_SECONDS, MAX_ATTEMPTS] + ) + const { rows } = await db.rawQuery( + `UPDATE seed_ranges SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1 + WHERE id = ( + SELECT id FROM seed_ranges + WHERE status = 'queued' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)) + ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED) + RETURNING id, group_name, next_id, end_id`, + [nodeId, LEASE_TTL_SECONDS] + ) + const row = rows[0] + return row ? { id: Number(row.id), group: row.group_name, start: Number(row.next_id), end: Number(row.end_id) } : null + } + + /** Stores a node's batch and moves the range's resume point; LeaseLost when the range is no longer this node's. */ + async batch(nodeId: number, rangeId: number, batch: SeedBatch) { + const range = await this.leased(nodeId, rangeId) + const stored = await SeedService.store(batch.files.map(normalizeSeedFile)) + await SeedService.storeFragments(range.group_name, batch.fragments) + await db.rawQuery( + `UPDATE seed_ranges SET next_id = greatest(next_id, least(?, end_id + 1)), heartbeat_at = now(), + headers = headers + ?, files = files + ?, fragments = fragments + ? + WHERE id = ?`, + [batch.next, batch.headers, batch.files.length, batch.fragments.length, rangeId] + ) + return stored + } + + async finish(nodeId: number, rangeId: number) { + await this.leased(nodeId, rangeId) + await db.from('seed_ranges').where('id', rangeId).update({ status: 'done', done_at: new Date(), heartbeat_at: new Date(), error: null }) + } + + /** A node gives a range back (it failed, or is shutting down); it's queued again unless it failed too often. */ + async release(nodeId: number, rangeId: number, error?: string) { + const range = await this.leased(nodeId, rangeId) + await db + .from('seed_ranges') + .where('id', rangeId) + .update({ status: range.attempts >= MAX_ATTEMPTS ? 'failed' : 'queued', node_id: null, error: error?.slice(0, 2000) ?? null }) + } + + private async leased(nodeId: number, rangeId: number) { + const range = await db.from('seed_ranges').where('id', rangeId).first() + if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new LeaseLost(`Range ${rangeId} is not leased to this node`) + return range + } + + /** + * Plans ranges for groups: from below what live indexing and the backfill + * cover (or below earlier seed ranges) down to `days` back, or to the + * server's oldest article. A group's backfill stops where its seed plan + * begins. Ranges are interleaved across groups, newest first. + */ + async plan(groups: string[], options: { days?: number; size: number }): Promise<{ group: string; from: number; to: number; ranges: number }[]> { + const perGroup: { group: string; ranges: [number, number][] }[] = [] + const summary: { group: string; from: number; to: number; ranges: number }[] = [] + const conn = await NntpService.acquire() + let failed = false + try { + for (const name of groups) { + const group = await db.from('groups').where('name', name).first() + if (!group) throw new Error(`Unknown group ${name}`) + const { first, last } = await NntpService.timed(NntpService.selectGroup(conn, name), `GROUP ${name}`) + const dateOf = async (n: bigint) => { + const { overviews } = await NntpService.timed(conn.xover(n, n + BigInt(PROBE)), `XOVER ${name} ${n}`) + const header = overviews.find(([number, h]: [number | null, any]) => number !== null && h?.date)?.[1] + const date = header ? Date.parse(header.date) : Number.NaN + return Number.isNaN(date) ? null : date + } + // Top: where live indexing began, or the backfill's position, or below earlier seed ranges. + let top = Number(await articleAtDate(first, last, new Date(group.created_at).getTime(), dateOf)) + const backfilling = group.backfill_next_id !== null && group.backfill_until_id !== null && Number(group.backfill_next_id) > Number(group.backfill_until_id) + if (backfilling) top = Math.min(top, Number(group.backfill_next_id)) + const seeded = await db.from('seed_ranges').where('group_name', name).min('start_id as low').first() + if (seeded?.low !== null && seeded?.low !== undefined) top = Math.min(top, Number(seeded.low)) + const bottom = options.days + ? Number(await articleAtDate(first, last, Date.now() - options.days * 24 * 60 * 60 * 1000, dateOf)) + : Number(first) + + const ranges: [number, number][] = [] + for (let end = top - 1; end >= bottom; end -= options.size) ranges.push([Math.max(bottom, end - options.size + 1), end]) + perGroup.push({ group: name, ranges }) + summary.push({ group: name, from: bottom, to: top - 1, ranges: ranges.length }) + // The seed plan takes over the rest of this group's backfill. + if (backfilling && ranges.length > 0) { + await db.from('groups').where('id', group.id).update({ backfill_until_id: group.backfill_next_id }) + } + } + } catch (error) { + failed = true + throw error + } finally { + NntpService.release(conn, failed) + } + + // Round-robin over groups, each newest first, so every group fills in from the present backwards. + const rows: Record[] = [] + for (let i = 0; perGroup.some((g) => i < g.ranges.length); i++) { + for (const g of perGroup) { + const range = g.ranges[i] + if (range) rows.push({ group_name: g.group, start_id: range[0], end_id: range[1], next_id: range[0] }) + } + } + for (let i = 0; i < rows.length; i += 5000) await db.table('seed_ranges').multiInsert(rows.slice(i, i + 5000)) + return summary + } + + async status() { + const { rows: ranges } = await db.rawQuery( + `SELECT status, count(*)::int AS ranges, sum(end_id - start_id + 1)::bigint AS articles, + sum(next_id - start_id)::bigint AS reached, sum(headers)::bigint AS headers, sum(files)::bigint AS files + FROM seed_ranges GROUP BY status ORDER BY status` + ) + const { rows: nodes } = await db.rawQuery( + `SELECT n.name, n.last_seen_at, n.disabled, + (SELECT count(*)::int FROM seed_ranges r WHERE r.node_id = n.id AND r.status = 'leased') AS leases, + (SELECT coalesce(sum(headers), 0)::bigint FROM seed_ranges r WHERE r.node_id = n.id) AS headers + FROM seed_nodes n ORDER BY n.name` + ) + const { rows: recent } = await db.rawQuery( + `SELECT coalesce(sum(headers), 0)::bigint AS headers FROM seed_ranges WHERE done_at > now() - interval '1 hour'` + ) + return { ranges, nodes, headersDoneLastHour: Number(recent[0].headers) } + } +} + +export default new SeedOrchestrator() diff --git a/app/services/SeedService.ts b/app/services/SeedService.ts index 4187912..e82350a 100644 --- a/app/services/SeedService.ts +++ b/app/services/SeedService.ts @@ -1,10 +1,11 @@ import db from '@adonisjs/lucid/services/db' -import type { AggregatedFile } from '#lib/aggregate/index' -import IngestService, { type IngestSegment } from '#services/IngestService' +import IngestService from '#services/IngestService' import FileStore from '#services/FileStore' import SfvService from '#services/SfvService' import ContentService from '#services/ContentService' import QueueService from '#services/QueueService' +import StagingService from '#services/StagingService' +import { fromFragment, type SeedFile, type SeedFragment } from '#lib/seed/wire' export interface SeedStoreResult { files: number @@ -15,49 +16,42 @@ export interface SeedStoreResult { } /** - * Stores files an edge node aggregated from headers (CONCEPT.md D37): their - * releases, the files with their segment lists, and one completed binary per - * file, as worker:file would have left it, so par2 naming, inspection and GC - * treat them like any other file. A batch is a few statements, not one per - * article. + * Stores what seeding nodes found (CONCEPT.md D37): per complete file its + * release, the file with its segment list, and one completed binary, as + * worker:file would have left it, so par2 naming, inspection and GC treat + * it like any other file; fragments take the regular ingest path. A batch + * is a few statements, not one per article. */ class SeedService { - async store(files: AggregatedFile[]): Promise { + async store(batch: SeedFile[]): Promise { const result: SeedStoreResult = { files: 0, existing: 0, releases: 0, inspections: 0 } + // A repost can complete the same file twice in one batch; ON CONFLICT DO UPDATE takes each row once. + const files = [...new Map(batch.map((file) => [file.fileKey, file])).values()] if (files.length === 0) return result const knex = db.connection().getWriteClient() const now = new Date() + const postedOf = (file: SeedFile) => (file.postedAt ? new Date(file.postedAt) : undefined) - // A file's identity and release come from its first segment, its date from the earliest. - const firstOf = (file: AggregatedFile) => file.segments[0] - const postedOf = (file: AggregatedFile) => { - const dates = file.segments.map((s) => s.postedAt?.getTime()).filter((t): t is number => Number.isFinite(t)) - return dates.length > 0 ? new Date(Math.min(...dates)) : undefined - } const releaseIds = await IngestService.upsertReleases( knex, - files.map((file) => ({ ...firstOf(file), postedAt: postedOf(file) })), + files.map((file) => ({ release: file.release, poster: file.poster, groups: file.groups, postedAt: postedOf(file) })), now ) result.releases = releaseIds.size - const rows = files.map((file) => { - const first = firstOf(file) - const postedAt = postedOf(file) - return { - file_key: file.fileKey, - filename: first.name, - poster: first.poster, - date: (postedAt ?? now).getTime(), - parts: file.segments.length, - segments: file.segments.map((s) => ({ number: s.number, id: s.messageId, size: s.bytes })), - groups: JSON.stringify(first.groups), - release_id: first.release ? (releaseIds.get(first.release.key) ?? null) : null, - size: file.segments.reduce((sum, s) => sum + s.bytes, 0), - created_at: now, - updated_at: now, - } - }) + const rows = files.map((file) => ({ + file_key: file.fileKey, + filename: file.name, + poster: file.poster, + date: (postedOf(file) ?? now).getTime(), + parts: file.segments.length, + segments: file.segments.map(([number, id, size]) => ({ number, id, size })), + groups: JSON.stringify(file.groups), + release_id: file.release ? (releaseIds.get(file.release.key) ?? null) : null, + size: file.segments.reduce((sum, [, , size]) => sum + size, 0), + created_at: now, + updated_at: now, + })) const inserted = await FileStore.insert(rows) result.files = inserted.length result.existing = rows.length - inserted.length @@ -67,19 +61,19 @@ class SeedService { fileIds.set(row.file_key, Number(row.id)) } // The completed binary worker:file would have left behind; an existing one just learns its file. - const binaries = rows.map((row, i) => ({ - file_key: row.file_key, - poster: row.poster, - subject: firstOf(files[i]).subject, - name: row.filename, - total_parts: firstOf(files[i]).totalParts, - groups: row.groups, - posted_at: postedOf(files[i]) ?? null, + const binaries = files.map((file, i) => ({ + file_key: file.fileKey, + poster: file.poster, + subject: file.subject, + name: file.name, + total_parts: file.totalParts, + groups: rows[i].groups, + posted_at: postedOf(file) ?? null, first_seen_at: now, last_seen_at: now, completed_at: now, - release_id: row.release_id, - file_id: fileIds.get(row.file_key) ?? null, + release_id: rows[i].release_id, + file_id: fileIds.get(file.fileKey) ?? null, })) binaries.sort((a, b) => a.file_key.localeCompare(b.file_key)) await knex('binaries') @@ -101,6 +95,16 @@ class SeedService { result.inspections = candidates.length return result } + + /** Segments of files that didn't complete on a node: staging (D35) for lone ones, binaries and parts for the rest. */ + async storeFragments(group: string, fragments: SeedFragment[]): Promise { + if (fragments.length === 0) return + const segments = fragments.map(fromFragment) + const { ingest } = StagingService.enabled(group) + ? await StagingService.filter(segments, { counters: StagingService.countersEnabled(group) }) + : { ingest: segments } + await IngestService.ingest(ingest) + } } export default new SeedService() diff --git a/commands/SeedNode.ts b/commands/SeedNode.ts new file mode 100644 index 0000000..3ad4bec --- /dev/null +++ b/commands/SeedNode.ts @@ -0,0 +1,194 @@ +import { gzipSync } from 'node:zlib' +import { setTimeout as sleep } from 'node:timers/promises' +import { BaseCommand, flags } from '@adonisjs/core/ace' +import { CommandOptions } from '@adonisjs/core/types/ace' +import nntpConfig from '#config/nntp' +import indexerConfig from '#config/indexer' +import { XoverClient } from '#lib/nntp/xover_client' +import { runRange } from '#lib/seed/runner' +import type { SeedBatch } from '#lib/seed/wire' +import type { OverviewHeader } from '#lib/ingest/segment' + +class LeaseLost extends Error {} +class Stopping extends Error {} + +interface Lease { + id: number + group: string + start: number + end: number +} + +/** + * A seeding node (CONCEPT.md D38): leases article ranges from the seed API + * over HTTPS, fetches and aggregates their headers over its own NNTP + * connection(s) (D37), and posts the results back in gzip-compressed + * batches. Outgoing requests only: no ports, database or Redis needed. + * + * SEED_URL=https://seed.example SEED_TOKEN=sn_… NNTP_HOST=… node ace seed:node --connections 4 + */ +export default class SeedNode extends BaseCommand { + public static commandName = 'seed:node' + public static description = 'Runs a seeding node: leases ranges from the seed API and submits what it aggregated.' + + public static options: CommandOptions = { + startApp: false, + staysAlive: true, + } + + @flags.string({ description: 'Seed API base URL (default: SEED_URL)' }) + declare url?: string + + @flags.string({ description: 'Node token (default: SEED_TOKEN)' }) + declare token?: string + + @flags.number({ description: 'Ranges worked on at once, one NNTP connection each (default: SEED_CONNECTIONS or 1)', default: Number(process.env.SEED_CONNECTIONS) || 1 }) + declare connections: number + + @flags.number({ description: 'Articles per XOVER', default: 20_000 }) + declare chunk: number + + @flags.number({ description: 'XOVER chunks per batch posted to the API', default: 5 }) + declare submitEvery: number + + @flags.boolean({ description: 'Stop when there is nothing to lease instead of waiting' }) + declare once: boolean + + private stopping = false + + public async run() { + const base = (this.url ?? process.env.SEED_URL ?? '').replace(/\/+$/, '') + const token = this.token ?? process.env.SEED_TOKEN ?? '' + if (!base || !token) { + this.logger.error('Set --url and --token (or SEED_URL and SEED_TOKEN).') + this.exitCode = 1 + return + } + const stop = () => { + if (!this.stopping) this.logger.info('Stopping: giving back the ranges in progress…') + this.stopping = true + } + process.on('SIGTERM', stop) + process.on('SIGINT', stop) + + const api = new SeedApi(base, token, (message) => this.logger.warning(message)) + this.logger.info(`Seeding node for ${base} with ${this.connections} connection(s).`) + await Promise.all(Array.from({ length: this.connections }, (_, i) => this.work(api, i + 1))) + this.logger.info('Seeding node stopped.') + } + + private async work(api: SeedApi, worker: number) { + const log = (message: string) => this.logger.info(`[${worker}] ${message}`) + while (!this.stopping) { + let lease: Lease | null + try { + lease = ((await api.post('/api/v1/seed/lease')) as { range: Lease } | null)?.range ?? null + } catch (error: any) { + if (error instanceof Unauthorized) throw error + log(`Lease failed: ${error.message}`) + await this.pause(60_000) + continue + } + if (!lease) { + if (this.once) return + await this.pause(60_000) + continue + } + + log(`${lease.group} ${lease.start}–${lease.end}`) + const started = Date.now() + let headers = 0 + let client: XoverClient | undefined + try { + client = await XoverClient.connect({ + host: nntpConfig.host, + port: nntpConfig.port, + secure: nntpConfig.secure, + user: nntpConfig.user, + password: nntpConfig.password?.release(), + timeoutMs: nntpConfig.commandTimeoutMs, + compress: true, + }) + await runRange({ + client, + group: lease.group, + start: lease.start, + end: lease.end, + chunk: this.chunk, + submitEvery: this.submitEvery, + isDeferred: (header: OverviewHeader) => indexerConfig.deferBodyMessageIds.some((re) => re.test(header['message-id'] ?? '')), + sink: async (batch: SeedBatch) => { + if (this.stopping) throw new Stopping() + await api.post(`/api/v1/seed/ranges/${lease!.id}/batches`, batch) + headers += batch.headers + }, + }) + await api.post(`/api/v1/seed/ranges/${lease.id}/done`) + const seconds = (Date.now() - started) / 1000 + log(`done: ${headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s (${Math.round(headers / seconds).toLocaleString('en')}/s)`) + } catch (error: any) { + if (error instanceof Unauthorized) throw error + if (error instanceof LeaseLost) { + log(`Range ${lease.id} passed to another node; dropping it.`) + continue + } + const reason = error instanceof Stopping ? 'node stopping' : error.message + if (!(error instanceof Stopping)) log(`Range ${lease.id} failed: ${reason}`) + await api.post(`/api/v1/seed/ranges/${lease.id}/release`, { error: reason }).catch(() => {}) + if (!this.stopping) await this.pause(30_000) + } finally { + client?.close() + } + } + } + + /** Sleeps, but wakes up early when the node is stopping. */ + private async pause(ms: number) { + for (let waited = 0; waited < ms && !this.stopping; waited += 1000) await sleep(1000) + } +} + +class Unauthorized extends Error {} + +/** The seed API client: JSON over HTTPS, gzip request bodies, retries for network errors and 5xx. */ +class SeedApi { + constructor( + private base: string, + private token: string, + private warn: (message: string) => void + ) {} + + async post(path: string, body?: unknown): Promise { + const payload = body === undefined ? undefined : gzipSync(JSON.stringify(body)) + for (let attempt = 1; ; attempt++) { + let status = 0 + let text = '' + try { + const response = await fetch(this.base + path, { + method: 'POST', + headers: { + 'authorization': `Bearer ${this.token}`, + 'accept': 'application/json', + ...(payload ? { 'content-type': 'application/json', 'content-encoding': 'gzip' } : {}), + }, + body: payload, + signal: AbortSignal.timeout(10 * 60 * 1000), + }) + status = response.status + text = await response.text() + } catch (error: any) { + text = error.message + } + if (status === 204) return null + if (status >= 200 && status < 300) return text ? JSON.parse(text) : null + if (status === 401) throw new Unauthorized('The seed API rejected this node token') + if (status === 409) throw new LeaseLost(text) + if (status >= 400 && status < 500) throw new Error(`Seed API ${path}: ${status} ${text.slice(0, 300)}`) + // Network errors and 5xx: storing a batch twice is harmless, so try again. + if (attempt >= 8) throw new Error(`Seed API ${path}: ${status || 'no response'} ${text.slice(0, 300)}`) + const wait = Math.min(120_000, 2000 * 2 ** (attempt - 1)) + this.warn(`Seed API ${path}: ${status || text}; retrying in ${wait / 1000} s`) + await sleep(wait) + } + } +} diff --git a/commands/SeedPlan.ts b/commands/SeedPlan.ts new file mode 100644 index 0000000..706cdc0 --- /dev/null +++ b/commands/SeedPlan.ts @@ -0,0 +1,41 @@ +import { BaseCommand, args, flags } from '@adonisjs/core/ace' +import { CommandOptions } from '@adonisjs/core/types/ace' +import NntpService from '#services/NntpService' +import SeedOrchestrator from '#services/SeedOrchestrator' + +/** + * Plans seed ranges (CONCEPT.md D38) for groups, from below what live + * indexing, the backfill and earlier plans cover down to `--days` back (or + * the server's oldest article). A group's backfill stops where its plan + * starts. Ranges are interleaved across groups, newest first. + * + * node ace seed:plan alt.binaries.ath alt.binaries.mom --days 365 + */ +export default class SeedPlan extends BaseCommand { + public static commandName = 'seed:plan' + public static description = 'Plans article ranges of groups for seeding nodes to lease.' + + public static options: CommandOptions = { + startApp: true, + } + + @args.spread({ description: 'Groups' }) + declare groups: string[] + + @flags.number({ description: 'How many days back (default: everything the server has)' }) + declare days?: number + + @flags.number({ description: 'Articles per range', default: 2_000_000 }) + declare size: number + + public async run() { + try { + const summary = await SeedOrchestrator.plan(this.groups, { days: this.days, size: this.size }) + for (const s of summary) { + this.logger.info(`${s.group}: ${s.ranges.toLocaleString('en')} ranges, articles ${s.from.toLocaleString('en')}–${s.to.toLocaleString('en')}.`) + } + } finally { + await NntpService.shutdown() + } + } +} diff --git a/commands/SeedRange.ts b/commands/SeedRange.ts index 8d3554b..e4f3568 100644 --- a/commands/SeedRange.ts +++ b/commands/SeedRange.ts @@ -1,21 +1,19 @@ import { BaseCommand, args, flags } from '@adonisjs/core/ace' import { CommandOptions } from '@adonisjs/core/types/ace' import nntpConfig from '#config/nntp' -import { XoverClient, type Overview } from '#lib/nntp/xover_client' -import IngestService, { segmentFromHeader, type IngestSegment } from '#services/IngestService' -import StagingService from '#services/StagingService' import SeedService from '#services/SeedService' import QueueService from '#services/QueueService' import { isDeferred } from '#services/DeferredService' -import { FileAggregator } from '#lib/aggregate/index' +import { XoverClient } from '#lib/nntp/xover_client' +import { runRange } from '#lib/seed/runner' +import type { SeedBatch } from '#lib/seed/wire' /** - * Edge aggregation (CONCEPT.md D37): indexes a range of a group's history - * over one NNTP connection (gzip-compressed overviews when the server has - * them), grouping headers into files in memory and - * storing one row per file. Files cut by the range edges, or that never - * complete, go through the regular ingest path as fragments. Header-only, - * like the backfill: obfuscated subjects are counted, not downloaded. + * Edge aggregation (CONCEPT.md D37) on this machine: indexes a range of a + * group's history over one NNTP connection (gzip-compressed overviews when + * the server has them), grouping headers into files in memory and storing + * one row per file directly. seed:node does the same for a remote node, + * through the seed API. Header-only, like the backfill. * * node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000 * node ace seed:range alt.binaries.ath --start 46000000000 --count 2000000 --dry-run @@ -49,17 +47,12 @@ export default class SeedRange extends BaseCommand { @flags.boolean({ description: 'Ask the server for gzip-compressed overviews', default: true }) declare compress: boolean - /** Fragment batches being stored the regular way, off the node's critical path. */ - private fragmentWork = new Set>() - private fragmentError?: unknown - public async run() { const first = Number(this.start) - const last = first + this.count - 1 if (!Number.isSafeInteger(first) || first < 1) throw new Error(`--start must be an article number, got ${this.start}`) - const aggregator = new FileAggregator() - const stats = { headers: 0, segments: 0, obfuscated: 0, deferred: 0, complete: 0, files: 0, existing: 0, fragments: 0, inspections: 0 } - const time = { fetch: 0, parse: 0, store: 0 } + const last = first + this.count - 1 + const stats = { headers: 0, obfuscated: 0, deferred: 0, complete: 0, files: 0, existing: 0, fragments: 0, inspections: 0 } + let storeMs = 0 const cpuStart = process.cpuUsage() const started = Date.now() @@ -74,56 +67,33 @@ export default class SeedRange extends BaseCommand { }) this.logger.info(`Connected${client.compressed ? ', overviews gzip-compressed' : ''}.`) try { - await client.group(this.group) - const fetch = async (from: number) => { - const to = Math.min(from + this.chunk - 1, last) - const t = Date.now() - const overviews = await client.xover(from, to) - time.fetch += Date.now() - t - return overviews.filter(([n, h]) => Number.isFinite(n) && h['message-id']) - } - - // One XOVER in flight while the previous batch is parsed and stored. - let from = first - const prefetch = (at: number) => { - const promise = fetch(at) - // Awaited after the current batch is stored; until then a failure must not count as unhandled. - promise.catch(() => {}) - return promise - } - let next: Promise | null = prefetch(from) - while (next) { - const overviews = await next - from += this.chunk - next = from <= last ? prefetch(from) : null - - let t = Date.now() - const segments: IngestSegment[] = [] - for (const [articleNumber, header] of overviews) { - const segment = segmentFromHeader(header, this.group, articleNumber) - if (segment) segments.push(segment) - else if (isDeferred(header)) stats.deferred++ - else stats.obfuscated++ - } - stats.headers += overviews.length - stats.segments += segments.length - const complete = aggregator.add(segments) - const fragments = aggregator.evict(this.idle) - stats.complete += complete.length - stats.fragments += fragments.length - time.parse += Date.now() - t - - t = Date.now() - if (!this.dryRun) await this.store(complete, fragments, stats) - time.store += Date.now() - t - this.progress(stats, started) - } - - const rest = aggregator.evict(0) - stats.fragments += rest.length - if (!this.dryRun) await this.store([], rest, stats) - await Promise.all(this.fragmentWork) - if (this.fragmentError) throw this.fragmentError + await runRange({ + client, + group: this.group, + start: first, + end: last, + chunk: this.chunk, + idle: this.idle, + submitEvery: 1, + isDeferred, + sink: async (batch: SeedBatch) => { + stats.headers += batch.headers + stats.obfuscated += batch.obfuscated + stats.deferred += batch.deferred + stats.complete += batch.files.length + stats.fragments += batch.fragments.length + if (!this.dryRun) { + const t = Date.now() + const stored = await SeedService.store(batch.files) + await SeedService.storeFragments(this.group, batch.fragments) + storeMs += Date.now() - t + stats.files += stored.files + stats.existing += stored.existing + stats.inspections += stored.inspections + } + this.progress(stats, started) + }, + }) } finally { client.close() await QueueService.closeAll() @@ -131,39 +101,14 @@ export default class SeedRange extends BaseCommand { const seconds = (Date.now() - started) / 1000 const cpu = process.cpuUsage(cpuStart) + const n = (x: number) => x.toLocaleString('en') this.logger.info( - `${this.group} ${first}-${last}: ${stats.headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s ` + - `(${Math.round(stats.headers / seconds).toLocaleString('en')}/s). ` + - `${stats.segments.toLocaleString('en')} readable, ${stats.obfuscated.toLocaleString('en')} obfuscated, ${stats.deferred.toLocaleString('en')} deferred. ` + - `${stats.complete.toLocaleString('en')} complete files (${stats.files.toLocaleString('en')} new, ${stats.existing.toLocaleString('en')} known), ` + - `${stats.fragments.toLocaleString('en')} fragment segments, ${stats.inspections.toLocaleString('en')} queued for inspection.` + `${this.group} ${first}-${last}: ${n(stats.headers)} headers in ${seconds.toFixed(0)} s (${n(Math.round(stats.headers / seconds))}/s). ` + + `${n(stats.obfuscated)} obfuscated, ${n(stats.deferred)} deferred. ` + + `${n(stats.complete)} complete files (${n(stats.files)} new, ${n(stats.existing)} known), ` + + `${n(stats.fragments)} fragment segments, ${n(stats.inspections)} queued for inspection.` ) - this.logger.info( - `Time: fetch ${(time.fetch / 1000).toFixed(0)} s (overlapped), parse+aggregate ${(time.parse / 1000).toFixed(0)} s, ` + - `store ${(time.store / 1000).toFixed(0)} s; node CPU ${((cpu.user + cpu.system) / 1e6).toFixed(0)} s.` - ) - } - - private async store(complete: Parameters[0], fragments: IngestSegment[], stats: Record) { - const stored = await SeedService.store(complete) - stats.files += stored.files - stats.existing += stored.existing - stats.inspections += stored.inspections - if (fragments.length === 0) return - // The regular path: staging (D35) for lone segments, binaries and parts for the rest. It runs - // beside fetching; with two batches in flight already, the node waits for one to finish. - while (this.fragmentWork.size >= 2) await Promise.race([...this.fragmentWork].map((w) => w.catch(() => {}))) - if (this.fragmentError) throw this.fragmentError - const work = (async () => { - const { ingest } = StagingService.enabled(this.group) - ? await StagingService.filter(fragments, { counters: StagingService.countersEnabled(this.group) }) - : { ingest: fragments } - await IngestService.ingest(ingest) - })() - this.fragmentWork.add(work) - work - .catch((error) => (this.fragmentError ??= error)) - .finally(() => this.fragmentWork.delete(work)) + this.logger.info(`Store ${(storeMs / 1000).toFixed(0)} s; node CPU ${((cpu.user + cpu.system) / 1e6).toFixed(0)} s.`) } private lastReport = 0 diff --git a/commands/SeedStatus.ts b/commands/SeedStatus.ts new file mode 100644 index 0000000..599205f --- /dev/null +++ b/commands/SeedStatus.ts @@ -0,0 +1,25 @@ +import { BaseCommand } from '@adonisjs/core/ace' +import { CommandOptions } from '@adonisjs/core/types/ace' +import SeedOrchestrator from '#services/SeedOrchestrator' + +/** Seeding progress per range status and per node (CONCEPT.md D38). */ +export default class SeedStatus extends BaseCommand { + public static commandName = 'seed:status' + public static description = 'Shows seeding progress per range status and per node.' + + public static options: CommandOptions = { + startApp: true, + } + + public async run() { + const status = await SeedOrchestrator.status() + const n = (x: unknown) => Number(x ?? 0).toLocaleString('en') + for (const r of status.ranges) { + this.logger.info(`${r.status}: ${n(r.ranges)} ranges, ${n(r.reached)} of ${n(r.articles)} articles reached, ${n(r.headers)} headers, ${n(r.files)} files`) + } + for (const node of status.nodes) { + this.logger.info(`node ${node.name}${node.disabled ? ' (disabled)' : ''}: ${node.leases} leases, ${n(node.headers)} headers, last seen ${node.last_seen_at ?? 'never'}`) + } + this.logger.info(`Headers in ranges finished in the last hour: ${n(status.headersDoneLastHour)}`) + } +} diff --git a/commands/SeedToken.ts b/commands/SeedToken.ts new file mode 100644 index 0000000..6cca706 --- /dev/null +++ b/commands/SeedToken.ts @@ -0,0 +1,37 @@ +import { BaseCommand, args, flags } from '@adonisjs/core/ace' +import { CommandOptions } from '@adonisjs/core/types/ace' +import db from '@adonisjs/lucid/services/db' +import SeedOrchestrator from '#services/SeedOrchestrator' + +/** + * Creates a seeding node, or gives it a new token (the old one stops + * working), and prints the token once (CONCEPT.md D38). + * + * node ace seed:token hetzner-1 + * node ace seed:token hetzner-1 --disable + */ +export default class SeedToken extends BaseCommand { + public static commandName = 'seed:token' + public static description = 'Creates a seeding node or replaces its token; prints the token once.' + + public static options: CommandOptions = { + startApp: true, + } + + @args.string({ description: 'Node name' }) + declare name: string + + @flags.boolean({ description: 'Disable the node instead (its leases pass to others after they expire)' }) + declare disable: boolean + + public async run() { + if (this.disable) { + const n = await db.from('seed_nodes').where('name', this.name).update({ disabled: true }) + this.logger.info(n ? `${this.name}: disabled.` : `${this.name}: no such node.`) + return + } + const token = await SeedOrchestrator.issueToken(this.name) + this.logger.info(`${this.name}: token (shown once, keep it on the node only):`) + console.log(token) + } +} diff --git a/config/bodyparser.ts b/config/bodyparser.ts index 7e836e6..079d02b 100644 --- a/config/bodyparser.ts +++ b/config/bodyparser.ts @@ -32,6 +32,11 @@ const bodyParserConfig = defineConfig({ */ convertEmptyStringsToNull: true, + /** + * Seeding nodes post batches of a few MB (CONCEPT.md D38); the limit applies after gzip. + */ + limit: '64mb', + /** * Content types handled by the JSON parser. */ diff --git a/database/migrations/1768620765140_seed_orchestrator.ts b/database/migrations/1768620765140_seed_orchestrator.ts new file mode 100644 index 0000000..56f550f --- /dev/null +++ b/database/migrations/1768620765140_seed_orchestrator.ts @@ -0,0 +1,47 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * The seed orchestrator (CONCEPT.md D38): nodes that lease article ranges + * over HTTPS and submit what they aggregated (D37), and the ranges planned + * for them. A range's id is its priority: plans insert newest first. + */ +export default class extends BaseSchema { + async up() { + this.schema.createTable('seed_nodes', (table) => { + table.increments('id') + table.text('name').notNullable().unique() + // sha256 of the bearer token; the token itself is shown once, when it's made. + table.string('token_hash', 64).notNullable().unique() + table.boolean('disabled').notNullable().defaultTo(false) + table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now()) + table.timestamp('last_seen_at', { useTz: true }).nullable() + }) + + this.schema.createTable('seed_ranges', (table) => { + table.bigIncrements('id') + table.text('group_name').notNullable() + table.bigInteger('start_id').notNullable() + table.bigInteger('end_id').notNullable() + // First article not reported yet; a re-leased range resumes here. + table.bigInteger('next_id').notNullable() + table.string('status', 8).notNullable().defaultTo('queued') + table.integer('node_id').nullable().references('id').inTable('seed_nodes').onDelete('SET NULL') + table.integer('attempts').notNullable().defaultTo(0) + table.timestamp('leased_at', { useTz: true }).nullable() + table.timestamp('heartbeat_at', { useTz: true }).nullable() + table.timestamp('done_at', { useTz: true }).nullable() + table.bigInteger('headers').notNullable().defaultTo(0) + table.bigInteger('files').notNullable().defaultTo(0) + table.bigInteger('fragments').notNullable().defaultTo(0) + table.text('error').nullable() + table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now()) + table.index(['group_name', 'start_id']) + }) + this.schema.raw(`CREATE INDEX seed_ranges_open_idx ON seed_ranges (id) WHERE status IN ('queued', 'leased')`) + } + + async down() { + this.schema.dropTable('seed_ranges') + this.schema.dropTable('seed_nodes') + } +} diff --git a/database/schema.ts b/database/schema.ts index 3b77b29..48d635a 100644 --- a/database/schema.ts +++ b/database/schema.ts @@ -348,6 +348,60 @@ export class ReleaseSchema extends BaseModel { declare source: string | null } +export class SeedNodeSchema extends BaseModel { + static $columns = ['createdAt', 'disabled', 'id', 'lastSeenAt', 'name', 'tokenHash'] as const + $columns = SeedNodeSchema.$columns + @column.dateTime({ autoCreate: true }) + declare createdAt: DateTime + @column() + declare disabled: boolean + @column({ isPrimary: true }) + declare id: number + @column.dateTime() + declare lastSeenAt: DateTime | null + @column() + declare name: string + @column() + declare tokenHash: string +} + +export class SeedRangeSchema extends BaseModel { + static $columns = ['attempts', 'createdAt', 'doneAt', 'endId', 'error', 'files', 'fragments', 'groupName', 'headers', 'heartbeatAt', 'id', 'leasedAt', 'nextId', 'nodeId', 'startId', 'status'] as const + $columns = SeedRangeSchema.$columns + @column() + declare attempts: number + @column.dateTime({ autoCreate: true }) + declare createdAt: DateTime + @column.dateTime() + declare doneAt: DateTime | null + @column() + declare endId: bigint | number + @column() + declare error: string | null + @column() + declare files: bigint | number + @column() + declare fragments: bigint | number + @column() + declare groupName: string + @column() + declare headers: bigint | number + @column.dateTime() + declare heartbeatAt: DateTime | null + @column({ isPrimary: true }) + declare id: bigint | number + @column.dateTime() + declare leasedAt: DateTime | null + @column() + declare nextId: bigint | number + @column() + declare nodeId: number | null + @column() + declare startId: bigint | number + @column() + declare status: string +} + export class SfvEntrySchema extends BaseModel { static $columns = ['crc', 'name', 'postedAt', 'poster', 'releaseId'] as const $columns = SfvEntrySchema.$columns diff --git a/lib/ingest/segment.ts b/lib/ingest/segment.ts new file mode 100644 index 0000000..c1589c6 --- /dev/null +++ b/lib/ingest/segment.ts @@ -0,0 +1,105 @@ +import { parseXrefGroups } from '../nntp/xref.js' +import { fileKey, parseSubject, weakFileKey } from '../subject/index.js' +import { releaseIdentity, type ReleaseIdentity } from '../release/index.js' +import { inferTotalParts, type YencMeta } from '../yenc/index.js' + +/** One segment, with the identity of the file it belongs to already resolved. */ +export interface IngestSegment { + fileKey: string + poster: string + /** Subject without segment counter. */ + subject: string + name: string + totalParts: number + number: number + /** Without angle brackets. */ + messageId: string + bytes: number + articleNumber?: bigint | number + postedAt?: Date + groups: string[] + /** + * The identity comes from the yEnc name only (obfuscated posts). Such names + * can be shared by several equal-sized files of one post, so a second + * article for the same part number marks the binary as conflicted. + */ + weakIdentity?: boolean + /** Release (download) this file belongs to, when it can be determined. */ + release?: ReleaseIdentity | null + /** Exact file size from =ybegin (body-derived segments only). */ + size?: number + /** yEnc CRCs (body-derived segments only), to verify split volumes. */ + pcrc32?: string + partSize?: number + fileCrc32?: string + /** Part 1 only: MD5 of the first 16 KiB of the file, as par2 records it. */ + md5_16k?: string +} + +/** An XOVER overview entry as returned by nntp-js. */ +export type OverviewHeader = Record + +function common(header: OverviewHeader, groupName: string, articleNumber?: bigint | number) { + const xrefGroups = parseXrefGroups(header.xref) + return { + poster: header.from ?? '', + messageId: (header['message-id'] ?? '').replace(/^<|>$/g, ''), + bytes: Number.parseInt(header[':bytes'] ?? '0', 10) || 0, + articleNumber, + postedAt: header.date ? new Date(header.date) : undefined, + groups: xrefGroups.length > 0 ? xrefGroups : [groupName], + } +} + +/** Builds a segment from the subject line; null when the subject is obfuscated. */ +export function segmentFromHeader( + header: OverviewHeader, + groupName: string, + articleNumber?: bigint | number +): IngestSegment | null { + const parsed = header.subject ? parseSubject(header.subject) : null + if (!parsed || !header['message-id']) return null + const base = common(header, groupName, articleNumber) + return { + ...base, + fileKey: fileKey(base.poster, parsed.base, parsed.totalSegments), + subject: parsed.base, + name: parsed.name, + totalParts: parsed.totalSegments, + number: parsed.segment, + release: releaseIdentity({ poster: base.poster, subject: parsed.base, name: parsed.name, weak: false }), + } +} + +/** + * Builds a segment for an obfuscated post from the yEnc header in its body. + * Subject and poster can be random per article, so the identity comes from + * the yEnc header only (see weakFileKey). + */ +export function segmentFromYenc( + header: OverviewHeader, + groupName: string, + meta: Pick & Partial>, + articleNumber?: bigint | number +): IngestSegment | null { + const base = common(header, groupName, articleNumber) + const totalParts = inferTotalParts(meta) + const number = meta.begin.part ?? 1 + if (!base.messageId || number < 1 || number > totalParts) return null + const subject = `"${meta.begin.name}" yEnc` + return { + ...base, + fileKey: weakFileKey(meta.begin.name, meta.begin.size, totalParts), + subject, + name: meta.begin.name, + totalParts, + number, + weakIdentity: true, + release: releaseIdentity({ poster: base.poster, subject, name: meta.begin.name, weak: true }), + size: meta.begin.size, + // Single-part posts only carry crc32, which is then also the part's CRC. + pcrc32: meta.end?.pcrc32 ?? (meta.part ? undefined : meta.end?.crc32), + partSize: meta.part ? meta.part.end - meta.part.begin + 1 : meta.begin.size, + fileCrc32: meta.end?.crc32, + } +} diff --git a/lib/seed/runner.ts b/lib/seed/runner.ts new file mode 100644 index 0000000..522dac8 --- /dev/null +++ b/lib/seed/runner.ts @@ -0,0 +1,78 @@ +import type { XoverClient, Overview } from '../nntp/xover_client.js' +import { segmentFromHeader, type IngestSegment, type OverviewHeader } from '../ingest/segment.js' +import { FileAggregator } from '../aggregate/index.js' +import { toFragment, toSeedFile, type SeedBatch } from './wire.js' + +export interface RunRangeOptions { + client: XoverClient + group: string + /** Inclusive article numbers. */ + start: number + end: number + /** Articles per XOVER. */ + chunk?: number + /** Batches without a new part before an open file leaves as fragments. */ + idle?: number + /** XOVER batches per SeedBatch handed to `sink`. */ + submitEvery?: number + /** Headers left to the deferred-body path (counted, not indexed). */ + isDeferred?: (header: OverviewHeader) => boolean + /** Receives batches in order; the next XOVER is already in flight meanwhile. */ + sink: (batch: SeedBatch) => Promise +} + +/** + * Indexes one article range on a seeding node (CONCEPT.md D37): XOVER in + * chunks over one connection (the next chunk in flight while the previous + * one is handled), headers parsed with the live rules, grouped into files + * in memory; complete files and given-up fragments go to `sink` in order. + * Shared by seed:range (stores directly) and seed:node (submits over HTTPS). + */ +export async function runRange(options: RunRangeOptions): Promise { + const chunk = options.chunk ?? 20_000 + const idle = options.idle ?? 5 + const submitEvery = options.submitEvery ?? 5 + const aggregator = new FileAggregator() + await options.client.group(options.group) + + const fetch = (from: number) => { + const promise = options.client.xover(from, Math.min(from + chunk - 1, options.end)) + // Awaited after the current batch is handled; until then a failure must not count as unhandled. + promise.catch(() => {}) + return promise + } + + let batch = emptyBatch(options.start) + let pending = 0 + let from = options.start + let next: Promise | null = fetch(from) + while (next) { + const overviews = await next + from += chunk + next = from <= options.end ? fetch(from) : null + + const segments: IngestSegment[] = [] + for (const [articleNumber, header] of overviews) { + if (!Number.isFinite(articleNumber) || !header['message-id']) continue + const segment = segmentFromHeader(header, options.group, articleNumber) + if (segment) segments.push(segment) + else if (options.isDeferred?.(header)) batch.deferred++ + else batch.obfuscated++ + } + batch.headers += overviews.length + batch.files.push(...aggregator.add(segments).map(toSeedFile)) + batch.fragments.push(...aggregator.evict(idle).map(toFragment)) + batch.next = Math.min(from, options.end + 1) + + if (++pending >= submitEvery || !next) { + if (!next) batch.fragments.push(...aggregator.evict(0).map(toFragment)) + await options.sink(batch) + batch = emptyBatch(batch.next) + pending = 0 + } + } +} + +function emptyBatch(next: number): SeedBatch { + return { next, headers: 0, obfuscated: 0, deferred: 0, files: [], fragments: [] } +} diff --git a/lib/seed/wire.ts b/lib/seed/wire.ts new file mode 100644 index 0000000..b6e2b70 --- /dev/null +++ b/lib/seed/wire.ts @@ -0,0 +1,73 @@ +import type { IngestSegment } from '../ingest/segment.js' +import type { ReleaseIdentity } from '../release/index.js' +import type { AggregatedFile } from '../aggregate/index.js' + +/** + * A complete file as a seeding node reports it (CONCEPT.md D37): what the + * file, its release and its binary need, once per file, and the segments + * as [part number, message-id, bytes]. + */ +export interface SeedFile { + fileKey: string + name: string + poster: string + subject: string + totalParts: number + groups: string[] + /** Earliest post date of its articles, ISO 8601. */ + postedAt: string | null + release: ReleaseIdentity | null + segments: [number, string, number][] +} + +/** A segment of a file that didn't complete on the node, for the regular ingest path. */ +export type SeedFragment = Omit & { postedAt?: string; articleNumber?: number } + +/** What a node sends per batch: the article it has reached, and what it found up to there. */ +export interface SeedBatch { + /** First article number not covered yet (everything below it, from the range start, is in this or earlier batches). */ + next: number + headers: number + obfuscated: number + deferred: number + files: SeedFile[] + fragments: SeedFragment[] +} + +export function toSeedFile(file: AggregatedFile): SeedFile { + const first = file.segments[0] + const times = file.segments.map((s) => s.postedAt?.getTime()).filter((t): t is number => Number.isFinite(t)) + return { + fileKey: file.fileKey, + name: first.name, + poster: first.poster, + subject: first.subject, + totalParts: first.totalParts, + groups: first.groups, + postedAt: times.length > 0 ? new Date(Math.min(...times)).toISOString() : null, + release: first.release ?? null, + segments: file.segments.map((s) => [s.number, s.messageId, s.bytes]), + } +} + +export function toFragment(segment: IngestSegment): SeedFragment { + const { postedAt, articleNumber, ...rest } = segment + const date = postedAt && !Number.isNaN(postedAt.getTime()) ? postedAt.toISOString() : undefined + return { ...rest, postedAt: date, articleNumber: articleNumber === undefined ? undefined : Number(articleNumber) } +} + +export function fromFragment(fragment: SeedFragment): IngestSegment { + // The JSON body parser turns empty strings into null; the columns want ''. + return { + ...fragment, + poster: fragment.poster ?? '', + subject: fragment.subject ?? '', + name: fragment.name ?? '', + postedAt: fragment.postedAt ? new Date(fragment.postedAt) : undefined, + } +} + +/** A file from a node's JSON, with empty strings back where the body parser made them null. */ +export function normalizeSeedFile(file: SeedFile): SeedFile { + return { ...file, name: file.name ?? '', poster: file.poster ?? '', subject: file.subject ?? '' } +} diff --git a/seed-node/README.md b/seed-node/README.md new file mode 100644 index 0000000..83b47ea --- /dev/null +++ b/seed-node/README.md @@ -0,0 +1,26 @@ +# Seeding node + +A seeding node indexes old article ranges for the indexer (CONCEPT.md D37, +D38). It asks the seed API for a range over HTTPS, fetches that range's +headers over its own NNTP connections, groups them into files, and posts the +result back. It only makes outgoing connections: no port forwarding, database +or Redis. + +On the indexer (homelab): + + docker compose exec web node ace seed:token my-server # prints the node's token once + docker compose exec web node ace seed:plan alt.binaries.ath --days 365 + docker compose exec web node ace seed:status + +On the node (any machine with Docker): + + cp node.env.example node.env # SEED_TOKEN, NNTP account, SEED_CONNECTIONS + docker compose up -d --build + docker compose logs -f + +`docker compose stop` gives the node's ranges back; a node that disappears +loses its ranges to other nodes after 15 minutes without a batch. Updating: +`docker compose up -d --build` again. + +Mind the NNTP account's connection limit: every node's SEED_CONNECTIONS +count against it, next to the indexer's own connections. diff --git a/seed-node/compose.yaml b/seed-node/compose.yaml new file mode 100644 index 0000000..ca35ca5 --- /dev/null +++ b/seed-node/compose.yaml @@ -0,0 +1,18 @@ +# A seeding node (CONCEPT.md D38): leases article ranges from the seed API over +# HTTPS, indexes their headers over its own NNTP connections, and posts the +# results back. Outgoing connections only. Builds the indexer image straight +# from the git repository. +# +# cp node.env.example node.env # fill in SEED_TOKEN and the NNTP account +# docker compose up -d --build +name: usenet-seed-node + +services: + seed-node: + build: https://gitea.exp.sforum.nl/daan/usenet-indexer.git#master + image: usenet-seed-node:latest + command: ["node", "ace", "seed:node"] + env_file: node.env + restart: unless-stopped + # On SIGTERM the node gives its ranges back (resumed from the last batch elsewhere). + stop_grace_period: 60s diff --git a/seed-node/node.env.example b/seed-node/node.env.example new file mode 100644 index 0000000..fbe103c --- /dev/null +++ b/seed-node/node.env.example @@ -0,0 +1,29 @@ +# --- Seed API: the orchestrator, and this node's token (`node ace seed:token ` on the indexer). +SEED_URL=https://seed.exp.sforum.nl +SEED_TOKEN=sn_replace_me +# Ranges worked on at once; each uses one NNTP connection of the account below. +SEED_CONNECTIONS=2 + +# --- NNTP account this node fetches headers with (mind its connection limit across all nodes). +NNTP_HOST=news.example.com +NNTP_PORT=563 +NNTP_SECURE=true +NNTP_USER=replace_me +NNTP_PASSWORD=replace_me + +# --- Required by the app's configuration, but unused on a seeding node (no database, Redis or web server). +NODE_ENV=production +TZ=UTC +PORT=3333 +HOST=0.0.0.0 +LOG_LEVEL=info +APP_KEY=unused-on-seed-nodes-unused-on-seed-nodes +APP_URL=http://localhost:3333 +SESSION_DRIVER=memory +DB_CONNECTION=pg +DB_HOST=localhost +DB_PORT=5432 +DB_USER=unused +DB_DATABASE=unused +REDIS_HOST=localhost +REDIS_PORT=6379 diff --git a/start/kernel.ts b/start/kernel.ts index 5514e48..599fda8 100644 --- a/start/kernel.ts +++ b/start/kernel.ts @@ -47,4 +47,5 @@ router.use([ */ export const middleware = router.named({ auth: () => import('#middleware/auth_middleware'), + seedNode: () => import('#middleware/seed_node_middleware'), }) diff --git a/start/routes.ts b/start/routes.ts index bf37e1d..77c3502 100644 --- a/start/routes.ts +++ b/start/routes.ts @@ -13,6 +13,7 @@ import { controllers } from '#generated/controllers' const ReleasesController = () => import('#controllers/releases_controller') const NewznabController = () => import('#controllers/newznab_controller') +const SeedController = () => import('#controllers/seed_controller') // Newznab API for Prowlarr / Sonarr / Radarr ("Generic Newznab", API path /api). router.get('/api', [NewznabController, 'handle']) @@ -48,3 +49,15 @@ router .use(middleware.auth()) }) .prefix('/api/v1') + +// Seed API for remote seeding nodes (CONCEPT.md D38); the only routes traefik exposes publicly. +router + .group(() => { + router.post('lease', [SeedController, 'lease']) + router.post('ranges/:id/batches', [SeedController, 'batch']).where('id', router.matchers.number()) + router.post('ranges/:id/done', [SeedController, 'done']).where('id', router.matchers.number()) + router.post('ranges/:id/release', [SeedController, 'release']).where('id', router.matchers.number()) + router.get('status', [SeedController, 'status']) + }) + .prefix('/api/v1/seed') + .use(middleware.seedNode()) diff --git a/tests/functional/seed.spec.ts b/tests/functional/seed.spec.ts index 241e47a..2661938 100644 --- a/tests/functional/seed.spec.ts +++ b/tests/functional/seed.spec.ts @@ -5,6 +5,7 @@ import { segmentFromHeader, type IngestSegment } from '#services/IngestService' import SeedService from '#services/SeedService' import QueueService from '#services/QueueService' import { FileAggregator } from '#lib/aggregate/index' +import { toSeedFile } from '#lib/seed/wire' import { segmentsOf } from '#services/FileStore' function header(subject: string, n: number) { @@ -25,7 +26,7 @@ test.group('SeedService', (group) => { const aggregator = new FileAggregator() const subjects = [1, 2].map((i) => `Seed.Test.S01E01.1080p - [1/2] - "Seed.Test.S01E01.1080p.part${i}.mkv" yEnc (%d/3)`) const segments = subjects.flatMap((s) => [1, 2, 3].map((n) => segmentFromHeader(header(s.replace('%d', String(n)), n), 'alt.binaries.test', n)!)) - const files = aggregator.add(segments) + const files = aggregator.add(segments).map(toSeedFile) assert.lengthOf(files, 2) const result = await SeedService.store(files) diff --git a/tests/functional/seed_api.spec.ts b/tests/functional/seed_api.spec.ts new file mode 100644 index 0000000..8c585be --- /dev/null +++ b/tests/functional/seed_api.spec.ts @@ -0,0 +1,77 @@ +import { test } from '@japa/runner' +import { gzipSync } from 'node:zlib' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import SeedOrchestrator from '#services/SeedOrchestrator' +import QueueService from '#services/QueueService' +import { segmentFromHeader } from '#lib/ingest/segment' +import { FileAggregator } from '#lib/aggregate/index' +import { toFragment, toSeedFile, type SeedBatch } from '#lib/seed/wire' +import type { IngestSegment } from '#lib/ingest/segment' + +function header(subject: string, n: number) { + return { 'subject': subject, 'from': 'api seeder ', 'date': 'Mon, 05 Oct 2026 10:00:00 +0000', 'message-id': ``, ':bytes': '500000' } +} + +function batch(next: number): SeedBatch { + const segments = [1, 2].map((n) => segmentFromHeader(header(`Api.Test.2026 - "Api.Test.2026.mkv" yEnc (${n}/2)`, n), 'alt.binaries.test', n)!) + const lone = segmentFromHeader(header('Api.Test.2026 - "Api.Test.2026.nfo" yEnc (1/3)', 9), 'alt.binaries.test', 9)! + const aggregator = new FileAggregator() + return { next, headers: 3, obfuscated: 0, deferred: 0, files: aggregator.add(segments).map(toSeedFile), fragments: [toFragment(lone)] } +} + +test.group('Seed API', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + group.teardown(() => QueueService.closeAll()) + + test('a node leases a range, posts a gzip batch, and finishes it', async ({ client, assert }) => { + const token = await SeedOrchestrator.issueToken('test-node') + await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 }) + const auth = { authorization: `Bearer ${token}` } + + ;(await client.post('/api/v1/seed/lease')).assertStatus(401) + const leased = await client.post('/api/v1/seed/lease').headers(auth) + leased.assertStatus(200) + const range = leased.body().range + assert.include(range, { group: 'alt.binaries.test', start: 1, end: 100 }) + + // As seed:node sends it: a gzip-compressed JSON body. + const posted = await fetch(`http://${process.env.HOST}:${process.env.PORT}/api/v1/seed/ranges/${range.id}/batches`, { + method: 'POST', + headers: { ...auth, 'content-type': 'application/json', 'content-encoding': 'gzip' }, + body: gzipSync(JSON.stringify(batch(51))), + }) + assert.equal(posted.status, 200) + assert.include(((await posted.json()) as any).stored, { files: 1 }) + const file = await db.from('files').where('filename', 'Api.Test.2026.mkv').first() + assert.equal(file.parts, 2) + // The lone .nfo segment took the regular path (staged or stored as a binary). + const fragment = (await db.from('staged_segments').count('* as n').first()).n + (await db.from('binaries').where('name', 'Api.Test.2026.nfo').count('* as n').first()).n + assert.isAbove(Number(fragment), 0) + + const row = await db.from('seed_ranges').where('id', range.id).first() + assert.equal(Number(row.next_id), 51) + assert.equal(Number(row.headers), 3) + + ;(await client.post(`/api/v1/seed/ranges/${range.id}/done`).headers(auth)).assertStatus(200) + assert.equal((await db.from('seed_ranges').where('id', range.id).first()).status, 'done') + ;(await client.post('/api/v1/seed/lease').headers(auth)).assertStatus(204) + }) + + test('a quiet lease passes to another node, and the first one gets 409', async ({ client, assert }) => { + const first = await SeedOrchestrator.issueToken('node-a') + const second = await SeedOrchestrator.issueToken('node-b') + await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 1 }) + const range = (await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${first}` })).body().range + await db.from('seed_ranges').where('id', range.id).update({ next_id: 40, heartbeat_at: new Date(Date.now() - 60 * 60 * 1000) }) + + const taken = (await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${second}` })).body().range + assert.equal(taken.id, range.id) + assert.equal(taken.start, 40) + const late = await client.post(`/api/v1/seed/ranges/${range.id}/batches`).headers({ authorization: `Bearer ${first}` }).json(batch(60)) + late.assertStatus(409) + + ;(await client.post(`/api/v1/seed/ranges/${range.id}/release`).headers({ authorization: `Bearer ${second}` }).json({ error: 'test' })).assertStatus(200) + assert.equal((await db.from('seed_ranges').where('id', range.id).first()).status, 'queued') + }) +})