213 lines
8.4 KiB
TypeScript
213 lines
8.4 KiB
TypeScript
import { gzipSync } from 'node:zlib'
|
||
import { setTimeout as sleep } from 'node:timers/promises'
|
||
import { BaseCommand, flags } from '@adonisjs/core/ace'
|
||
import { CommandOptions } from '@adonisjs/core/types/ace'
|
||
import nntpConfig from '#config/nntp'
|
||
import indexerConfig from '#config/indexer'
|
||
import { XoverClient } from '#lib/nntp/xover_client'
|
||
import { runRange } from '#lib/seed/runner'
|
||
import type { SeedBatch } from '#lib/seed/wire'
|
||
import type { OverviewHeader } from '#lib/ingest/segment'
|
||
|
||
class LeaseLost extends Error {}
|
||
class Stopping extends Error {}
|
||
|
||
interface Lease {
|
||
id: number
|
||
group: string
|
||
start: number
|
||
end: number
|
||
}
|
||
|
||
/**
|
||
* A seeding node (CONCEPT.md D38): leases article ranges from the seed API
|
||
* over HTTPS, fetches and aggregates their headers over its own NNTP
|
||
* connection(s) (D37), and posts the results back in gzip-compressed
|
||
* batches. Outgoing requests only: no ports, database or Redis needed.
|
||
*
|
||
* SEED_URL=https://seed.example SEED_TOKEN=sn_… NNTP_HOST=… node ace seed:node --connections 4
|
||
*/
|
||
export default class SeedNode extends BaseCommand {
|
||
public static commandName = 'seed:node'
|
||
public static description = 'Runs a seeding node: leases ranges from the seed API and submits what it aggregated.'
|
||
|
||
public static options: CommandOptions = {
|
||
startApp: false,
|
||
staysAlive: true,
|
||
}
|
||
|
||
@flags.string({ description: 'Seed API base URL (default: SEED_URL)' })
|
||
declare url?: string
|
||
|
||
@flags.string({ description: 'Node token (default: SEED_TOKEN)' })
|
||
declare token?: string
|
||
|
||
@flags.number({ description: 'Ranges worked on at once, one NNTP connection each (default: SEED_CONNECTIONS or 1)', default: Number(process.env.SEED_CONNECTIONS) || 1 })
|
||
declare connections: number
|
||
|
||
@flags.number({ description: 'Articles per XOVER', default: 20_000 })
|
||
declare chunk: number
|
||
|
||
@flags.number({ description: 'XOVER chunks per batch posted to the API', default: 5 })
|
||
declare submitEvery: number
|
||
|
||
@flags.boolean({ description: 'Stop when there is nothing to lease instead of waiting' })
|
||
declare once: boolean
|
||
|
||
private stopping = false
|
||
|
||
public async run() {
|
||
const base = (this.url ?? process.env.SEED_URL ?? '').replace(/\/+$/, '')
|
||
const token = this.token ?? process.env.SEED_TOKEN ?? ''
|
||
if (!base || !token) {
|
||
this.logger.error('Set --url and --token (or SEED_URL and SEED_TOKEN).')
|
||
this.exitCode = 1
|
||
return
|
||
}
|
||
const stop = () => {
|
||
if (!this.stopping) this.logger.info('Stopping: giving back the ranges in progress…')
|
||
this.stopping = true
|
||
}
|
||
process.on('SIGTERM', stop)
|
||
process.on('SIGINT', stop)
|
||
|
||
const api = new SeedApi(base, token, (message) => this.logger.warning(message))
|
||
this.logger.info(`Seeding node for ${base} with ${this.connections} connection(s).`)
|
||
await Promise.all(Array.from({ length: this.connections }, (_, i) => this.work(api, i + 1)))
|
||
this.logger.info('Seeding node stopped.')
|
||
}
|
||
|
||
private async work(api: SeedApi, worker: number) {
|
||
const log = (message: string) => this.logger.info(`[${worker}] ${message}`)
|
||
while (!this.stopping) {
|
||
let lease: Lease | null
|
||
try {
|
||
lease = ((await api.post('/api/v1/seed/lease')) as { range: Lease } | null)?.range ?? null
|
||
} catch (error: any) {
|
||
if (error instanceof Unauthorized) throw error
|
||
log(`Lease failed: ${error.message}`)
|
||
await this.pause(60_000)
|
||
continue
|
||
}
|
||
if (!lease) {
|
||
if (this.once) return
|
||
await this.pause(60_000)
|
||
continue
|
||
}
|
||
|
||
log(`${lease.group} ${lease.start}–${lease.end}`)
|
||
const started = Date.now()
|
||
let headers = 0
|
||
let client: XoverClient | undefined
|
||
let chain: Promise<void> = Promise.resolve()
|
||
const posting = new Set<Promise<void>>()
|
||
// 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
|
||
}
|
||
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()
|
||
}
|
||
}
|
||
}
|
||
|
||
/** Sleeps, but wakes up early when the node is stopping. */
|
||
private async pause(ms: number) {
|
||
for (let waited = 0; waited < ms && !this.stopping; waited += 1000) await sleep(1000)
|
||
}
|
||
}
|
||
|
||
class Unauthorized extends Error {}
|
||
|
||
/** The seed API client: JSON over HTTPS, gzip request bodies, retries for network errors and 5xx. */
|
||
class SeedApi {
|
||
constructor(
|
||
private base: string,
|
||
private token: string,
|
||
private warn: (message: string) => void
|
||
) {}
|
||
|
||
async post(path: string, body?: unknown): Promise<unknown> {
|
||
const payload = body === undefined ? undefined : gzipSync(JSON.stringify(body))
|
||
for (let attempt = 1; ; attempt++) {
|
||
let status = 0
|
||
let text = ''
|
||
try {
|
||
const response = await fetch(this.base + path, {
|
||
method: 'POST',
|
||
headers: {
|
||
'authorization': `Bearer ${this.token}`,
|
||
'accept': 'application/json',
|
||
...(payload ? { 'content-type': 'application/json', 'content-encoding': 'gzip' } : {}),
|
||
},
|
||
body: payload,
|
||
signal: AbortSignal.timeout(10 * 60 * 1000),
|
||
})
|
||
status = response.status
|
||
text = await response.text()
|
||
} catch (error: any) {
|
||
text = error.message
|
||
}
|
||
if (status === 204) return null
|
||
if (status >= 200 && status < 300) return text ? JSON.parse(text) : null
|
||
if (status === 401) throw new Unauthorized('The seed API rejected this node token')
|
||
if (status === 409) throw new LeaseLost(text)
|
||
// 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)}`)
|
||
// 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)}`)
|
||
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)
|
||
}
|
||
}
|
||
}
|