diff --git a/app/controllers/seed_controller.ts b/app/controllers/seed_controller.ts index 05b7629..8e244fc 100644 --- a/app/controllers/seed_controller.ts +++ b/app/controllers/seed_controller.ts @@ -1,5 +1,6 @@ import type { HttpContext } from '@adonisjs/core/http' import SeedOrchestrator, { LeaseLost } from '#services/SeedOrchestrator' +import SeedFetchPool from '#services/SeedFetchPool' import type { SeedBatch } from '#lib/seed/wire' /** @@ -31,6 +32,27 @@ export default class SeedController { }) } + /** + * GET /api/v1/seed/ranges/:id/chunk?from=&to= → the overviews of those + * articles as a zlib/gzip stream (relay nodes fetch through the indexer). + */ + async chunk({ seedNode, params, request, response }: HttpContext) { + if (SeedFetchPool.size === 0) return response.serviceUnavailable({ error: 'Relay is off (SEED_NNTP_CONNECTIONS=0); fetch directly' }) + const from = Number(request.input('from')) + const to = Number(request.input('to')) + if (!Number.isSafeInteger(from) || !Number.isSafeInteger(to)) return response.badRequest({ error: 'Expected ?from=&to=' }) + return this.leased(response, async () => { + try { + const raw = await SeedOrchestrator.chunk(seedNode.id, Number(params.id), from, to) + response.header('content-type', 'application/octet-stream') + return response.send(raw) + } catch (error) { + if (error instanceof RangeError) return response.badRequest({ error: error.message }) + throw error + } + }) + } + /** POST /api/v1/seed/ranges/:id/done */ async done({ seedNode, params, response }: HttpContext) { return this.leased(response, async () => { diff --git a/app/services/SeedFetchPool.ts b/app/services/SeedFetchPool.ts new file mode 100644 index 0000000..45eabfa --- /dev/null +++ b/app/services/SeedFetchPool.ts @@ -0,0 +1,91 @@ +import nntpConfig from '#config/nntp' +import indexerConfig from '#config/indexer' +import { XoverClient } from '#lib/nntp/xover_client' + +interface Slot { + client: XoverClient + group?: string + usedAt: number +} + +/** A pooled connection unused this long is replaced: the news server may have dropped it. */ +const IDLE_MS = 2 * 60 * 1000 + +/** + * NNTP connections in the seed API for relay nodes (CONCEPT.md D38): the + * indexer fetches overview chunks and passes the gzip stream on unopened, so + * nodes need no news server account and every connection comes from this + * machine's IP. SEED_NNTP_CONNECTIONS (0: relay off). + */ +class SeedFetchPool { + private idle: Slot[] = [] + private open = 0 + private waiters: (() => void)[] = [] + + get size() { + return indexerConfig.seed.nntpConnections + } + + async fetch(group: string, from: number, to: number): Promise { + const slot = await this.acquire() + let ok = false + try { + if (slot.group !== group) { + await slot.client.group(group) + slot.group = group + } + const raw = await slot.client.xoverRaw(from, to) + ok = true + return raw + } finally { + this.release(slot, ok) + } + } + + private async acquire(): Promise { + while (true) { + const slot = this.idle.pop() + if (slot) { + if (Date.now() - slot.usedAt < IDLE_MS) return slot + slot.client.close() + this.open-- + continue + } + if (this.open < this.size) { + this.open++ + try { + const client = await XoverClient.connect({ + host: nntpConfig.host, + port: nntpConfig.port, + secure: nntpConfig.secure, + servername: nntpConfig.tlsServername, + user: nntpConfig.user, + password: nntpConfig.password?.release(), + timeoutMs: nntpConfig.commandTimeoutMs, + compress: true, + }) + return { client, usedAt: Date.now() } + } catch (error) { + this.open-- + this.waiters.shift()?.() + throw error + } + } + await new Promise((resolve) => this.waiters.push(resolve)) + } + } + + private release(slot: Slot, ok: boolean) { + if (ok) { + slot.usedAt = Date.now() + this.idle.push(slot) + } else { + // A failed command can leave the connection mid-reply: start afresh. + slot.client.close() + this.open-- + } + this.waiters.shift()?.() + } +} + +export default new SeedFetchPool() diff --git a/app/services/SeedOrchestrator.ts b/app/services/SeedOrchestrator.ts index 7778051..13efaad 100644 --- a/app/services/SeedOrchestrator.ts +++ b/app/services/SeedOrchestrator.ts @@ -3,6 +3,7 @@ import db from '@adonisjs/lucid/services/db' import NntpService from '#services/NntpService' import SeedService from '#services/SeedService' import BackfillService from '#services/BackfillService' +import SeedFetchPool from '#services/SeedFetchPool' import { articleAtDate } from '#lib/nntp/date_search' import { normalizeSeedFile, type SeedBatch } from '#lib/seed/wire' @@ -92,6 +93,19 @@ class SeedOrchestrator { return stored } + /** + * Relay: overviews of `from`..`to` of a leased range, fetched here and handed + * on as the gzip stream the news server sent. Keeps the lease alive. + */ + async chunk(nodeId: number, rangeId: number, from: number, to: number): Promise { + const range = await this.leased(nodeId, rangeId) + const start = Math.max(from, Number(range.start_id)) + const end = Math.min(to, Number(range.end_id), start + 49_999) + if (end < start) throw new RangeError(`Articles ${from}–${to} are outside range ${rangeId}`) + await db.from('seed_ranges').where('id', rangeId).update({ heartbeat_at: new Date() }) + return SeedFetchPool.fetch(range.group_name, start, end) + } + 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 }) diff --git a/commands/SeedNode.ts b/commands/SeedNode.ts index c7ba69f..ce11e8c 100644 --- a/commands/SeedNode.ts +++ b/commands/SeedNode.ts @@ -1,11 +1,11 @@ -import { gzipSync } from 'node:zlib' +import { gzipSync, unzipSync } 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 { XoverClient, parseOverviews, type Overview } from '#lib/nntp/xover_client' +import { runRange, type OverviewSource } from '#lib/seed/runner' import type { SeedBatch } from '#lib/seed/wire' import type { OverviewHeader } from '#lib/ingest/segment' @@ -56,6 +56,13 @@ export default class SeedNode extends BaseCommand { @flags.number({ description: 'XOVER chunks per batch posted to the API', default: 5 }) declare submitEvery: number + @flags.string({ + description: + 'relay: headers come through the seed API (no NNTP account needed); direct: own NNTP connections (default: SEED_FETCH or relay)', + default: process.env.SEED_FETCH || 'relay', + }) + declare fetch: string + @flags.boolean({ description: 'Stop when there is nothing to lease instead of waiting' }) declare once: boolean @@ -77,7 +84,9 @@ export default class SeedNode extends BaseCommand { 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).`) + this.logger.info( + `Seeding node for ${base}: ${this.connections} range(s) at once, ${this.fetch === 'direct' ? 'own NNTP connections' : 'headers through the seed API'}.` + ) await Promise.all(Array.from({ length: this.connections }, (_, i) => this.work(api, i + 1))) this.logger.info('Seeding node stopped.') } @@ -97,6 +106,10 @@ export default class SeedNode extends BaseCommand { await this.pause(60_000) continue } + if (this.fetch !== 'direct') { + await this.workRange(api, new RelaySource(api, lease.id), lease, log) + continue + } if (!client) { try { client = await this.connect() @@ -151,7 +164,7 @@ export default class SeedNode extends BaseCommand { /** Works one leased range; false when the NNTP connection should be replaced. */ private async workRange( api: SeedApi, - client: XoverClient, + client: OverviewSource, lease: Lease, log: (message: string) => void ): Promise { @@ -230,38 +243,66 @@ class SeedApi { async post(path: string, body?: unknown): Promise { const payload = body === undefined ? undefined : gzipSync(JSON.stringify(body)) + const result = await this.request('POST', path, payload) + if (result === null) return null + const text = result.toString('utf8') + return text ? JSON.parse(text) : null + } + + /** A binary GET (a relay chunk). */ + async get(path: string): Promise { + return (await this.request('GET', path)) ?? Buffer.alloc(0) + } + + private async request(method: string, path: string, payload?: Buffer): Promise { for (let attempt = 1; ; attempt++) { let status = 0 - let text = '' + let data = Buffer.alloc(0) try { const response = await fetch(this.base + path, { - method: 'POST', + method, 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() + data = Buffer.from(await response.arrayBuffer()) } catch (error: any) { - text = error.message + data = Buffer.from(error.message) } + const text = () => data.toString('utf8').slice(0, 300) if (status === 204) return null - if (status >= 200 && status < 300) return text ? JSON.parse(text) : null + if (status >= 200 && status < 300) return data if (status === 401) throw new Unauthorized('The seed API rejected this node token') - if (status === 409) throw new LeaseLost(text) + if (status === 409) throw new LeaseLost(text()) // 404 comes from the proxy while the seed API restarts (a deploy): this API has no 404s of its own. if (status >= 400 && status < 500 && status !== 404) - throw new Error(`Seed API ${path}: ${status} ${text.slice(0, 300)}`) + throw new Error(`Seed API ${path}: ${status} ${text()}`) // Network errors, 404 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)}`) + if (attempt >= 8) throw new Error(`Seed API ${path}: ${status || 'no response'} ${text()}`) const wait = Math.min(120_000, 2000 * 2 ** (attempt - 1)) - this.warn(`Seed API ${path}: ${status || text}; retrying in ${wait / 1000} s`) + this.warn(`Seed API ${path}: ${status || text()}; retrying in ${wait / 1000} s`) await sleep(wait) } } } + +/** A range's overviews through the seed API's relay (CONCEPT.md D38): no news server account on the node. */ +class RelaySource implements OverviewSource { + constructor( + private api: SeedApi, + private rangeId: number + ) {} + + async group() {} + + async xover(from: number, to: number): Promise { + const raw = await this.api.get( + `/api/v1/seed/ranges/${this.rangeId}/chunk?from=${from}&to=${to}` + ) + return raw.length > 0 ? parseOverviews(unzipSync(raw).toString('utf8')) : [] + } +} diff --git a/config/indexer.ts b/config/indexer.ts index 7d124a1..733104c 100644 --- a/config/indexer.ts +++ b/config/indexer.ts @@ -63,6 +63,11 @@ export default { * Header backfill (`group:backfill`, CONCEPT.md D32): low-priority, * header-only fetch ranges walking back through a group's history. */ + seed: { + /** NNTP connections the seed API fetches with for relay nodes (CONCEPT.md D38); 0: relay off. */ + nntpConnections: env.get('SEED_NNTP_CONNECTIONS', 0), + }, + backfill: { /** Backfill ranges waiting or running at a time, over all groups. */ maxQueued: env.get('BACKFILL_MAX_QUEUED', 40), diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index de91ffc..11b6e66 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -1,6 +1,6 @@ import net from 'node:net' import tls from 'node:tls' -import { unzipSync } from 'node:zlib' +import { deflateSync, unzipSync } from 'node:zlib' export interface XoverClientOptions { host: string @@ -90,6 +90,20 @@ export class XoverClient { return this.timed(() => this.overviews(from, to)) } + /** + * Overviews of `from`..`to` as one zlib/gzip stream, as the server sent it when it + * compresses (plain replies are deflated here): the seed API passes these to nodes + * without unpacking them (CONCEPT.md D38). `parseOverviews(unzipSync(raw).toString())`. + */ + async xoverRaw(from: number, to: number): Promise { + return this.timed(async () => { + const reply = await this.command(`XOVER ${from}-${to}`) + if (reply.startsWith('420') || reply.startsWith('423')) return deflateSync(Buffer.alloc(0)) + await this.expect(reply, ['224']) + return reply.includes('COMPRESS=GZIP') ? (await this.compressedRaw()).raw : deflateSync(Buffer.from(await this.plainBody())) + }) + } + /** A reply that stalls for timeoutMs fails the connection; idle time between commands doesn't count. */ private async timed(exchange: () => Promise): Promise { this.sock.setTimeout(this.timeoutMs) @@ -179,15 +193,20 @@ export class XoverClient { } } - /** A zlib/gzip stream followed by a "." line: complete once it ends in ".\r\n" and inflates. */ private async compressedBody(): Promise { + return (await this.compressedRaw()).text + } + + /** A zlib/gzip stream followed by a "." line: complete once it ends in ".\r\n" and inflates. */ + private async compressedRaw(): Promise<{ raw: Buffer; text: string }> { while (true) { const all = this.buffer() if (all.length > DOT_LINE.length && all.subarray(all.length - DOT_LINE.length).equals(DOT_LINE)) { + const raw = all.subarray(0, all.length - DOT_LINE.length) try { - const text = unzipSync(all.subarray(0, all.length - DOT_LINE.length)).toString('utf8') + const text = unzipSync(raw).toString('utf8') this.take(all.length) - return text + return { raw: Buffer.from(raw), text } } catch { // ".\r\n" inside the compressed data: wait for more. } diff --git a/lib/seed/runner.ts b/lib/seed/runner.ts index 522dac8..d23facb 100644 --- a/lib/seed/runner.ts +++ b/lib/seed/runner.ts @@ -1,10 +1,16 @@ -import type { XoverClient, Overview } from '../nntp/xover_client.js' +import type { 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' +/** Where a range's overviews come from: an NNTP connection, or the seed API's relay. */ +export interface OverviewSource { + group(name: string): Promise + xover(from: number, to: number): Promise +} + export interface RunRangeOptions { - client: XoverClient + client: OverviewSource group: string /** Inclusive article numbers. */ start: number diff --git a/seed-node/README.md b/seed-node/README.md index 5e17825..2c210d2 100644 --- a/seed-node/README.md +++ b/seed-node/README.md @@ -1,10 +1,18 @@ # 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 +D38). It asks the seed API for a range over HTTPS, gets that range's headers +(by default through the seed API, which fetches them from the news server and +passes them on gzip-compressed), groups them into files, and posts the results back. It only makes outgoing connections: no port forwarding, -database or Redis. +database, Redis or news server account. Only a token and +`SEED_CONNECTIONS` need setting. + +Headers through the seed API (`SEED_FETCH=relay`, the default) use the +indexer's news server account: `SEED_NNTP_CONNECTIONS` in the indexer's +`.env` sets how many connections it fetches with for all nodes together. +With `SEED_FETCH=direct` a node uses its own NNTP account instead (fill in +the NNTP lines in `node.env`). ## What a node needs diff --git a/seed-node/node.env.example b/seed-node/node.env.example index cf92388..b836bd8 100644 --- a/seed-node/node.env.example +++ b/seed-node/node.env.example @@ -1,17 +1,19 @@ # --- 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 +# Ranges worked on at once (about 0.4 CPU core each). +SEED_CONNECTIONS=4 +# relay: headers come through the seed API, no news server account needed (the default). +# direct: the node fetches with its own NNTP account; then fill in the NNTP_* lines below. +SEED_FETCH=relay -# --- NNTP account this node fetches headers with (mind its connection limit across all nodes). -NNTP_HOST=news.example.com +# --- Only for SEED_FETCH=direct: the NNTP account (counts against its connection and IP limits). +NNTP_HOST=localhost NNTP_PORT=563 NNTP_SECURE=true -NNTP_USER=replace_me -NNTP_PASSWORD=replace_me -# Only when NNTP_HOST is a tunnel or proxy (e.g. localhost:1563 forwarded to news.eweka.nl:563): -# the hostname the server's TLS certificate is for. +NNTP_USER=unused +NNTP_PASSWORD=unused +# Through a tunnel or proxy: the hostname the server's TLS certificate is for. # NNTP_TLS_SERVERNAME=news.eweka.nl # --- Required by the app's configuration, but unused on a seeding node (no database, Redis or web server). diff --git a/start/env.ts b/start/env.ts index ff6146d..8719642 100644 --- a/start/env.ts +++ b/start/env.ts @@ -49,6 +49,7 @@ export default await Env.create(new URL('../', import.meta.url), { NNTP_POOL_SIZE: Env.schema.number.optional(), NNTP_COMMAND_TIMEOUT_MS: Env.schema.number.optional(), NNTP_TLS_SERVERNAME: Env.schema.string.optional(), + SEED_NNTP_CONNECTIONS: Env.schema.number.optional(), // Indexer (config/indexer.ts) BACKFILL_MAX_QUEUED: Env.schema.number.optional(), diff --git a/start/routes.ts b/start/routes.ts index 77c3502..b0cd4b6 100644 --- a/start/routes.ts +++ b/start/routes.ts @@ -55,6 +55,7 @@ router .group(() => { router.post('lease', [SeedController, 'lease']) router.post('ranges/:id/batches', [SeedController, 'batch']).where('id', router.matchers.number()) + router.get('ranges/:id/chunk', [SeedController, 'chunk']).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']) diff --git a/tests/functional/seed_api.spec.ts b/tests/functional/seed_api.spec.ts index add6f05..3f22ff7 100644 --- a/tests/functional/seed_api.spec.ts +++ b/tests/functional/seed_api.spec.ts @@ -5,6 +5,8 @@ import testUtils from '@adonisjs/core/services/test_utils' import SeedOrchestrator from '#services/SeedOrchestrator' import QueueService from '#services/QueueService' import indexerConfig from '#config/indexer' +import SeedFetchPool from '#services/SeedFetchPool' +import { deflateSync } from 'node:zlib' import { segmentFromHeader } from '#lib/ingest/segment' import { FileAggregator } from '#lib/aggregate/index' import { toFragment, toSeedFile, type SeedBatch } from '#lib/seed/wire' @@ -99,4 +101,31 @@ test.group('Seed API', (group) => { assert.equal(row.status, 'queued') assert.equal(row.attempts, 0) }) + + test('relay: a leased range gets its overview chunks through the API', async ({ client, assert }) => { + const token = await SeedOrchestrator.issueToken('relay-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}` } + const range = (await client.post('/api/v1/seed/lease').headers(auth)).body().range + const fetch = SeedFetchPool.fetch + const connections = indexerConfig.seed.nntpConnections + const asked: unknown[] = [] + SeedFetchPool.fetch = async (group, from, to) => { + asked.push([group, from, to]) + return deflateSync(Buffer.from('5\tsubject\r\n')) + } + try { + indexerConfig.seed.nntpConnections = 0 + ;(await client.get(`/api/v1/seed/ranges/${range.id}/chunk`).qs({ from: 1, to: 50 }).headers(auth)).assertStatus(503) + indexerConfig.seed.nntpConnections = 1 + const chunk = await client.get(`/api/v1/seed/ranges/${range.id}/chunk`).qs({ from: 50, to: 500 }).headers(auth) + chunk.assertStatus(200) + // Clamped to the range. + assert.deepEqual(asked, [['alt.binaries.test', 50, 100]]) + ;(await client.get(`/api/v1/seed/ranges/${range.id}/chunk`).qs({ from: 200, to: 300 }).headers(auth)).assertStatus(400) + } finally { + SeedFetchPool.fetch = fetch + indexerConfig.seed.nntpConnections = connections + } + }) }) diff --git a/tests/unit/xover_client.spec.ts b/tests/unit/xover_client.spec.ts index dad2196..0e80c02 100644 --- a/tests/unit/xover_client.spec.ts +++ b/tests/unit/xover_client.spec.ts @@ -2,6 +2,7 @@ import { test } from '@japa/runner' import net from 'node:net' import { deflateSync } from 'node:zlib' import { XoverClient, parseOverviews } from '#lib/nntp/xover_client' +import { unzipSync } from 'node:zlib' const LINES = [ '101\tShow.S01E01 - "a.mkv" yEnc (1/2)\tposter \tMon, 05 Oct 2026 10:00:00 +0000\t\t\t700000\t5000\tXref: news.example a.b.test:101', @@ -86,4 +87,20 @@ test.group('nntp / XoverClient', () => { close() } }) + + for (const compress of [false, true]) { + test(`hands out overviews as one zlib stream (${compress ? 'as the server sent it' : 'deflated here'})`, async ({ assert }) => { + const { port, close } = await server() + const client = await XoverClient.connect({ host: '127.0.0.1', port, user: 'u', password: 'p', compress }) + try { + await client.group('a.b.test') + const raw = await client.xoverRaw(101, 102) + assert.deepEqual(parseOverviews(unzipSync(raw).toString('utf8')).map(([n]) => n), [101, 102]) + assert.lengthOf(parseOverviews(unzipSync(await client.xoverRaw(1, 1)).toString('utf8')), 0) + } finally { + client.close() + close() + } + }) + } })