diff --git a/commands/SeedNode.ts b/commands/SeedNode.ts index 3ad4bec..19b2bee 100644 --- a/commands/SeedNode.ts +++ b/commands/SeedNode.ts @@ -99,6 +99,10 @@ export default class SeedNode extends BaseCommand { 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, @@ -117,12 +121,23 @@ export default class SeedNode extends BaseCommand { 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() - await api.post(`/api/v1/seed/ranges/${lease!.id}/batches`, batch) - headers += batch.headers + 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)`) @@ -134,6 +149,8 @@ export default class SeedNode extends BaseCommand { } 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 {