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>
81 lines
3.9 KiB
TypeScript
81 lines
3.9 KiB
TypeScript
import db from '@adonisjs/lucid/services/db'
|
|
import coordinatorConfig from '#config/coordinator'
|
|
import RuleSets from '#services/rule_sets'
|
|
import Spool from '#services/spool'
|
|
import type { Lease } from '#lib/edge/index'
|
|
|
|
export class LeaseLost extends Error {}
|
|
|
|
/**
|
|
* Ranges for workers (REBUILD.md, coordinator): the next range with chunks
|
|
* spooled, by priority (oldest era first), else the next planned one (the
|
|
* prefetcher fetches leased ranges first; the worker's stream waits). A
|
|
* worker never waits for prefetching to choose its range: refused leases
|
|
* left workers idle for 30 s, so fewer ranges were leased and fewer
|
|
* prefetched (LOG.md 2026-10-09). A lease whose worker went quiet for
|
|
* LEASE_TTL_SECONDS is taken over and resumes at its resume point.
|
|
*/
|
|
class Leases {
|
|
async lease(nodeId: number): Promise<Lease | null> {
|
|
const rules = await RuleSets.current()
|
|
const take = (spooled: boolean) =>
|
|
db.rawQuery(
|
|
`UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ?
|
|
WHERE r.id = (
|
|
SELECT id FROM ranges
|
|
WHERE ${spooled ? 'spooled_chunks > 0' : 'true'} AND next_id <= end_id
|
|
AND (status = 'planned' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)))
|
|
ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED)
|
|
RETURNING r.id, r.group_id, r.start_id, r.end_id, r.next_id`,
|
|
[nodeId, rules.version, coordinatorConfig.leaseTtlSeconds]
|
|
)
|
|
let { rows } = await take(true)
|
|
if (rows.length === 0) ({ rows } = await take(false))
|
|
const row = rows[0]
|
|
if (!row) return null
|
|
await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
|
|
const group = await db.from('groups').where('id', row.group_id).select('name').firstOrFail()
|
|
return { range: { id: Number(row.id), group: group.name, start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id) }, rules }
|
|
}
|
|
|
|
async leased(rangeId: number, nodeId: number) {
|
|
const range = await db.from('ranges').where('id', rangeId).first()
|
|
if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new LeaseLost(`Range ${rangeId} is not leased to this node`)
|
|
return range
|
|
}
|
|
|
|
async heartbeat(rangeId: number, nodeId: number) {
|
|
await this.leased(rangeId, nodeId)
|
|
await db.from('ranges').where('id', rangeId).update({ heartbeat_at: new Date() })
|
|
await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
|
|
}
|
|
|
|
/** The worker reached the range's end: done, and its spooled chunks go. */
|
|
async finish(rangeId: number, nodeId: number) {
|
|
const range = await this.leased(rangeId, nodeId)
|
|
if (Number(range.next_id) <= Number(range.end_id)) throw new LeaseLost(`Range ${rangeId} is at ${range.next_id}, not past its end ${range.end_id}`)
|
|
await db.from('ranges').where('id', rangeId).update({ status: 'done', done_at: new Date(), heartbeat_at: new Date() })
|
|
await Spool.remove(rangeId)
|
|
}
|
|
|
|
/**
|
|
* A worker got a chunk that doesn't inflate: the spooled chunk goes and
|
|
* the range back to the queue (the prefetcher fetches leased ranges'
|
|
* missing chunks again; the next worker resumes at the range's resume point).
|
|
*/
|
|
async corrupt(rangeId: number, nodeId: number, from: number) {
|
|
const range = await this.leased(rangeId, nodeId)
|
|
const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find((s) => s.from === from)
|
|
if (span) await Spool.removeChunk(rangeId, span)
|
|
await db.from('ranges').where('id', rangeId).update({ status: 'planned', node_id: null })
|
|
}
|
|
|
|
/** A stopping worker gives its range back; the next one resumes it. */
|
|
async release(rangeId: number, nodeId: number) {
|
|
await this.leased(rangeId, nodeId)
|
|
await db.from('ranges').where('id', rangeId).update({ status: 'planned', node_id: null })
|
|
}
|
|
}
|
|
|
|
export default new Leases()
|