Daan MeijerandClaude Opus 5.5
d5f168d8d2
- groups:add / ranges:plan / nodes:create / coordinator:run commands - the node API: lease, the range's chunks as one framed stream from the spool, batches, heartbeat, finish, release (node guard, token per node) - edge:node: needs COORDINATOR_URL and NODE_TOKEN only (placeholders for the rest), posts a batch every 10 chunks, gives its range back on SIGTERM - releases are gathered per batch: 1.04 chunk rows per release at 10 chunks per batch (1.35 when built per XOVER chunk) - partitions are checked per batch instead of remembered Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
79 lines
2.3 KiB
TypeScript
79 lines
2.3 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))
|
|
}
|
|
|
|
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()
|