From 31fc204c762c6b3ae84eb3a59343d4e96dc75415 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Tue, 27 Oct 2026 09:19:16 +0100 Subject: [PATCH] The range stream polls the corpus as well as the spool while it waits The corpus fill claims chunks of a leased range ahead of the leased run (the prefetcher's claim keys are per range and chunk), stores them in the corpus only, and the run leaves them there. The stream looked at the corpus once per chunk and then polled the spool only: when the store came seconds after that look, the stream waited its whole hour and the worker lost the group and its open batch (three groups on 2026-10-27, LOG.md (118)). Now the wait loop checks both. Co-Authored-By: Claude Fable 5.1 --- REBUILD.md | 7 +++++++ app/controllers/node_controller.ts | 26 ++++++++++++++---------- tests/functional/node_api.spec.ts | 32 ++++++++++++++++++++++++++++++ 3 files changed, 54 insertions(+), 11 deletions(-) diff --git a/REBUILD.md b/REBUILD.md index 4e62abd..626d574 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -551,6 +551,13 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md 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. + The stream polls the corpus as well as the spool while it waits for a + chunk: the corpus fill claims chunks of the same range ahead of the + leased run (the prefetcher's claims are per range and chunk), stores + them in the corpus only, and the run leaves them there; a stream that + looked at the corpus once, seconds before that store, waited its whole + hour on the spool and the worker lost the group and its open batch + (three groups on the first morning). 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/controllers/node_controller.ts b/app/controllers/node_controller.ts index fb50ad7..d217e71 100644 --- a/app/controllers/node_controller.ts +++ b/app/controllers/node_controller.ts @@ -5,7 +5,7 @@ import db from '@adonisjs/lucid/services/db' import Leases, { LeaseLost } from '#services/leases' import GroupLeases from '#services/group_leases' import BatchStore, { StaleBatch } from '#services/batch_store' -import Spool from '#services/spool' +import Spool, { type ChunkSpan } from '#services/spool' import Corpus from '#services/corpus' import { KEEPALIVE, encodeFrame, fromWire, type WireBatch } from '#lib/edge/index' @@ -99,19 +99,22 @@ export default class NodeController { const spans = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }, Number(range.next_id)) const group = range.capture ? await db.from('groups').where('id', range.group_id).select('name').first() : null const corpusRange = group ? { group: group.name as string, start: Number(range.start_id), end: Number(range.end_id) } : null + // The chunk from the corpus or the spool, whichever has it. Both are polled while waiting: the corpus fill + // claims a chunk ahead of the leased run (the prefetcher's claims are per range and chunk), stores it in the + // corpus only, and the run then leaves it there — a stream that looked at the corpus once, before that + // store, waited the whole hour on the spool (three groups on 2026-10-27, LOG.md (118)). + const stored = async (span: ChunkSpan) => + corpusRange && (await Corpus.has(corpusRange, span)) ? Corpus.read(corpusRange, span) : null async function* frames() { for (const span of spans) { - if (corpusRange && (await Corpus.has(corpusRange, span))) { - const data = await Corpus.read(corpusRange, span) - if (data) { - yield encodeFrame({ ...span, data }) - await Leases.heartbeat(rangeId, nodeId) - continue - } - } const waitUntil = Date.now() + CHUNK_WAIT_MS let quietSince = Date.now() - while (!(await Spool.has(rangeId, span))) { + let data = await stored(span) + while (!data) { + if (await Spool.has(rangeId, span)) { + data = await Spool.read(rangeId, span) + break + } 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() @@ -119,8 +122,9 @@ export default class NodeController { await Leases.heartbeat(rangeId, nodeId) } await sleep(500) + data = await stored(span) } - yield encodeFrame({ ...span, data: await Spool.read(rangeId, span) }) + yield encodeFrame({ ...span, data }) // Streaming keeps the lease alive. await Leases.heartbeat(rangeId, nodeId) } diff --git a/tests/functional/node_api.spec.ts b/tests/functional/node_api.spec.ts index 0aadb87..8992634 100644 --- a/tests/functional/node_api.spec.ts +++ b/tests/functional/node_api.spec.ts @@ -153,6 +153,38 @@ test.group('Node API', (group) => { } }) + test('a chunk the corpus fill stores while the stream waits for it comes from the corpus', async ({ client, assert }) => { + const corpus = await mkdtemp(join(tmpdir(), 'corpus-')) + const saved = coordinatorConfig.corpus.dir + coordinatorConfig.corpus.dir = corpus + try { + const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id') + // A capture range not in the corpus: its first chunk spooled by the leased run, the second claimed by + // the corpus fill, which stores it in the corpus only, after the stream started. + const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 200, next_id: 1, capture: true }).returning('id') + const rangeId = Number(r.id) + await Spool.write(rangeId, { from: 1, to: 100 }, chunk(11)) + const node = await Node.create({ name: 'n1' }) + const token = (await Node.accessTokens.create(node)).value!.release() + await db.from('ranges').where('id', rangeId).update({ status: 'leased', node_id: node.id, leased_at: new Date(), heartbeat_at: new Date() }) + + const pending = client.get(`/api/v1/node/ranges/${rangeId}/stream`).bearerToken(token) + setTimeout(() => Corpus.write({ group: GROUP, start: 1, end: 200 }, { from: 101, to: 200 }, chunk(111), 2), 800) + const stream = await pending + stream.assertStatus(200) + const raw = stream.body() as unknown + const body: Buffer = Buffer.isBuffer(raw) ? raw : Buffer.from(stream.text(), 'binary') + const frames = [] + for await (const frame of decodeFrames([body])) frames.push(frame) + assert.deepEqual(frames.map((f) => [f.from, f.to]), [[1, 100], [101, 200]]) + assert.equal(parseOverviews(unzipSync(frames[1].data).toString('utf8')).length, 3) + assert.isFalse(await Spool.has(rangeId, { from: 101, to: 200 })) + } finally { + coordinatorConfig.corpus.dir = saved + await rm(corpus, { recursive: true, force: true }) + } + }) + test('a chunk that does not inflate: reported, removed from the spool, the range goes back', async ({ client, assert }) => { const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id') const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: 1, end_id: 300, next_id: 1, spooled_chunks: 3 }).returning('id')