From 4af5e06400266722ac35e3316fad884e9d58b533 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 26 Oct 2026 23:26:31 +0100 Subject: [PATCH] LEASE_CAPTURE_ONLY: group leases of the corpus windows' ranges only The lease bounds alone don't select a window: the history planner's oldest buckets are priority 0-2 as well, so the window worker at 0..0 also took 2026 history at 0 (25k ranges in 40 groups, fetched live) while 1,739 ranges of the 2016 corpus window sat at voyager 2's leftover priorities (done when the window was marked, planned again by the reset). With LEASE_CAPTURE_ONLY the lease, next and the prefetcher's run-ahead take capture ranges only; the run-ahead now also keeps to the lease bounds. Test: a group with history at 0 only isn't leased, the history range after the window isn't handed out. Co-Authored-By: Claude Fable 5.1 --- REBUILD.md | 7 +++ app/services/group_leases.ts | 63 ++++++++++++++++++++++----- app/services/prefetcher.ts | 21 ++++++--- config/coordinator.ts | 7 ++- start/env.ts | 1 + tests/functional/group_leases.spec.ts | 26 +++++++++++ 6 files changed, 106 insertions(+), 19 deletions(-) diff --git a/REBUILD.md b/REBUILD.md index 034c7a8..4e62abd 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -544,6 +544,13 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md (a live reply cut short; the spool doesn't check) is reported after the batch before it went out and the range asked for again from that chunk: the window stays (the first run lost it, and the open batch). + The lease bounds alone don't select a window: the history planner's + oldest buckets are priority 0–2 too, so the first run also took 2026 + history at 0 (25k ranges in 40 groups, fetched live) while 1,739 of + the 2016 corpus ranges sat at voyager 2's leftover priorities (done + when the window was marked, planned again by the reset): hence + `LEASE_CAPTURE_ONLY` (the lease, next and the prefetcher's run-ahead + take capture ranges only) and the one-off repair of the priorities. 6. **The corpus stays as it is.** Measured on 300 unfiltered chunks (30 M lines, LOG.md (102)): 43% of lines lack a "(k/N)" counter, and of diff --git a/app/services/group_leases.ts b/app/services/group_leases.ts index 05dca47..4675e0c 100644 --- a/app/services/group_leases.ts +++ b/app/services/group_leases.ts @@ -31,6 +31,11 @@ class GroupLeases { return coordinatorConfig.snapshotDir } + /** LEASE_CAPTURE_ONLY: only the corpus windows' ranges (the history's first buckets share their priorities). */ + private get captureOnly(): string { + return coordinatorConfig.leaseCaptureOnly ? 'AND r.capture' : '' + } + async lease(nodeId: number): Promise { const rules = await RuleSets.current() const ttl = coordinatorConfig.leaseTtlSeconds @@ -43,7 +48,7 @@ class GroupLeases { const { rows } = await trx.rawQuery( `SELECT g.id, g.name, g.priority, l.node_id AS holder, l.snapshot_next, todo.priority AS todo_priority FROM groups g - JOIN LATERAL (SELECT min(r.priority) AS priority FROM ranges r WHERE r.group_id = g.id AND r.status = 'planned' AND r.next_id <= r.end_id AND r.priority BETWEEN ? AND ? ${history}) todo ON true + JOIN LATERAL (SELECT min(r.priority) AS priority FROM ranges r WHERE r.group_id = g.id AND r.status = 'planned' AND r.next_id <= r.end_id AND r.priority BETWEEN ? AND ? ${history} ${this.captureOnly}) todo ON true LEFT JOIN group_leases l ON l.group_id = g.id WHERE g.active AND todo.priority IS NOT NULL AND (l.node_id IS NULL OR l.heartbeat_at < now() - make_interval(secs => ?)) @@ -62,7 +67,10 @@ class GroupLeases { [row.snapshot_next, groupId, row.snapshot_next, row.snapshot_next] ) } else { - await trx.rawQuery(`UPDATE ranges SET status = 'planned', node_id = NULL WHERE group_id = ? AND status = 'leased'`, [groupId]) + await trx.rawQuery( + `UPDATE ranges SET status = 'planned', node_id = NULL WHERE group_id = ? AND status = 'leased'`, + [groupId] + ) } await trx.rawQuery( `INSERT INTO group_leases (group_id, node_id, leased_at, heartbeat_at, range_id) @@ -82,7 +90,8 @@ class GroupLeases { /** The group's lease row, when this node holds it. */ async held(groupId: number, nodeId: number) { const lease = await db.from('group_leases').where('group_id', groupId).first() - if (!lease || Number(lease.node_id) !== nodeId) throw new LeaseLost(`Group ${groupId} is not held by this node`) + if (!lease || Number(lease.node_id) !== nodeId) + throw new LeaseLost(`Group ${groupId} is not held by this node`) return lease } @@ -91,15 +100,21 @@ class GroupLeases { await this.held(groupId, nodeId) const rules = await RuleSets.current() const { leasePriorityMin: min, leasePriorityMax: max } = coordinatorConfig - const mine = await db.from('ranges').where('group_id', groupId).where('status', 'leased').where('node_id', nodeId).orderBy('start_id').first() + const mine = await db + .from('ranges') + .where('group_id', groupId) + .where('status', 'leased') + .where('node_id', nodeId) + .orderBy('start_id') + .first() const row = mine ?? ( await 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 group_id = ? AND status = 'planned' AND next_id <= end_id AND priority BETWEEN ? AND ? - ORDER BY start_id LIMIT 1 FOR UPDATE SKIP LOCKED) + SELECT id FROM ranges r WHERE r.group_id = ? AND r.status = 'planned' AND r.next_id <= r.end_id AND r.priority BETWEEN ? AND ? ${this.captureOnly} + ORDER BY r.start_id LIMIT 1 FOR UPDATE SKIP LOCKED) RETURNING r.id, r.start_id, r.end_id, r.next_id`, [nodeId, rules.version, groupId, min, max] ) @@ -109,7 +124,14 @@ class GroupLeases { .where('group_id', groupId) .update({ range_id: row ? Number(row.id) : null, heartbeat_at: new Date() }) if (!row) return null - return { range: { id: Number(row.id), start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id) } } + return { + range: { + id: Number(row.id), + start: Number(row.start_id), + end: Number(row.end_id), + next: Number(row.next_id), + }, + } } async heartbeat(groupId: number, nodeId: number) { @@ -120,14 +142,25 @@ class GroupLeases { /** A range of the group finished by its holder: counted on the lease (Leases.finish does the range). */ async rangeDone(groupId: number, headers: number) { - await db.rawQuery(`UPDATE group_leases SET ranges_done = ranges_done + 1, headers = headers + ?, range_id = NULL WHERE group_id = ?`, [headers, groupId]) + await db.rawQuery( + `UPDATE group_leases SET ranges_done = ranges_done + 1, headers = headers + ?, range_id = NULL WHERE group_id = ?`, + [headers, groupId] + ) } /** The worker gives the group back (after its snapshot): its range stays at its resume point for the next holder. */ async release(groupId: number, nodeId: number) { await this.held(groupId, nodeId) - await db.from('ranges').where('group_id', groupId).where('status', 'leased').where('node_id', nodeId).update({ status: 'planned', node_id: null }) - await db.from('group_leases').where('group_id', groupId).update({ node_id: null, range_id: null }) + await db + .from('ranges') + .where('group_id', groupId) + .where('status', 'leased') + .where('node_id', nodeId) + .update({ status: 'planned', node_id: null }) + await db + .from('group_leases') + .where('group_id', groupId) + .update({ node_id: null, range_id: null }) } private snapshotPath(groupId: number) { @@ -144,11 +177,17 @@ class GroupLeases { const path = this.snapshotPath(groupId) await pipeline(body, createWriteStream(`${path}.part`)) await rename(`${path}.part`, path) - await db.from('group_leases').where('group_id', groupId).update({ snapshot_next: next, snapshot_at: new Date(), heartbeat_at: new Date() }) + await db + .from('group_leases') + .where('group_id', groupId) + .update({ snapshot_next: next, snapshot_at: new Date(), heartbeat_at: new Date() }) } /** The snapshot's stream and size, or null when there is none. */ - async getSnapshot(groupId: number, nodeId: number): Promise<{ stream: Readable; size: number; next: number } | null> { + async getSnapshot( + groupId: number, + nodeId: number + ): Promise<{ stream: Readable; size: number; next: number } | null> { const lease = await this.held(groupId, nodeId) if (lease.snapshot_next === null) return null const path = this.snapshotPath(groupId) diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index 07f93cc..2d1aedc 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -430,7 +430,11 @@ export class Prefetcher { await db .from('group_leases') .whereNotNull('node_id') - .where('heartbeat_at', '>=', new Date(Date.now() - coordinatorConfig.leaseTtlSeconds * 1000)) + .where( + 'heartbeat_at', + '>=', + new Date(Date.now() - coordinatorConfig.leaseTtlSeconds * 1000) + ) .select('group_id') ).map((l) => Number(l.group_id)) // Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease. @@ -456,8 +460,10 @@ export class Prefetcher { const want = Math.max(ahead - Number(ready), short) // The catalog at its cap on the NVMe: no history fetched ahead (CatalogSpace). const full = await CatalogSpace.full() - // Ahead of each held group's cursor: its next ranges in article order, a range of every held group before any - // group's second; without held groups (no window worker running) nothing is fetched ahead for workers. + // Ahead of each held group's cursor: its next ranges in article order (within what the group leases hand out: + // LEASE_MIN/MAX_PRIORITY, LEASE_CAPTURE_ONLY), a range of every held group before any group's second; without + // held groups (no window worker running) nothing is fetched ahead for workers. + const { leasePriorityMin, leasePriorityMax, leaseCaptureOnly } = coordinatorConfig const fresh = room && want > 0 && held.length > 0 ? await rows(db.query()) @@ -465,8 +471,12 @@ export class Prefetcher { .where('r.spooled_chunks', 0) .whereRaw('r.next_id <= r.end_id') .whereIn('r.group_id', held) + .whereBetween('r.priority', [leasePriorityMin, leasePriorityMax]) + .where((q: any) => (leaseCaptureOnly ? q.where('r.capture', true) : q)) .where((q: any) => (full ? q.where('r.priority', '<', 0) : q)) - .orderByRaw('row_number() OVER (PARTITION BY r.group_id ORDER BY r.start_id), r.priority, r.start_id') + .orderByRaw( + 'row_number() OVER (PARTITION BY r.group_id ORDER BY r.start_id), r.priority, r.start_id' + ) .limit(want) : [] const targets: Target[] = [] @@ -478,7 +488,8 @@ export class Prefetcher { // A chunk already in the corpus isn't spooled: the stream serves it from there (REBUILD.md design 1). const corpusRange = row.capture === true ? { group: row.group as string, start, end } : null for (const span of Spool.chunksOf({ start, end }, Number(row.next_id))) { - if (this.claimed.has(`${row.id}:${span.from}`) || (await Spool.has(Number(row.id), span))) continue + if (this.claimed.has(`${row.id}:${span.from}`) || (await Spool.has(Number(row.id), span))) + continue if (corpusRange && (await Corpus.has(corpusRange, span))) continue missing.push(span) } diff --git a/config/coordinator.ts b/config/coordinator.ts index f38fa4f..8ea9530 100644 --- a/config/coordinator.ts +++ b/config/coordinator.ts @@ -33,11 +33,14 @@ export default { leaseTtlSeconds: env.get('LEASE_TTL_SECONDS', 15 * 60), /** * The ranges group leases hand out (REBUILD.md design 1), by priority: the tail is negative, the capture windows - * 0–2 (2016, 2021, 2025), the rest of the history above (its position in the group, or 4000+). Bounds so a - * deployment can run one window first (homelab 2026-10-26: 0..0, the 2016 window); unbounded by default. + * 0–2 (2016, 2021, 2025), the history by its position in the group (0 oldest … 1000 newest: its first buckets + * share 0–2 with the windows) or 4000+. Bounds so a deployment can run one window first, with + * LEASE_CAPTURE_ONLY so the history's 0 isn't taken along (homelab 2026-10-26: 0..0 and capture only, the 2016 + * window; LOG.md (118)); unbounded and any range by default. */ leasePriorityMin: env.get('LEASE_MIN_PRIORITY', -1_000_000), leasePriorityMax: env.get('LEASE_MAX_PRIORITY', 1_000_000), + leaseCaptureOnly: env.get('LEASE_CAPTURE_ONLY', false), /** NNTP connections the coordinator fetches headers with: at most (its share, nntpShares), and at least while sharing. */ nntpConnections: env.get('NNTP_CONNECTIONS', 4), nntpMinConnections: env.get('NNTP_MIN_CONNECTIONS', 8), diff --git a/start/env.ts b/start/env.ts index 38931ee..42f7ff7 100644 --- a/start/env.ts +++ b/start/env.ts @@ -77,6 +77,7 @@ export default await Env.create(new URL('../', import.meta.url), { SNAPSHOT_DIR: Env.schema.string.optional(), LEASE_MIN_PRIORITY: Env.schema.number.optional(), LEASE_MAX_PRIORITY: Env.schema.number.optional(), + LEASE_CAPTURE_ONLY: Env.schema.boolean.optional(), PENDING_SHARE: Env.schema.number.optional(), PENDING_MIN_PROBES: Env.schema.number.optional(), PENDING_MAX_NAMED: Env.schema.number.optional(), diff --git a/tests/functional/group_leases.spec.ts b/tests/functional/group_leases.spec.ts index b82de71..bf9aa52 100644 --- a/tests/functional/group_leases.spec.ts +++ b/tests/functional/group_leases.spec.ts @@ -133,6 +133,32 @@ test.group('Group leases', (g) => { assert.equal((await GroupLeases.next(a, n1))?.range.id, r4) }) + test('capture only: the history ranges sharing a window priority are neither leased nor handed out', async ({ + assert, + }) => { + // Group e: history at priority 0 only (the planner's oldest bucket); group f: a 2016 window range (capture) + // between two history ranges at 0. + const e = await group('alt.binaries.gl.e', 1) + await range(e, 1, 0) + const f = await group('alt.binaries.gl.f', 2) + await range(f, 1, 0) + const f2 = await range(f, 101, 0, { capture: true }) + await range(f, 201, 0) + const n1 = await node('gl-8') + const was = coordinatorConfig.leaseCaptureOnly + coordinatorConfig.leaseCaptureOnly = true + try { + assert.equal((await GroupLeases.lease(n1))?.group.id, f, 'the group with a capture range, not the older one') + assert.equal((await GroupLeases.next(f, n1))?.range.id, f2) + await db.from('ranges').where('id', f2).update({ status: 'done', next_id: 201 }) + assert.isNull(await GroupLeases.next(f, n1), 'the history range after the window is not handed out') + assert.isNull(await GroupLeases.lease(await node('gl-9'))) + } finally { + coordinatorConfig.leaseCaptureOnly = was + } + assert.equal((await GroupLeases.lease(await node('gl-10')))?.group.id, e, 'any range again without the flag') + }) + test('a snapshot older than a done range makes that range be read again from its start', async ({ assert }) => { const a = await group('alt.binaries.gl.d', 1) const r1 = await range(a, 1, 0)