From 016b9c1f73db412f89500118ca27ab03557e2daf Mon Sep 17 00:00:00 2001 From: ntorbinskiy Date: Thu, 17 Sep 2026 12:28:52 +0300 Subject: [PATCH 1/7] feat(reins): classify index failures as transient or permanent --- .../reins/source/domain/indexFailure.spec.ts | 90 ++++++++++++++++++ .../reins/source/domain/indexFailure.ts | 91 +++++++++++++++++++ 2 files changed, 181 insertions(+) create mode 100644 api/src/slices/reins/source/domain/indexFailure.spec.ts create mode 100644 api/src/slices/reins/source/domain/indexFailure.ts diff --git a/api/src/slices/reins/source/domain/indexFailure.spec.ts b/api/src/slices/reins/source/domain/indexFailure.spec.ts new file mode 100644 index 00000000..2e952f01 --- /dev/null +++ b/api/src/slices/reins/source/domain/indexFailure.spec.ts @@ -0,0 +1,90 @@ +import { + classifyIndexFailure, + MAX_INDEX_RETRIES, + nextRetryAt, + retryDelayMs, +} from './indexFailure'; + +describe('classifyIndexFailure', () => { + // Messages as they appeared in the dev logs and on source rows; the first is + // the 2026-09-17 outage verbatim. + const transient = [ + 'RetryError[]', + 'BedrockConnectionError: connection error', + 'ServiceUnavailableException: Bedrock is unable to process your request. Please try again later.', + 'ThrottlingException: Too many requests, please wait before trying again.', + 'Request timed out after 300s', + 'LightRAG /documents/upload failed: 502 Bad Gateway', + 'LightRAG /documents/text failed: 503', + 'fetch failed', + 'read ECONNRESET', + 'Identical content already exists under another filename. Original doc_id: doc-9f, Status: failed', + "Document storage already contains 'manual.pdf' (Status: failed)", + 'LightRAG failed to process it', + ]; + + it.each(transient)('retries: %s', (message) => { + expect(classifyIndexFailure(message)).toBe('transient'); + }); + + const permanent = [ + 'File content contains only whitespace characters', + 'Unsupported file type: .exe', + 'Source src-1 has no content', + 'text extraction failed: file could not be read as a PDF', + 'LightRAG reported no state for this document', + 'LightRAG /documents/upload failed: 400 Bad Request', + // A refused re-upload whose original is fine is adopted, never retried; + // if it reaches here something else is wrong and a person should look. + 'Identical content already exists under another filename. Original doc_id: doc-9f, Status: processed', + 'something nobody has seen before', + ]; + + it.each(permanent)('gives up on: %s', (message) => { + expect(classifyIndexFailure(message)).toBe('permanent'); + }); + + it('treats no message at all as permanent', () => { + expect(classifyIndexFailure(null)).toBe('permanent'); + }); +}); + +describe('retryDelayMs', () => { + it('waits 5, 15 and 60 minutes before the three retries', () => { + expect(retryDelayMs(1)).toBe(5 * 60_000); + expect(retryDelayMs(2)).toBe(15 * 60_000); + expect(retryDelayMs(3)).toBe(60 * 60_000); + }); + + it('has no fourth retry', () => { + expect(MAX_INDEX_RETRIES).toBe(3); + expect(retryDelayMs(4)).toBeNull(); + }); + + it('rejects nonsense attempt numbers instead of indexing off the table', () => { + expect(retryDelayMs(0)).toBeNull(); + expect(retryDelayMs(-1)).toBeNull(); + expect(retryDelayMs(1.5)).toBeNull(); + }); +}); + +describe('nextRetryAt', () => { + const now = new Date('2026-09-17T00:20:00Z'); + + it('schedules a transient failure after the pause for that attempt', () => { + expect(nextRetryAt('RetryError[...]', 1, now)).toEqual( + new Date('2026-09-17T00:25:00Z'), + ); + expect(nextRetryAt('RetryError[...]', 3, now)).toEqual( + new Date('2026-09-17T01:20:00Z'), + ); + }); + + it('stops once the retries are spent', () => { + expect(nextRetryAt('RetryError[...]', 4, now)).toBeNull(); + }); + + it('never schedules a permanent failure', () => { + expect(nextRetryAt('only whitespace', 1, now)).toBeNull(); + }); +}); diff --git a/api/src/slices/reins/source/domain/indexFailure.ts b/api/src/slices/reins/source/domain/indexFailure.ts new file mode 100644 index 00000000..817ddf94 --- /dev/null +++ b/api/src/slices/reins/source/domain/indexFailure.ts @@ -0,0 +1,91 @@ +/** + * Whether an index failure is worth retrying without a person. + * + * LightRAG reports a document as failed for two very different reasons. One is + * the document: no text layer, unsupported type, nothing to index. The other + * is the world around it: Bedrock answering 503 for a quarter of an hour, a + * pod being replaced mid-batch, a connection reset. LightRAG's own retry runs + * for seconds, so the second kind ends up recorded exactly like the first, and + * on 2026-09-17 nine perfectly good documents sat red overnight because of it. + * + * The distinction is made on the message, which is all either side leaves + * behind. Unknown messages are permanent on purpose: a wrong "permanent" costs + * a person one look, a wrong "transient" hides a broken file behind "Retrying" + * for over an hour. Grow the list from a log line, not from a guess. + */ +export type IndexFailureKindTypes = 'transient' | 'permanent'; + +/** + * Automatic retries after the first failure. Together with the pauses below + * they span 80 minutes, which comfortably outlasts the 10-15 minute Bedrock + * waves seen on dev; after that the row reads `failed` for good. + */ +export const MAX_INDEX_RETRIES = 3; + +const RETRY_DELAYS_MS: readonly number[] = [5, 15, 60].map( + (minutes) => minutes * 60_000, +); + +const TRANSIENT_MARKERS: readonly RegExp[] = [ + // LightRAG wraps the exhausted tenacity retry as + // `RetryError[]`. + /RetryError/i, + /BedrockConnectionError/i, + /ServiceUnavailable/i, + /Bedrock is unable to process/i, + /ThrottlingException/i, + /Too many requests/i, + /rate limit/i, + /timed? ?out/i, + /connection error/i, + /ECONNRESET|ECONNREFUSED|EAI_AGAIN/, + /fetch failed/i, + // LightragClientError renders an HTTP failure as `LightRAG failed: + // `; a gateway or unavailable status is LightRAG being + // restarted or overloaded, not the document. + /LightRAG \S+ failed: 50[234]\b/, + // LightRAG holds the document as failed but left no reason on it. It + // accepted the document; one bounded round of reprocessing is the right + // first move. + /LightRAG failed to process it/, + // A refused re-upload naming a failed original, in either of LightRAG's two + // wordings ("Original doc_id: X, Status: failed", "Document storage already + // contains 'name' (Status: failed)"). Reprocessing that original is exactly + // the cure, so the refusal is a symptom of a transient failure, not a new + // permanent one. + /Status:\s*failed/i, +]; + +export function classifyIndexFailure( + message: string | null, +): IndexFailureKindTypes { + if (message === null) return 'permanent'; + return TRANSIENT_MARKERS.some((marker) => marker.test(message)) + ? 'transient' + : 'permanent'; +} + +/** + * How long to wait before retry number `attempt` (1-based: the pause after the + * first failure is `retryDelayMs(1)`), or null once the retries are spent. + */ +export function retryDelayMs(attempt: number): number | null { + if (!Number.isInteger(attempt) || attempt < 1) return null; + if (attempt > MAX_INDEX_RETRIES) return null; + return RETRY_DELAYS_MS[attempt - 1]; +} + +/** + * When a failure recorded now may be retried, or null when it may not: the + * message is permanent, or `attempts` (this failure included) has used up the + * retries. + */ +export function nextRetryAt( + message: string | null, + attempts: number, + now: Date, +): Date | null { + if (classifyIndexFailure(message) === 'permanent') return null; + const delay = retryDelayMs(attempts); + return delay === null ? null : new Date(now.getTime() + delay); +} From c100784a3ba6b3a80d7b49a978b68a7c56350a47 Mon Sep 17 00:00:00 2001 From: ntorbinskiy Date: Thu, 17 Sep 2026 12:37:20 +0300 Subject: [PATCH 2/7] feat(reins): remember index attempts and schedule retries on the row --- .../migration.sql | 10 ++ .../domain/textExtraction.service.spec.ts | 2 + .../knowledge/domain/indexReconcile.spec.ts | 4 + .../knowledge/knowledge.isolation.spec.ts | 2 + .../domain/migration.service.spec.ts | 2 + .../reins/source/data/source.gateway.spec.ts | 142 +++++++++++++++++- .../reins/source/data/source.gateway.ts | 95 +++++++++--- .../reins/source/data/source.mapper.spec.ts | 49 ++++-- .../slices/reins/source/data/source.mapper.ts | 9 +- .../source/data/sourceArchive.writer.spec.ts | 2 + .../reins/source/domain/source.service.ts | 2 + .../reins/source/domain/source.types.ts | 22 ++- .../slices/reins/source/dtos/source.dto.ts | 13 ++ api/src/slices/reins/source/source.prisma | 7 + 14 files changed, 316 insertions(+), 45 deletions(-) create mode 100644 api/prisma/migrations/20260917120000_source_index_retry/migration.sql diff --git a/api/prisma/migrations/20260917120000_source_index_retry/migration.sql b/api/prisma/migrations/20260917120000_source_index_retry/migration.sql new file mode 100644 index 00000000..05fa4c7f --- /dev/null +++ b/api/prisma/migrations/20260917120000_source_index_retry/migration.sql @@ -0,0 +1,10 @@ +-- Automatic retry of index failures that pass on their own (a model outage, a +-- lost connection). `indexAttempts` counts the failures since the last success +-- or manual retry and sets the pause before the next try; `indexRetryAt` is +-- when the reconciler may try again, null when it never will. Existing failed +-- rows keep null: nothing is retried behind anyone's back by the migration. +ALTER TABLE "Source" ADD COLUMN "indexAttempts" INTEGER NOT NULL DEFAULT 0, +ADD COLUMN "indexRetryAt" TIMESTAMP(3); + +-- The reconciler asks for due rows every minute across every knowledge. +CREATE INDEX "Source_indexRetryAt_idx" ON "Source"("indexRetryAt"); diff --git a/api/src/slices/reins/extraction/domain/textExtraction.service.spec.ts b/api/src/slices/reins/extraction/domain/textExtraction.service.spec.ts index be5b59fb..6d13edac 100644 --- a/api/src/slices/reins/extraction/domain/textExtraction.service.spec.ts +++ b/api/src/slices/reins/extraction/domain/textExtraction.service.spec.ts @@ -33,6 +33,8 @@ function pdf(id: string, overrides: Partial = {}): ISourceData { indexState: 'queued', indexError: 'left over from an earlier run', indexedAt: null, + indexAttempts: 0, + indexRetryAt: null, textState: 'none', textUrl: null, textError: null, diff --git a/api/src/slices/reins/knowledge/domain/indexReconcile.spec.ts b/api/src/slices/reins/knowledge/domain/indexReconcile.spec.ts index e8a8d380..e7c72405 100644 --- a/api/src/slices/reins/knowledge/domain/indexReconcile.spec.ts +++ b/api/src/slices/reins/knowledge/domain/indexReconcile.spec.ts @@ -19,6 +19,8 @@ function makeSource(id: string): ISourceData { indexState: 'queued', indexError: null, indexedAt: null, + indexAttempts: 0, + indexRetryAt: null, textState: 'none', textUrl: null, textError: null, @@ -34,6 +36,7 @@ function confirmed(id: string): ISourceIndexOutcome { status: 'indexed', indexed: true, error: null, + retryAt: null, }; } @@ -44,6 +47,7 @@ function moving(id: string): ISourceIndexOutcome { status: 'pending', indexed: false, error: 'still in LightRAG pipeline', + retryAt: null, }; } diff --git a/api/src/slices/reins/knowledge/knowledge.isolation.spec.ts b/api/src/slices/reins/knowledge/knowledge.isolation.spec.ts index 750e7ade..2705975c 100644 --- a/api/src/slices/reins/knowledge/knowledge.isolation.spec.ts +++ b/api/src/slices/reins/knowledge/knowledge.isolation.spec.ts @@ -68,6 +68,8 @@ function source( indexState: 'indexed', indexError: null, indexedAt: new Date(0), + indexAttempts: 0, + indexRetryAt: null, textState: 'none', textUrl: null, textError: null, diff --git a/api/src/slices/reins/migration/domain/migration.service.spec.ts b/api/src/slices/reins/migration/domain/migration.service.spec.ts index e5e52bf4..4c4a95ea 100644 --- a/api/src/slices/reins/migration/domain/migration.service.spec.ts +++ b/api/src/slices/reins/migration/domain/migration.service.spec.ts @@ -138,6 +138,8 @@ describe('instance isolation opt-in gate', () => { indexState: 'queued', indexError: null, indexedAt: null, + indexAttempts: 0, + indexRetryAt: null, textState: 'none', textUrl: null, textError: null, diff --git a/api/src/slices/reins/source/data/source.gateway.spec.ts b/api/src/slices/reins/source/data/source.gateway.spec.ts index 12b0d2c6..31b6758b 100644 --- a/api/src/slices/reins/source/data/source.gateway.spec.ts +++ b/api/src/slices/reins/source/data/source.gateway.spec.ts @@ -28,6 +28,8 @@ function makeSource(overrides: Partial = {}): ISourceData { indexState: 'queued', indexError: null, indexedAt: null, + indexAttempts: 0, + indexRetryAt: null, textState: 'none', textUrl: null, textError: null, @@ -57,29 +59,46 @@ function failed(message: string): ITrackStatus { interface IRowPatch { lightragDocId?: string | null; + indexState?: string; indexError?: string | null; indexedAt?: Date | null; + indexAttempts?: number; + indexRetryAt?: Date | null; } -// Tracks the three columns indexSources writes, applying only the keys each -// update actually sends - the way Prisma does - so a `{ indexError }` write -// cannot be mistaken for clearing the doc id. +// Tracks the columns the gateway writes, applying only the keys each update +// actually sends - the way Prisma does - so a `{ indexError }` write cannot +// be mistaken for clearing the doc id. function makePrismaStub(docIds: Record = {}) { + const states: Record = {}; const errors: Record = {}; const indexedAt: Record = {}; + const attempts: Record = {}; + const retryAt: Record = {}; return { docIds, + states, errors, indexedAt, + attempts, + retryAt, source: { findUnique: jest.fn(({ where }: { where: { id: string } }) => - Promise.resolve({ lightragDocId: docIds[where.id] ?? null }), + Promise.resolve({ + id: where.id, + knowledgeId: 'knowledge-1', + lightragDocId: docIds[where.id] ?? null, + indexAttempts: attempts[where.id] ?? 0, + }), ), update: jest.fn( ({ where, data }: { where: { id: string }; data: IRowPatch }) => { if ('lightragDocId' in data) docIds[where.id] = data.lightragDocId!; + if ('indexState' in data) states[where.id] = data.indexState!; if ('indexError' in data) errors[where.id] = data.indexError!; if ('indexedAt' in data) indexedAt[where.id] = data.indexedAt!; + if ('indexAttempts' in data) attempts[where.id] = data.indexAttempts!; + if ('indexRetryAt' in data) retryAt[where.id] = data.indexRetryAt!; return Promise.resolve({ id: where.id }); }, ), @@ -176,6 +195,7 @@ describe('SourceGateway.indexSources', () => { status: 'indexed', indexed: true, error: null, + retryAt: null, }, ]); expect(prisma.docIds['src-1']).toBe('track-1'); @@ -279,6 +299,7 @@ describe('SourceGateway.indexSources', () => { status: 'failed', indexed: false, error: 'embedding request rejected', + retryAt: null, }, ]); // No confirmation timestamp, so `indexed` stays false and the next run @@ -317,6 +338,7 @@ describe('SourceGateway.indexSources', () => { status: 'indexed', indexed: true, error: null, + retryAt: null, }, ]); expect(prisma.docIds['src-1']).toBe('track-existing'); @@ -463,6 +485,7 @@ describe('SourceGateway.indexSources', () => { status: 'indexed', indexed: true, error: null, + retryAt: null, }); expect(prisma.docIds['src-1']).toBe('doc-c8d0423fb8bc5700de256d6cb7fe89c8'); }); @@ -534,6 +557,7 @@ describe('SourceGateway.indexSources', () => { status: 'indexed', indexed: true, error: null, + retryAt: null, }); expect(prisma.docIds['src-1']).toBe('doc-stored'); }); @@ -605,6 +629,116 @@ describe('SourceGateway.indexSources', () => { }); }); +describe('SourceGateway: what a failure earns', () => { + const NOW = new Date('2026-09-17T00:20:00Z'); + + beforeEach(() => { + jest.useFakeTimers(); + jest.setSystemTime(NOW); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('schedules a retry after a failure that will pass on its own', async () => { + // The 2026-09-17 outage: Bedrock 503 for a quarter of an hour, LightRAG's + // own retry exhausted in seconds, nine good documents red until morning. + const prisma = makePrismaStub(); + const lightrag = makeLightragStub([ + failed('RetryError[]'), + ]); + const gateway = makeGateway(prisma, lightrag); + + const run = gateway.indexSources([makeSource()]); + await jest.advanceTimersByTimeAsync(POLL_MS); + const outcomes = await run; + + // Measured from when the failure was seen (after the first poll), not + // from when the run started. + const fiveMinutesOn = new Date(Date.now() + 5 * 60_000); + expect(prisma.attempts['src-1']).toBe(1); + expect(prisma.retryAt['src-1']).toEqual(fiveMinutesOn); + expect(outcomes[0].retryAt).toEqual(fiveMinutesOn); + // The handle stays: reprocessing the document LightRAG holds is the retry. + expect(prisma.docIds['src-1']).toBe('track-1'); + }); + + it('waits longer each time and gives up after the third retry', async () => { + const prisma = makePrismaStub(); + prisma.attempts['src-1'] = 3; + const lightrag = makeLightragStub([failed('RetryError[...]')]); + const gateway = makeGateway(prisma, lightrag); + + const run = gateway.indexSources([makeSource()]); + await jest.advanceTimersByTimeAsync(POLL_MS); + const outcomes = await run; + + expect(prisma.attempts['src-1']).toBe(4); + expect(prisma.retryAt['src-1']).toBeNull(); + expect(outcomes[0].retryAt).toBeNull(); + expect(prisma.errors['src-1']).toBe('RetryError[...]'); + }); + + it('never schedules a failure that is about the document', async () => { + const prisma = makePrismaStub(); + const lightrag = makeLightragStub([ + failed('File content contains only whitespace characters'), + ]); + const gateway = makeGateway(prisma, lightrag); + + const run = gateway.indexSources([makeSource()]); + await jest.advanceTimersByTimeAsync(POLL_MS); + await run; + + expect(prisma.attempts['src-1']).toBe(1); + expect(prisma.retryAt['src-1']).toBeNull(); + }); + + it('forgets the failures once LightRAG confirms the document', async () => { + const prisma = makePrismaStub({ 'src-1': 'track-existing' }); + prisma.attempts['src-1'] = 2; + prisma.retryAt['src-1'] = NOW; + const lightrag = makeLightragStub([processed()]); + const gateway = makeGateway(prisma, lightrag); + + await gateway.indexSources([makeSource({ indexError: 'RetryError[...]' })]); + + expect(prisma.attempts['src-1']).toBe(0); + expect(prisma.retryAt['src-1']).toBeNull(); + }); + + it('records a failure the single-row path meets the same way', async () => { + const prisma = makePrismaStub({ 'src-1': 'track-1' }); + const lightrag = makeLightragStub([failed('RetryError[...]')]); + const gateway = makeGateway(prisma, lightrag); + + await gateway.waitForSourceIndexed('src-1'); + + expect(prisma.states['src-1']).toBe('failed'); + expect(prisma.attempts['src-1']).toBe(1); + expect(prisma.retryAt['src-1']).toEqual(new Date('2026-09-17T00:25:00Z')); + }); + + it('leaves a slow document in flight for the reconciler instead of failing it', async () => { + // A 1 MB manual outlives the single-row wait routinely. Calling that a + // failure hid a working document behind a red badge and, worse, let the + // next click re-upload it into a duplicate refusal. + const prisma = makePrismaStub({ 'src-1': 'track-1' }); + const lightrag = makeLightragStub([stillProcessing()]); + const gateway = makeGateway(prisma, lightrag); + + const wait = gateway.waitForSourceIndexed('src-1'); + await jest.advanceTimersByTimeAsync(15 * 60_000 + POLL_MS); + await wait; + + expect(prisma.states['src-1']).toBe('processing'); + expect(prisma.errors['src-1']).toBeNull(); + expect(prisma.docIds['src-1']).toBe('track-1'); + expect(prisma.attempts['src-1'] ?? 0).toBe(0); + }); +}); + describe('SourceGateway.indexSource (one row, the Reindex button)', () => { it('does not upload a PDF whose text is still being extracted', async () => { const prisma = makePrismaStub({ 'src-1': null }); diff --git a/api/src/slices/reins/source/data/source.gateway.ts b/api/src/slices/reins/source/data/source.gateway.ts index 046bb9f3..6f0b3b07 100644 --- a/api/src/slices/reins/source/data/source.gateway.ts +++ b/api/src/slices/reins/source/data/source.gateway.ts @@ -35,6 +35,7 @@ import { SourceTypes, } from '../domain/source.types'; import { indexBudgetMs, pollIntervalMs } from '../domain/indexBudget'; +import { nextRetryAt } from '../domain/indexFailure'; import { SourceMapper } from './source.mapper'; // LightRAG processes ingested documents in a background pipeline. How long one @@ -43,8 +44,8 @@ import { SourceMapper } from './source.mapper'; // per-document figure can work. // The per-source wait (waitForSourceIndexed) has no batch to size a budget // from, so it keeps a flat deadline. Generous: a large PDF through entity -// extraction takes minutes, and an expired deadline marks the source failed -// (retryable), never silently indexed. +// extraction takes minutes, and an expired deadline leaves the source in +// flight for the reconciler to confirm, never silently indexed. const TRACK_POLL_INTERVAL_MS = 3_000; const TRACK_POLL_TIMEOUT_MS = 15 * 60 * 1000; @@ -134,8 +135,14 @@ function whereForStatus( switch (status) { case 'indexed': return { indexedAt: { not: null } }; + case 'retrying': + return { + indexedAt: null, + indexError: { not: null }, + indexRetryAt: { not: null }, + }; case 'failed': - return { indexedAt: null, indexError: { not: null } }; + return { indexedAt: null, indexError: { not: null }, indexRetryAt: null }; case 'pending': return { indexedAt: null, indexError: null }; case undefined: @@ -535,7 +542,7 @@ export class SourceGateway extends ISourceGateway { source.id, block.wait ? this.stillProcessing(source, block.reason) - : await this.fail(source, block.reason), + : await this.recordFailure(source, block.reason), ); continue; } @@ -565,7 +572,7 @@ export class SourceGateway extends ISourceGateway { continue; } - outcomes.set(source.id, await this.fail(source, message)); + outcomes.set(source.id, await this.recordFailure(source, message)); } } @@ -575,7 +582,7 @@ export class SourceGateway extends ISourceGateway { for (const s of sources) { results.push( outcomes.get(s.id) ?? - (await this.fail(s, 'LightRAG reported no state for this document')), + (await this.recordFailure(s, 'LightRAG reported no state for this document')), ); } return results; @@ -641,13 +648,18 @@ export class SourceGateway extends ISourceGateway { return outcomes; } - private failed(source: ISourceData, error: string): ISourceIndexOutcome { + private failed( + source: ISourceData, + error: string, + retryAt: Date | null = null, + ): ISourceIndexOutcome { return { sourceId: source.id, name: source.name, status: 'failed', indexed: false, error, + retryAt, }; } @@ -661,6 +673,7 @@ export class SourceGateway extends ISourceGateway { status: 'pending', indexed: false, error: reason, + retryAt: null, }; } @@ -671,6 +684,7 @@ export class SourceGateway extends ISourceGateway { status: 'indexed', indexed: true, error: null, + retryAt: null, }; } @@ -690,6 +704,8 @@ export class SourceGateway extends ISourceGateway { indexState: 'indexed', indexedAt: new Date(), indexError: null, + indexAttempts: 0, + indexRetryAt: null, }, }); return this.indexed(source); @@ -719,24 +735,39 @@ export class SourceGateway extends ISourceGateway { try { docId = await this.ingestByType(source); } catch (err) { - await this.updateIndexState(source.id, { - indexState: 'failed', - indexError: errorMessage(err), - }); + await this.recordFailure(source, errorMessage(err)); throw err; } await this.rememberHandle(source.id, docId); } - private async fail( + /** + * The one place a failure lands on a row. Besides the state and the message + * it decides whether the reconciler gets to try again: a transient reason + * (see indexFailure.ts) earns a retry after a pause that grows with each + * attempt; a permanent one, or a spent budget, leaves `indexRetryAt` null + * and the row honestly `failed`. + */ + private async recordFailure( source: ISourceData, error: string, ): Promise { + const row = await this.prisma.source.findUnique({ + where: { id: source.id }, + select: { indexAttempts: true }, + }); + const attempts = (row?.indexAttempts ?? 0) + 1; + const retryAt = nextRetryAt(error, attempts, new Date()); await this.prisma.source.update({ where: { id: source.id }, - data: { indexState: 'failed', indexError: error }, + data: { + indexState: 'failed', + indexError: error, + indexAttempts: attempts, + indexRetryAt: retryAt, + }, }); - return this.failed(source, error); + return this.failed(source, error, retryAt); } /** @@ -757,6 +788,7 @@ export class SourceGateway extends ISourceGateway { lightragDocId: handle, indexState: 'processing', indexError: null, + indexRetryAt: null, }, }); return this.stillProcessing(source, reason); @@ -934,7 +966,7 @@ export class SourceGateway extends ISourceGateway { outcomes.set( source.id, - await this.fail( + await this.recordFailure( source, failure.errorMessage ?? 'LightRAG failed to process it', ), @@ -974,7 +1006,8 @@ export class SourceGateway extends ISourceGateway { where: { id: sourceId }, }); if (!record) throw new NotFoundException(`Source ${sourceId} not found`); - if (!record.lightragDocId) { + const handle = record.lightragDocId; + if (!handle) { return this.mapper.toEntity(record); } @@ -982,7 +1015,7 @@ export class SourceGateway extends ISourceGateway { while (Date.now() < deadline) { const track = await this.lightrag.getTrackStatus( record.knowledgeId, - record.lightragDocId, + handle, ); // One track id can cover several documents (an archive upload); the // source is indexed only when every one of them is processed, failed as @@ -990,10 +1023,10 @@ export class SourceGateway extends ISourceGateway { // track yet - keep waiting. const failure = track.documents.find((d) => d.status === 'failed'); if (failure) { - await this.updateIndexState(sourceId, { - indexState: 'failed', - indexError: failure.errorMessage ?? 'processing failed', - }); + await this.recordFailure( + this.mapper.toEntity(record), + failure.errorMessage ?? 'LightRAG failed to process it', + ); return this.requireEntity(sourceId); } if ( @@ -1009,10 +1042,14 @@ export class SourceGateway extends ISourceGateway { } await sleep(TRACK_POLL_INTERVAL_MS); } - await this.updateIndexState(sourceId, { - indexState: 'failed', - indexError: `processing did not finish within ${TRACK_POLL_TIMEOUT_MS / 60000} minutes`, - }); + // Not a fault of the row: LightRAG is still working on it, the way a batch + // run leaves a slow document. Keeping the handle is what lets the + // reconciler confirm it when it lands. + await this.markInFlight( + this.mapper.toEntity(record), + handle, + `still processing after ${TRACK_POLL_TIMEOUT_MS / 60000} min - the reconciler will confirm it`, + ); return this.requireEntity(sourceId); } @@ -1026,6 +1063,12 @@ export class SourceGateway extends ISourceGateway { indexState: patch.indexState, ...(patch.indexError !== undefined && { indexError: patch.indexError }), ...(patch.indexedAt !== undefined && { indexedAt: patch.indexedAt }), + ...(patch.indexAttempts !== undefined && { + indexAttempts: patch.indexAttempts, + }), + ...(patch.indexRetryAt !== undefined && { + indexRetryAt: patch.indexRetryAt, + }), }, }); } @@ -1083,6 +1126,8 @@ export class SourceGateway extends ISourceGateway { indexState: 'queued', indexError: null, indexedAt: null, + indexAttempts: 0, + indexRetryAt: null, }); if (docId !== null) { await this.lightrag.deleteDocumentsByTrackIds(source.knowledgeId, [docId]); diff --git a/api/src/slices/reins/source/data/source.mapper.spec.ts b/api/src/slices/reins/source/data/source.mapper.spec.ts index b396c0b4..6eb4c12d 100644 --- a/api/src/slices/reins/source/data/source.mapper.spec.ts +++ b/api/src/slices/reins/source/data/source.mapper.spec.ts @@ -3,30 +3,59 @@ import { deriveIndexStatus } from './source.mapper'; describe('deriveIndexStatus', () => { it('is indexed once processing was confirmed, even with a stale error', () => { // LightRAG confirmed the content is searchable; an older error must not - // paint the row red. + // paint the row red, and a leftover retry slot means nothing either. expect( - deriveIndexStatus({ indexedAt: new Date(), indexError: 'old' }), + deriveIndexStatus({ + indexedAt: new Date(), + indexError: 'old', + indexRetryAt: new Date(), + }), ).toBe('indexed'); }); - it('is failed when the last run recorded an error and nothing succeeded', () => { + it('is failed when the last run recorded an error and nothing will retry it', () => { expect( - deriveIndexStatus({ indexedAt: null, indexError: 'rejected' }), + deriveIndexStatus({ + indexedAt: null, + indexError: 'rejected', + indexRetryAt: null, + }), ).toBe('failed'); }); + it('is retrying while the reconciler still owes the row another attempt', () => { + // Failed, yes, but for a reason that passes; showing that as a plain + // failure is what sent people to re-upload documents that were about to + // recover on their own. + expect( + deriveIndexStatus({ + indexedAt: null, + indexError: 'RetryError[...]', + indexRetryAt: new Date(), + }), + ).toBe('retrying'); + }); + it('is pending when nothing has been confirmed and nothing failed', () => { - expect(deriveIndexStatus({ indexedAt: null, indexError: null })).toBe( - 'pending', - ); + expect( + deriveIndexStatus({ + indexedAt: null, + indexError: null, + indexRetryAt: null, + }), + ).toBe('pending'); }); it('stays pending while a document sits in the pipeline with a handle', () => { // The row carries lightragDocId as a resume handle from ingest time, which // deriveIndexStatus deliberately ignores: only indexedAt proves the // document is searchable. - expect(deriveIndexStatus({ indexedAt: null, indexError: null })).toBe( - 'pending', - ); + expect( + deriveIndexStatus({ + indexedAt: null, + indexError: null, + indexRetryAt: null, + }), + ).toBe('pending'); }); }); diff --git a/api/src/slices/reins/source/data/source.mapper.ts b/api/src/slices/reins/source/data/source.mapper.ts index b4b02e5a..57c4f546 100644 --- a/api/src/slices/reins/source/data/source.mapper.ts +++ b/api/src/slices/reins/source/data/source.mapper.ts @@ -37,10 +37,13 @@ function parseSourceType(value: string): SourceTypes { export function deriveIndexStatus(record: { indexedAt: Date | null; indexError: string | null; + indexRetryAt: Date | null; }): SourceIndexStatusTypes { if (record.indexedAt !== null) return 'indexed'; - if (record.indexError !== null) return 'failed'; - return 'pending'; + if (record.indexError === null) return 'pending'; + // A scheduled retry is the reconciler's promise to try again; only a + // failure nobody will touch reads as failed. + return record.indexRetryAt !== null ? 'retrying' : 'failed'; } const TEXT_STATES: readonly SourceTextStateTypes[] = ['none', 'pending', 'ready', 'failed']; @@ -75,6 +78,8 @@ export class SourceMapper { indexState: parseIndexState(record.indexState), indexError: record.indexError ?? null, indexedAt: record.indexedAt ?? null, + indexAttempts: record.indexAttempts, + indexRetryAt: record.indexRetryAt ?? null, textState: parseTextState(record.textState), textUrl: record.textUrl ?? null, textError: record.textError ?? null, diff --git a/api/src/slices/reins/source/data/sourceArchive.writer.spec.ts b/api/src/slices/reins/source/data/sourceArchive.writer.spec.ts index 9041448f..1fe3907a 100644 --- a/api/src/slices/reins/source/data/sourceArchive.writer.spec.ts +++ b/api/src/slices/reins/source/data/sourceArchive.writer.spec.ts @@ -24,6 +24,8 @@ function makeSource(over: Partial = {}): ISourceData { indexState: 'indexed', indexError: null, indexedAt: new Date(0), + indexAttempts: 0, + indexRetryAt: null, textState: 'none', textUrl: null, textError: null, diff --git a/api/src/slices/reins/source/domain/source.service.ts b/api/src/slices/reins/source/domain/source.service.ts index 132eb77c..fc1b658a 100644 --- a/api/src/slices/reins/source/domain/source.service.ts +++ b/api/src/slices/reins/source/domain/source.service.ts @@ -470,6 +470,8 @@ export class SourceService { return this.gateway.updateIndexState(sourceId, { indexState: 'queued', indexError: null, + indexAttempts: 0, + indexRetryAt: null, }); } diff --git a/api/src/slices/reins/source/domain/source.types.ts b/api/src/slices/reins/source/domain/source.types.ts index 030e1ecd..5a80b422 100644 --- a/api/src/slices/reins/source/domain/source.types.ts +++ b/api/src/slices/reins/source/domain/source.types.ts @@ -4,11 +4,17 @@ export type SourceTypes = 'file' | 'url' | 'text'; /** * Per-source view of the last index run. `indexed` = LightRAG confirmed the - * document as processed; `failed` = the last run reported an error for it and - * nothing has succeeded since; `pending` = never sent, or sent and still - * waiting for a verdict. + * document as processed; `failed` = the last run reported an error for it, + * nothing has succeeded since and nothing will be tried without a person; + * `retrying` = it failed for a reason that passes (a model outage, a lost + * connection) and the reconciler will try again at `indexRetryAt`; + * `pending` = never sent, or sent and still waiting for a verdict. */ -export type SourceIndexStatusTypes = 'indexed' | 'pending' | 'failed'; +export type SourceIndexStatusTypes = + | 'indexed' + | 'pending' + | 'retrying' + | 'failed'; /** * Stored per-source ingestion state, the migration's resume marker: @@ -44,6 +50,10 @@ export interface ISourceData { indexState: SourceIndexStateTypes; indexError: string | null; indexedAt: Date | null; + /** Failed attempts since the last success or manual retry. */ + indexAttempts: number; + /** When the reconciler may retry a failed row; null when it never will. */ + indexRetryAt: Date | null; textState: SourceTextStateTypes; textUrl: string | null; textError: string | null; @@ -61,6 +71,8 @@ export interface ISourceIndexStatePatch { indexState: SourceIndexStateTypes; indexError?: string | null; indexedAt?: Date | null; + indexAttempts?: number; + indexRetryAt?: Date | null; } export interface ISourceFilter { @@ -135,6 +147,8 @@ export interface ISourceIndexOutcome { indexed: boolean; /** Set for `failed`, and for `pending` as the reason the wait ended. */ error: string | null; + /** For `failed`: when the reconciler will try again, null when it will not. */ + retryAt: Date | null; } export interface ICreateSourceData { diff --git a/api/src/slices/reins/source/dtos/source.dto.ts b/api/src/slices/reins/source/dtos/source.dto.ts index 34f2c05e..947a478d 100644 --- a/api/src/slices/reins/source/dtos/source.dto.ts +++ b/api/src/slices/reins/source/dtos/source.dto.ts @@ -10,6 +10,7 @@ import { export const SOURCE_INDEX_STATUSES: readonly SourceIndexStatusTypes[] = [ 'indexed', 'pending', + 'retrying', 'failed', ]; @@ -37,6 +38,18 @@ export class SourceDto implements Omit { }) indexError: string | null; @ApiProperty({ type: String, nullable: true }) indexedAt: Date | null; + @ApiProperty({ + description: + 'Failed attempts since the source last indexed or was retried by hand; the reconciler stops retrying after three.', + }) + indexAttempts: number; + @ApiProperty({ + type: String, + nullable: true, + description: + 'When the reconciler will retry a failed source on its own; null once it will not (permanent failure, or the retries are spent).', + }) + indexRetryAt: Date | null; @ApiProperty({ enum: ['none', 'pending', 'ready', 'failed'], description: diff --git a/api/src/slices/reins/source/source.prisma b/api/src/slices/reins/source/source.prisma index 992b8117..ce53ffff 100644 --- a/api/src/slices/reins/source/source.prisma +++ b/api/src/slices/reins/source/source.prisma @@ -16,6 +16,12 @@ model Source { indexState String @default("queued") indexError String? indexedAt DateTime? + // Failed attempts since the source was last indexed or manually retried; + // sets the pause before the next automatic retry. Back to 0 on success. + indexAttempts Int @default(0) + // When the reconciler may retry this failed row; null when it must not + // (permanent failure, retries spent, or the row is not failed). + indexRetryAt DateTime? // Text extracted for a PDF that has no text layer (reins/extraction): // none | pending | ready | failed. `none` is both "not a PDF" and "the PDF // has its own text"; only `ready` carries a textUrl, and that text - not @@ -30,5 +36,6 @@ model Source { @@index([knowledgeId, createdAt]) @@index([lightragDocId]) @@index([indexState]) + @@index([indexRetryAt]) @@index([textState]) } From 33302f01ba35f621e5f396da4933b76c6dc124f8 Mon Sep 17 00:00:00 2001 From: ntorbinskiy Date: Thu, 17 Sep 2026 12:43:43 +0300 Subject: [PATCH 3/7] feat(reins): reconciler retries transient index failures on an idle pipeline --- .../domain/indexReconcile.service.ts | 173 ++++++++++--- .../knowledge/domain/indexReconcile.spec.ts | 140 ++++++++++- .../lightrag/data/lightragHttp.client.ts | 11 + .../reins/lightrag/domain/lightrag.types.ts | 4 + .../reins/source/data/source.gateway.spec.ts | 236 +++++++++++++++++- .../reins/source/data/source.gateway.ts | 197 +++++++++++++-- .../reins/source/domain/source.gateway.ts | 10 + .../reins/source/domain/source.service.ts | 9 + .../reins/source/domain/source.types.ts | 20 ++ 9 files changed, 732 insertions(+), 68 deletions(-) diff --git a/api/src/slices/reins/knowledge/domain/indexReconcile.service.ts b/api/src/slices/reins/knowledge/domain/indexReconcile.service.ts index 4a9695c7..170d8c45 100644 --- a/api/src/slices/reins/knowledge/domain/indexReconcile.service.ts +++ b/api/src/slices/reins/knowledge/domain/indexReconcile.service.ts @@ -5,15 +5,21 @@ import { OnModuleInit, } from '@nestjs/common'; import { SourceService } from '../../source/domain/source.service'; +import { + ISourceData, + ISourceRetryOutcome, + SourceRetryActionTypes, +} from '../../source/domain/source.types'; import { ILightragClient } from '../../lightrag/domain/lightrag.client'; const DEFAULT_INTERVAL_SEC = 60; /** - * How long to leave an idle pipeline alone after nudging it. A nudge only - * re-queues work LightRAG already holds, so the cost of one is a request; the - * cooldown exists so a genuinely unprocessable backlog cannot turn every - * reconcile pass into a restart. + * How long to leave a pipeline alone after nudging it. A nudge only re-queues + * work LightRAG already holds, so the cost of one is a request; the cooldown + * exists so a genuinely unprocessable backlog cannot turn every reconcile + * pass into a restart. Shared by both paths that nudge (a stalled pipeline, + * a due retry) so they cannot double up on one base. */ const RESTART_COOLDOWN_MS = 10 * 60 * 1000; @@ -21,9 +27,33 @@ function errorMessage(err: unknown): string { return err instanceof Error ? err.message : String(err); } +function groupByKnowledge(sources: ISourceData[]): Map { + const groups = new Map(); + for (const source of sources) { + const group = groups.get(source.knowledgeId); + if (group) group.push(source); + else groups.set(source.knowledgeId, [source]); + } + return groups; +} + +function countActions( + outcomes: ISourceRetryOutcome[], +): Record { + const counts: Record = { + reprocess: 0, + resent: 0, + indexed: 0, + failed: 0, + }; + for (const outcome of outcomes) counts[outcome.action] += 1; + return counts; +} + /** * Confirms documents LightRAG finished after the index run that submitted them - * had already stopped waiting. + * had already stopped waiting, and retries the ones that failed for a reason + * that passes. * * Why this exists: an index run waits a bounded time (see indexBudgetMs) and * then leaves whatever is still moving to be picked up later. On real content @@ -34,10 +64,19 @@ function errorMessage(err: unknown): string { * Index again and hope the timing works out. * * So a timer does it instead. Each pass asks LightRAG about the handles it - * already stored, stamps the ones that came back processed, and leaves the - * rest alone. It never uploads anything - re-sending a document is the Index - * action's job - so a pass costs a couple of status reads and cannot start - * work that costs money. + * already stored, stamps the ones that came back processed, records the ones + * LightRAG gave up on, and leaves the rest alone. + * + * The retry step is the second reason. Bedrock fails in waves of ten to + * fifteen minutes; LightRAG's own retry runs for seconds and marks the + * document failed, and a re-upload is refused as a duplicate of that failed + * copy, so nothing a person could press recovered it. A failure that reads + * as transient (indexFailure.ts) gets a retry slot on its row; when it comes + * due and the base's pipeline is idle, the row goes back in flight and the + * pipeline is nudged to reprocess what it holds. Only a document LightRAG no + * longer holds is uploaded again, and only within the row's attempt budget: + * that is the one way a pass can start work that costs money, and it is + * bounded by rows a person already asked to index. * * Ranch has no scheduler facility, so this is a plain `setInterval` in an * OnModuleInit service, the same pattern as chatSync.service and @@ -48,7 +87,8 @@ export class IndexReconcileService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(IndexReconcileService.name); private timer?: ReturnType; private running = false; - private lastRestartAt = 0; + /** Per knowledge: when its pipeline was last nudged, by either path. */ + private readonly lastNudgeAt = new Map(); constructor( private readonly sources: SourceService, @@ -91,25 +131,80 @@ export class IndexReconcileService implements OnModuleInit, OnModuleDestroy { this.running = true; try { const pending = await this.sources.findUnconfirmed(); - if (pending.length === 0) return 0; + let confirmed = 0; + // Per knowledge: sources still moving after this pass. Each base has + // its own pipeline to ask about, so the count is kept per base. + const stalled = new Map(); - const outcomes = await this.sources.confirmProcessed(pending); - const confirmed = outcomes.filter((o) => o.indexed).length; - const stillMoving = outcomes.filter((o) => o.status === 'pending').length; - - if (confirmed > 0) { - this.logger.log( - `reconcile confirmed ${confirmed} source(s); ${stillMoving} still in the pipeline`, - ); + if (pending.length > 0) { + const baseOf = new Map(pending.map((s) => [s.id, s.knowledgeId])); + const outcomes = await this.sources.confirmProcessed(pending); + confirmed = outcomes.filter((o) => o.indexed).length; + let stillMoving = 0; + for (const outcome of outcomes) { + if (outcome.status !== 'pending') continue; + stillMoving += 1; + const base = baseOf.get(outcome.sourceId); + if (base !== undefined) { + stalled.set(base, (stalled.get(base) ?? 0) + 1); + } + } + if (confirmed > 0) { + this.logger.log( + `reconcile confirmed ${confirmed} source(s); ${stillMoving} still in the pipeline`, + ); + } } - await this.restartIfStalled(stillMoving); + await this.retryDue(); + await this.restartIfStalled(stalled); return confirmed; } finally { this.running = false; } } + /** + * Failed rows whose retry slot has come. Each base is handled on its own: + * a busy pipeline is left to finish (the rows stay due for the next pass), + * a base nudged within the cooldown waits it out, and otherwise the rows + * go back in flight and the pipeline is told to reprocess what it holds. + * Failures here are swallowed per base: the retry is a bonus on top of a + * reconcile pass, never a reason to lose one. + */ + private async retryDue(): Promise { + const due = await this.sources.findDueForRetry(new Date()); + if (due.length === 0) return; + + for (const [knowledgeId, rows] of groupByKnowledge(due)) { + if (this.nudgedRecently(knowledgeId)) continue; + try { + const status = await this.lightrag.getPipelineStatus(knowledgeId); + if (status.busy) { + this.logger.log( + `${rows.length} source(s) due for retry in ${knowledgeId}; pipeline busy, next pass`, + ); + continue; + } + const outcomes = await this.sources.retryFailed(rows); + const counts = countActions(outcomes); + if (counts.reprocess > 0) { + // Stamped before the call: a restart that throws half-way may still + // have started the pipeline. + this.lastNudgeAt.set(knowledgeId, Date.now()); + await this.lightrag.restartPipeline(knowledgeId); + } + this.logger.warn( + `retrying ${rows.length} source(s) in ${knowledgeId}: ${counts.reprocess} reprocessed, ${counts.resent} re-sent, ${counts.indexed} already processed, ${counts.failed} failed again`, + ); + } catch (err) { + this.logger.error( + `retry in ${knowledgeId} failed: ${errorMessage(err)}`, + ); + } + } + } + /** * Documents waiting on a pipeline that is not running. * @@ -126,24 +221,28 @@ export class IndexReconcileService implements OnModuleInit, OnModuleDestroy { * remedy. Failures are swallowed: recovery is a bonus on top of a reconcile * pass, never a reason to lose one. */ - private async restartIfStalled(stillMoving: number): Promise { - if (stillMoving === 0) return; - if (Date.now() - this.lastRestartAt < RESTART_COOLDOWN_MS) return; + private async restartIfStalled(stalled: Map): Promise { + for (const [knowledgeId, count] of stalled) { + if (this.nudgedRecently(knowledgeId)) continue; + try { + const status = await this.lightrag.getPipelineStatus(knowledgeId); + if (status.busy) continue; - try { - const status = await this.lightrag.getPipelineStatus(); - if (status.busy) return; - - // Stamped before the call, not after: a restart that throws half-way may - // still have started the pipeline, and retrying it every minute is worse - // than waiting out one cooldown. - this.lastRestartAt = Date.now(); - await this.lightrag.restartPipeline(); - this.logger.warn( - `pipeline idle with ${stillMoving} source(s) waiting; re-queued the backlog`, - ); - } catch (err) { - this.logger.error(`pipeline restart failed: ${errorMessage(err)}`); + this.lastNudgeAt.set(knowledgeId, Date.now()); + await this.lightrag.restartPipeline(knowledgeId); + this.logger.warn( + `pipeline of ${knowledgeId} idle with ${count} source(s) waiting; re-queued the backlog`, + ); + } catch (err) { + this.logger.error( + `pipeline restart for ${knowledgeId} failed: ${errorMessage(err)}`, + ); + } } } + + private nudgedRecently(knowledgeId: string): boolean { + const at = this.lastNudgeAt.get(knowledgeId); + return at !== undefined && Date.now() - at < RESTART_COOLDOWN_MS; + } } diff --git a/api/src/slices/reins/knowledge/domain/indexReconcile.spec.ts b/api/src/slices/reins/knowledge/domain/indexReconcile.spec.ts index e7c72405..0d2f9a72 100644 --- a/api/src/slices/reins/knowledge/domain/indexReconcile.spec.ts +++ b/api/src/slices/reins/knowledge/domain/indexReconcile.spec.ts @@ -1,6 +1,11 @@ import { IndexReconcileService } from './indexReconcile.service'; import { SourceService } from '../../source/domain/source.service'; -import { ISourceData, ISourceIndexOutcome } from '../../source/domain'; +import { + ISourceData, + ISourceIndexOutcome, + ISourceRetryOutcome, + SourceRetryActionTypes, +} from '../../source/domain'; import { ILightragClient } from '../../lightrag/domain/lightrag.client'; import { IPipelineStatus } from '../../lightrag/domain/lightrag.types'; @@ -64,8 +69,12 @@ function makeService( restartPipeline: jest.fn(() => Promise.resolve()), }, ): IndexReconcileService { + const quiet: Partial = { + findDueForRetry: jest.fn(() => Promise.resolve([])), + retryFailed: jest.fn(() => Promise.resolve([])), + }; return new IndexReconcileService( - stub as SourceService, + { ...quiet, ...stub } as SourceService, lightrag as ILightragClient, ); } @@ -220,3 +229,130 @@ describe('IndexReconcileService: recovering a stalled pipeline', () => { expect(await service.reconcile()).toBe(1); }); }); + +describe('IndexReconcileService: retrying what failed for a passing reason', () => { + function due(id: string): ISourceData { + return { + ...makeSource(id), + indexState: 'failed', + indexStatus: 'retrying', + indexError: 'RetryError[]', + indexAttempts: 1, + indexRetryAt: new Date(0), + }; + } + + function retried( + id: string, + action: SourceRetryActionTypes, + ): ISourceRetryOutcome { + return { sourceId: id, name: `${id}.md`, action, error: null }; + } + + it('puts due rows back on an idle pipeline and nudges it', async () => { + const retryFailed = jest.fn(() => Promise.resolve([retried('src-1', 'reprocess')])); + const restartPipeline = jest.fn(() => Promise.resolve()); + const service = makeService( + { + findUnconfirmed: jest.fn(() => Promise.resolve([])), + findDueForRetry: jest.fn(() => Promise.resolve([due('src-1')])), + retryFailed, + }, + { + getPipelineStatus: jest.fn(() => Promise.resolve(pipeline(false))), + restartPipeline, + }, + ); + + await service.reconcile(); + + expect(retryFailed).toHaveBeenCalledWith([due('src-1')]); + expect(restartPipeline).toHaveBeenCalledWith('knowledge-1'); + }); + + it('leaves a busy pipeline to finish; the rows stay due', async () => { + const retryFailed = jest.fn(() => Promise.resolve([])); + const restartPipeline = jest.fn(() => Promise.resolve()); + const service = makeService( + { + findUnconfirmed: jest.fn(() => Promise.resolve([])), + findDueForRetry: jest.fn(() => Promise.resolve([due('src-1')])), + retryFailed, + }, + { + getPipelineStatus: jest.fn(() => Promise.resolve(pipeline(true))), + restartPipeline, + }, + ); + + await service.reconcile(); + + expect(retryFailed).not.toHaveBeenCalled(); + expect(restartPipeline).not.toHaveBeenCalled(); + }); + + it('does not nudge when nothing was left for the pipeline to reprocess', async () => { + const restartPipeline = jest.fn(() => Promise.resolve()); + const service = makeService( + { + findUnconfirmed: jest.fn(() => Promise.resolve([])), + findDueForRetry: jest.fn(() => Promise.resolve([due('src-1')])), + retryFailed: jest.fn(() => Promise.resolve([retried('src-1', 'resent')])), + }, + { + getPipelineStatus: jest.fn(() => Promise.resolve(pipeline(false))), + restartPipeline, + }, + ); + + await service.reconcile(); + + expect(restartPipeline).not.toHaveBeenCalled(); + }); + + it('shares the cooldown with the stall nudge on the same base', async () => { + const retryFailed = jest.fn(() => Promise.resolve([retried('src-2', 'reprocess')])); + const restartPipeline = jest.fn(() => Promise.resolve()); + const findDueForRetry = jest + .fn() + .mockResolvedValueOnce([]) + .mockResolvedValue([due('src-2')]); + const service = makeService( + { + findUnconfirmed: jest.fn(() => Promise.resolve([makeSource('src-1')])), + confirmProcessed: jest.fn(() => Promise.resolve([moving('src-1')])), + findDueForRetry, + retryFailed, + }, + { + getPipelineStatus: jest.fn(() => Promise.resolve(pipeline(false))), + restartPipeline, + }, + ); + + // First pass: the stall nudge fires. Second pass, a minute later: a retry + // comes due on the same base and has to wait the cooldown out. + await service.reconcile(); + await service.reconcile(); + + expect(restartPipeline).toHaveBeenCalledTimes(1); + expect(retryFailed).not.toHaveBeenCalled(); + }); + + it('keeps its confirmations when the retry step throws', async () => { + const service = makeService( + { + findUnconfirmed: jest.fn(() => Promise.resolve([makeSource('src-1')])), + confirmProcessed: jest.fn(() => Promise.resolve([confirmed('src-1')])), + findDueForRetry: jest.fn(() => Promise.resolve([due('src-2')])), + retryFailed: jest.fn(() => Promise.reject(new Error('502'))), + }, + { + getPipelineStatus: jest.fn(() => Promise.resolve(pipeline(false))), + restartPipeline: jest.fn(() => Promise.resolve()), + }, + ); + + expect(await service.reconcile()).toBe(1); + }); +}); diff --git a/api/src/slices/reins/lightrag/data/lightragHttp.client.ts b/api/src/slices/reins/lightrag/data/lightragHttp.client.ts index edae4b62..aae01aa1 100644 --- a/api/src/slices/reins/lightrag/data/lightragHttp.client.ts +++ b/api/src/slices/reins/lightrag/data/lightragHttp.client.ts @@ -520,12 +520,23 @@ function extractDocuments(body: unknown): IDocumentRecord[] { // bucket name is the fallback (they agree in practice). status: toProcessingStatus(doc.status ?? statusName), filePath: typeof doc.file_path === 'string' ? doc.file_path : null, + errorMessage: + typeof doc.error_msg === 'string' && doc.error_msg !== '' + ? doc.error_msg + : null, + updatedAt: parseTimestamp(doc.updated_at), }); } } return out; } +function parseTimestamp(value: unknown): Date | null { + if (typeof value !== 'string') return null; + const ms = Date.parse(value); + return Number.isNaN(ms) ? null : new Date(ms); +} + function toProcessingStatus(value: unknown): DocumentProcessingStatusTypes { if (value === 'processed' || value === 'failed' || value === 'pending') { return value; diff --git a/api/src/slices/reins/lightrag/domain/lightrag.types.ts b/api/src/slices/reins/lightrag/domain/lightrag.types.ts index 668e7a8b..a79aeb70 100644 --- a/api/src/slices/reins/lightrag/domain/lightrag.types.ts +++ b/api/src/slices/reins/lightrag/domain/lightrag.types.ts @@ -62,6 +62,10 @@ export interface IDocumentRecord { id: string; status: DocumentProcessingStatusTypes; filePath: string | null; + /** LightRAG's reason for a failed document, when it left one. */ + errorMessage: string | null; + /** When LightRAG last changed the document's status; null when it did not say. */ + updatedAt: Date | null; } export interface IQueryReference { diff --git a/api/src/slices/reins/source/data/source.gateway.spec.ts b/api/src/slices/reins/source/data/source.gateway.spec.ts index 31b6758b..b1484b5d 100644 --- a/api/src/slices/reins/source/data/source.gateway.spec.ts +++ b/api/src/slices/reins/source/data/source.gateway.spec.ts @@ -422,7 +422,7 @@ describe('SourceGateway.indexSources', () => { const prisma = makePrismaStub({ 'src-1': 'doc-existing' }); const lightrag = makeLightragStub( [processed()], - [{ id: 'doc-existing', status: 'processed', filePath: 'notes.txt' }], + [{ id: 'doc-existing', status: 'processed', filePath: 'notes.txt', errorMessage: null, updatedAt: null }], ); const gateway = makeGateway(prisma, lightrag); @@ -441,7 +441,7 @@ describe('SourceGateway.indexSources', () => { const prisma = makePrismaStub({ 'src-1': 'track-existing' }); const lightrag = makeLightragStub( [processed()], - [{ id: 'doc-other', status: 'processed', filePath: 'other.txt' }], + [{ id: 'doc-other', status: 'processed', filePath: 'other.txt', errorMessage: null, updatedAt: null }], ); const gateway = makeGateway(prisma, lightrag); @@ -520,6 +520,8 @@ describe('SourceGateway.indexSources', () => { id: 'doc-c8d0423fb8bc5700de256d6cb7fe89c8', status: 'processed', filePath: 'notes.txt', + errorMessage: null, + updatedAt: null, }, ], ); @@ -537,7 +539,7 @@ describe('SourceGateway.indexSources', () => { const prisma = makePrismaStub(); const lightrag = makeLightragStub( [processed()], - [{ id: 'doc-stored', status: 'processed', filePath: 'notes.txt' }], + [{ id: 'doc-stored', status: 'processed', filePath: 'notes.txt', errorMessage: null, updatedAt: null }], ); // LightRAG names only the file in this refusal, never the doc id, so the // id has to come from the listing. Ranch reaches this state whenever it @@ -566,7 +568,7 @@ describe('SourceGateway.indexSources', () => { const prisma = makePrismaStub(); const lightrag = makeLightragStub( [processed()], - [{ id: 'doc-stored', status: 'failed', filePath: 'notes.txt' }], + [{ id: 'doc-stored', status: 'failed', filePath: 'notes.txt', errorMessage: null, updatedAt: null }], ); lightrag.ingestText.mockRejectedValueOnce( new Error( @@ -608,7 +610,7 @@ describe('SourceGateway.indexSources', () => { const prisma = makePrismaStub(); const lightrag = makeLightragStub( [processed()], - [{ id: 'doc-stored', status: 'processing', filePath: 'notes.txt' }], + [{ id: 'doc-stored', status: 'processing', filePath: 'notes.txt', errorMessage: null, updatedAt: null }], ); // Same 409 as the adopt-by-filename case, but the stored copy has not // finished yet. Reporting a failure here made every overlapping run red @@ -784,3 +786,227 @@ describe('SourceGateway.indexSource (one row, the Reindex button)', () => { }); }); + +describe('SourceGateway.confirmProcessed: a document LightRAG gave up on', () => { + const VERDICT = new Date('2026-09-17T00:30:00Z'); + + function heldFailed( + id: string, + errorMessage: string | null, + updatedAt: Date | null = VERDICT, + ): IDocumentRecord { + return { id, status: 'failed', filePath: 'notes.txt', errorMessage, updatedAt }; + } + + it('records the failure and keeps the handle the retry will reprocess', async () => { + // Before: the handle was dropped and nothing written, so the row sat at + // `processing` with no way back - "Indexing…" for four days. + const prisma = makePrismaStub({ 'src-1': 'doc-1' }); + const lightrag = makeLightragStub( + [], + [heldFailed('doc-1', 'RetryError[]')], + ); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.confirmProcessed([ + makeSource({ indexState: 'processing', updatedAt: new Date(0) }), + ]); + + expect(outcomes[0].status).toBe('failed'); + expect(outcomes[0].retryAt).toBeInstanceOf(Date); + expect(prisma.states['src-1']).toBe('failed'); + expect(prisma.errors['src-1']).toContain('BedrockConnectionError'); + expect(prisma.attempts['src-1']).toBe(1); + expect(prisma.docIds['src-1']).toBe('doc-1'); + }); + + it('adopts the processed original a refusal names', async () => { + const prisma = makePrismaStub({ 'src-1': 'dup-1' }); + const lightrag = makeLightragStub( + [], + [ + heldFailed( + 'dup-1', + 'Identical content already exists under another filename. Original doc_id: doc-9, Status: processed', + ), + ], + ); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.confirmProcessed([ + makeSource({ indexState: 'processing', updatedAt: new Date(0) }), + ]); + + expect(outcomes[0].indexed).toBe(true); + expect(prisma.docIds['src-1']).toBe('doc-9'); + expect(prisma.indexedAt['src-1']).toBeInstanceOf(Date); + }); + + it('leaves alone a row re-queued after the verdict it is looking at', async () => { + // The retry put the row back in flight at 00:40; LightRAG's failed + // status is from 00:30 and the pipeline has simply not reached the + // document yet. Recording that as a new failure would spend an attempt + // on nothing. + const prisma = makePrismaStub({ 'src-1': 'doc-1' }); + const lightrag = makeLightragStub([], [heldFailed('doc-1', 'RetryError[...]')]); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.confirmProcessed([ + makeSource({ + indexState: 'processing', + updatedAt: new Date('2026-09-17T00:40:00Z'), + }), + ]); + + expect(outcomes[0].status).toBe('pending'); + expect(prisma.source.update).not.toHaveBeenCalled(); + }); + + it('treats a verdict with no timestamp as fresh', async () => { + const prisma = makePrismaStub({ 'src-1': 'doc-1' }); + const lightrag = makeLightragStub([], [heldFailed('doc-1', 'RetryError[...]', null)]); + const gateway = makeGateway(prisma, lightrag); + + await gateway.confirmProcessed([ + makeSource({ + indexState: 'processing', + updatedAt: new Date('2026-09-17T00:40:00Z'), + }), + ]); + + expect(prisma.states['src-1']).toBe('failed'); + }); +}); + +describe('SourceGateway.retryFailed', () => { + function heldAs( + id: string, + status: IDocumentRecord['status'], + ): IDocumentRecord { + return { id, status, filePath: 'notes.txt', errorMessage: null, updatedAt: null }; + } + + function failedRow(overrides: Partial = {}): ISourceData { + return makeSource({ + indexState: 'failed', + indexStatus: 'retrying', + indexError: 'RetryError[...]', + indexAttempts: 1, + indexRetryAt: new Date(0), + ...overrides, + }); + } + + it('puts a document LightRAG still holds back in flight for a reprocess', async () => { + const prisma = makePrismaStub({ 'src-1': 'doc-1' }); + const lightrag = makeLightragStub([], [heldAs('doc-1', 'failed')]); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.retryFailed([failedRow()]); + + expect(outcomes[0].action).toBe('reprocess'); + expect(prisma.states['src-1']).toBe('processing'); + expect(prisma.errors['src-1']).toBeNull(); + expect(prisma.retryAt['src-1']).toBeNull(); + expect(prisma.docIds['src-1']).toBe('doc-1'); + expect(lightrag.ingestText).not.toHaveBeenCalled(); + }); + + it('takes the failed original a refusal named as the handle', async () => { + // The row's own handle is the rejected duplicate, which LightRAG will + // never process; the original is what a reprocess finishes. + const prisma = makePrismaStub({ 'src-1': 'dup-1' }); + const lightrag = makeLightragStub([], [heldAs('doc-orig', 'failed')]); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.retryFailed([ + failedRow({ + indexError: + 'Identical content already exists under another filename. Original doc_id: doc-orig, Status: failed', + }), + ]); + + expect(outcomes[0].action).toBe('reprocess'); + expect(prisma.docIds['src-1']).toBe('doc-orig'); + }); + + it('resolves a refusal by filename through the listing', async () => { + const prisma = makePrismaStub({ 'src-1': null }); + const lightrag = makeLightragStub([], [heldAs('doc-stored', 'failed')]); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.retryFailed([ + failedRow({ + indexError: "LightRAG /documents/upload failed: 409 Document storage already contains 'notes.txt' (Status: failed)", + }), + ]); + + expect(outcomes[0].action).toBe('reprocess'); + expect(prisma.docIds['src-1']).toBe('doc-stored'); + }); + + it('uploads again when LightRAG holds nothing under the handle', async () => { + const prisma = makePrismaStub({ 'src-1': 'gone' }); + const lightrag = makeLightragStub([], []); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.retryFailed([failedRow()]); + + expect(outcomes[0].action).toBe('resent'); + expect(lightrag.ingestText).toHaveBeenCalledTimes(1); + expect(prisma.docIds['src-1']).toBe('track-1'); + expect(prisma.states['src-1']).toBe('processing'); + }); + + it('stamps a document that turned out processed after all', async () => { + const prisma = makePrismaStub({ 'src-1': 'doc-1' }); + const lightrag = makeLightragStub([], [heldAs('doc-1', 'processed')]); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.retryFailed([failedRow()]); + + expect(outcomes[0].action).toBe('indexed'); + expect(prisma.indexedAt['src-1']).toBeInstanceOf(Date); + expect(prisma.attempts['src-1']).toBe(0); + }); + + it('reports a re-upload that failed again, recorded by the upload path', async () => { + const prisma = makePrismaStub({ 'src-1': null }); + const lightrag = makeLightragStub([], []); + lightrag.ingestText.mockRejectedValueOnce(new Error('fetch failed')); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.retryFailed([failedRow()]); + + expect(outcomes[0]).toMatchObject({ action: 'failed', error: 'fetch failed' }); + expect(prisma.states['src-1']).toBe('failed'); + expect(prisma.attempts['src-1']).toBe(1); + }); +}); + +describe('SourceGateway.indexSources: a row the reconciler still owes a retry', () => { + it('reports it without uploading or spending the attempt', async () => { + const retryAt = new Date('2026-09-17T00:25:00Z'); + const prisma = makePrismaStub({ 'src-1': 'doc-1' }); + const lightrag = makeLightragStub( + [], + [{ id: 'doc-1', status: 'failed', filePath: 'notes.txt', errorMessage: 'RetryError[...]', updatedAt: null }], + ); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.indexSources([ + makeSource({ + indexState: 'failed', + indexStatus: 'retrying', + indexError: 'RetryError[...]', + indexAttempts: 1, + indexRetryAt: retryAt, + }), + ]); + + expect(outcomes[0]).toMatchObject({ status: 'failed', retryAt }); + expect(lightrag.ingestText).not.toHaveBeenCalled(); + expect(prisma.source.update).not.toHaveBeenCalled(); + expect(prisma.docIds['src-1']).toBe('doc-1'); + }); +}); diff --git a/api/src/slices/reins/source/data/source.gateway.ts b/api/src/slices/reins/source/data/source.gateway.ts index 6f0b3b07..3455dbe6 100644 --- a/api/src/slices/reins/source/data/source.gateway.ts +++ b/api/src/slices/reins/source/data/source.gateway.ts @@ -29,6 +29,7 @@ import { IUploadSourceStreamInput, IUploadedSourceFile, ISourceIndexOutcome, + ISourceRetryOutcome, ISourceTextStatePatch, SourceTextStateTypes, ISourceBreakdown, @@ -85,9 +86,25 @@ type IExistingIndexCheck = // row pending with a handle, and this is where it finally becomes indexed. | { kind: 'indexed'; docId: string } | { kind: 'inFlight'; trackId: string } + // LightRAG holds the document and gave up on it. `failedAt` is LightRAG's + // own timestamp for that verdict, so a caller can tell a fresh failure from + // one the row was already re-queued over. + | { + kind: 'failed'; + docId: string; + error: string | null; + failedAt: Date | null; + } | { kind: 'stale' } | { kind: 'unknown'; error: string }; +/** A document as either status endpoint describes it. */ +interface IHeldDocument { + status: DocumentProcessingStatusTypes; + errorMessage: string | null; + updatedAt: Date | null; +} + // "Identical content already exists under another filename. Original doc_id: // doc-abc123, Status: processed" - only worth adopting when that original is // itself processed; a failed original has nothing to offer. @@ -102,7 +119,7 @@ const DUPLICATE_OF = /Original doc_id:\s*(\S+?),?\s*Status:\s*(\w+)/i; const ALREADY_STORED = /Document storage already contains ['"]([^'"]+)['"]/i; interface IDocumentSnapshot { - byId: Map; + byId: Map; byName: Map; } @@ -532,6 +549,25 @@ export class SourceGateway extends ISourceGateway { outcomes.set(source.id, this.failed(source, existing.error)); continue; } + if (existing.kind === 'failed') { + // LightRAG gave up on the copy it holds. If the reconciler already + // owes this row a retry, uploading now would only collect the + // duplicate refusal and spend one of its attempts: report and leave + // it. Otherwise drop the claim and send it again, as for a document + // LightRAG lost. + if (source.indexRetryAt !== null) { + outcomes.set( + source.id, + this.failed( + source, + existing.error ?? 'LightRAG failed to process it', + source.indexRetryAt, + ), + ); + continue; + } + await this.forgetDocId(source.id); + } // A scanned PDF is only worth sending once its text exists. Sending the // file itself would fail inside LightRAG with "only whitespace", which @@ -633,6 +669,10 @@ export class SourceGateway extends ISourceGateway { ); continue; } + if (existing.kind === 'failed') { + outcomes.push(await this.recordHeldFailure(source, existing)); + continue; + } // 'stale' or 'unknown'. Neither is this pass's business: re-sending a // document is what the Index action is for, and an unreachable LightRAG // resolves itself. Report without writing. @@ -648,6 +688,106 @@ export class SourceGateway extends ISourceGateway { return outcomes; } + /** + * LightRAG holds the row's document as failed. Three readings, in order: the + * refusal names a processed original (adopt it, the content is searchable); + * LightRAG's verdict predates the row's last write, which means the row was + * re-queued after that verdict and the pipeline has not reached it yet + * (still in flight); or a fresh failure, recorded with the retry it earns. + * Before this the failure was never written at all: the handle was dropped + * and the row sat at `processing` with nothing to confirm, which is how a + * base read "indexing" for four days over one document. + */ + private async recordHeldFailure( + source: ISourceData, + held: { docId: string; error: string | null; failedAt: Date | null }, + ): Promise { + const adopted = adoptableDocId(held.error); + if (adopted !== null) return this.succeed(source, adopted); + if (held.failedAt !== null && held.failedAt < source.updatedAt) { + return this.stillProcessing(source, 'queued for reprocessing'); + } + return this.recordFailure( + source, + held.error ?? 'LightRAG failed to process it', + ); + } + + async findDueForRetry(now: Date): Promise { + const records = await this.prisma.source.findMany({ + where: { indexState: 'failed', indexRetryAt: { lte: now } }, + orderBy: { indexRetryAt: 'asc' }, + }); + return records.map((r) => this.mapper.toEntity(r)); + } + + async retryFailed(sources: ISourceData[]): Promise { + const snapshots = new Map(); + const outcomes: ISourceRetryOutcome[] = []; + for (const source of sources) { + let known = snapshots.get(source.knowledgeId); + if (!known) { + known = await this.snapshotDocuments(source.knowledgeId); + snapshots.set(source.knowledgeId, known); + } + outcomes.push(await this.retryOne(source, known)); + } + return outcomes; + } + + private async retryOne( + source: ISourceData, + known: IDocumentSnapshot, + ): Promise { + const record = await this.prisma.source.findUnique({ + where: { id: source.id }, + select: { lightragDocId: true }, + }); + // A refused re-upload names the copy LightRAG really holds. That copy is + // what a reprocess finishes, so it becomes the handle; the rejected + // upload's own id leads nowhere. + const handle = + this.documentNamedByRefusal(source.indexError, known) ?? + record?.lightragDocId ?? + null; + const held = handle === null ? undefined : known.byId.get(handle); + if (handle === null || held === undefined) { + // Nothing to reprocess: the failure happened before LightRAG kept + // anything (it was restarting, the upload bounced). Send it again. + try { + await this.indexSource(source); + return this.retried(source, 'resent', null); + } catch (err) { + // indexSource recorded it on the row already. + return this.retried(source, 'failed', errorMessage(err)); + } + } + if (held.status === 'processed') { + await this.succeed(source, handle); + return this.retried(source, 'indexed', null); + } + await this.markInFlight(source, handle, 'queued for reprocessing'); + return this.retried(source, 'reprocess', null); + } + + private documentNamedByRefusal( + message: string | null, + known: IDocumentSnapshot, + ): string | null { + if (message === null) return null; + const duplicate = DUPLICATE_OF.exec(message); + if (duplicate !== null) return duplicate[1]; + return this.resolveStoredByName(message, known.byName)?.id ?? null; + } + + private retried( + source: ISourceData, + action: ISourceRetryOutcome['action'], + error: string | null, + ): ISourceRetryOutcome { + return { sourceId: source.id, name: source.name, action, error }; + } + private failed( source: ISourceData, error: string, @@ -730,6 +870,7 @@ export class SourceGateway extends ISourceGateway { await this.updateIndexState(source.id, { indexState: 'processing', indexError: null, + indexRetryAt: null, }); let docId: string; try { @@ -812,7 +953,7 @@ export class SourceGateway extends ISourceGateway { byName: new Map(), }; for (const doc of documents) { - snapshot.byId.set(doc.id, doc.status); + snapshot.byId.set(doc.id, doc); if (doc.filePath !== null) { snapshot.byName.set(normalizeName(doc.filePath), doc); } @@ -860,7 +1001,7 @@ export class SourceGateway extends ISourceGateway { */ private async checkExistingIndex( source: ISourceData, - known: Map, + known: Map, ): Promise { // Ask about any source that carries a handle, not only ones we call // indexed: the handle is written at ingest time, so a source left pending @@ -882,44 +1023,52 @@ export class SourceGateway extends ISourceGateway { // for it, and a real share of the load that made it slow. Only a // handle the snapshot does not know - a track id from an ingest - is // worth a call. - const fromSnapshot = this.statusesFromSnapshot(storedId, known); - const statuses = - fromSnapshot.length > 0 - ? fromSnapshot - : ( - await this.lightrag.getTrackStatus(source.knowledgeId, storedId) - ).documents.map((d) => d.status); - - if (statuses.length === 0) { + const fromSnapshot = known.get(storedId); + const documents: IHeldDocument[] = fromSnapshot + ? [fromSnapshot] + : ( + await this.lightrag.getTrackStatus(source.knowledgeId, storedId) + ).documents.map((d) => ({ + status: d.status, + errorMessage: d.errorMessage, + // A track id is this row's own upload, so a failure on it is + // always fresh; only the listing carries a timestamp. + updatedAt: null, + })); + + if (documents.length === 0) { await this.forgetDocId(source.id); return { kind: 'stale' }; } - if (statuses.every((s) => s === 'processed')) { + if (documents.every((d) => d.status === 'processed')) { return { kind: 'indexed', docId: storedId }; } - if (statuses.some((s) => s === 'pending' || s === 'processing')) { + if ( + documents.some((d) => d.status === 'pending' || d.status === 'processing') + ) { return { kind: 'inFlight', trackId: storedId }; } + const failure = documents.find((d) => d.status === 'failed'); + if (failure) { + return { + kind: 'failed', + docId: storedId, + error: failure.errorMessage, + failedAt: failure.updatedAt, + }; + } } catch (err) { // Cannot tell either way, so keep the existing claim and report the // source as unverified for this run rather than re-ingesting blindly. return { kind: 'unknown', error: errorMessage(err) }; } - // Failed, or a state we do not treat as in flight: drop the claim so this - // run re-sends it. + // A state we do not treat as in flight: drop the claim so this run + // re-sends it. await this.forgetDocId(source.id); return { kind: 'stale' }; } - private statusesFromSnapshot( - docId: string, - known: Map, - ): DocumentProcessingStatusTypes[] { - const status = known.get(docId); - return status ? [status] : []; - } - private async forgetDocId(sourceId: string): Promise { await this.prisma.source.update({ where: { id: sourceId }, diff --git a/api/src/slices/reins/source/domain/source.gateway.ts b/api/src/slices/reins/source/domain/source.gateway.ts index 053aa2d0..5d8808cc 100644 --- a/api/src/slices/reins/source/domain/source.gateway.ts +++ b/api/src/slices/reins/source/domain/source.gateway.ts @@ -11,6 +11,7 @@ import { IUploadSourceStreamInput, IUploadedSourceFile, ISourceIndexOutcome, + ISourceRetryOutcome, ISourceBreakdown, ISourceTextStatePatch, SourceTextStateTypes, @@ -65,6 +66,15 @@ export abstract class ISourceGateway { abstract confirmProcessed( sources: ISourceData[], ): Promise; + /** Failed rows whose scheduled retry is due, across every knowledge. */ + abstract findDueForRetry(now: Date): Promise; + /** + * Give each failed row another go, the cheapest way that can work: adopt + * the document a refusal named, put a row LightRAG still holds back in + * flight for a reprocess (the caller nudges the pipeline), or upload again + * when LightRAG has nothing. Never waits. + */ + abstract retryFailed(sources: ISourceData[]): Promise; /** Hands the source to the retrieval service and marks it processing. */ abstract indexSource(source: ISourceData): Promise; /** diff --git a/api/src/slices/reins/source/domain/source.service.ts b/api/src/slices/reins/source/domain/source.service.ts index fc1b658a..f1e887d3 100644 --- a/api/src/slices/reins/source/domain/source.service.ts +++ b/api/src/slices/reins/source/domain/source.service.ts @@ -21,6 +21,7 @@ import { ISourceFilter, ISourceIndexOutcome, ISourcePage, + ISourceRetryOutcome, ISourceSelection, ISourceBreakdown, } from './source.types'; @@ -456,6 +457,14 @@ export class SourceService { return this.gateway.confirmProcessed(sources); } + findDueForRetry(now: Date): Promise { + return this.gateway.findDueForRetry(now); + } + + retryFailed(sources: ISourceData[]): Promise { + return this.gateway.retryFailed(sources); + } + /** * Hand the source over AND wait until the retrieval service reports it * processed — "indexed" means searchable, not merely submitted. Returns diff --git a/api/src/slices/reins/source/domain/source.types.ts b/api/src/slices/reins/source/domain/source.types.ts index 5a80b422..6ccf7c28 100644 --- a/api/src/slices/reins/source/domain/source.types.ts +++ b/api/src/slices/reins/source/domain/source.types.ts @@ -151,6 +151,26 @@ export interface ISourceIndexOutcome { retryAt: Date | null; } +/** + * What a retry did with one failed row. `reprocess`: LightRAG still holds the + * document, the row is back in flight and the caller has to nudge the + * pipeline; `resent`: LightRAG had nothing, so the document went up again; + * `indexed`: it had finished after all and the row is stamped; `failed`: the + * re-upload itself failed, recorded on the row with whatever retry it earns. + */ +export type SourceRetryActionTypes = + | 'reprocess' + | 'resent' + | 'indexed' + | 'failed'; + +export interface ISourceRetryOutcome { + sourceId: string; + name: string; + action: SourceRetryActionTypes; + error: string | null; +} + export interface ICreateSourceData { knowledgeId: string; type: SourceTypes; From 4b9c623c5f118e97cc23aa49a637d57723aa261c Mon Sep 17 00:00:00 2001 From: ntorbinskiy Date: Thu, 17 Sep 2026 12:45:16 +0300 Subject: [PATCH 4/7] feat(reins): Reindex hands a document LightRAG still holds to the retry path --- .../knowledge/domain/failureSummary.spec.ts | 17 +++- .../reins/knowledge/domain/failureSummary.ts | 11 ++- .../reins/source/data/source.gateway.spec.ts | 85 +++++++++++++++++++ .../reins/source/data/source.gateway.ts | 27 ++++++ .../reins/source/domain/source.gateway.ts | 9 ++ .../reins/source/domain/source.service.ts | 7 ++ 6 files changed, 152 insertions(+), 4 deletions(-) diff --git a/api/src/slices/reins/knowledge/domain/failureSummary.spec.ts b/api/src/slices/reins/knowledge/domain/failureSummary.spec.ts index 8659b5ae..b13e195b 100644 --- a/api/src/slices/reins/knowledge/domain/failureSummary.spec.ts +++ b/api/src/slices/reins/knowledge/domain/failureSummary.spec.ts @@ -4,6 +4,7 @@ function failures(n: number) { return Array.from({ length: n }, (_, i) => ({ name: `doc-${i + 1}.pdf`, error: 'only whitespace', + retryAt: null, })); } @@ -28,8 +29,18 @@ describe('summarizeFailures', () => { }); it('falls back to a wording for a failure with no message', () => { - expect(summarizeFailures([{ name: 'a.pdf', error: null }])).toBe( - '1 source(s) failed: a.pdf (unknown error)', - ); + expect( + summarizeFailures([{ name: 'a.pdf', error: null, retryAt: null }]), + ).toBe('1 source(s) failed: a.pdf (unknown error)'); + }); + + it('says how many of them the reconciler will retry on its own', () => { + const line = summarizeFailures([ + { name: 'a.pdf', error: 'RetryError[...]', retryAt: new Date() }, + { name: 'b.pdf', error: 'RetryError[...]', retryAt: new Date() }, + { name: 'c.pdf', error: 'only whitespace', retryAt: null }, + ])!; + expect(line.startsWith('3 source(s) failed: a.pdf')).toBe(true); + expect(line.endsWith('; 2 of them will be retried automatically')).toBe(true); }); }); diff --git a/api/src/slices/reins/knowledge/domain/failureSummary.ts b/api/src/slices/reins/knowledge/domain/failureSummary.ts index 486921fa..d7fbb99d 100644 --- a/api/src/slices/reins/knowledge/domain/failureSummary.ts +++ b/api/src/slices/reins/knowledge/domain/failureSummary.ts @@ -4,6 +4,8 @@ export const FAILURE_SUMMARY_LIMIT = 5; export interface IFailureLine { name: string; error: string | null; + /** Set when the reconciler will retry this one on its own. */ + retryAt: Date | null; } /** @@ -23,5 +25,12 @@ export function summarizeFailures(failures: IFailureLine[]): string | null { rest > 0 ? `; and ${rest} more, all listed under Sources > Failed` : ''; - return `${failures.length} source(s) failed: ${shown}${tail}`; + // Said once, in numbers, so the line does not send anyone to re-upload a + // document that is about to recover by itself. + const retrying = failures.filter((f) => f.retryAt !== null).length; + const retryNote = + retrying > 0 + ? `; ${retrying} of them will be retried automatically` + : ''; + return `${failures.length} source(s) failed: ${shown}${tail}${retryNote}`; } diff --git a/api/src/slices/reins/source/data/source.gateway.spec.ts b/api/src/slices/reins/source/data/source.gateway.spec.ts index b1484b5d..a12f869c 100644 --- a/api/src/slices/reins/source/data/source.gateway.spec.ts +++ b/api/src/slices/reins/source/data/source.gateway.spec.ts @@ -1010,3 +1010,88 @@ describe('SourceGateway.indexSources: a row the reconciler still owes a retry', expect(prisma.docIds['src-1']).toBe('doc-1'); }); }); + +describe('SourceGateway.requestRetry (the Retry button on a row)', () => { + function makeRecordStub(row: { + lightragDocId: string | null; + indexError: string | null; + indexedAt: Date | null; + }) { + const prisma = makePrismaStub({ 'src-1': row.lightragDocId }); + prisma.source.findUnique.mockImplementation(() => + Promise.resolve({ + id: 'src-1', + knowledgeId: 'knowledge-1', + lightragDocId: row.lightragDocId, + indexError: row.indexError, + indexedAt: row.indexedAt, + indexAttempts: 3, + }), + ); + return prisma; + } + + it('hands a row LightRAG still holds to the reconciler, due now, count reset', async () => { + const prisma = makeRecordStub({ + lightragDocId: 'doc-1', + indexError: 'RetryError[...]', + indexedAt: null, + }); + const gateway = makeGateway(prisma, makeLightragStub([])); + + expect(await gateway.requestRetry(makeSource())).toBe(true); + expect(prisma.attempts['src-1']).toBe(0); + expect(prisma.retryAt['src-1']).toBeInstanceOf(Date); + expect(prisma.states['src-1']).toBe('failed'); + }); + + it('reads a duplicate refusal as LightRAG holding the document', async () => { + const prisma = makeRecordStub({ + lightragDocId: null, + indexError: + 'Identical content already exists under another filename. Original doc_id: doc-9, Status: failed', + indexedAt: null, + }); + const gateway = makeGateway(prisma, makeLightragStub([])); + + expect(await gateway.requestRetry(makeSource())).toBe(true); + }); + + it('leaves a row with nothing to reprocess to the upload path', async () => { + const prisma = makeRecordStub({ + lightragDocId: null, + indexError: 'fetch failed', + indexedAt: null, + }); + const gateway = makeGateway(prisma, makeLightragStub([])); + + expect(await gateway.requestRetry(makeSource())).toBe(false); + expect(prisma.source.update).not.toHaveBeenCalled(); + }); + + it('leaves an indexed row to the upload path even with a stale error', async () => { + const prisma = makeRecordStub({ + lightragDocId: 'doc-1', + indexError: 'old', + indexedAt: new Date(0), + }); + const gateway = makeGateway(prisma, makeLightragStub([])); + + expect(await gateway.requestRetry(makeSource())).toBe(false); + }); +}); + +describe('SourceGateway.waitForSourceIndexed: a refusal naming a processed original', () => { + it('adopts it instead of recording a failure', async () => { + const prisma = makePrismaStub({ 'src-1': 'track-1' }); + const lightrag = makeLightragStub([duplicateOf('doc-9', 'processed')]); + const gateway = makeGateway(prisma, lightrag); + + const result = await gateway.waitForSourceIndexed('src-1'); + + expect(prisma.docIds['src-1']).toBe('doc-9'); + expect(prisma.indexedAt['src-1']).toBeInstanceOf(Date); + expect(prisma.errors['src-1']).toBeNull(); + expect(result.id).toBe('src-1'); + }); +}); diff --git a/api/src/slices/reins/source/data/source.gateway.ts b/api/src/slices/reins/source/data/source.gateway.ts index 3455dbe6..0e6d8210 100644 --- a/api/src/slices/reins/source/data/source.gateway.ts +++ b/api/src/slices/reins/source/data/source.gateway.ts @@ -713,6 +713,26 @@ export class SourceGateway extends ISourceGateway { ); } + async requestRetry(source: ISourceData): Promise { + const record = await this.prisma.source.findUnique({ + where: { id: source.id }, + select: { lightragDocId: true, indexError: true, indexedAt: true }, + }); + if (!record || record.indexError === null || record.indexedAt !== null) { + return false; + } + const heldByLightrag = + record.lightragDocId !== null || + DUPLICATE_OF.test(record.indexError) || + ALREADY_STORED.test(record.indexError); + if (!heldByLightrag) return false; + await this.prisma.source.update({ + where: { id: source.id }, + data: { indexState: 'failed', indexAttempts: 0, indexRetryAt: new Date() }, + }); + return true; + } + async findDueForRetry(now: Date): Promise { const records = await this.prisma.source.findMany({ where: { indexState: 'failed', indexRetryAt: { lte: now } }, @@ -1172,6 +1192,13 @@ export class SourceGateway extends ISourceGateway { // track yet - keep waiting. const failure = track.documents.find((d) => d.status === 'failed'); if (failure) { + // Same reading as the batch path: a refusal naming a processed + // original means the content is searchable under that id. + const adopted = adoptableDocId(failure.errorMessage); + if (adopted !== null) { + await this.succeed(this.mapper.toEntity(record), adopted); + return this.requireEntity(sourceId); + } await this.recordFailure( this.mapper.toEntity(record), failure.errorMessage ?? 'LightRAG failed to process it', diff --git a/api/src/slices/reins/source/domain/source.gateway.ts b/api/src/slices/reins/source/domain/source.gateway.ts index 5d8808cc..65abd043 100644 --- a/api/src/slices/reins/source/domain/source.gateway.ts +++ b/api/src/slices/reins/source/domain/source.gateway.ts @@ -75,6 +75,15 @@ export abstract class ISourceGateway { * when LightRAG has nothing. Never waits. */ abstract retryFailed(sources: ISourceData[]): Promise; + /** + * A person retrying a failed row. Returns true when the row was handed to + * the reconciler with a fresh attempt count, due now, because LightRAG + * still holds its document (the only path that can work then: a re-upload + * is refused as a duplicate, a delete is refused while the pipeline is + * busy); false when there is nothing to reprocess and the caller should + * upload it again. + */ + abstract requestRetry(source: ISourceData): Promise; /** Hands the source to the retrieval service and marks it processing. */ abstract indexSource(source: ISourceData): Promise; /** diff --git a/api/src/slices/reins/source/domain/source.service.ts b/api/src/slices/reins/source/domain/source.service.ts index f1e887d3..cef6534c 100644 --- a/api/src/slices/reins/source/domain/source.service.ts +++ b/api/src/slices/reins/source/domain/source.service.ts @@ -487,12 +487,19 @@ export class SourceService { /** * Retry a single failed source without touching the rest of the batch * (FR-032). Runs in the background; per-source state reports the outcome. + * + * A document LightRAG still holds as failed cannot be re-sent (the upload + * is refused as a duplicate of that copy) and cannot be deleted while the + * pipeline is busy, so for those the row is handed to the reconciler with + * a fresh attempt count: its next pass reprocesses the copy LightRAG has. + * The row reads `retrying` straight away. */ async reindexSource(knowledgeId: string, sourceId: string): Promise { const source = await this.gateway.findById(sourceId); if (!source || source.knowledgeId !== knowledgeId) { throw new NotFoundException(`Source ${sourceId} not found`); } + if (await this.gateway.requestRetry(source)) return; await this.requeueSource(sourceId); void this.indexSourceAndWait({ ...source, indexState: 'queued' }).catch( (err) => { From 592e8780550198763a11528bba338065fbb23edb Mon Sep 17 00:00:00 2001 From: ntorbinskiy Date: Thu, 17 Sep 2026 12:46:45 +0300 Subject: [PATCH 5/7] feat(admin): show a retrying source as such and offer Retry on a row --- .../knowledge/SourceStatusBadge.vue | 2 + .../components/knowledge/sources/Provider.vue | 43 ++++++++++++++++++- admin/slices/reins/data/knowledge.mapper.ts | 3 ++ admin/slices/reins/domain/format.ts | 10 +++++ admin/slices/reins/domain/knowledge.types.ts | 10 ++++- 5 files changed, 65 insertions(+), 3 deletions(-) diff --git a/admin/slices/reins/components/knowledge/SourceStatusBadge.vue b/admin/slices/reins/components/knowledge/SourceStatusBadge.vue index 5cb695ca..dd4560a1 100644 --- a/admin/slices/reins/components/knowledge/SourceStatusBadge.vue +++ b/admin/slices/reins/components/knowledge/SourceStatusBadge.vue @@ -10,6 +10,8 @@ const label = computed(() => { return 'Indexed'; case 'failed': return 'Failed'; + case 'retrying': + return 'Retrying'; case 'pending': return 'Pending'; default: diff --git a/admin/slices/reins/components/knowledge/sources/Provider.vue b/admin/slices/reins/components/knowledge/sources/Provider.vue index 871ef8f0..9c2dfa2a 100644 --- a/admin/slices/reins/components/knowledge/sources/Provider.vue +++ b/admin/slices/reins/components/knowledge/sources/Provider.vue @@ -9,8 +9,13 @@ import type { SourceIndexStatus, SourceType, } from '#reins/stores/knowledge'; -import { Download, Eye, ScanText, Trash2 } from 'lucide-vue-next'; -import { errorMessageOf, formatBytes, formatDate } from '#reins/domain'; +import { Download, Eye, RefreshCw, ScanText, Trash2 } from 'lucide-vue-next'; +import { + errorMessageOf, + formatBytes, + formatDate, + formatTime, +} from '#reins/domain'; import { Badge } from '#theme/components/ui/badge'; import { Button } from '#theme/components/ui/button'; import { Checkbox } from '#theme/components/ui/checkbox'; @@ -46,8 +51,12 @@ const STATUS_OPTIONS: { value: SourceIndexStatus | 'all'; label: string }[] = [ { value: 'all', label: 'All statuses' }, { value: 'indexed', label: 'Indexed' }, { value: 'pending', label: 'Pending' }, + { value: 'retrying', label: 'Retrying' }, { value: 'failed', label: 'Failed' }, ]; +// One upload plus the API's three automatic retries (MAX_INDEX_RETRIES in +// reins/source/domain/indexFailure.ts); only shown, never enforced here. +const MAX_INDEX_ATTEMPTS = 4; const TYPE_OPTIONS: { value: SourceType | 'all'; label: string }[] = [ { value: 'all', label: 'All types' }, { value: 'file', label: 'File' }, @@ -279,6 +288,15 @@ async function handleReextract(source: ISource) { await load(); } +// A failed row gets another go with a fresh attempt count. When LightRAG +// still holds the document the API schedules it for its reconciler (the row +// reads "Retrying" at once); otherwise it is uploaded again in the background. +async function handleRetry(source: ISource) { + await store.reindexSource(knowledgeId.value, source.id); + await load(); + if (refresh) await refresh(); +} + async function onAdded() { // The new rows land at the end of the list (oldest first), so a user sitting // on a later page or a filter would not see them; go back to a clean view. @@ -439,11 +457,22 @@ async function onAdded() { > {{ s.textError }} + + Attempt {{ s.indexAttempts + 1 }} of {{ MAX_INDEX_ATTEMPTS }}, + next try {{ formatTime(s.indexRetryAt) }} + + {{ s.indexError }} @@ -485,6 +514,16 @@ async function onAdded() { Re-extract text from {{ s.name }} +