From 39480c486f7b5e5cc4ba712df84c1ca83e95a89a Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Sun, 18 Oct 2026 11:20:09 +0200 Subject: [PATCH] NNTP shares: the account's connections shared by priority The coordinator (38) and the prober (8) had a fixed split; with history paused the coordinator's stood idle while the prober's were its limit. nntp_shares: each process syncs every 15 s (priority, minimum, what it could use, what it has open) and gets what it may open: minimums first, then by priority what each wants, growing only into connections the others closed or gave up. The pool's size and the prober's lanes follow the grant; the coordinator doubles its want while work waits for a connection, else asks for its peak in use and a margin; the prober wants its maximum unless the queue is dry. Claims scale with the lanes. Co-Authored-By: Claude Opus 5.5 --- app/services/nntp_pool.ts | 65 ++++++-- app/services/nntp_shares.ts | 90 +++++++++++ app/services/prefetcher.ts | 5 + app/services/prober.ts | 153 +++++++++++------- commands/coordinator_run.ts | 20 ++- config/coordinator.ts | 19 ++- .../migrations/1770000004400_nntp_shares.ts | 26 +++ deploy/compose.yaml | 13 +- start/env.ts | 4 + tests/functional/nntp_shares.spec.ts | 53 ++++++ 10 files changed, 373 insertions(+), 75 deletions(-) create mode 100644 app/services/nntp_shares.ts create mode 100644 database/migrations/1770000004400_nntp_shares.ts create mode 100644 tests/functional/nntp_shares.spec.ts diff --git a/app/services/nntp_pool.ts b/app/services/nntp_pool.ts index 631f361..a7d7e98 100644 --- a/app/services/nntp_pool.ts +++ b/app/services/nntp_pool.ts @@ -2,12 +2,17 @@ import env from '#start/env' import logger from '@adonisjs/core/services/logger' import coordinatorConfig from '#config/coordinator' import { XoverClient } from '#lib/nntp/index' +import type { NntpShare } from '#services/nntp_shares' + +/** Connections wanted beyond the most in use at once since the last sync (the planner's, a hedge's). */ +const SHARE_MARGIN = 2 /** * The coordinator's header connections (REBUILD.md, decision e: workers * have no NNTP account). At most NNTP_CONNECTIONS, reused; a connection * that failed, or is older than NNTP_CONNECTION_MAX_MINUTES, is closed and - * replaced on the next use. + * replaced on the next use. Sharing the account (share()), at most its + * share: over it, connections close as they come free. */ class NntpPool { private idle: XoverClient[] = [] @@ -17,9 +22,40 @@ class NntpPool { private openedAt = new Map() /** When a connection gets closed once free: its max age ±25%, so they don't all reconnect at once. */ private retireAt = new Map() + private limit = coordinatorConfig.nntpConnections + /** Since the last sync: the most connections in use at once, and whether work waited for one (starved()). */ + private peak = 0 + private short = false get size() { - return coordinatorConfig.nntpConnections + return this.limit + } + + /** Work that could use another connection didn't get one (the prefetcher with all its loops running). */ + starved() { + this.short = true + } + + /** + * Tells the share what the pool could use and has open, and takes the share's grant as its size. After work waited: + * twice its size (history going on gets its maximum in a few syncs; a tail round, done in seconds, doesn't take + * the prober's connections for one); else the most in use at once since the last sync and a margin. + */ + async share(share: NntpShare) { + const want = this.short || this.waiting.length > 0 ? Math.max(this.limit * 2, this.limit + SHARE_MARGIN) : this.peak + SHARE_MARGIN + this.short = false + this.peak = this.busy.size + const granted = await share.sync(want, this.open) + const was = this.limit + this.limit = granted + // Over the share: the idle ones close now, the busy ones once free (release()). + while (this.open > this.limit && this.idle.length > 0) this.retire(this.idle.pop()!) + // Under it with work waiting: connections for it. + while (this.waiting.length > 0 && this.open < this.limit) { + const next = this.waiting.shift()! + this.acquire().then(next, () => this.waiting.unshift(next)) + } + return { was, granted } } async use(work: (client: XoverClient) => Promise): Promise { @@ -44,6 +80,7 @@ class NntpPool { private async run(client: XoverClient, work: (client: XoverClient) => Promise): Promise { this.busy.add(client) + this.peak = Math.max(this.peak, this.busy.size) let broken = false try { return await work(client) @@ -84,15 +121,12 @@ class NntpPool { } private release(client: XoverClient, broken: boolean) { + if (this.open > this.limit) { + this.retire(client) + return + } if (broken || Date.now() >= (this.retireAt.get(client) ?? Infinity)) { - this.open-- - this.openedAt.delete(client) - this.retireAt.delete(client) - try { - client.close() - } catch (error: any) { - logger.debug(`Closing a failed NNTP connection: ${error.message}`) - } + this.retire(client) // Someone waiting gets a fresh connection. const next = this.waiting.shift() if (next) this.acquire().then(next, () => this.waiting.unshift(next)) @@ -103,6 +137,17 @@ class NntpPool { else this.idle.push(client) } + private retire(client: XoverClient) { + this.open-- + this.openedAt.delete(client) + this.retireAt.delete(client) + try { + client.close() + } catch (error: any) { + logger.debug(`Closing an NNTP connection: ${error.message}`) + } + } + /** Closes every connection, also those in use: their commands fail. */ private plain: Promise | null = null private plainQueue: Promise = Promise.resolve() diff --git a/app/services/nntp_shares.ts b/app/services/nntp_shares.ts new file mode 100644 index 0000000..9e9b46e --- /dev/null +++ b/app/services/nntp_shares.ts @@ -0,0 +1,90 @@ +import db from '@adonisjs/lucid/services/db' +import coordinatorConfig from '#config/coordinator' + +/** A share whose process hasn't synced for this long is gone: its connections closed with it. */ +const STALE_SECONDS = 60 +/** Syncs one at a time (pg_advisory_xact_lock). */ +const LOCK = 2_026_101_875 + +export interface Share { + name: string + priority: number + /** Kept whatever the others want (as far as the total goes, by priority). */ + min: number + /** What it could use now. */ + want: number + /** Connections it has open (or may have: a lane without one yet counts). */ + open: number +} + +/** What each share should have: its minimum first, then by priority what it wants, up to the total. */ +export function targets(total: number, shares: Share[]): Map { + const ordered = [...shares].sort((a, b) => b.priority - a.priority || a.name.localeCompare(b.name)) + const target = new Map() + let left = total + for (const s of ordered) { + const t = Math.min(s.min, left) + target.set(s.name, t) + left -= t + } + for (const s of ordered) { + const more = Math.min(Math.max(s.want - target.get(s.name)!, 0), left) + target.set(s.name, target.get(s.name)! + more) + left -= more + } + return target +} + +/** + * The connections `self` may have: its target, but no more than the others leave over. Another share holds what it + * has open or was granted, whichever is more (it may open up to its grant until its next sync tells it otherwise), + * so the account never goes over the total: a share grows only into what the others closed or gave up. + */ +export function grant(total: number, self: Share, others: (Share & { granted: number })[]): number { + const target = targets(total, [self, ...others]).get(self.name)! + const held = others.reduce((sum, o) => sum + Math.max(o.open, o.granted), 0) + return Math.max(0, Math.min(target, total - held)) +} + +/** + * A process's share of the NNTP account's connections (LOG.md 2026-10-18 (75)): the coordinator and the prober had a + * fixed split, and with history paused the coordinator's 38 stood idle while the prober's 8 were its limit. Each + * syncs every nntpShares.everyMs (what it wants and has open) and gets what it may open; one wanting less closes + * connections as they come free, and the other grows as those show up closed. + */ +export class NntpShare { + granted = 0 + + constructor( + readonly name: string, + private priority: number, + readonly min: number, + readonly max: number + ) {} + + async sync(want: number, open: number): Promise { + const self: Share = { name: this.name, priority: this.priority, min: this.min, want: Math.min(Math.max(want, this.min), this.max), open } + const total = coordinatorConfig.nntpShares.total + this.granted = await db.transaction(async (trx) => { + await trx.rawQuery(`SELECT pg_advisory_xact_lock(?)`, [LOCK]) + const { rows } = await trx.rawQuery( + `SELECT name, priority, min, want, open, granted FROM nntp_shares WHERE name <> ? AND heartbeat_at > now() - make_interval(secs => ?)`, + [this.name, STALE_SECONDS] + ) + const granted = grant(total, self, rows) + await trx.rawQuery( + `INSERT INTO nntp_shares (name, priority, min, want, open, granted, heartbeat_at) VALUES (?, ?, ?, ?, ?, ?, now()) + ON CONFLICT (name) DO UPDATE SET priority = excluded.priority, min = excluded.min, want = excluded.want, + open = excluded.open, granted = excluded.granted, heartbeat_at = now()`, + [this.name, this.priority, this.min, self.want, open, granted] + ) + return granted + }) + return this.granted + } + + /** Stopping: its connections are closed, the others needn't wait for the row to go stale. */ + async leave() { + await db.from('nntp_shares').where('name', this.name).delete() + } +} diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index 3a3a41d..846a042 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -131,6 +131,8 @@ export class Prefetcher { .catch((error) => logger.warn(`Prefetching: ${error.message}`)) .finally(() => this.running--) } + // Work left with every loop running: the pool's share could be bigger. + if (!this.stopped && this.running >= this.loops && this.peek()) NntpPool.starved() } stop() { @@ -145,6 +147,9 @@ export class Prefetcher { /** One connection's loop: chunks until there is nothing left to fetch. */ private async work() { while (!this.stopped) { + // The pool's share got smaller: fewer loops (waiting on the pool, they'd count as work short of a connection). + // Two leaving at once is one too many for a tick (TICK_MS), whose loop comes back. + if (this.running > this.loops) return const next = this.take() if (!next) return const { target, span } = next diff --git a/app/services/prober.ts b/app/services/prober.ts index a07269d..650c861 100644 --- a/app/services/prober.ts +++ b/app/services/prober.ts @@ -6,6 +6,7 @@ import coordinatorConfig from '#config/coordinator' import { XoverClient } from '#lib/nntp/index' import { YencError } from '#lib/yenc/index' import { inspect, innerName, PROBE_VERSION, type Inspection } from '#lib/probe/index' +import { NntpShare } from '#services/nntp_shares' interface Candidate { id: number @@ -46,6 +47,12 @@ export class Prober { private refilling: Promise | null = null private stopped = false private stats = Prober.freshStats() + /** Lanes running (each one connection, once it has work); at most the share's grant. Counted apart from the set: a lane leaving takes itself off at once. */ + private lanes = new Set>() + private running = 0 + /** The last lane to look found nothing to claim: the prober wants its minimum only. */ + private dry = false + private share = new NntpShare('prober', coordinatorConfig.nntpShares.proberPriority, coordinatorConfig.probe.minConnections, coordinatorConfig.probe.connections) /** Per minute: what was read, and where the lanes' time went (ms summed over lanes; flushes ms on their own). */ private static freshStats() { @@ -56,8 +63,7 @@ export class Prober { // A previous run's claims: one prober at a time. const { rowCount } = await db.rawQuery(`UPDATE probe_candidates SET status = 'waiting' WHERE status = 'taken'`) if (rowCount) logger.info(`Probing: ${rowCount} candidates from a previous run back in the queue.`) - const connections = coordinatorConfig.probe.connections - logger.info(`Probing with ${connections} connections.`) + logger.info(`Probing with ${this.share.min}–${this.share.max} connections (its share of the account's).`) const flusher = (async () => { while (!this.stopped || this.outcomes.length > 0) { await sleep(FLUSH_MS) @@ -67,7 +73,26 @@ export class Prober { this.report() } })() - await Promise.all(Array.from({ length: connections }, () => this.lane())) + // Lanes as the share allows: the coordinator's history comes first (app/services/nntp_shares.ts). + while (!this.stopped) { + try { + const was = this.share.granted + const granted = await this.share.sync(this.dry ? 0 : this.share.max, this.running) + if (granted !== was) logger.info(`Probing: ${granted} connections (was ${was}).`) + while (this.running < granted) { + this.running++ + const lane: Promise = this.lane() + .catch((error) => logger.error(`Probing: a lane failed: ${error.message}`)) + .finally(() => this.lanes.delete(lane)) + this.lanes.add(lane) + } + } catch (error: any) { + logger.error(`Probing: NNTP share failed: ${error.message}`) + } + for (let waited = 0; waited < coordinatorConfig.nntpShares.everyMs && !this.stopped; waited += 1000) await sleep(1000) + } + await Promise.all(this.lanes) + await this.share.leave().catch(() => {}) // What was claimed and not probed goes back. for (const candidate of this.queue.splice(0)) this.outcomes.push({ candidate, status: 'waiting' }) this.stopped = true @@ -78,68 +103,82 @@ export class Prober { this.stopped = true } - /** One connection's loop: PROBE_PIPELINE articles at a time (pipelined), reconnecting after a failure. */ + /** + * One connection's loop: PROBE_PIPELINE articles at a time (pipelined), reconnecting after a failure; done when + * there are more lanes than the share (its connection closed, the coordinator takes it on its next sync). + */ private async lane() { let client: XoverClient | null = null - while (!this.stopped) { - const batch: Candidate[] = [] - const waited = Date.now() - while (batch.length < coordinatorConfig.probe.pipeline) { - const candidate = await this.next() - if (!candidate) break - batch.push(candidate) - } - this.stats.waitMs += Date.now() - waited - if (batch.length === 0) { - this.stats.idleMs += 30_000 - await sleep(30_000) - continue - } - try { - client ??= 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: false, - timeoutMs: coordinatorConfig.nntpTimeoutMs, - }) - const fetched = Date.now() - const bodies = await client.bodies(batch.map((c) => c.messageId)) - this.stats.fetchMs += Date.now() - fetched - const inspected = Date.now() - for (const [i, candidate] of batch.entries()) { - const body = bodies[i] - this.stats.articles++ - if (body === null) { - // Taken down or expired: final. - this.outcomes.push({ candidate, status: 'missing' }) - continue + try { + while (!this.stopped) { + if (this.running > this.share.granted) break + const batch: Candidate[] = [] + const waited = Date.now() + while (batch.length < coordinatorConfig.probe.pipeline) { + const candidate = await this.next() + if (!candidate) break + batch.push(candidate) + } + this.stats.waitMs += Date.now() - waited + this.dry = batch.length === 0 + if (batch.length === 0) { + this.stats.idleMs += 30_000 + await sleep(30_000) + continue + } + try { + client ??= 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: false, + timeoutMs: coordinatorConfig.nntpTimeoutMs, + }) + const fetched = Date.now() + const bodies = await client.bodies(batch.map((c) => c.messageId)) + this.stats.fetchMs += Date.now() - fetched + const inspected = Date.now() + for (const [i, candidate] of batch.entries()) { + const body = bodies[i] + this.stats.articles++ + if (body === null) { + // Taken down or expired: final. + this.outcomes.push({ candidate, status: 'missing' }) + continue + } + this.stats.bytes += body.length + try { + this.outcomes.push({ candidate, status: 'done', inspection: inspect(body) }) + } catch (error: any) { + // Not yEnc, a broken one, or data a parser chokes on: a final answer, not a connection problem. + if (!(error instanceof YencError)) logger.warn(`Probing ${candidate.messageId}: ${error.message}`) + this.outcomes.push({ candidate, status: 'error' }) + } } - this.stats.bytes += body.length - try { - this.outcomes.push({ candidate, status: 'done', inspection: inspect(body) }) - } catch (error: any) { - // Not yEnc, a broken one, or data a parser chokes on: a final answer, not a connection problem. - if (!(error instanceof YencError)) logger.warn(`Probing ${candidate.messageId}: ${error.message}`) - this.outcomes.push({ candidate, status: 'error' }) - } + this.stats.inspectMs += Date.now() - inspected + } catch (error: any) { + logger.warn(`Probing ${batch[0].messageId} and ${batch.length - 1} more: ${error.message}`) + for (const candidate of batch) this.outcomes.push({ candidate, status: 'waiting' }) + client?.close() + client = null + await sleep(5000) } - this.stats.inspectMs += Date.now() - inspected - } catch (error: any) { - logger.warn(`Probing ${batch[0].messageId} and ${batch.length - 1} more: ${error.message}`) - for (const candidate of batch) this.outcomes.push({ candidate, status: 'waiting' }) - client?.close() - client = null - await sleep(5000) } + } finally { + client?.close() + this.running-- } - client?.close() + } + + /** Candidates claimed at a time: a few rounds of every lane's pipeline (a claim takes seconds; 400 lasted ~2 s at 37 lanes). */ + private get batch() { + return Math.max(coordinatorConfig.probe.batch, this.running * coordinatorConfig.probe.pipeline * 4) } private async next(): Promise { - if (this.queue.length < coordinatorConfig.probe.batch / 2) { + if (this.queue.length < this.batch / 2) { this.refilling ??= this.claim().finally(() => (this.refilling = null)) if (this.queue.length === 0) await this.refilling } @@ -163,7 +202,7 @@ export class Prober { private async claimTiers() { for (const { kind, unreadable } of TIERS) { - const want = coordinatorConfig.probe.batch - this.queue.length + const want = this.batch - this.queue.length if (want <= 0) return // The starting points chosen here: in the query Postgres drew random() once for all slices. // Over the claimable ones of this tier only: the rest of the queue isn't settled yet. diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index 55e2f4f..e8ecace 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -8,6 +8,7 @@ import Settler from '#services/settler' import RangeBlocks from '#services/range_blocks' import Leases from '#services/leases' import NntpPool from '#services/nntp_pool' +import { NntpShare } from '#services/nntp_shares' import coordinatorConfig from '#config/coordinator' const TICK_MS = 2000 @@ -43,6 +44,7 @@ export default class CoordinatorRun extends BaseCommand { const [index, count] = (this.shard ?? '1/1').split('/').map(Number) if (!(count >= 1 && index >= 1 && index <= count)) throw new Error(`--shard: "i/n" with 1 <= i <= n, not "${this.shard}"`) const prefetcher = new Prefetcher({ index: index - 1, count }) + const share = new NntpShare(count > 1 ? `coordinator ${index}/${count}` : 'coordinator', coordinatorConfig.nntpShares.coordinatorPriority, coordinatorConfig.nntpMinConnections, coordinatorConfig.nntpConnections) const main = index === 1 let stopping = false let stopped: () => void @@ -55,6 +57,21 @@ export default class CoordinatorRun extends BaseCommand { await done }) this.logger.info(count > 1 ? `Coordinator started (shard ${index}/${count}${main ? '' : ', prefetching only'}).` : 'Coordinator started.') + // The pool's size: its share of the account's connections, the prober the rest (app/services/nntp_shares.ts). + // Before the first connection: until then it's NNTP_CONNECTIONS. + const first = await NntpPool.share(share) + this.logger.info(`NNTP share: ${first.granted} connections.`) + const sharing = (async () => { + while (!stopping) { + try { + const { was, granted } = await NntpPool.share(share) + if (granted !== was) this.logger.info(`NNTP share: ${granted} connections (was ${was}).`) + } catch (error: any) { + this.logger.error(`NNTP share failed: ${error.message}`) + } + for (let waited = 0; waited < coordinatorConfig.nntpShares.everyMs && !stopping; waited += TICK_MS) await sleep(TICK_MS) + } + })() // Two loops: the prefetcher's ticks don't wait for merging or planning (a merge round // of up to MERGE_BUDGET_MS left the connections idle once their targets were fetched). const prefetching = (async () => { @@ -126,12 +143,13 @@ export default class CoordinatorRun extends BaseCommand { } await sleep(TICK_MS) } - await Promise.all([prefetching, blocks]) + await Promise.all([prefetching, blocks, sharing]) // The chunks in flight, for at most STOP_WAIT_MS (a connection the server holds back can take minutes); then closed. const waitUntil = Date.now() + STOP_WAIT_MS while (prefetcher.active > 0 && Date.now() < waitUntil) await sleep(200) if (prefetcher.active > 0) this.logger.info(`Closing ${prefetcher.active} connections still fetching.`) await NntpPool.close() + await share.leave().catch(() => {}) this.logger.info('Coordinator stopped.') stopped!() await this.terminate() diff --git a/config/coordinator.ts b/config/coordinator.ts index f9f75ec..57ca3a8 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -31,8 +31,21 @@ export default { }, /** 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. */ + /** NNTP connections the coordinator fetches headers with: at most (its share, nntpShares), and at least while sharing. */ nntpConnections: env.get('NNTP_CONNECTIONS', 4), + nntpMinConnections: env.get('NNTP_MIN_CONNECTIONS', 8), + /** + * The account's connections, shared by the coordinator and the prober by priority (app/services/nntp_shares.ts): + * all of them less the reserved ones (the spotter's and the spot inspector's 1 each, the coordinator's plain one, + * and 2 for a hedged request's or a reconnect's new connection while the old one closes). + */ + nntpShares: { + total: env.get('NNTP_ACCOUNT_CONNECTIONS', 50) - env.get('NNTP_RESERVED_CONNECTIONS', 5), + everyMs: 15_000, + /** Headers before probes: history takes what it can use, the prober the rest. */ + coordinatorPriority: 2, + proberPriority: 1, + }, /** * A reply with no data for this long fails, and its connection is replaced. * Eweka sometimes stops answering on long-lived connections (all 40 quiet, @@ -70,11 +83,13 @@ export default { nntpConnectionMaxMs: env.get('NNTP_CONNECTION_MAX_MINUTES', 0) * 60 * 1000, probe: { /** - * NNTP connections the prober reads articles with, besides the + * At most (its share, nntpShares): NNTP connections the prober reads articles with, besides the * coordinator's: the account has 50, the coordinator 40 + 1 plain, and a * couple stay free for downloads. ~7 MB/s each (~10 articles/s). */ connections: env.get('PROBE_CONNECTIONS', 6), + /** Of them, always (its share's minimum; nntpShares). */ + minConnections: env.get('PROBE_MIN_CONNECTIONS', 2), /** Candidates claimed at a time. */ batch: 400, /** BODY commands in flight per connection (pipelined; XoverClient.bodies). */ diff --git a/database/migrations/1770000004400_nntp_shares.ts b/database/migrations/1770000004400_nntp_shares.ts new file mode 100644 index 0000000..f88b66a --- /dev/null +++ b/database/migrations/1770000004400_nntp_shares.ts @@ -0,0 +1,26 @@ +import { BaseSchema } from '@adonisjs/lucid/schema' + +/** + * nntp_shares: the account's connections shared by the processes that read articles (app/services/nntp_shares.ts). + * Each process's row says what it could use and has open, and what it may open; a row without a heartbeat for a + * minute is a process gone (its connections closed with it). The split was fixed per process: with history paused + * the coordinator's 38 stood idle while the prober's 8 were the limit (LOG.md 2026-10-18 (75)). + */ +export default class extends BaseSchema { + async up() { + this.schema.raw(` + CREATE TABLE IF NOT EXISTS nntp_shares ( + name text PRIMARY KEY, + priority integer NOT NULL, + min integer NOT NULL, + want integer NOT NULL, + open integer NOT NULL, + granted integer NOT NULL, + heartbeat_at timestamptz NOT NULL DEFAULT now() + )`) + } + + async down() { + this.schema.raw(`DROP TABLE IF EXISTS nntp_shares`) + } +} diff --git a/deploy/compose.yaml b/deploy/compose.yaml index 32a23f3..9a5dd3c 100644 --- a/deploy/compose.yaml +++ b/deploy/compose.yaml @@ -115,11 +115,14 @@ services: <<: *app-env NNTP_CONNECTIONS: ${NNTP_CONNECTIONS:-38} - # Phase 2's prober: reads the probe queue's articles over PROBE_CONNECTIONS - # connections of its own. The account has 50: the coordinator's 38 (+ its - # plain one), the prober's 8, the spotter's and the spot inspector's 1 each, - # 2 to spare (a hedged request closes its connection and opens another) - # (with 10 the spot inspector got 502 Too many connections). + # Phase 2's prober: reads the probe queue's articles over connections of its + # own. The account has 50 (NNTP_ACCOUNT_CONNECTIONS): 5 reserved (the + # coordinator's plain one, the spotter's and the spot inspector's 1 each, 2 + # to spare: a hedged request closes its connection and opens another; with + # 10 the spot inspector got 502 Too many connections), the other 45 shared + # by the coordinator (NNTP_MIN_CONNECTIONS–NNTP_CONNECTIONS) and the prober + # (PROBE_MIN_CONNECTIONS–PROBE_CONNECTIONS), headers first + # (app/services/nntp_shares.ts). prober: <<: *app command: ['node', 'ace', 'probe:run'] diff --git a/start/env.ts b/start/env.ts index 0b3be12..75f3579 100644 --- a/start/env.ts +++ b/start/env.ts @@ -43,6 +43,9 @@ export default await Env.create(new URL('../', import.meta.url), { NNTP_PASSWORD: Env.schema.string.optional(), NNTP_SECURE: Env.schema.boolean.optional(), NNTP_CONNECTIONS: Env.schema.number.optional(), + NNTP_MIN_CONNECTIONS: Env.schema.number.optional(), + NNTP_ACCOUNT_CONNECTIONS: Env.schema.number.optional(), + NNTP_RESERVED_CONNECTIONS: Env.schema.number.optional(), NNTP_TIMEOUT_MS: Env.schema.number.optional(), NNTP_HEDGE_AFTER_SECONDS: Env.schema.number.optional(), NNTP_HEDGE_CONNECTIONS: Env.schema.number.optional(), @@ -74,6 +77,7 @@ export default await Env.create(new URL('../', import.meta.url), { SPOOL_MAX_GB: Env.schema.number.optional(), PREFETCH_PER_RANGE: Env.schema.number.optional(), PROBE_CONNECTIONS: Env.schema.number.optional(), + PROBE_MIN_CONNECTIONS: Env.schema.number.optional(), HTTP_ROUTES: Env.schema.enum.optional(['all', 'node', 'public'] as const), PREFETCH_PER_GROUP: Env.schema.number.optional(), }) diff --git a/tests/functional/nntp_shares.spec.ts b/tests/functional/nntp_shares.spec.ts new file mode 100644 index 0000000..e633af3 --- /dev/null +++ b/tests/functional/nntp_shares.spec.ts @@ -0,0 +1,53 @@ +import { test } from '@japa/runner' +import db from '@adonisjs/lucid/services/db' +import testUtils from '@adonisjs/core/services/test_utils' +import coordinatorConfig from '#config/coordinator' +import { grant, targets, NntpShare } from '#services/nntp_shares' + +test.group('NNTP shares', (group) => { + group.each.setup(() => testUtils.db().withGlobalTransaction()) + + test('targets: minimums first, then by priority what each wants', ({ assert }) => { + const coordinator = { name: 'coordinator', priority: 2, min: 8, want: 38, open: 0 } + const prober = { name: 'prober', priority: 1, min: 2, want: 40, open: 0 } + // History running: the coordinator all it wants, the prober the rest. + assert.deepEqual(Object.fromEntries(targets(45, [prober, coordinator])), { coordinator: 38, prober: 7 }) + // History paused: the coordinator its minimum (or what it uses), the prober the rest. + assert.deepEqual(Object.fromEntries(targets(45, [prober, { ...coordinator, want: 8 }])), { coordinator: 8, prober: 37 }) + // Wanting more than its minimum doesn't take another's minimum. + assert.deepEqual(Object.fromEntries(targets(45, [prober, { ...coordinator, want: 60 }])), { coordinator: 43, prober: 2 }) + }) + + test('grant: growing only into connections the others closed or gave up', ({ assert }) => { + const prober = { name: 'prober', priority: 1, min: 2, want: 40, open: 37, granted: 37 } + const coordinator = { name: 'coordinator', priority: 2, min: 8, want: 38, open: 8 } + // History came back: the prober still holds 37, the coordinator keeps its 8 for now. + assert.equal(grant(45, coordinator, [prober]), 8) + // The prober syncs: its target is 7, its grant goes down at once. + assert.equal(grant(45, { ...prober, open: 37 }, [{ ...coordinator, granted: 8 }]), 7) + // Its connections not closed yet: still held. + assert.equal(grant(45, coordinator, [{ ...prober, granted: 7, open: 30 }]), 15) + // Closed: the coordinator gets all it wants. + assert.equal(grant(45, coordinator, [{ ...prober, granted: 7, open: 7 }]), 38) + }) + + test('sync: each process its row; a process gone (no heartbeat) holds nothing', async ({ assert }) => { + const total = coordinatorConfig.nntpShares.total + const coordinator = new NntpShare('coordinator', 2, 8, 38) + const prober = new NntpShare('prober', 1, 2, total) + assert.equal(await coordinator.sync(8, 0), 8) + assert.equal(await prober.sync(total, 0), total - 8) + // The coordinator wants all: the prober holds its grant until it has synced and closed them. + assert.equal(await coordinator.sync(38, 8), 8) + assert.equal(await prober.sync(total, total - 8), total - 38) + assert.equal(await coordinator.sync(38, 8), 8) + assert.equal(await prober.sync(total, total - 38), total - 38) + assert.equal(await coordinator.sync(38, 8), 38) + + // The prober's process died: after a minute without a sync its share is free. + await db.from('nntp_shares').where('name', 'prober').update({ open: total, granted: total, heartbeat_at: new Date(Date.now() - 120_000) }) + assert.equal(await coordinator.sync(38, 38), 38) + await coordinator.leave() + assert.equal(await prober.sync(total, 0), total) + }) +})