Daan MeijerandClaude Opus 5.5
a979b5d11c
On homelab (1 Gbit, 20 connections, ~630k articles/s) the coordinator used 1.3-1.7 cores, mostly GC of the text it inflated only to know a compressed reply was complete, plus a copy of every reply. A raw reply is now complete when it ends in ".\r\n" (no copy): 0.4-0.5 cores at the same rate. ".\r\n" can also end a network read inside the compressed data (~1 in 16M reads, a few dozen times over the whole target). Such a chunk doesn't inflate on the worker: it reports it (POST ranges/:id/corrupt), the spooled chunk is removed and the range goes back to the queue; the prefetcher fetches it again. The rest of that reply breaks the connection's next command, so the connection is replaced. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
84 lines
2.5 KiB
TypeScript
84 lines
2.5 KiB
TypeScript
import { mkdir, readFile, readdir, rename, rm, stat, writeFile } from 'node:fs/promises'
|
|
import { join } from 'node:path'
|
|
import coordinatorConfig from '#config/coordinator'
|
|
|
|
export interface ChunkSpan {
|
|
from: number
|
|
to: number
|
|
}
|
|
|
|
/**
|
|
* Prefetched XOVER chunks on disk (REBUILD.md, prefetch): `<dir>/<range id>/<from>-<to>.gz`,
|
|
* each as the news server sent it. Chunks follow a fixed grid from the
|
|
* range's start, so a resume point (always at a batch boundary) is a chunk's start.
|
|
*/
|
|
class Spool {
|
|
get dir() {
|
|
return coordinatorConfig.spool.dir
|
|
}
|
|
|
|
/** The chunks of a range from `from` (its resume point) to its end. */
|
|
chunksOf(range: { start: number; end: number }, from = range.start): ChunkSpan[] {
|
|
const size = coordinatorConfig.chunkArticles
|
|
const spans: ChunkSpan[] = []
|
|
for (let at = range.start; at <= range.end; at += size) {
|
|
const span = { from: at, to: Math.min(at + size - 1, range.end) }
|
|
if (span.to >= from) spans.push(span)
|
|
}
|
|
return spans
|
|
}
|
|
|
|
private path(rangeId: number, span: ChunkSpan) {
|
|
return join(this.dir, String(rangeId), `${span.from}-${span.to}.gz`)
|
|
}
|
|
|
|
async has(rangeId: number, span: ChunkSpan) {
|
|
try {
|
|
await stat(this.path(rangeId, span))
|
|
return true
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
|
|
async write(rangeId: number, span: ChunkSpan, data: Buffer) {
|
|
const file = this.path(rangeId, span)
|
|
await mkdir(join(this.dir, String(rangeId)), { recursive: true })
|
|
// Written under another name first: a reader never sees half a chunk.
|
|
await writeFile(`${file}.part`, data)
|
|
await rename(`${file}.part`, file)
|
|
}
|
|
|
|
read(rangeId: number, span: ChunkSpan) {
|
|
return readFile(this.path(rangeId, span))
|
|
}
|
|
|
|
/** One chunk, e.g. one that doesn't inflate (a false end of reply; see XoverClient.compressedRaw). */
|
|
async removeChunk(rangeId: number, span: ChunkSpan) {
|
|
await rm(this.path(rangeId, span), { force: true })
|
|
}
|
|
|
|
async remove(rangeId: number) {
|
|
await rm(join(this.dir, String(rangeId)), { recursive: true, force: true })
|
|
}
|
|
|
|
/** Bytes in the spool. */
|
|
async size(): Promise<number> {
|
|
let total = 0
|
|
let ranges: string[] = []
|
|
try {
|
|
ranges = await readdir(this.dir)
|
|
} catch {
|
|
return 0
|
|
}
|
|
for (const range of ranges) {
|
|
for (const file of await readdir(join(this.dir, range)).catch(() => [])) {
|
|
total += (await stat(join(this.dir, range, file)).catch(() => ({ size: 0 }))).size
|
|
}
|
|
}
|
|
return total
|
|
}
|
|
}
|
|
|
|
export default new Spool()
|