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)