diff --git a/app/controllers/seed_controller.ts b/app/controllers/seed_controller.ts index 9b991a6..05b7629 100644 --- a/app/controllers/seed_controller.ts +++ b/app/controllers/seed_controller.ts @@ -39,11 +39,12 @@ export default class SeedController { }) } - /** POST /api/v1/seed/ranges/:id/release { error? } */ + /** POST /api/v1/seed/ranges/:id/release { error?, uncounted? } */ async release({ seedNode, params, request, response }: HttpContext) { const error = request.input('error') + const uncounted = request.input('uncounted') === true return this.leased(response, async () => { - await SeedOrchestrator.release(seedNode.id, Number(params.id), typeof error === 'string' ? error : undefined) + await SeedOrchestrator.release(seedNode.id, Number(params.id), typeof error === 'string' ? error : undefined, uncounted) return { ok: true } }) } diff --git a/app/services/SeedOrchestrator.ts b/app/services/SeedOrchestrator.ts index 4b90cf8..7778051 100644 --- a/app/services/SeedOrchestrator.ts +++ b/app/services/SeedOrchestrator.ts @@ -97,13 +97,18 @@ class SeedOrchestrator { await db.from('seed_ranges').where('id', rangeId).update({ status: 'done', done_at: new Date(), heartbeat_at: new Date(), error: null }) } - /** A node gives a range back (it failed, or is shutting down); it's queued again unless it failed too often. */ - async release(nodeId: number, rangeId: number, error?: string) { + /** + * A node gives a range back (it failed, or is shutting down); it's queued + * again unless it failed too often. `uncounted`: the node never started on + * it (couldn't reach its news server), so the lease doesn't count as an attempt. + */ + async release(nodeId: number, rangeId: number, error?: string, uncounted = false) { const range = await this.leased(nodeId, rangeId) + const attempts = uncounted ? Math.max(0, range.attempts - 1) : range.attempts await db .from('seed_ranges') .where('id', rangeId) - .update({ status: range.attempts >= MAX_ATTEMPTS ? 'failed' : 'queued', node_id: null, error: error?.slice(0, 2000) ?? null }) + .update({ status: attempts >= MAX_ATTEMPTS ? 'failed' : 'queued', node_id: null, attempts, error: error?.slice(0, 2000) ?? null }) } private async leased(nodeId: number, rangeId: number) { diff --git a/commands/SeedNode.ts b/commands/SeedNode.ts index c0d8900..c7ba69f 100644 --- a/commands/SeedNode.ts +++ b/commands/SeedNode.ts @@ -84,33 +84,40 @@ export default class SeedNode extends BaseCommand { private async work(api: SeedApi, worker: number) { const log = (message: string) => this.logger.info(`[${worker}] ${message}`) - // 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. + // The NNTP connection is opened for a lease and kept while leases keep coming; with + // nothing to lease it's closed, so an idle node holds no connections. 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() + const lease = await this.lease(api, log) + if (!lease) { + client?.close() client = undefined + if (this.once) return + await this.pause(60_000) + continue } if (!client) { try { client = await this.connect() } catch (error: any) { + // Never started: the lease doesn't count as one of the range's attempts. const ipLimit = /simultaneous IP/i.test(error.message) log(`NNTP login failed: ${error.message}; trying again in ${ipLimit ? 5 : 1} min`) + await api + .post(`/api/v1/seed/ranges/${lease.id}/release`, { + error: `NNTP login: ${error.message}`, + uncounted: true, + }) + .catch(() => {}) await this.pause(ipLimit ? 300_000 : 60_000) continue } } - const outcome = await this.range(api, client, log) - if (outcome === 'replace') { + if (!(await this.workRange(api, client, lease, log))) { client.close() client = undefined } - if (outcome !== 'idle') idleSince = Date.now() } } finally { client?.close() @@ -130,30 +137,24 @@ export default class SeedNode extends BaseCommand { }) } - /** - * 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 + /** The next range, or null when there is none (or the API can't be reached; logged). */ + private async lease(api: SeedApi, log: (message: string) => void): Promise { try { - lease = ((await api.post('/api/v1/seed/lease')) as { range: Lease } | null)?.range ?? null + return ((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' + return null } + } + /** Works one leased range; false when the NNTP connection should be replaced. */ + private async workRange( + api: SeedApi, + client: XoverClient, + lease: Lease, + log: (message: string) => void + ): Promise { log(`${lease.group} ${lease.start}–${lease.end}`) const started = Date.now() let headers = 0 @@ -178,7 +179,7 @@ export default class SeedNode extends BaseCommand { 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) + await api.post(`/api/v1/seed/ranges/${lease.id}/batches`, batch) headers += batch.headers }) chain = post @@ -193,13 +194,13 @@ export default class SeedNode extends BaseCommand { log( `done: ${headers.toLocaleString('en')} headers in ${seconds.toFixed(0)} s (${Math.round(headers / seconds).toLocaleString('en')}/s)` ) - return 'done' + return true } 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' + return false } const reason = error instanceof Stopping ? 'node stopping' : error.message if (!(error instanceof Stopping)) log(`Range ${lease.id} failed: ${reason}`) @@ -207,7 +208,7 @@ export default class SeedNode extends BaseCommand { 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' + return false } } diff --git a/tests/functional/seed_api.spec.ts b/tests/functional/seed_api.spec.ts index 60c1a63..add6f05 100644 --- a/tests/functional/seed_api.spec.ts +++ b/tests/functional/seed_api.spec.ts @@ -88,4 +88,15 @@ test.group('Seed API', (group) => { } ;(await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${token}` })).assertStatus(200) }) + + test('a range given back before it started does not use up an attempt', async ({ client, assert }) => { + const token = await SeedOrchestrator.issueToken('no-news-server') + 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 + ;(await client.post(`/api/v1/seed/ranges/${range.id}/release`).headers(auth).json({ error: 'NNTP login: 502', uncounted: true })).assertStatus(200) + const row = await db.from('seed_ranges').where('id', range.id).first() + assert.equal(row.status, 'queued') + assert.equal(row.attempts, 0) + }) })