fix(KB): persist accumulated chunk count across batched ZIM dispatches

The continuation dispatch in EmbedFileJob did not pass the running chunk
count forward, so each batch started with job.data.chunks undefined.
On the final batch, totalChunks collapsed to just that batch's result
and KbIngestState.markIndexed stored a value far below what Qdrant
actually held.

Add chunksSoFar to EmbedFileJobParams and thread it through the
continuation chain so the final markIndexed call reflects the true
total across all batches.

Closes #933
This commit is contained in:
Metbcy 2026-05-31 17:29:39 +00:00 committed by jakeaturner
parent f553a0d57d
commit d175259219
No known key found for this signature in database
GPG Key ID: B1072EBDEECE328D
1 changed files with 15 additions and 4 deletions

View File

@ -18,6 +18,11 @@ export interface EmbedFileJobParams {
batchOffset?: number // Current batch offset (for ZIM files)
totalArticles?: number // Total articles in ZIM (for progress tracking)
isFinalBatch?: boolean // Whether this is the last batch (prevents premature deletion)
// Running total of chunks embedded across prior batches in this dispatch chain.
// Carried forward so the final batch can persist an accurate `chunks_embedded`
// count via KbIngestState.markIndexed (see #933 — without this, only the last
// batch's chunk count was stored while Qdrant held the full set).
chunksSoFar?: number
}
export class EmbedFileJob {
@ -159,13 +164,16 @@ export class EmbedFileJob {
await new Promise((resolve) => setTimeout(resolve, EmbedFileJob.CPU_BATCH_DELAY_MS))
}
// Dispatch next batch (not final yet)
// Dispatch next batch (not final yet). Carry forward the running
// chunk count so the final batch can persist an accurate total (#933).
const chunksSoFarNext = (job.data.chunksSoFar || 0) + (result.chunks || 0)
await EmbedFileJob.dispatch({
filePath,
fileName,
batchOffset: nextOffset,
totalArticles: totalArticles || result.totalArticles,
isFinalBatch: false, // Explicitly not final
chunksSoFar: chunksSoFarNext,
})
// Calculate progress based on articles processed
@ -178,7 +186,7 @@ export class EmbedFileJob {
...job.data,
status: 'batch_completed',
lastBatchAt: Date.now(),
chunks: (job.data.chunks || 0) + (result.chunks || 0),
chunks: chunksSoFarNext,
})
return {
@ -192,8 +200,11 @@ export class EmbedFileJob {
}
}
// Final batch or non-batched file - mark as complete
const totalChunks = (job.data.chunks || 0) + (result.chunks || 0)
// Final batch or non-batched file - mark as complete.
// chunksSoFar carries the accumulated count from prior dispatched batches
// (each continuation passes it forward — see EmbedFileJobParams). For a
// non-batched file it is undefined and we just count this single result.
const totalChunks = (job.data.chunksSoFar || 0) + (result.chunks || 0)
await this.safeUpdateProgress(job, 100)
await job.updateData({
...job.data,