diff --git a/app/controllers/node_controller.ts b/app/controllers/node_controller.ts index dd3c458..af0c9bc 100644 --- a/app/controllers/node_controller.ts +++ b/app/controllers/node_controller.ts @@ -4,10 +4,12 @@ import { setTimeout as sleep } from 'node:timers/promises' import Leases, { LeaseLost } from '#services/leases' import BatchStore, { StaleBatch } from '#services/batch_store' import Spool from '#services/spool' -import { encodeFrame, fromWire, type WireBatch } from '#lib/edge/index' +import { KEEPALIVE, encodeFrame, fromWire, type WireBatch } from '#lib/edge/index' /** How long the stream waits for a chunk the prefetcher hasn't spooled yet. */ const CHUNK_WAIT_MS = 15 * 60 * 1000 +/** While waiting, an empty frame this often: an HTTP client gives up on a quiet body (Node's after 5 minutes). */ +const KEEPALIVE_MS = 20_000 /** * The edge workers' API (REBUILD.md, parts 1–2): a lease, the range's @@ -34,8 +36,14 @@ export default class NodeController { async function* frames() { for (const span of spans) { const waitUntil = Date.now() + CHUNK_WAIT_MS + let quietSince = Date.now() while (!(await Spool.has(rangeId, span))) { if (Date.now() > waitUntil) throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} wasn't spooled in time`) + if (Date.now() - quietSince >= KEEPALIVE_MS) { + quietSince = Date.now() + yield encodeFrame(KEEPALIVE) + await Leases.heartbeat(rangeId, nodeId) + } await sleep(500) } yield encodeFrame({ ...span, data: await Spool.read(rangeId, span) }) diff --git a/app/services/prefetcher.ts b/app/services/prefetcher.ts index 31f021b..a7a2c45 100644 --- a/app/services/prefetcher.ts +++ b/app/services/prefetcher.ts @@ -9,12 +9,14 @@ interface Target { group: string /** Chunks not spooled yet, in order. */ missing: ChunkSpan[] - /** All of the range's chunks from its resume point. */ + /** All of the range's chunks, from its start. */ total: number } /** How often the targets are read again from the database. */ const REFRESH_MS = 2000 +/** How often the spool's size is measured (a stat per file). */ +const SIZE_MS = 30_000 /** * Keeps the news server busy ahead of the workers (REBUILD.md, prefetch). @@ -31,6 +33,8 @@ export class Prefetcher { private running = 0 private refreshedAt = 0 private stopped = false + private spoolBytes = 0 + private sizedAt = 0 /** One round: refreshes the targets now and then, and puts idle connections to work. */ async tick() { @@ -65,7 +69,10 @@ export class Prefetcher { await client.group(target.group) return client.xoverRaw(span.from, span.to) }) + const existed = await Spool.has(target.id, span) await Spool.write(target.id, span, data) + if (existed) continue + this.spoolBytes += data.length const { rows } = await db.rawQuery( `UPDATE ranges SET spooled_chunks = spooled_chunks + 1 WHERE id = ? RETURNING spooled_chunks`, [target.id] @@ -94,20 +101,31 @@ export class Prefetcher { } private async readTargets(): Promise { - const rows = (q: any) => - q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').whereNull('r.spooled_at').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id') - // Leased ranges whose chunks aren't all there: a worker is waiting for them. + if (Date.now() - this.sizedAt >= SIZE_MS) { + this.sizedAt = Date.now() + this.spoolBytes = await Spool.size() + } + const rows = (q: any) => q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id') + // Leased ranges, checked against the spool's files whatever spooled_at says: a worker waits for them. const leased = await rows(db.query()).where('r.status', 'leased').orderBy('r.leased_at') - const [{ leases }] = await db.from('ranges').where('status', 'leased').count('* as leases') - const ahead = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * Number(leases)) - const room = (await Spool.size()) < coordinatorConfig.spool.maxBytes - const planned = room ? await rows(db.query()).where('r.status', 'planned').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead) : [] + // Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease. + const ahead = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * leased.length) + const started = await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', '>', 0).whereNull('r.spooled_at').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]) + const [{ ready }] = await db.from('ranges').where('status', 'planned').where('spooled_chunks', '>', 0).count('* as ready') + const room = this.spoolBytes < coordinatorConfig.spool.maxBytes + const fresh = room && Number(ready) < ahead + ? await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', 0).orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(ahead - Number(ready)) + : [] const targets: Target[] = [] - for (const row of [...leased, ...planned]) { - const spans = Spool.chunksOf({ start: Number(row.start_id), end: Number(row.end_id) }, Number(row.next_id)) + for (const row of [...leased, ...(room ? started : []), ...fresh]) { + const start = Number(row.start_id) + const end = Number(row.end_id) + const total = Spool.chunksOf({ start, end }).length const missing: ChunkSpan[] = [] - for (const span of spans) if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) missing.push(span) - targets.push({ id: Number(row.id), group: row.group, missing, total: spans.length }) + 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))) missing.push(span) + } + if (missing.length > 0) targets.push({ id: Number(row.id), group: row.group, missing, total }) } return targets } diff --git a/commands/edge_node.ts b/commands/edge_node.ts index b0b7c4a..3ecd144 100644 --- a/commands/edge_node.ts +++ b/commands/edge_node.ts @@ -1,10 +1,12 @@ import { BaseCommand, flags } from '@adonisjs/core/ace' import type { CommandOptions } from '@adonisjs/core/types/ace' import { setTimeout as sleep } from 'node:timers/promises' -import { gzipSync, unzipSync } from 'node:zlib' -import { RangeIndexer, decodeFrames, toWire, type EdgeBatch, type Lease } from '#lib/edge/index' +import { promisify } from 'node:util' +import { gzipSync, unzip } from 'node:zlib' +import { RangeIndexer, decodeFrames, isKeepalive, toWire, type EdgeBatch, type Lease } from '#lib/edge/index' const IDLE_MS = 30_000 +const unzipAsync = promisify(unzip) const RETRIES = 8 class LeaseLost extends Error {} @@ -66,31 +68,54 @@ export default class EdgeNode extends BaseCommand { private async runRange(api: Api, lease: Lease) { const { range, rules } = lease const started = Date.now() - const indexer = new RangeIndexer({ group: range.group, start: range.start, end: range.end, rules }) + const indexer = new RangeIndexer({ group: range.group, start: range.start, end: range.end, resumeAt: range.next, rules }) const totals = { headers: 0, releases: 0, candidates: 0, bytes: 0, postMs: 0 } let chunks = 0 - const post = async (batch: EdgeBatch) => { - const at = Date.now() - const stored = await api.batch(range.id, batch) - totals.postMs += Date.now() - at - totals.bytes += stored.bytes - totals.headers += batch.headers - totals.releases += stored.releases - totals.candidates += stored.candidates + let reachedEnd = false + // Posts go in order, but the worker indexes on meanwhile (a post took 1–16 s; LOG.md 2026-10-09). + let posting: Promise = Promise.resolve() + const post = (batch: EdgeBatch) => { + const previous = posting + posting = (async () => { + await previous + const at = Date.now() + const stored = await api.batch(range.id, batch) + totals.postMs += Date.now() - at + totals.bytes += stored.bytes + totals.headers += batch.headers + totals.releases += stored.releases + totals.candidates += stored.candidates + })() + // A failed post stops the range at the next chunk (a lost lease: nothing more will be stored). + posting.catch((error) => (postError ??= error)) + } + let postError: Error | null = null + const handle = async (chunk: { to: number; text: Promise }) => { + if (postError) throw postError + indexer.addText((await chunk.text).toString('utf8'), chunk.to + 1) + chunks++ + reachedEnd = chunk.to >= range.end + if (reachedEnd) indexer.finish() + if (reachedEnd || chunks % this.batchChunks === 0) post(indexer.take()) } try { + // Inflating runs on a thread-pool thread: the next chunk inflates while this one is indexed. + let pending: { to: number; text: Promise } | null = null for await (const frame of decodeFrames(await api.stream(range.id))) { - indexer.addText(unzipSync(frame.data).toString('utf8'), frame.to + 1) - chunks++ - const last = frame.to >= range.end - if (last) indexer.finish() - if (last || chunks % this.batchChunks === 0) await post(indexer.take()) - if (this.stopping && !last) { - // Back to the queue; the next worker resumes at the last batch. - if (chunks % this.batchChunks !== 0) this.logger.info(`Range ${range.id}: giving it back (${chunks % this.batchChunks} chunks since the last batch are read again).`) - await api.call(range.id, 'release') - return - } + if (isKeepalive(frame)) continue + const next = { to: frame.to, text: unzipAsync(frame.data) } + next.text.catch(() => {}) + if (pending) await handle(pending) + pending = next + if (this.stopping) break + } + if (pending && !this.stopping) await handle(pending) + await posting + if (!reachedEnd) { + // Stopping: back to the queue; the next worker resumes at the last stored batch. + this.logger.info(`Range ${range.id}: giving it back.`) + await api.call(range.id, 'release') + return } await api.call(range.id, 'finish') const seconds = (Date.now() - started) / 1000 @@ -100,6 +125,7 @@ export default class EdgeNode extends BaseCommand { `posted ${(totals.bytes / 1e6).toFixed(1)} MB in ${(totals.postMs / 1000).toFixed(1)} s.` ) } catch (error: any) { + await posting.catch(() => {}) if (error instanceof LeaseLost) { this.logger.warning(`Range ${range.id}: ${error.message}`) return diff --git a/lib/edge/index.ts b/lib/edge/index.ts index ee46770..6d1c751 100644 --- a/lib/edge/index.ts +++ b/lib/edge/index.ts @@ -1,4 +1,4 @@ export { RangeIndexer, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js' -export { encodeFrame, decodeFrames, type Frame } from './stream.js' +export { encodeFrame, decodeFrames, isKeepalive, KEEPALIVE, type Frame } from './stream.js' export { toWire, fromWire, type WireBatch } from './wire.js' export type { Lease } from './lease.js' diff --git a/lib/edge/range_indexer.ts b/lib/edge/range_indexer.ts index bc00652..753363b 100644 --- a/lib/edge/range_indexer.ts +++ b/lib/edge/range_indexer.ts @@ -83,6 +83,8 @@ export interface RangeIndexerOptions { /** Inclusive article numbers. */ start: number end: number + /** Where this worker starts: the range's resume point (a range taken over midway); its first batch starts here. */ + resumeAt?: number rules: Rules /** Chunks without a new part before an open file is given up (as a fragment near an edge). */ idleChunks?: number @@ -112,7 +114,7 @@ export class RangeIndexer { this.rules = new RuleSet(options.rules) this.idle = options.idleChunks ?? 5 this.edge = options.edgeArticles ?? 50_000 - this.batch = this.empty(options.start) + this.batch = this.empty(options.resumeAt ?? options.start) } /** One chunk of overview lines (parsed); `next` is the first article after it. */ diff --git a/lib/edge/stream.ts b/lib/edge/stream.ts index 7cc3cce..663342d 100644 --- a/lib/edge/stream.ts +++ b/lib/edge/stream.ts @@ -11,6 +11,13 @@ export interface Frame { const HEADER = 20 +/** An empty frame, sent while the next chunk is still being fetched, so the response doesn't go quiet. */ +export const KEEPALIVE: Frame = { from: 0, to: 0, data: Buffer.alloc(0) } + +export function isKeepalive(frame: Frame) { + return frame.data.length === 0 && frame.from === 0 +} + export function encodeFrame(frame: Frame): Buffer { const head = Buffer.alloc(HEADER) head.writeBigUInt64BE(BigInt(frame.from), 0) diff --git a/tests/unit/edge/range_indexer.spec.ts b/tests/unit/edge/range_indexer.spec.ts index 7381b53..be7d13a 100644 --- a/tests/unit/edge/range_indexer.spec.ts +++ b/tests/unit/edge/range_indexer.spec.ts @@ -65,3 +65,11 @@ test.group('edge / RangeIndexer', () => { assert.lengthOf(batch.releases, 0) }) }) + +test.group('edge / taking over a range', () => { + test("a worker that resumes a range midway starts its first batch at the range's resume point", ({ assert }) => { + const indexer = new RangeIndexer({ group: GROUP, start: 1, end: 1_000_000, resumeAt: 500_001, rules }) + indexer.add(post(500_001, 'A.Release.2020.mkv', 1), 600_000) + assert.deepEqual([indexer.take().from, indexer.take().from], [500_001, 600_000]) + }) +}) diff --git a/tests/unit/edge/stream.spec.ts b/tests/unit/edge/stream.spec.ts new file mode 100644 index 0000000..1f64db7 --- /dev/null +++ b/tests/unit/edge/stream.spec.ts @@ -0,0 +1,28 @@ +import { test } from '@japa/runner' +import { KEEPALIVE, decodeFrames, encodeFrame, isKeepalive } from '#lib/edge/index' + +test.group('edge / stream frames', () => { + test('frames survive being split anywhere; keep-alive frames are recognised', async ({ assert }) => { + const whole = Buffer.concat([ + encodeFrame({ from: 160_000_000_001, to: 160_000_050_000, data: Buffer.from('first') }), + encodeFrame(KEEPALIVE), + encodeFrame({ from: 160_000_050_001, to: 160_000_100_000, data: Buffer.from('second chunk') }), + ]) + const pieces = [whole.subarray(0, 7), whole.subarray(7, 30), whole.subarray(30)] + const frames = [] + for await (const frame of decodeFrames(pieces)) frames.push(frame) + assert.deepEqual(frames.map((f) => [f.from, f.to, f.data.toString(), isKeepalive(f)]), [ + [160_000_000_001, 160_000_050_000, 'first', false], + [0, 0, '', true], + [160_000_050_001, 160_000_100_000, 'second chunk', false], + ]) + }) + + test('a stream that ends inside a frame is an error', async ({ assert }) => { + const cut = encodeFrame({ from: 1, to: 2, data: Buffer.from('abc') }).subarray(0, 21) + await assert.rejects(async () => { + for await (const _ of decodeFrames([cut])) { + } + }, /inside a frame/) + }) +})