diff --git a/app/services/SeedFetchPool.ts b/app/services/SeedFetchPool.ts index 45eabfa..8e03b90 100644 --- a/app/services/SeedFetchPool.ts +++ b/app/services/SeedFetchPool.ts @@ -61,7 +61,7 @@ class SeedFetchPool { servername: nntpConfig.tlsServername, user: nntpConfig.user, password: nntpConfig.password?.release(), - timeoutMs: nntpConfig.commandTimeoutMs, + timeoutMs: indexerConfig.seed.nntpTimeoutMs, compress: true, }) return { client, usedAt: Date.now() } diff --git a/app/services/SeedOrchestrator.ts b/app/services/SeedOrchestrator.ts index 13efaad..b6dc703 100644 --- a/app/services/SeedOrchestrator.ts +++ b/app/services/SeedOrchestrator.ts @@ -60,6 +60,12 @@ class SeedOrchestrator { */ async lease(nodeId: number): Promise { if ((await BackfillService.room()) !== true) return null + // Fully reported but never finished (the node went away at the very end): nothing left to fetch. + await db.rawQuery( + `UPDATE seed_ranges SET status = 'done', done_at = now(), node_id = NULL + WHERE next_id > end_id AND (status = 'queued' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)))`, + [LEASE_TTL_SECONDS] + ) // Ranges that broke node after node stop being handed out. await db.rawQuery( `UPDATE seed_ranges SET status = 'failed', error = coalesce(error, 'lease expired too often') diff --git a/config/indexer.ts b/config/indexer.ts index 733104c..acc709b 100644 --- a/config/indexer.ts +++ b/config/indexer.ts @@ -66,6 +66,8 @@ export default { seed: { /** NNTP connections the seed API fetches with for relay nodes (CONCEPT.md D38); 0: relay off. */ nntpConnections: env.get('SEED_NNTP_CONNECTIONS', 0), + /** A relay XOVER may take this long: some old ranges come at ~1k headers/s (and slower when cold). */ + nntpTimeoutMs: env.get('SEED_NNTP_TIMEOUT_MS', 10 * 60 * 1000), }, backfill: { diff --git a/lib/seed/runner.ts b/lib/seed/runner.ts index d23facb..08e8cb5 100644 --- a/lib/seed/runner.ts +++ b/lib/seed/runner.ts @@ -39,6 +39,11 @@ export async function runRange(options: RunRangeOptions): Promise { const idle = options.idle ?? 5 const submitEvery = options.submitEvery ?? 5 const aggregator = new FileAggregator() + // A range fully reported before but not finished (its node went away right at the end) resumes past its end. + if (options.start > options.end) { + await options.sink(emptyBatch(options.end + 1)) + return + } await options.client.group(options.group) const fetch = (from: number) => { diff --git a/start/env.ts b/start/env.ts index 8719642..bfc4852 100644 --- a/start/env.ts +++ b/start/env.ts @@ -50,6 +50,7 @@ export default await Env.create(new URL('../', import.meta.url), { NNTP_COMMAND_TIMEOUT_MS: Env.schema.number.optional(), NNTP_TLS_SERVERNAME: Env.schema.string.optional(), SEED_NNTP_CONNECTIONS: Env.schema.number.optional(), + SEED_NNTP_TIMEOUT_MS: Env.schema.number.optional(), // Indexer (config/indexer.ts) BACKFILL_MAX_QUEUED: Env.schema.number.optional(), diff --git a/tests/functional/seed_api.spec.ts b/tests/functional/seed_api.spec.ts index 3f22ff7..3769b91 100644 --- a/tests/functional/seed_api.spec.ts +++ b/tests/functional/seed_api.spec.ts @@ -128,4 +128,11 @@ test.group('Seed API', (group) => { indexerConfig.seed.nntpConnections = connections } }) + + test('a range reported to its end but never finished is marked done, not leased', async ({ client, assert }) => { + const token = await SeedOrchestrator.issueToken('late-node') + const [range] = await db.table('seed_ranges').insert({ group_name: 'alt.binaries.test', start_id: 1, end_id: 100, next_id: 101 }).returning('id') + ;(await client.post('/api/v1/seed/lease').headers({ authorization: `Bearer ${token}` })).assertStatus(204) + assert.equal((await db.from('seed_ranges').where('id', range.id).first()).status, 'done') + }) }) diff --git a/tests/unit/aggregate.spec.ts b/tests/unit/aggregate.spec.ts index e1f7c60..fabeada 100644 --- a/tests/unit/aggregate.spec.ts +++ b/tests/unit/aggregate.spec.ts @@ -37,3 +37,20 @@ test.group('aggregate / FileAggregator', () => { assert.equal(aggregator.openFiles, 0) }) }) + +test.group('seed / runRange', () => { + test('a range already past its end only reports that it is done', async ({ assert }) => { + const { runRange } = await import('#lib/seed/runner') + const batches: any[] = [] + await runRange({ + client: { group: async () => assert.fail('no GROUP'), xover: async () => assert.fail('no XOVER') }, + group: 'a.b.test', + start: 101, + end: 100, + sink: async (batch) => { + batches.push(batch) + }, + }) + assert.deepEqual(batches.map((b) => [b.next, b.headers, b.files.length]), [[101, 0, 0]]) + }) +})