From 70036847cf9aeef407b1a729c31486f57e8ca770 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 5 Oct 2026 14:14:15 +0200 Subject: [PATCH] Seeding nodes log in before leasing; TLS through tunnels - seed:node logs in to NNTP before taking a range and keeps the connection across ranges (replaced after 2 minutes idle or a failed range). A node that can't reach its news server takes no range: two ranges were set aside as failed while an office node's tunnel was being set up (ECONNREFUSED 127.0.0.1:1563, five leases each). After an IP-address limit refusal it waits 5 minutes. - NNTP_TLS_SERVERNAME: the hostname the TLS certificate is checked against when NNTP_HOST is a tunnel ("Hostname/IP does not match certificate's altnames: Host: localhost"). - seed-node/README: the IP-address limit and tunnelling. Co-Authored-By: Claude Opus 5.5 --- commands/SeedNode.ts | 216 +++++++++++++++++++++++-------------- commands/SeedRange.ts | 1 + config/nntp.ts | 2 + lib/nntp/xover_client.ts | 4 +- seed-node/README.md | 14 ++- seed-node/node.env.example | 3 + start/env.ts | 1 + 7 files changed, 156 insertions(+), 85 deletions(-) diff --git a/commands/SeedNode.ts b/commands/SeedNode.ts index 76bf2de..c0d8900 100644 --- a/commands/SeedNode.ts +++ b/commands/SeedNode.ts @@ -29,7 +29,8 @@ interface Lease { */ 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 description = + 'Runs a seeding node: leases ranges from the seed API and submits what it aggregated.' public static options: CommandOptions = { startApp: false, @@ -42,7 +43,11 @@ export default class SeedNode extends BaseCommand { @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 }) + @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 }) @@ -79,83 +84,130 @@ export default class SeedNode extends BaseCommand { 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 - let chain: Promise = Promise.resolve() - const posting = new Set>() - // Throws once a post has failed, so the range stops fetching. - const failedSoFar = () => Promise.race([chain, Promise.resolve()]) - 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'] ?? '')), - // Batches are posted in order, one after the other, while fetching goes on; with two - // waiting, fetching waits too. A failed post stops the chain, so no batch is stored - // past a missing one (the range would resume beyond it). - sink: async (batch: SeedBatch) => { - if (this.stopping) throw new Stopping() - while (posting.size >= 2) await Promise.race(posting) - const post = chain.then(async () => { - await api.post(`/api/v1/seed/ranges/${lease!.id}/batches`, batch) - headers += batch.headers - }) - chain = post - posting.add(post) - post.catch(() => {}).finally(() => posting.delete(post)) - await failedSoFar() - }, - }) - await chain - 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 + // Logged in before leasing, so a node that can't reach its news server takes no range + // (each lease counts as an attempt; a range is set aside after five). Kept across ranges. + let client: XoverClient | undefined + let idleSince = Date.now() + try { + while (!this.stopping) { + // A news server may drop a connection that sat idle (nothing to lease): log in afresh. + if (client && Date.now() - idleSince > 2 * 60_000) { + client.close() + client = undefined } - const reason = error instanceof Stopping ? 'node stopping' : error.message - if (!(error instanceof Stopping)) log(`Range ${lease.id} failed: ${reason}`) - // Let posts in flight finish (or fail) first, so the range resumes after what was stored. - await chain.catch(() => {}) - await api.post(`/api/v1/seed/ranges/${lease.id}/release`, { error: reason }).catch(() => {}) - if (!this.stopping) await this.pause(30_000) - } finally { - client?.close() + if (!client) { + try { + client = await this.connect() + } catch (error: any) { + const ipLimit = /simultaneous IP/i.test(error.message) + log(`NNTP login failed: ${error.message}; trying again in ${ipLimit ? 5 : 1} min`) + await this.pause(ipLimit ? 300_000 : 60_000) + continue + } + } + const outcome = await this.range(api, client, log) + if (outcome === 'replace') { + client.close() + client = undefined + } + if (outcome !== 'idle') idleSince = Date.now() } + } finally { + client?.close() + } + } + + private connect() { + return 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, + }) + } + + /** + * Leases and works one range: 'done', 'idle' (nothing leased; the connection + * wasn't used), or 'replace' (the connection may be in an odd state). + */ + private async range( + api: SeedApi, + client: XoverClient, + log: (message: string) => void + ): Promise<'done' | 'idle' | 'replace'> { + 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) + return 'idle' + } + if (!lease) { + if (this.once) this.stopping = true + else await this.pause(60_000) + return 'idle' + } + + log(`${lease.group} ${lease.start}–${lease.end}`) + const started = Date.now() + let headers = 0 + let chain: Promise = Promise.resolve() + const posting = new Set>() + // Throws once a post has failed, so the range stops fetching. + const failedSoFar = () => Promise.race([chain, Promise.resolve()]) + try { + 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'] ?? '')), + // Batches are posted in order, one after the other, while fetching goes on; with two + // waiting, fetching waits too. A failed post stops the chain, so no batch is stored + // past a missing one (the range would resume beyond it). + sink: async (batch: SeedBatch) => { + if (this.stopping) throw new Stopping() + while (posting.size >= 2) await Promise.race(posting) + const post = chain.then(async () => { + await api.post(`/api/v1/seed/ranges/${lease!.id}/batches`, batch) + headers += batch.headers + }) + chain = post + posting.add(post) + post.catch(() => {}).finally(() => posting.delete(post)) + await failedSoFar() + }, + }) + await chain + 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)` + ) + return 'done' + } catch (error: any) { + if (error instanceof Unauthorized) throw error + // An XOVER may still be in flight: the connection is replaced either way. + if (error instanceof LeaseLost) { + log(`Range ${lease.id} passed to another node; dropping it.`) + return 'replace' + } + const reason = error instanceof Stopping ? 'node stopping' : error.message + if (!(error instanceof Stopping)) log(`Range ${lease.id} failed: ${reason}`) + // Let posts in flight finish (or fail) first, so the range resumes after what was stored. + await chain.catch(() => {}) + await api.post(`/api/v1/seed/ranges/${lease.id}/release`, { error: reason }).catch(() => {}) + if (!this.stopping) await this.pause(30_000) + return 'replace' } } @@ -184,8 +236,8 @@ class SeedApi { const response = await fetch(this.base + path, { method: 'POST', headers: { - 'authorization': `Bearer ${this.token}`, - 'accept': 'application/json', + authorization: `Bearer ${this.token}`, + accept: 'application/json', ...(payload ? { 'content-type': 'application/json', 'content-encoding': 'gzip' } : {}), }, body: payload, @@ -201,9 +253,11 @@ class SeedApi { if (status === 401) throw new Unauthorized('The seed API rejected this node token') 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)}`) + if (status >= 400 && status < 500 && status !== 404) + throw new Error(`Seed API ${path}: ${status} ${text.slice(0, 300)}`) // 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.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/SeedRange.ts b/commands/SeedRange.ts index e4f3568..ade9884 100644 --- a/commands/SeedRange.ts +++ b/commands/SeedRange.ts @@ -60,6 +60,7 @@ export default class SeedRange extends BaseCommand { host: nntpConfig.host, port: nntpConfig.port, secure: nntpConfig.secure, + servername: nntpConfig.tlsServername, user: nntpConfig.user, password: nntpConfig.password?.release(), timeoutMs: nntpConfig.commandTimeoutMs, diff --git a/config/nntp.ts b/config/nntp.ts index 18cb77a..7e5420e 100644 --- a/config/nntp.ts +++ b/config/nntp.ts @@ -10,4 +10,6 @@ export default { poolSize: env.get('NNTP_POOL_SIZE', 10), /** A command without an answer this long counts as a dead connection (XOVER of 20k headers takes ~0.2 s). */ commandTimeoutMs: env.get('NNTP_COMMAND_TIMEOUT_MS', 120_000), + /** TLS certificate hostname when NNTP_HOST is a tunnel (seeding nodes; lib/nntp/xover_client). */ + tlsServername: env.get('NNTP_TLS_SERVERNAME'), } diff --git a/lib/nntp/xover_client.ts b/lib/nntp/xover_client.ts index b56b428..de91ffc 100644 --- a/lib/nntp/xover_client.ts +++ b/lib/nntp/xover_client.ts @@ -8,6 +8,8 @@ export interface XoverClientOptions { user?: string password?: string secure?: boolean + /** Hostname the TLS certificate must match, when connecting through a tunnel (default: host). */ + servername?: string /** A command whose reply stalls this long fails, and the connection with it. */ timeoutMs?: number /** Ask for gzip-compressed overview data (XFEATURE COMPRESS GZIP), when the server has it. */ @@ -58,7 +60,7 @@ export class XoverClient { static async connect(options: XoverClientOptions): Promise { const sock = options.secure - ? tls.connect({ host: options.host, port: options.port, servername: options.host }) + ? tls.connect({ host: options.host, port: options.port, servername: options.servername || options.host }) : net.connect({ host: options.host, port: options.port }) const client = new XoverClient(sock, options.timeoutMs ?? 120_000) sock.setTimeout(client.timeoutMs) diff --git a/seed-node/README.md b/seed-node/README.md index e77f480..5e17825 100644 --- a/seed-node/README.md +++ b/seed-node/README.md @@ -16,8 +16,14 @@ database or Redis. gzip-compressed); a few hundred MB of RAM per node. Disk: about 1.5 GB for the image. - An NNTP account. Every connection a node opens counts against that - account's connection limit, next to the ~40 the indexer itself uses if it's - the same account. Too many gives `481`/`502` errors at login. + account's connection limit, next to the ~20 the indexer itself uses if it's + the same account. Too many gives `481`/`502` errors at login. Some + providers also limit the number of IP addresses at once (Eweka: 2, as + `502 (max number of simultaneous IP addresses reached: 2)`): a node behind + the same public IP as the indexer doesn't count extra, others do. A node + can share an IP by tunnelling its NNTP traffic (e.g. SSH + `-L 1563:news.eweka.nl:563` to the indexer's machine); then set + `NNTP_HOST` to the tunnel and `NNTP_TLS_SERVERNAME=news.eweka.nl`. ## 1. On the indexer (homelab): a token for the node @@ -44,6 +50,7 @@ for the same name replaces the token; `--disable` revokes it. | `SEED_CONNECTIONS` | ranges worked on at once, one NNTP connection each. Start with 2–4. | | `NNTP_HOST`, `NNTP_PORT`, `NNTP_SECURE` | the news server; over the internet use TLS: port `563`, `NNTP_SECURE=true` (for Eweka: `news.eweka.nl`) | | `NNTP_USER`, `NNTP_PASSWORD` | the NNTP account | +| `NNTP_TLS_SERVERNAME` | only through a tunnel: the news server's real hostname, for the TLS certificate check | Leave the block below "Required by the app's configuration, but unused" as it is: the app checks these settings at startup, but a node never uses them. @@ -90,6 +97,7 @@ articles reached, and per node its leases and headers. |---|---| | `The seed API rejected this node token` | wrong or revoked `SEED_TOKEN`; the node stops | | `Seed API …: 404 …; retrying` | the seed API is restarting (a deploy on homelab); the node retries | -| `NNTP: unexpected reply "481 …"` or `"502 …"` | NNTP login refused: wrong account, or too many connections on it; lower `SEED_CONNECTIONS` | +| `NNTP login failed: …` | the node didn't take a range and tries again in a minute (5 minutes for the IP-address limit); `481`/`502`: wrong account, too many connections (lower `SEED_CONNECTIONS`) or too many IP addresses | +| `Hostname/IP does not match certificate's altnames: Host: localhost` | TLS through a tunnel: set `NNTP_TLS_SERVERNAME` to the news server's hostname | | `Range … failed: …` | the range is given back and retried later, possibly by another node | | `Range … passed to another node; dropping it.` | this node was too slow to report (15 minutes); harmless | diff --git a/seed-node/node.env.example b/seed-node/node.env.example index fbe103c..cf92388 100644 --- a/seed-node/node.env.example +++ b/seed-node/node.env.example @@ -10,6 +10,9 @@ 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_TLS_SERVERNAME=news.eweka.nl # --- Required by the app's configuration, but unused on a seeding node (no database, Redis or web server). NODE_ENV=production diff --git a/start/env.ts b/start/env.ts index d015643..ff6146d 100644 --- a/start/env.ts +++ b/start/env.ts @@ -48,6 +48,7 @@ export default await Env.create(new URL('../', import.meta.url), { NNTP_SECURE: Env.schema.boolean(), NNTP_POOL_SIZE: Env.schema.number.optional(), NNTP_COMMAND_TIMEOUT_MS: Env.schema.number.optional(), + NNTP_TLS_SERVERNAME: Env.schema.string.optional(), // Indexer (config/indexer.ts) BACKFILL_MAX_QUEUED: Env.schema.number.optional(),