seed:node posts batches in order beside fetching
At most two posts waiting; a failed post stops the chain, so nothing is stored past a missing batch. The first remote node did 7k headers/s while waiting for every post. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
8c36744088
commit
e2acfae98d
1 file changed
+19
-2
+19
-2
@@ -99,6 +99,10 @@ export default class SeedNode extends BaseCommand {
|
||||
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,
|
||||
@@ -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 {
|
||||
|
||||
Reference in new issue
Block a user