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')