From d5f168d8d25e779b8a5a036a833d0e2aff0775b4 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Fri, 9 Oct 2026 00:31:06 +0200 Subject: [PATCH] Rebuild: coordinator (planner, NNTP pool, spool, prefetcher, leases, rule sets), node API and the edge:node worker - groups:add / ranges:plan / nodes:create / coordinator:run commands - the node API: lease, the range's chunks as one framed stream from the spool, batches, heartbeat, finish, release (node guard, token per node) - edge:node: needs COORDINATOR_URL and NODE_TOKEN only (placeholders for the rest), posts a batch every 10 chunks, gives its range back on SIGTERM - releases are gathered per batch: 1.04 chunk rows per release at 10 chunks per batch (1.35 when built per XOVER chunk) - partitions are checked per batch instead of remembered Co-Authored-By: Claude Opus 5.5 --- .adonisjs/client/manifest.d.ts | 1 + .adonisjs/client/registry/index.ts | 36 +++++ .adonisjs/client/registry/schema.d.ts | 72 +++++++++ .adonisjs/client/registry/tree.d.ts | 10 ++ .adonisjs/server/controllers.ts | 1 + .adonisjs/server/routes.d.ts | 13 ++ app/controllers/access_tokens_controller.ts | 2 +- app/controllers/node_controller.ts | 94 ++++++++++++ app/controllers/profile_controller.ts | 2 +- app/models/node.ts | 8 + app/services/chunk_partitions.ts | 12 +- app/services/leases.ts | 61 ++++++++ app/services/nntp_pool.ts | 80 ++++++++++ app/services/planner.ts | 64 ++++++++ app/services/prefetcher.ts | 90 +++++++++++ app/services/rule_sets.ts | 36 +++++ app/services/spool.ts | 78 ++++++++++ bin/console.ts | 13 ++ commands/coordinator_run.ts | 51 +++++++ commands/edge_bench.ts | 14 +- commands/edge_node.ts | 157 ++++++++++++++++++++ commands/groups_add.ts | 30 ++++ commands/nodes_create.ts | 19 +++ commands/ranges_plan.ts | 20 +++ config/auth.ts | 10 ++ config/bodyparser.ts | 3 +- config/coordinator.ts | 24 +++ lib/edge/index.ts | 3 + lib/edge/lease.ts | 7 + lib/edge/range_indexer.ts | 15 +- lib/edge/stream.ts | 35 +++++ lib/edge/wire.ts | 23 +++ lib/nntp/xover_client.ts | 15 ++ start/env.ts | 11 ++ start/routes.ts | 14 ++ tests/functional/node_api.spec.ts | 83 +++++++++++ 36 files changed, 1193 insertions(+), 14 deletions(-) create mode 100644 app/controllers/node_controller.ts create mode 100644 app/models/node.ts create mode 100644 app/services/leases.ts create mode 100644 app/services/nntp_pool.ts create mode 100644 app/services/planner.ts create mode 100644 app/services/prefetcher.ts create mode 100644 app/services/rule_sets.ts create mode 100644 app/services/spool.ts create mode 100644 commands/coordinator_run.ts create mode 100644 commands/edge_node.ts create mode 100644 commands/groups_add.ts create mode 100644 commands/nodes_create.ts create mode 100644 commands/ranges_plan.ts create mode 100644 config/coordinator.ts create mode 100644 lib/edge/lease.ts create mode 100644 lib/edge/stream.ts create mode 100644 lib/edge/wire.ts create mode 100644 tests/functional/node_api.spec.ts diff --git a/.adonisjs/client/manifest.d.ts b/.adonisjs/client/manifest.d.ts index 9fc3356..40e0c09 100644 --- a/.adonisjs/client/manifest.d.ts +++ b/.adonisjs/client/manifest.d.ts @@ -5,5 +5,6 @@ /// /// +/// /// /// diff --git a/.adonisjs/client/registry/index.ts b/.adonisjs/client/registry/index.ts index 298b217..53494a3 100644 --- a/.adonisjs/client/registry/index.ts +++ b/.adonisjs/client/registry/index.ts @@ -30,6 +30,42 @@ const routes = { tokens: [{"old":"/api/v1/account/logout","type":0,"val":"api","end":""},{"old":"/api/v1/account/logout","type":0,"val":"v1","end":""},{"old":"/api/v1/account/logout","type":0,"val":"account","end":""},{"old":"/api/v1/account/logout","type":0,"val":"logout","end":""}], types: placeholder as Registry['profile.access_tokens.destroy']['types'], }, + 'node.node.lease': { + methods: ["POST"], + pattern: '/api/v1/node/lease', + tokens: [{"old":"/api/v1/node/lease","type":0,"val":"api","end":""},{"old":"/api/v1/node/lease","type":0,"val":"v1","end":""},{"old":"/api/v1/node/lease","type":0,"val":"node","end":""},{"old":"/api/v1/node/lease","type":0,"val":"lease","end":""}], + types: placeholder as Registry['node.node.lease']['types'], + }, + 'node.node.stream': { + methods: ["GET","HEAD"], + pattern: '/api/v1/node/ranges/:id/stream', + tokens: [{"old":"/api/v1/node/ranges/:id/stream","type":0,"val":"api","end":""},{"old":"/api/v1/node/ranges/:id/stream","type":0,"val":"v1","end":""},{"old":"/api/v1/node/ranges/:id/stream","type":0,"val":"node","end":""},{"old":"/api/v1/node/ranges/:id/stream","type":0,"val":"ranges","end":""},{"old":"/api/v1/node/ranges/:id/stream","type":1,"val":"id","end":""},{"old":"/api/v1/node/ranges/:id/stream","type":0,"val":"stream","end":""}], + types: placeholder as Registry['node.node.stream']['types'], + }, + 'node.node.batch': { + methods: ["POST"], + pattern: '/api/v1/node/ranges/:id/batches', + tokens: [{"old":"/api/v1/node/ranges/:id/batches","type":0,"val":"api","end":""},{"old":"/api/v1/node/ranges/:id/batches","type":0,"val":"v1","end":""},{"old":"/api/v1/node/ranges/:id/batches","type":0,"val":"node","end":""},{"old":"/api/v1/node/ranges/:id/batches","type":0,"val":"ranges","end":""},{"old":"/api/v1/node/ranges/:id/batches","type":1,"val":"id","end":""},{"old":"/api/v1/node/ranges/:id/batches","type":0,"val":"batches","end":""}], + types: placeholder as Registry['node.node.batch']['types'], + }, + 'node.node.heartbeat': { + methods: ["POST"], + pattern: '/api/v1/node/ranges/:id/heartbeat', + tokens: [{"old":"/api/v1/node/ranges/:id/heartbeat","type":0,"val":"api","end":""},{"old":"/api/v1/node/ranges/:id/heartbeat","type":0,"val":"v1","end":""},{"old":"/api/v1/node/ranges/:id/heartbeat","type":0,"val":"node","end":""},{"old":"/api/v1/node/ranges/:id/heartbeat","type":0,"val":"ranges","end":""},{"old":"/api/v1/node/ranges/:id/heartbeat","type":1,"val":"id","end":""},{"old":"/api/v1/node/ranges/:id/heartbeat","type":0,"val":"heartbeat","end":""}], + types: placeholder as Registry['node.node.heartbeat']['types'], + }, + 'node.node.finish': { + methods: ["POST"], + pattern: '/api/v1/node/ranges/:id/finish', + tokens: [{"old":"/api/v1/node/ranges/:id/finish","type":0,"val":"api","end":""},{"old":"/api/v1/node/ranges/:id/finish","type":0,"val":"v1","end":""},{"old":"/api/v1/node/ranges/:id/finish","type":0,"val":"node","end":""},{"old":"/api/v1/node/ranges/:id/finish","type":0,"val":"ranges","end":""},{"old":"/api/v1/node/ranges/:id/finish","type":1,"val":"id","end":""},{"old":"/api/v1/node/ranges/:id/finish","type":0,"val":"finish","end":""}], + types: placeholder as Registry['node.node.finish']['types'], + }, + 'node.node.release': { + methods: ["POST"], + pattern: '/api/v1/node/ranges/:id/release', + tokens: [{"old":"/api/v1/node/ranges/:id/release","type":0,"val":"api","end":""},{"old":"/api/v1/node/ranges/:id/release","type":0,"val":"v1","end":""},{"old":"/api/v1/node/ranges/:id/release","type":0,"val":"node","end":""},{"old":"/api/v1/node/ranges/:id/release","type":0,"val":"ranges","end":""},{"old":"/api/v1/node/ranges/:id/release","type":1,"val":"id","end":""},{"old":"/api/v1/node/ranges/:id/release","type":0,"val":"release","end":""}], + types: placeholder as Registry['node.node.release']['types'], + }, } as const satisfies Record export { routes } diff --git a/.adonisjs/client/registry/schema.d.ts b/.adonisjs/client/registry/schema.d.ts index 82c02fd..b4e89e8 100644 --- a/.adonisjs/client/registry/schema.d.ts +++ b/.adonisjs/client/registry/schema.d.ts @@ -55,4 +55,76 @@ export interface Registry { errorResponse: ExtractErrorResponse>> } } + 'node.node.lease': { + methods: ["POST"] + pattern: '/api/v1/node/lease' + types: { + body: {} + paramsTuple: [] + params: {} + query: {} + response: ExtractResponse>> + errorResponse: ExtractErrorResponse>> + } + } + 'node.node.stream': { + methods: ["GET","HEAD"] + pattern: '/api/v1/node/ranges/:id/stream' + types: { + body: {} + paramsTuple: [ParamValue] + params: { id: ParamValue } + query: {} + response: ExtractResponse>> + errorResponse: ExtractErrorResponse>> + } + } + 'node.node.batch': { + methods: ["POST"] + pattern: '/api/v1/node/ranges/:id/batches' + types: { + body: {} + paramsTuple: [ParamValue] + params: { id: ParamValue } + query: {} + response: ExtractResponse>> + errorResponse: ExtractErrorResponse>> + } + } + 'node.node.heartbeat': { + methods: ["POST"] + pattern: '/api/v1/node/ranges/:id/heartbeat' + types: { + body: {} + paramsTuple: [ParamValue] + params: { id: ParamValue } + query: {} + response: ExtractResponse>> + errorResponse: ExtractErrorResponse>> + } + } + 'node.node.finish': { + methods: ["POST"] + pattern: '/api/v1/node/ranges/:id/finish' + types: { + body: {} + paramsTuple: [ParamValue] + params: { id: ParamValue } + query: {} + response: ExtractResponse>> + errorResponse: ExtractErrorResponse>> + } + } + 'node.node.release': { + methods: ["POST"] + pattern: '/api/v1/node/ranges/:id/release' + types: { + body: {} + paramsTuple: [ParamValue] + params: { id: ParamValue } + query: {} + response: ExtractResponse>> + errorResponse: ExtractErrorResponse>> + } + } } diff --git a/.adonisjs/client/registry/tree.d.ts b/.adonisjs/client/registry/tree.d.ts index 3964805..c5ff012 100644 --- a/.adonisjs/client/registry/tree.d.ts +++ b/.adonisjs/client/registry/tree.d.ts @@ -18,4 +18,14 @@ export interface ApiDefinition { destroy: typeof routes['profile.access_tokens.destroy'] } } + node: { + node: { + lease: typeof routes['node.node.lease'] + stream: typeof routes['node.node.stream'] + batch: typeof routes['node.node.batch'] + heartbeat: typeof routes['node.node.heartbeat'] + finish: typeof routes['node.node.finish'] + release: typeof routes['node.node.release'] + } + } } diff --git a/.adonisjs/server/controllers.ts b/.adonisjs/server/controllers.ts index 202eb41..0ad50db 100644 --- a/.adonisjs/server/controllers.ts +++ b/.adonisjs/server/controllers.ts @@ -6,5 +6,6 @@ export const controllers = { AccessTokens: () => import('#controllers/access_tokens_controller'), NewAccount: () => import('#controllers/new_account_controller'), + Node: () => import('#controllers/node_controller'), Profile: () => import('#controllers/profile_controller'), } diff --git a/.adonisjs/server/routes.d.ts b/.adonisjs/server/routes.d.ts index 9aa155f..da4dce0 100644 --- a/.adonisjs/server/routes.d.ts +++ b/.adonisjs/server/routes.d.ts @@ -8,17 +8,30 @@ export type ScannedRoutes = { 'auth.access_tokens.store': { paramsTuple?: []; params?: {} } 'profile.profile.show': { paramsTuple?: []; params?: {} } 'profile.access_tokens.destroy': { paramsTuple?: []; params?: {} } + 'node.node.lease': { paramsTuple?: []; params?: {} } + 'node.node.stream': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } + 'node.node.batch': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } + 'node.node.heartbeat': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } + 'node.node.finish': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } + 'node.node.release': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } } GET: { 'profile.profile.show': { paramsTuple?: []; params?: {} } + 'node.node.stream': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } } HEAD: { 'profile.profile.show': { paramsTuple?: []; params?: {} } + 'node.node.stream': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } } POST: { 'auth.new_account.store': { paramsTuple?: []; params?: {} } 'auth.access_tokens.store': { paramsTuple?: []; params?: {} } 'profile.access_tokens.destroy': { paramsTuple?: []; params?: {} } + 'node.node.lease': { paramsTuple?: []; params?: {} } + 'node.node.batch': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } + 'node.node.heartbeat': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } + 'node.node.finish': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } + 'node.node.release': { paramsTuple: [ParamValue]; params: {'id': ParamValue} } } } declare module '@adonisjs/core/types/http' { diff --git a/app/controllers/access_tokens_controller.ts b/app/controllers/access_tokens_controller.ts index 153267c..60eb424 100644 --- a/app/controllers/access_tokens_controller.ts +++ b/app/controllers/access_tokens_controller.ts @@ -17,7 +17,7 @@ export default class AccessTokensController { } async destroy({ auth }: HttpContext) { - const user = auth.getUserOrFail() + const user = auth.use('api').getUserOrFail() if (user.currentAccessToken) { await User.accessTokens.delete(user, user.currentAccessToken.identifier) } diff --git a/app/controllers/node_controller.ts b/app/controllers/node_controller.ts new file mode 100644 index 0000000..dd3c458 --- /dev/null +++ b/app/controllers/node_controller.ts @@ -0,0 +1,94 @@ +import type { HttpContext } from '@adonisjs/core/http' +import { Readable } from 'node:stream' +import { setTimeout as sleep } from 'node:timers/promises' +import Leases, { LeaseLost } from '#services/leases' +import BatchStore, { StaleBatch } from '#services/batch_store' +import Spool from '#services/spool' +import { encodeFrame, fromWire, type WireBatch } from '#lib/edge/index' + +/** How long the stream waits for a chunk the prefetcher hasn't spooled yet. */ +const CHUNK_WAIT_MS = 15 * 60 * 1000 + +/** + * The edge workers' API (REBUILD.md, parts 1–2): a lease, the range's + * chunks as one stream, results per batch, and the end of the range. + */ +export default class NodeController { + private node(ctx: HttpContext) { + return Number(ctx.auth.use('node').getUserOrFail().id) + } + + async lease(ctx: HttpContext) { + const lease = await Leases.lease(this.node(ctx)) + if (!lease) return ctx.response.noContent() + return lease + } + + /** The range's chunks from its resume point, waiting for those still being fetched. */ + async stream(ctx: HttpContext) { + const nodeId = this.node(ctx) + const rangeId = Number(ctx.params.id) + const range = await this.leased(ctx, rangeId, nodeId) + if (!range) return + const spans = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }, Number(range.next_id)) + async function* frames() { + for (const span of spans) { + const waitUntil = Date.now() + CHUNK_WAIT_MS + while (!(await Spool.has(rangeId, span))) { + if (Date.now() > waitUntil) throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} wasn't spooled in time`) + await sleep(500) + } + yield encodeFrame({ ...span, data: await Spool.read(rangeId, span) }) + // Streaming keeps the lease alive. + await Leases.heartbeat(rangeId, nodeId) + } + } + ctx.response.header('Content-Type', 'application/octet-stream') + ctx.response.stream(Readable.from(frames())) + } + + async batch(ctx: HttpContext) { + const nodeId = this.node(ctx) + const rangeId = Number(ctx.params.id) + try { + return await BatchStore.store(rangeId, nodeId, fromWire(ctx.request.body() as WireBatch)) + } catch (error) { + if (error instanceof StaleBatch) return ctx.response.conflict({ error: error.message }) + throw error + } + } + + async heartbeat(ctx: HttpContext) { + return this.lifecycle(ctx, (rangeId, nodeId) => Leases.heartbeat(rangeId, nodeId)) + } + + async finish(ctx: HttpContext) { + return this.lifecycle(ctx, (rangeId, nodeId) => Leases.finish(rangeId, nodeId)) + } + + async release(ctx: HttpContext) { + return this.lifecycle(ctx, (rangeId, nodeId) => Leases.release(rangeId, nodeId)) + } + + private async lifecycle(ctx: HttpContext, action: (rangeId: number, nodeId: number) => Promise) { + try { + await action(Number(ctx.params.id), this.node(ctx)) + return { ok: true } + } catch (error) { + if (error instanceof LeaseLost) return ctx.response.conflict({ error: error.message }) + throw error + } + } + + private async leased(ctx: HttpContext, rangeId: number, nodeId: number) { + try { + return await Leases.leased(rangeId, nodeId) + } catch (error) { + if (error instanceof LeaseLost) { + ctx.response.conflict({ error: error.message }) + return null + } + throw error + } + } +} diff --git a/app/controllers/profile_controller.ts b/app/controllers/profile_controller.ts index 1fe82f2..d0c8208 100644 --- a/app/controllers/profile_controller.ts +++ b/app/controllers/profile_controller.ts @@ -3,6 +3,6 @@ import type { HttpContext } from '@adonisjs/core/http' export default class ProfileController { async show({ auth, serialize }: HttpContext) { - return serialize(UserTransformer.transform(auth.getUserOrFail())) + return serialize(UserTransformer.transform(auth.use('api').getUserOrFail())) } } diff --git a/app/models/node.ts b/app/models/node.ts new file mode 100644 index 0000000..78dc47e --- /dev/null +++ b/app/models/node.ts @@ -0,0 +1,8 @@ +import { NodeSchema } from '#database/schema' +import { type AccessToken, DbAccessTokensProvider } from '@adonisjs/auth/access_tokens' + +/** An edge worker (REBUILD.md, part 2); it authenticates with an access token (the `node` guard). */ +export default class Node extends NodeSchema { + static accessTokens = DbAccessTokensProvider.forModel(Node, { table: 'node_access_tokens' }) + declare currentAccessToken?: AccessToken +} diff --git a/app/services/chunk_partitions.ts b/app/services/chunk_partitions.ts index db0dd02..addb2fb 100644 --- a/app/services/chunk_partitions.ts +++ b/app/services/chunk_partitions.ts @@ -7,15 +7,14 @@ const AHEAD = 2 /** * release_chunks is partitioned by month of writing (REBUILD.md, decision a). * Inserts fail without a partition for now(), so the current and the next - * months' partitions are created before they are needed; checked once per - * process and month. + * months' partitions are created before they are needed. */ class ChunkPartitions { - private ensuredFor: string | null = null - async ensure(now = new Date()) { - const month = now.toISOString().slice(0, 7) - if (this.ensuredFor === month) return + // Checked each time, not remembered: a rolled-back transaction can take a created partition along. + const current = `release_chunks_${now.toISOString().slice(0, 7).replace('-', '')}` + const { rows } = await db.rawQuery(`SELECT to_regclass(?) IS NOT NULL AS present`, [current]) + if (rows[0].present) return const tablespace = env.get('CHUNK_TABLESPACE') for (let i = 0; i <= AHEAD; i++) { const from = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + i, 1)) @@ -25,7 +24,6 @@ class ChunkPartitions { `CREATE TABLE IF NOT EXISTS ${name} PARTITION OF release_chunks FOR VALUES FROM ('${from.toISOString()}') TO ('${to.toISOString()}')${tablespace ? ` TABLESPACE ${tablespace}` : ''}` ) } - this.ensuredFor = month } } diff --git a/app/services/leases.ts b/app/services/leases.ts new file mode 100644 index 0000000..2bb8f40 --- /dev/null +++ b/app/services/leases.ts @@ -0,0 +1,61 @@ +import db from '@adonisjs/lucid/services/db' +import coordinatorConfig from '#config/coordinator' +import RuleSets from '#services/rule_sets' +import Spool from '#services/spool' +import type { Lease } from '#lib/edge/index' + +export class LeaseLost extends Error {} + +/** + * Ranges for workers (REBUILD.md, coordinator): the next range whose first + * chunk is spooled, by priority (oldest era first); a lease whose worker + * went quiet for LEASE_TTL_SECONDS is taken over and resumes at its resume + * point. + */ +class Leases { + async lease(nodeId: number): Promise { + const rules = await RuleSets.current() + const { rows } = await db.rawQuery( + `UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ? + WHERE r.id = ( + SELECT id FROM ranges + WHERE spooled_chunks > 0 AND (status = 'planned' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?))) + ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED) + RETURNING r.id, r.group_id, r.start_id, r.end_id, r.next_id`, + [nodeId, rules.version, coordinatorConfig.leaseTtlSeconds] + ) + const row = rows[0] + if (!row) return null + await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() }) + const group = await db.from('groups').where('id', row.group_id).select('name').firstOrFail() + return { range: { id: Number(row.id), group: group.name, start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id) }, rules } + } + + async leased(rangeId: number, nodeId: number) { + const range = await db.from('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 + } + + async heartbeat(rangeId: number, nodeId: number) { + await this.leased(rangeId, nodeId) + await db.from('ranges').where('id', rangeId).update({ heartbeat_at: new Date() }) + await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() }) + } + + /** The worker reached the range's end: done, and its spooled chunks go. */ + async finish(rangeId: number, nodeId: number) { + const range = await this.leased(rangeId, nodeId) + if (Number(range.next_id) <= Number(range.end_id)) throw new LeaseLost(`Range ${rangeId} is at ${range.next_id}, not past its end ${range.end_id}`) + await db.from('ranges').where('id', rangeId).update({ status: 'done', done_at: new Date(), heartbeat_at: new Date() }) + await Spool.remove(rangeId) + } + + /** A stopping worker gives its range back; the next one resumes it. */ + async release(rangeId: number, nodeId: number) { + await this.leased(rangeId, nodeId) + await db.from('ranges').where('id', rangeId).update({ status: 'planned', node_id: null }) + } +} + +export default new Leases() diff --git a/app/services/nntp_pool.ts b/app/services/nntp_pool.ts new file mode 100644 index 0000000..d21927f --- /dev/null +++ b/app/services/nntp_pool.ts @@ -0,0 +1,80 @@ +import env from '#start/env' +import logger from '@adonisjs/core/services/logger' +import coordinatorConfig from '#config/coordinator' +import { XoverClient } from '#lib/nntp/index' + +/** + * The coordinator's header connections (REBUILD.md, decision e: workers + * have no NNTP account). At most NNTP_CONNECTIONS, reused; a connection + * that failed is closed and replaced on the next use. + */ +class NntpPool { + private idle: XoverClient[] = [] + private open = 0 + private waiting: ((client: XoverClient) => void)[] = [] + + get size() { + return coordinatorConfig.nntpConnections + } + + async use(work: (client: XoverClient) => Promise): Promise { + const client = await this.acquire() + let broken = false + try { + return await work(client) + } catch (error) { + broken = true + throw error + } finally { + this.release(client, broken) + } + } + + private async acquire(): Promise { + const ready = this.idle.pop() + if (ready) return ready + if (this.open < this.size) { + this.open++ + try { + return await XoverClient.connect({ + host: env.get('NNTP_HOST')!, + port: env.get('NNTP_PORT', 563), + user: env.get('NNTP_USER'), + password: env.get('NNTP_PASSWORD'), + secure: env.get('NNTP_SECURE', true), + compress: true, + timeoutMs: coordinatorConfig.nntpTimeoutMs, + }) + } catch (error) { + this.open-- + throw error + } + } + return new Promise((resolve) => this.waiting.push(resolve)) + } + + private release(client: XoverClient, broken: boolean) { + if (broken) { + this.open-- + try { + client.close() + } catch (error: any) { + logger.debug(`Closing a failed NNTP connection: ${error.message}`) + } + // Someone waiting gets a fresh connection. + const next = this.waiting.shift() + if (next) this.acquire().then(next, () => this.waiting.unshift(next)) + return + } + const next = this.waiting.shift() + if (next) next(client) + else this.idle.push(client) + } + + async close() { + for (const client of this.idle.splice(0)) client.close() + this.open = 0 + } +} + +export default new NntpPool() diff --git a/app/services/planner.ts b/app/services/planner.ts new file mode 100644 index 0000000..9ac9478 --- /dev/null +++ b/app/services/planner.ts @@ -0,0 +1,64 @@ +import db from '@adonisjs/lucid/services/db' +import coordinatorConfig from '#config/coordinator' +import NntpPool from '#services/nntp_pool' + +/** + * The indexed groups and their ranges (REBUILD.md, group list, coordinator). + * New groups get priorities after the existing ones (owner rule). A group's + * span from GROUP is cut into ranges of RANGE_ARTICLES; a range's priority is + * its position in its group's history (0 oldest … 1000 newest), so all groups + * start with their oldest era. Planning again only adds ranges for new articles. + */ +class Planner { + async addGroups(names: string[]): Promise { + let added = 0 + for (const name of names) { + const { rows } = await db.rawQuery( + `INSERT INTO groups (name, priority) SELECT ?, coalesce(max(priority), 0) + 1 FROM groups ON CONFLICT (name) DO NOTHING RETURNING id`, + [name] + ) + added += rows.length + } + return added + } + + /** The `top` groups matching `pattern` by article span, largest first (LIST ACTIVE). */ + async largestGroups(pattern: string, top: number): Promise { + const groups = await NntpPool.use((client) => client.listActive(pattern)) + return groups + .filter((g) => g.high > g.low) + .sort((a, b) => b.high - b.low - (a.high - a.low)) + .slice(0, top) + .map((g) => g.name) + } + + async planGroup(group: { id: number; name: string }): Promise { + const info = await NntpPool.use((client) => client.group(group.name)) + const size = coordinatorConfig.rangeArticles + const last = await db.from('ranges').where('group_id', group.id).max('end_id as end').first() + const from = Math.max(info.low, Number(last?.end ?? 0) + 1) + const span = Math.max(1, info.high - info.low) + const rows = [] + for (let start = from; start <= info.high; start += size) { + rows.push({ + group_id: group.id, + start_id: start, + end_id: Math.min(start + size - 1, info.high), + next_id: start, + priority: Math.floor((1000 * (start - info.low)) / span), + }) + } + for (let i = 0; i < rows.length; i += 2000) await db.table('ranges').multiInsert(rows.slice(i, i + 2000)) + await db.from('groups').where('id', group.id).update({ low: info.low, high: info.high, planned_at: new Date() }) + return rows.length + } + + async planAll(): Promise<{ groups: number; ranges: number }> { + const groups = await db.from('groups').where('active', true).orderBy('priority').select('id', 'name') + let ranges = 0 + for (const group of groups) ranges += await this.planGroup({ id: Number(group.id), name: group.name }) + return { groups: groups.length, ranges } + } +} + +export default new Planner() diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts new file mode 100644 index 0000000..ccf6c43 --- /dev/null +++ b/app/services/prefetcher.ts @@ -0,0 +1,90 @@ +import db from '@adonisjs/lucid/services/db' +import logger from '@adonisjs/core/services/logger' +import coordinatorConfig from '#config/coordinator' +import NntpPool from '#services/nntp_pool' +import Spool from '#services/spool' + +interface Target { + id: number + group: string + start: number + end: number + next: number + leased: boolean +} + +/** + * Keeps the news server busy ahead of the workers (REBUILD.md, prefetch): + * the missing chunks of leased ranges first, then the next planned ranges, + * about SPOOL_AHEAD_PER_LEASE per active lease (at least SPOOL_MIN_AHEAD), + * within SPOOL_MAX_GB. One range per connection at a time, in order. + */ +export class Prefetcher { + private busy = new Set() + private stopped = false + + /** One round: starts work for free connections; returns how many ranges it started. */ + async tick(): Promise { + const free = NntpPool.size - this.busy.size + if (free <= 0) return 0 + const targets = await this.targets(free) + for (const target of targets) { + this.busy.add(target.id) + this.spoolRange(target) + .catch((error) => logger.warn(`Spooling range ${target.id}: ${error.message}`)) + .finally(() => this.busy.delete(target.id)) + } + return targets.length + } + + stop() { + this.stopped = true + } + + get active() { + return this.busy.size + } + + private async targets(limit: number): Promise { + const busy = [...this.busy] + const select = (q: any) => + q + .from('ranges as r') + .join('groups as g', 'g.id', 'r.group_id') + .whereNull('r.spooled_at') + .whereNotIn('r.id', busy.length > 0 ? busy : [0]) + .select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id', 'r.status') + // Leased ranges whose chunks aren't all there: a worker is waiting for them. + const leased = await select(db.query()).where('r.status', 'leased').orderBy('r.leased_at').limit(limit) + const out: Target[] = leased.map(toTarget) + if (out.length >= limit) return out + + const [{ leases }] = await db.from('ranges').where('status', 'leased').count('* as leases') + const [{ ahead }] = await db.from('ranges').where('status', 'planned').where('spooled_chunks', '>', 0).count('* as ahead') + const wanted = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * Number(leases)) - Number(ahead) + if (wanted <= 0 || (await Spool.size()) >= coordinatorConfig.spool.maxBytes) return out + const planned = await select(db.query()).where('r.status', 'planned').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(Math.min(limit - out.length, wanted)) + return [...out, ...planned.map(toTarget)] + } + + private async spoolRange(target: Target) { + let count = 0 + for (const span of Spool.chunksOf({ start: target.start, end: target.end }, target.next)) { + if (this.stopped) return + if (!(await Spool.has(target.id, span))) { + const data = await NntpPool.use(async (client) => { + await client.group(target.group) + return client.xoverRaw(span.from, span.to) + }) + await Spool.write(target.id, span, data) + } + count++ + await db.from('ranges').where('id', target.id).update({ spooled_chunks: count }) + } + await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() }) + } +} + +function toTarget(row: any): Target { + return { id: Number(row.id), group: row.group, start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id), leased: row.status === 'leased' } +} diff --git a/app/services/rule_sets.ts b/app/services/rule_sets.ts new file mode 100644 index 0000000..731b088 --- /dev/null +++ b/app/services/rule_sets.ts @@ -0,0 +1,36 @@ +import db from '@adonisjs/lucid/services/db' +import { DEFAULT_RULES, type Rules } from '#lib/rules/index' + +type StoredRules = Omit + +/** + * The rule set workers get with a lease (REBUILD.md, rules and recovery): + * the newest version, plus the indexed groups' priorities (owner rule). + * Version 1 (DEFAULT_RULES, no dropped signatures) is created when there is none. + */ +class RuleSets { + async current(): Promise { + let row = await db.from('rule_sets').orderBy('version', 'desc').first() + if (!row) { + const { version: _v, ...rules } = DEFAULT_RULES + await db.table('rule_sets').insert({ version: 1, rules: JSON.stringify({ ...rules, droppedSignatures: [] }), note: 'default' }).onConflict('version').ignore() + row = await db.from('rule_sets').orderBy('version', 'desc').first() + } + const groups = await db.from('groups').where('active', true).select('name', 'priority') + return { + ...(row.rules as StoredRules), + version: Number(row.version), + priorities: Object.fromEntries(groups.map((g) => [g.name, Number(g.priority)])), + } + } + + /** A new version: what changes is recorded on the ranges that run with it. */ + async publish(rules: StoredRules, note?: string): Promise { + const latest = await db.from('rule_sets').max('version as v').first() + const version = Number(latest?.v ?? 0) + 1 + await db.table('rule_sets').insert({ version, rules: JSON.stringify(rules), note: note ?? null }) + return version + } +} + +export default new RuleSets() diff --git a/app/services/spool.ts b/app/services/spool.ts new file mode 100644 index 0000000..8b35292 --- /dev/null +++ b/app/services/spool.ts @@ -0,0 +1,78 @@ +import { mkdir, readFile, readdir, rename, rm, stat, writeFile } from 'node:fs/promises' +import { join } from 'node:path' +import coordinatorConfig from '#config/coordinator' + +export interface ChunkSpan { + from: number + to: number +} + +/** + * Prefetched XOVER chunks on disk (REBUILD.md, prefetch): `//-.gz`, + * each as the news server sent it. Chunks follow a fixed grid from the + * range's start, so a resume point (always at a batch boundary) is a chunk's start. + */ +class Spool { + get dir() { + return coordinatorConfig.spool.dir + } + + /** The chunks of a range from `from` (its resume point) to its end. */ + chunksOf(range: { start: number; end: number }, from = range.start): ChunkSpan[] { + const size = coordinatorConfig.chunkArticles + const spans: ChunkSpan[] = [] + for (let at = range.start; at <= range.end; at += size) { + const span = { from: at, to: Math.min(at + size - 1, range.end) } + if (span.to >= from) spans.push(span) + } + return spans + } + + private path(rangeId: number, span: ChunkSpan) { + return join(this.dir, String(rangeId), `${span.from}-${span.to}.gz`) + } + + async has(rangeId: number, span: ChunkSpan) { + try { + await stat(this.path(rangeId, span)) + return true + } catch { + return false + } + } + + async write(rangeId: number, span: ChunkSpan, data: Buffer) { + const file = this.path(rangeId, span) + await mkdir(join(this.dir, String(rangeId)), { recursive: true }) + // Written under another name first: a reader never sees half a chunk. + await writeFile(`${file}.part`, data) + await rename(`${file}.part`, file) + } + + read(rangeId: number, span: ChunkSpan) { + return readFile(this.path(rangeId, span)) + } + + async remove(rangeId: number) { + await rm(join(this.dir, String(rangeId)), { recursive: true, force: true }) + } + + /** Bytes in the spool. */ + async size(): Promise { + let total = 0 + let ranges: string[] = [] + try { + ranges = await readdir(this.dir) + } catch { + return 0 + } + for (const range of ranges) { + for (const file of await readdir(join(this.dir, range)).catch(() => [])) { + total += (await stat(join(this.dir, range, file)).catch(() => ({ size: 0 }))).size + } + } + return total + } +} + +export default new Spool() diff --git a/bin/console.ts b/bin/console.ts index 23f042f..c07b42a 100644 --- a/bin/console.ts +++ b/bin/console.ts @@ -11,6 +11,19 @@ | */ +/** + * Edge workers (edge:node) have no database or web server: what start/env + * requires for those gets a placeholder, so a node's .env needs only the + * coordinator's URL and its token. + */ +if (process.argv[2] === 'edge:node') { + const placeholders: Record = { + NODE_ENV: 'production', PORT: '0', HOST: '127.0.0.1', LOG_LEVEL: 'info', APP_KEY: 'edge-node-has-no-app-key-0000000000', APP_URL: 'http://127.0.0.1', + SESSION_DRIVER: 'memory', DB_HOST: '127.0.0.1', DB_PORT: '5432', DB_USER: 'none', DB_DATABASE: 'none', + } + for (const [key, value] of Object.entries(placeholders)) process.env[key] ??= value +} + await import('reflect-metadata') const { Ignitor, prettyPrintError } = await import('@adonisjs/core/ignitor') diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts new file mode 100644 index 0000000..a5a9b5e --- /dev/null +++ b/commands/coordinator_run.ts @@ -0,0 +1,51 @@ +import { BaseCommand } from '@adonisjs/core/ace' +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 NntpPool from '#services/nntp_pool' + +const TICK_MS = 2000 +const PLAN_EVERY_MS = 60 * 60 * 1000 + +/** + * The coordinator's background work (REBUILD.md, part 1): the prefetcher, + * and planning ranges for new articles every hour. The API (leases, stream, + * batches) runs in the web server. + */ +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 } + + async run() { + const prefetcher = new Prefetcher() + let stopping = false + let stopped: () => void + const done = new Promise((resolve) => (stopped = resolve)) + // bin/console turns SIGTERM into app.terminate: wait for the spooling in flight first. + this.app.terminating(async () => { + stopping = true + prefetcher.stop() + await done + }) + let plannedAt = 0 + this.logger.info('Coordinator started.') + while (!stopping) { + try { + if (Date.now() - plannedAt >= PLAN_EVERY_MS) { + plannedAt = Date.now() + const { groups, ranges } = await Planner.planAll() + if (ranges > 0) this.logger.info(`Planned ${ranges} ranges over ${groups} groups.`) + } + await prefetcher.tick() + } catch (error: any) { + this.logger.error(`Coordinator round failed: ${error.message}`) + } + await sleep(TICK_MS) + } + while (prefetcher.active > 0) await sleep(200) + await NntpPool.close() + stopped!() + } +} diff --git a/commands/edge_bench.ts b/commands/edge_bench.ts index 72b76dc..3422c25 100644 --- a/commands/edge_bench.ts +++ b/commands/edge_bench.ts @@ -37,6 +37,9 @@ export default class EdgeBench extends BaseCommand { @flags.number({ description: 'Articles per XOVER', default: 50_000 }) declare chunk: number + @flags.number({ description: 'Chunks per batch (results taken every N chunks)', default: 10 }) + declare batchChunks: number + @flags.number({ description: 'Chunks without a new part before an open file is given up', default: 5 }) declare idle: number @@ -87,6 +90,8 @@ export default class EdgeBench extends BaseCommand { let peakRss = 0 const out = { headers: 0, releases: 0, files: 0, segments: 0, payload: 0, candidates: 0, fragments: 0, samples: 0 } const counts = new Map() + const keys = new Set() + let batches = 0 const cpuStart = process.cpuUsage() const wall = Date.now() for (const file of files) { @@ -102,12 +107,18 @@ export default class EdgeBench extends BaseCommand { at = performance.now() const next = Number(file.split('-')[2].replace('.gz', '')) + 1 indexer.add(overviews, next) - if (file === files.at(-1)) indexer.finish() + const last = file === files.at(-1) + if (last) indexer.finish() + t.index += performance.now() - at + if (!last && (files.indexOf(file) + 1) % this.batchChunks !== 0) continue + at = performance.now() const batch = indexer.take() t.index += performance.now() - at out.headers += batch.headers out.releases += batch.releases.length + batches++ for (const r of batch.releases) { + keys.add(r.key) out.files += r.files out.segments += r.segments out.payload += r.payload.length @@ -124,6 +135,7 @@ export default class EdgeBench extends BaseCommand { this.logger.info(`${group} ${start}–${end}: ${files.length} chunks, ${(compressed / 1e6).toFixed(0)} MB compressed, ${(text / 1e6).toFixed(0)} MB text`) this.logger.info(`${out.headers.toLocaleString('en')} headers in ${(total / 1000).toFixed(1)} s (${Math.round(out.headers / (total / 1000)).toLocaleString('en')}/s on one core; CPU ${cpuS.toFixed(1)} s, wall ${((Date.now() - wall) / 1000).toFixed(1)} s)`) this.logger.info(` inflate ${(t.inflate / 1000).toFixed(1)} s, parse ${(t.parse / 1000).toFixed(1)} s, index ${(t.index / 1000).toFixed(1)} s; peak RSS ${(peakRss / 1e6).toFixed(0)} MB`) + this.logger.info(` ${batches} batches of ${this.batchChunks} chunks: ${out.releases} release chunks for ${keys.size} releases (${(out.releases / Math.max(1, keys.size)).toFixed(2)} chunks per release)`) this.logger.info(` kept: ${out.releases} release chunks, ${out.files} files, ${out.segments} segments, payload ${(out.payload / 1e6).toFixed(1)} MB (${(out.payload / Math.max(1, out.segments)).toFixed(1)} B/segment); ${out.candidates} candidates, ${out.fragments} fragments, ${out.samples} samples`) for (const [k, v] of [...counts].sort((a, b) => b[1] - a[1]).slice(0, 8)) this.logger.info(` dropped ${k}: ${v.toLocaleString('en')} articles`) } diff --git a/commands/edge_node.ts b/commands/edge_node.ts new file mode 100644 index 0000000..5d96fcf --- /dev/null +++ b/commands/edge_node.ts @@ -0,0 +1,157 @@ +import { BaseCommand, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import { setTimeout as sleep } from 'node:timers/promises' +import { unzipSync } from 'node:zlib' +import { parseOverviews } from '#lib/nntp/index' +import { RangeIndexer, decodeFrames, toWire, type EdgeBatch, type Lease } from '#lib/edge/index' + +const IDLE_MS = 30_000 +const RETRIES = 8 + +class LeaseLost extends Error {} + +/** + * An edge worker (REBUILD.md, part 2): leases a range from the coordinator, + * streams its chunks, indexes them, posts a batch every --batch-chunks + * chunks (each moves the range's resume point), and finishes the range. + * Needs COORDINATOR_URL and NODE_TOKEN only. On SIGTERM it gives its range + * back after the current chunk; the next worker resumes from the last batch. + */ +export default class EdgeNode extends BaseCommand { + static commandName = 'edge:node' + static description = 'Indexes ranges for the coordinator' + static options: CommandOptions = { startApp: true, staysAlive: true } + + @flags.number({ description: 'Chunks per posted batch (500k articles: ~4% of releases get a second chunk; LOG.md 2026-10-09)', default: 10 }) + declare batchChunks: number + + @flags.number({ description: 'Stop after this many ranges (default: keep going)' }) + declare ranges?: number + + private stopping = false + + async run() { + const url = process.env.COORDINATOR_URL + const token = process.env.NODE_TOKEN + if (!url || !token) throw new Error('COORDINATOR_URL and NODE_TOKEN are needed') + let stopped: () => void + const done = new Promise((resolve) => (stopped = resolve)) + this.app.terminating(async () => { + this.stopping = true + await done + }) + const api = new Api(url.replace(/\/$/, ''), token) + let ranges = 0 + try { + while (!this.stopping && (!this.ranges || ranges < this.ranges)) { + let lease: Lease | null = null + try { + lease = await api.lease() + } catch (error: any) { + this.logger.warning(`Lease failed: ${error.message}`) + } + if (!lease) { + if (this.ranges) break + await sleep(IDLE_MS) + continue + } + await this.runRange(api, lease) + ranges++ + } + } finally { + stopped!() + } + await this.terminate() + } + + private async runRange(api: Api, lease: Lease) { + const { range, rules } = lease + const started = Date.now() + const indexer = new RangeIndexer({ group: range.group, start: range.start, end: range.end, rules }) + const totals = { headers: 0, releases: 0, candidates: 0 } + let chunks = 0 + const post = async (batch: EdgeBatch) => { + const stored = await api.batch(range.id, batch) + totals.headers += batch.headers + totals.releases += stored.releases + totals.candidates += stored.candidates + } + try { + for await (const frame of decodeFrames(await api.stream(range.id))) { + indexer.add(parseOverviews(unzipSync(frame.data).toString('utf8')), frame.to + 1) + chunks++ + const last = frame.to >= range.end + if (last) indexer.finish() + if (last || chunks % this.batchChunks === 0) await post(indexer.take()) + if (this.stopping && !last) { + // Back to the queue; the next worker resumes at the last batch. + if (chunks % this.batchChunks !== 0) this.logger.info(`Range ${range.id}: giving it back (${chunks % this.batchChunks} chunks since the last batch are read again).`) + await api.call(range.id, 'release') + return + } + } + await api.call(range.id, 'finish') + const seconds = (Date.now() - started) / 1000 + this.logger.info( + `${range.group} ${range.start}–${range.end}: ${totals.headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s ` + + `(${Math.round(totals.headers / seconds).toLocaleString('en')}/s), ${totals.releases} release chunks, ${totals.candidates} probe candidates.` + ) + } catch (error: any) { + if (error instanceof LeaseLost) { + this.logger.warning(`Range ${range.id}: ${error.message}`) + return + } + this.logger.error(`Range ${range.id} failed: ${error.message}; giving it back.`) + await api.call(range.id, 'release').catch(() => {}) + } + } +} + +/** The coordinator's node API, with retries for network trouble and 5xx. */ +class Api { + constructor( + private base: string, + private token: string + ) {} + + private async request(path: string, init: RequestInit = {}): Promise { + for (let attempt = 1; ; attempt++) { + try { + const response = await fetch(`${this.base}/api/v1/node/${path}`, { + ...init, + headers: { 'Authorization': `Bearer ${this.token}`, 'Content-Type': 'application/json', 'Accept': 'application/json', ...init.headers }, + }) + if (response.status === 409) throw new LeaseLost(((await response.json().catch(() => ({}))) as { error?: string }).error ?? 'lease lost') + if (response.status < 500) return response + if (attempt >= RETRIES) throw new Error(`${path}: HTTP ${response.status}`) + } catch (error) { + if (error instanceof LeaseLost || attempt >= RETRIES) throw error + } + await sleep(Math.min(60_000, 1000 * 2 ** attempt)) + } + } + + async lease(): Promise { + const response = await this.request('lease', { method: 'POST', body: '{}' }) + if (response.status === 204) return null + if (!response.ok) throw new Error(`lease: HTTP ${response.status}`) + return response.json() as Promise + } + + async stream(rangeId: number): Promise> { + const response = await this.request(`ranges/${rangeId}/stream`, { headers: { Accept: 'application/octet-stream' } }) + if (!response.ok || !response.body) throw new Error(`stream: HTTP ${response.status}`) + return response.body as unknown as AsyncIterable + } + + async batch(rangeId: number, batch: EdgeBatch): Promise<{ releases: number; candidates: number }> { + const response = await this.request(`ranges/${rangeId}/batches`, { method: 'POST', body: JSON.stringify(toWire(batch)) }) + if (!response.ok) throw new Error(`batch: HTTP ${response.status}`) + return response.json() as Promise<{ releases: number; candidates: number }> + } + + async call(rangeId: number, action: 'finish' | 'release' | 'heartbeat') { + const response = await this.request(`ranges/${rangeId}/${action}`, { method: 'POST', body: '{}' }) + if (!response.ok) throw new Error(`${action}: HTTP ${response.status}`) + } +} diff --git a/commands/groups_add.ts b/commands/groups_add.ts new file mode 100644 index 0000000..d979950 --- /dev/null +++ b/commands/groups_add.ts @@ -0,0 +1,30 @@ +import { BaseCommand, args, flags } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import Planner from '#services/planner' +import NntpPool from '#services/nntp_pool' + +/** Adds groups to index, after the existing ones in priority (owner rule). */ +export default class GroupsAdd extends BaseCommand { + static commandName = 'groups:add' + static description = 'Adds groups to index (by name, or the largest by LIST ACTIVE)' + static options: CommandOptions = { startApp: true } + + @args.spread({ description: 'Group names', required: false }) + declare names?: string[] + + @flags.number({ description: 'The largest N groups matching --pattern' }) + declare top?: number + + @flags.string({ description: 'LIST ACTIVE pattern for --top', default: 'alt.binaries.*' }) + declare pattern: string + + async run() { + try { + const names = this.top ? await Planner.largestGroups(this.pattern, this.top) : (this.names ?? []) + const added = await Planner.addGroups(names) + this.logger.info(`${added} of ${names.length} groups added.`) + } finally { + await NntpPool.close() + } + } +} diff --git a/commands/nodes_create.ts b/commands/nodes_create.ts new file mode 100644 index 0000000..6d4859c --- /dev/null +++ b/commands/nodes_create.ts @@ -0,0 +1,19 @@ +import { BaseCommand, args } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import Node from '#models/node' + +/** Registers an edge worker and prints its token (shown once). */ +export default class NodesCreate extends BaseCommand { + static commandName = 'nodes:create' + static description = 'Registers an edge worker and prints its token' + static options: CommandOptions = { startApp: true } + + @args.string({ description: 'Node name' }) + declare name: string + + async run() { + const node = await Node.firstOrCreate({ name: this.name }, { name: this.name }) + const token = await Node.accessTokens.create(node) + this.logger.info(`Node ${node.name} (#${node.id}). Token: ${token.value!.release()}`) + } +} diff --git a/commands/ranges_plan.ts b/commands/ranges_plan.ts new file mode 100644 index 0000000..273b8a3 --- /dev/null +++ b/commands/ranges_plan.ts @@ -0,0 +1,20 @@ +import { BaseCommand } from '@adonisjs/core/ace' +import type { CommandOptions } from '@adonisjs/core/types/ace' +import Planner from '#services/planner' +import NntpPool from '#services/nntp_pool' + +/** Cuts every active group's history (and new articles since) into ranges. */ +export default class RangesPlan extends BaseCommand { + static commandName = 'ranges:plan' + static description = "Plans ranges for the indexed groups' articles" + static options: CommandOptions = { startApp: true } + + async run() { + try { + const { groups, ranges } = await Planner.planAll() + this.logger.info(`${ranges} ranges planned over ${groups} groups.`) + } finally { + await NntpPool.close() + } + } +} diff --git a/config/auth.ts b/config/auth.ts index 686e0ae..fc91c24 100644 --- a/config/auth.ts +++ b/config/auth.ts @@ -20,6 +20,16 @@ const authConfig = defineConfig({ }), }), + /** + * Edge workers (REBUILD.md, part 2): a token per node. + */ + node: tokensGuard({ + provider: tokensUserProvider({ + tokens: 'accessTokens', + model: () => import('#models/node'), + }), + }), + /** * Session-based guard for browser authentication. */ diff --git a/config/bodyparser.ts b/config/bodyparser.ts index 7e836e6..b774fcf 100644 --- a/config/bodyparser.ts +++ b/config/bodyparser.ts @@ -66,7 +66,8 @@ const bodyParserConfig = defineConfig({ /** * Maximum accepted payload size for multipart requests. */ - limit: '20mb', + // Edge worker batches carry their releases' chunks (a few MB, base64). + limit: '64mb', /** * Content types handled by the multipart parser. diff --git a/config/coordinator.ts b/config/coordinator.ts new file mode 100644 index 0000000..2492fe8 --- /dev/null +++ b/config/coordinator.ts @@ -0,0 +1,24 @@ +import env from '#start/env' +import app from '@adonisjs/core/services/app' + +/** The coordinator (REBUILD.md, part 1). */ +export default { + /** Articles per range. */ + rangeArticles: env.get('RANGE_ARTICLES', 2_000_000), + /** Articles per XOVER chunk (spool files, stream frames). */ + chunkArticles: env.get('CHUNK_ARTICLES', 50_000), + /** A leased range without a heartbeat for this long goes back to the queue. */ + leaseTtlSeconds: env.get('LEASE_TTL_SECONDS', 15 * 60), + /** NNTP connections the coordinator fetches headers with. */ + nntpConnections: env.get('NNTP_CONNECTIONS', 4), + /** A reply with no data for this long fails (old ranges answer slowly). */ + nntpTimeoutMs: env.get('NNTP_TIMEOUT_MS', 10 * 60 * 1000), + spool: { + dir: env.get('SPOOL_DIR', app.tmpPath('spool')), + /** Ranges kept fetched ahead, per active lease. */ + aheadPerLease: env.get('SPOOL_AHEAD_PER_LEASE', 2), + /** At least this many ranges ahead, so the first leases start at once. */ + minAhead: env.get('SPOOL_MIN_AHEAD', 2), + maxBytes: env.get('SPOOL_MAX_GB', 10) * 1e9, + }, +} diff --git a/lib/edge/index.ts b/lib/edge/index.ts index d68c3b6..ee46770 100644 --- a/lib/edge/index.ts +++ b/lib/edge/index.ts @@ -1 +1,4 @@ export { RangeIndexer, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js' +export { encodeFrame, decodeFrames, type Frame } from './stream.js' +export { toWire, fromWire, type WireBatch } from './wire.js' +export type { Lease } from './lease.js' diff --git a/lib/edge/lease.ts b/lib/edge/lease.ts new file mode 100644 index 0000000..1e86d24 --- /dev/null +++ b/lib/edge/lease.ts @@ -0,0 +1,7 @@ +import type { Rules } from '../rules/index.js' + +/** What a worker gets with a range (the coordinator's lease). */ +export interface Lease { + range: { id: number; group: string; start: number; end: number; next: number } + rules: Rules +} diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts index a408f0d..e34adfb 100644 --- a/lib/edge/range_indexer.ts +++ b/lib/edge/range_indexer.ts @@ -143,6 +143,7 @@ export class RangeIndexer { /** What was found since the last take, and a fresh batch. */ take(): EdgeBatch { + this.judgeReleases() const out = this.batch out.counts = [...this.counts.values()] this.counts.clear() @@ -154,14 +155,22 @@ export class RangeIndexer { return this.aggregator.openFiles } + /** Completed files, gathered per release until the batch is taken: one chunk per release per batch. */ + private completed = new Map[]>() + private files(files: AggregatedFile[]) { - const byRelease = new Map[]>() for (const file of files) { const key = releaseKeyOf(file.segments[0]) - const list = byRelease.get(key) + const list = this.completed.get(key) if (list) list.push(file) - else byRelease.set(key, [file]) + else this.completed.set(key, [file]) } + } + + /** Judges and encodes the releases completed in this batch. */ + private judgeReleases() { + const byRelease = this.completed + this.completed = new Map() for (const [key, list] of byRelease) { const verdict = this.rules.judge( key, diff --git a/lib/edge/stream.ts b/lib/edge/stream.ts new file mode 100644 index 0000000..7cc3cce --- /dev/null +++ b/lib/edge/stream.ts @@ -0,0 +1,35 @@ +/** + * A range's chunks on the wire (REBUILD.md, prefetch): one streaming + * response, a frame per chunk: from (8 bytes), to (8 bytes), length + * (4 bytes), then the chunk as the news server sent it (compressed XOVER). + */ +export interface Frame { + from: number + to: number + data: Buffer +} + +const HEADER = 20 + +export function encodeFrame(frame: Frame): Buffer { + const head = Buffer.alloc(HEADER) + head.writeBigUInt64BE(BigInt(frame.from), 0) + head.writeBigUInt64BE(BigInt(frame.to), 8) + head.writeUInt32BE(frame.data.length, 16) + return Buffer.concat([head, frame.data]) +} + +/** The frames of a byte stream, as they complete. */ +export async function* decodeFrames(stream: AsyncIterable | Iterable): AsyncGenerator { + let buffer = Buffer.alloc(0) + for await (const piece of stream) { + buffer = buffer.length === 0 ? Buffer.from(piece) : Buffer.concat([buffer, piece]) + while (buffer.length >= HEADER) { + const length = buffer.readUInt32BE(16) + if (buffer.length < HEADER + length) break + yield { from: Number(buffer.readBigUInt64BE(0)), to: Number(buffer.readBigUInt64BE(8)), data: buffer.subarray(HEADER, HEADER + length) } + buffer = buffer.subarray(HEADER + length) + } + } + if (buffer.length > 0) throw new Error(`Stream ended inside a frame (${buffer.length} bytes left)`) +} diff --git a/lib/edge/wire.ts b/lib/edge/wire.ts new file mode 100644 index 0000000..b6f0652 --- /dev/null +++ b/lib/edge/wire.ts @@ -0,0 +1,23 @@ +import type { EdgeBatch } from './range_indexer.js' + +/** An EdgeBatch as JSON: chunk payloads in base64. */ +export type WireBatch = Omit & { + releases: (Omit & { payload: string })[] + samples: (Omit & { payload: string })[] +} + +export function toWire(batch: EdgeBatch): WireBatch { + return { + ...batch, + releases: batch.releases.map((r) => ({ ...r, payload: r.payload.toString('base64') })), + samples: batch.samples.map((s) => ({ ...s, payload: s.payload.toString('base64') })), + } +} + +export function fromWire(wire: WireBatch): EdgeBatch { + return { + ...wire, + releases: wire.releases.map((r) => ({ ...r, payload: Buffer.from(r.payload, 'base64') })), + samples: wire.samples.map((s) => ({ ...s, payload: Buffer.from(s.payload, 'base64') })), + } +} diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index e592cee..0f3de19 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -98,6 +98,21 @@ export class XoverClient { * compresses (plain replies are deflated here): the seed API passes these to nodes * without unpacking them (CONCEPT.md D38). `parseOverviews(unzipSync(raw).toString())`. */ + /** LIST ACTIVE : each group's name, high and low article (low is often just 2 on big servers; GROUP gives the real one). */ + async listActive(pattern: string): Promise<{ name: string; high: number; low: number }[]> { + return this.timed(async () => { + const reply = await this.command(`LIST ACTIVE ${pattern}`) + await this.expect(reply, ['215']) + return (await this.plainBody()) + .split('\r\n') + .filter(Boolean) + .map((line) => { + const [name, high, low] = line.split(' ') + return { name, high: Number(high), low: Number(low) } + }) + }) + } + async xoverRaw(from: number, to: number): Promise { return this.timed(async () => { const reply = await this.command(`XOVER ${from}-${to}`) diff --git a/start/env.ts b/start/env.ts index 1262ffb..a4246af 100644 --- a/start/env.ts +++ b/start/env.ts @@ -42,4 +42,15 @@ export default await Env.create(new URL('../', import.meta.url), { NNTP_USER: Env.schema.string.optional(), NNTP_PASSWORD: Env.schema.string.optional(), NNTP_SECURE: Env.schema.boolean.optional(), + NNTP_CONNECTIONS: Env.schema.number.optional(), + NNTP_TIMEOUT_MS: Env.schema.number.optional(), + + // Coordinator + RANGE_ARTICLES: Env.schema.number.optional(), + CHUNK_ARTICLES: Env.schema.number.optional(), + LEASE_TTL_SECONDS: Env.schema.number.optional(), + SPOOL_DIR: Env.schema.string.optional(), + SPOOL_AHEAD_PER_LEASE: Env.schema.number.optional(), + SPOOL_MIN_AHEAD: Env.schema.number.optional(), + SPOOL_MAX_GB: Env.schema.number.optional(), }) diff --git a/start/routes.ts b/start/routes.ts index 29fc17f..b8bcf5e 100644 --- a/start/routes.ts +++ b/start/routes.ts @@ -33,5 +33,19 @@ router .prefix('account') .as('profile') .use(middleware.auth()) + + // Edge workers (REBUILD.md, parts 1–2). + router + .group(() => { + router.post('lease', [controllers.Node, 'lease']) + router.get('ranges/:id/stream', [controllers.Node, 'stream']) + router.post('ranges/:id/batches', [controllers.Node, 'batch']) + router.post('ranges/:id/heartbeat', [controllers.Node, 'heartbeat']) + router.post('ranges/:id/finish', [controllers.Node, 'finish']) + router.post('ranges/:id/release', [controllers.Node, 'release']) + }) + .prefix('node') + .as('node') + .use(middleware.auth({ guards: ['node'] })) }) .prefix('/api/v1') diff --git a/tests/functional/node_api.spec.ts b/tests/functional/node_api.spec.ts new file mode 100644 index 0000000..d4c5d14 --- /dev/null +++ b/tests/functional/node_api.spec.ts @@ -0,0 +1,83 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import { deflateSync } from 'node:zlib' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import Node from '#models/node' +import Spool from '#services/spool' +import coordinatorConfig from '#config/coordinator' +import { decodeFrames, fromWire, toWire, RangeIndexer, type Lease } from '#lib/edge/index' +import { parseOverviews } from '#lib/nntp/index' +import { unzipSync } from 'node:zlib' + +const GROUP = 'alt.binaries.test' + +/** One overview chunk: a two-part release posted at `at`. */ +function chunk(at: number): Buffer { + const lines = [1, 2].map( + (n) => `${at + n - 1}\tSome.Show.S01E0${at % 7} [1/1] - "Some.Show.S01E0${at % 7}.1080p.mkv" yEnc (${n}/2)\tp \tMon, 05 Oct 2026 10:00:00 +0000\t\t\t700000\t5000\tXref: news.example ${GROUP}:${at + n - 1}` + ) + return deflateSync(Buffer.from(lines.join('\r\n') + '\r\n')) +} + +test.group('Node API', (group) => { + const config = { ...coordinatorConfig, spool: { ...coordinatorConfig.spool } } + let dir = '' + group.each.setup(async () => { + dir = await mkdtemp(join(tmpdir(), 'spool-')) + coordinatorConfig.spool.dir = dir + coordinatorConfig.chunkArticles = 100 + return testUtils.db().withGlobalTransaction() + }) + group.each.teardown(async () => { + coordinatorConfig.spool.dir = config.spool.dir + coordinatorConfig.chunkArticles = config.chunkArticles + await rm(dir, { recursive: true, force: true }) + }) + + test('lease, stream from the spool, post batches, finish', async ({ client, assert }) => { + const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id') + const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 300, next_id: 1, spooled_chunks: 3 }).returning('id') + const rangeId = Number(r.id) + for (const span of Spool.chunksOf({ start: 1, end: 300 })) await Spool.write(rangeId, span, chunk(span.from + 10)) + const node = await Node.create({ name: 'n1' }) + const token = (await Node.accessTokens.create(node)).value!.release() + + // No token, no lease. + ;(await client.post('/api/v1/node/lease')).assertStatus(401) + + const lease = await client.post('/api/v1/node/lease').bearerToken(token) + lease.assertStatus(200) + const leased = lease.body() as Lease + assert.deepEqual(leased.range, { id: rangeId, group: GROUP, start: 1, end: 300, next: 1 }) + assert.equal(leased.rules.priorities[GROUP], 1) + // Nothing else spooled: no second lease. + ;(await client.post('/api/v1/node/lease').bearerToken(token)).assertStatus(204) + + const stream = await client.get(`/api/v1/node/ranges/${rangeId}/stream`).bearerToken(token) + stream.assertStatus(200) + const raw = stream.body() as unknown + const body: Buffer = Buffer.isBuffer(raw) ? raw : Buffer.from(stream.text(), 'binary') + const frames = [] + for await (const frame of decodeFrames([body])) frames.push(frame) + assert.deepEqual(frames.map((f) => [f.from, f.to]), [[1, 100], [101, 200], [201, 300]]) + + const indexer = new RangeIndexer({ group: GROUP, start: 1, end: 300, rules: leased.rules }) + for (const f of frames) indexer.add(parseOverviews(unzipSync(f.data).toString('utf8')), f.to + 1) + indexer.finish() + const batch = toWire(indexer.take()) + assert.deepEqual(fromWire(batch).releases.length, 3) + const posted = await client.post(`/api/v1/node/ranges/${rangeId}/batches`).bearerToken(token).json(batch) + posted.assertStatus(200) + assert.include(posted.body(), { releases: 3, newReleases: 3 }) + // The same batch again: refused. + ;(await client.post(`/api/v1/node/ranges/${rangeId}/batches`).bearerToken(token).json(batch)).assertStatus(409) + + ;(await client.post(`/api/v1/node/ranges/${rangeId}/finish`).bearerToken(token)).assertStatus(200) + const range = await db.from('ranges').where('id', rangeId).first() + assert.equal(range.status, 'done') + assert.isFalse(await Spool.has(rangeId, { from: 1, to: 100 })) + }) +})