From dc317b66241ef8fe2dd2163e8ee0364c5987b8b0 Mon Sep 17 00:00:00 2001 From: ntorbinskiy Date: Fri, 18 Sep 2026 16:05:40 +0300 Subject: [PATCH 1/2] fix(reins): bound every LightRAG call so a hung LightRAG fails fast (CLEAN-100) --- .../lightrag/data/lightragHttp.client.spec.ts | 134 ++++++++ .../lightrag/data/lightragHttp.client.ts | 305 ++++++++++++------ .../reins/lightrag/domain/lightrag.types.ts | 17 + 3 files changed, 349 insertions(+), 107 deletions(-) create mode 100644 api/src/slices/reins/lightrag/data/lightragHttp.client.spec.ts diff --git a/api/src/slices/reins/lightrag/data/lightragHttp.client.spec.ts b/api/src/slices/reins/lightrag/data/lightragHttp.client.spec.ts new file mode 100644 index 00000000..a157c5f6 --- /dev/null +++ b/api/src/slices/reins/lightrag/data/lightragHttp.client.spec.ts @@ -0,0 +1,134 @@ +import { LightragHttpClient } from './lightragHttp.client'; +import { LightragTimeoutError } from '../domain/lightrag.types'; + +type FetchImpl = typeof fetch; + +type FetchMock = jest.Mock, [string, RequestInit?]>; + +/** A LightRAG that takes the connection and never answers, until aborted. */ +function hangingFetch(): FetchMock { + return jest.fn( + (_url: string, init?: RequestInit) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => { + reject( + Object.assign(new Error('This operation was aborted'), { + name: 'AbortError', + }), + ); + }); + }), + ); +} + +function makeClient(fetchImpl: FetchMock): LightragHttpClient { + return new LightragHttpClient({ + resolveConfig: () => + Promise.resolve({ url: 'http://lightrag:9621', apiKey: 'k', enabled: true }), + fetchImpl: fetchImpl as unknown as FetchImpl, + }); +} + +describe('LightragHttpClient: a LightRAG that never answers', () => { + beforeEach(() => { + jest.useFakeTimers(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('gives up on a query before the MCP client does, and says what it waited for', async () => { + // The MCP client stops at 60 s with a transport error that reads as "MCP + // is broken". Stopping first is what lets the tool answer in words. + const client = makeClient(hangingFetch()); + + const pending = client.query({ knowledgeId: 'k1', query: 'anything' }); + const outcome = expect(pending).rejects.toThrow( + 'LightRAG /query timed out after 50 s', + ); + await jest.advanceTimersByTimeAsync(50_000); + + await outcome; + await expect(pending).rejects.toBeInstanceOf(LightragTimeoutError); + }); + + it('does not give up on a query a second early', async () => { + const client = makeClient(hangingFetch()); + let settled = false; + + const pending = client + .query({ knowledgeId: 'k1', query: 'anything' }) + .catch(() => undefined) + .finally(() => { + settled = true; + }); + await jest.advanceTimersByTimeAsync(49_000); + expect(settled).toBe(false); + + await jest.advanceTimersByTimeAsync(1_000); + await pending; + expect(settled).toBe(true); + }); + + it('bounds the status reads the reconciler makes inside its lock', async () => { + // One unbounded listDocuments and no reconcile pass ever runs again. + const client = makeClient(hangingFetch()); + + const documents = expect(client.listDocuments('k1')).rejects.toThrow( + 'LightRAG /documents timed out after 30 s', + ); + const pipeline = expect(client.getPipelineStatus('k1')).rejects.toThrow( + 'LightRAG /documents/pipeline_status timed out after 30 s', + ); + const track = expect( + client.getTrackStatus('k1', 'track-9'), + ).rejects.toBeInstanceOf(LightragTimeoutError); + await jest.advanceTimersByTimeAsync(30_000); + + await Promise.all([documents, pipeline, track]); + }); + + it('hands the abort signal to fetch so the socket is released too', async () => { + const fetchImpl = hangingFetch(); + const client = makeClient(fetchImpl); + + const pending = client.listDocuments('k1').catch(() => undefined); + await jest.advanceTimersByTimeAsync(30_000); + await pending; + + const [, init] = fetchImpl.mock.calls[0]; + expect(init?.signal?.aborted).toBe(true); + }); +}); + +describe('LightragHttpClient: a LightRAG that answers', () => { + it('returns the answer and leaves no timer behind', async () => { + jest.useFakeTimers(); + try { + const fetchImpl: FetchMock = jest.fn((_url: string) => + Promise.resolve( + new Response(JSON.stringify({ statuses: {} }), { status: 200 }), + ), + ); + const client = makeClient(fetchImpl); + + expect(await client.listDocuments('k1')).toEqual([]); + expect(jest.getTimerCount()).toBe(0); + } finally { + jest.useRealTimers(); + } + }); + + it('passes any other failure through untouched', async () => { + const fetchImpl: FetchMock = jest.fn((_url: string) => + Promise.reject(new TypeError('fetch failed')), + ); + const client = makeClient(fetchImpl); + + const failure = client.listDocuments('k1'); + + await expect(failure).rejects.toThrow('fetch failed'); + await expect(failure).rejects.not.toBeInstanceOf(LightragTimeoutError); + }); +}); diff --git a/api/src/slices/reins/lightrag/data/lightragHttp.client.ts b/api/src/slices/reins/lightrag/data/lightragHttp.client.ts index bfb42315..2c31d294 100644 --- a/api/src/slices/reins/lightrag/data/lightragHttp.client.ts +++ b/api/src/slices/reins/lightrag/data/lightragHttp.client.ts @@ -20,10 +20,29 @@ import { IDocumentRecord, IPipelineStatus, LightragClientError, + LightragTimeoutError, } from '../domain/lightrag.types'; type FetchImpl = typeof fetch; +/** + * How long each kind of call may take before LightRAG counts as hung. None of + * these used to be bounded, so a LightRAG that accepted the connection and + * never answered held its caller for as long as the socket lived: agents saw + * the MCP client's own 60 s limit ("MCP error -32001"), and the index + * reconciler, which awaits these inside its lock, would never run again. + * The query limit sits under that 60 s on purpose, so the agent gets a + * sentence from the tool instead of a transport error. + */ +const HEALTH_TIMEOUT_MS = 2_000; +const READ_TIMEOUT_MS = 30_000; +const QUERY_TIMEOUT_MS = 50_000; +const GRAPH_TIMEOUT_MS = 60_000; +const WRITE_TIMEOUT_MS = 120_000; +// Deleting a large document recomputes the descriptions of every entity it +// touched; minutes are normal. +const DELETE_TIMEOUT_MS = 300_000; + export interface LightragRequestConfig { url: string; apiKey: string; @@ -68,22 +87,42 @@ export class LightragHttpClient extends ILightragClient { this.fetchImpl = options.fetchImpl ?? fetch; } - async health(): Promise { - const cfg = await this.requireEnabled(); + /** + * Runs one exchange with LightRAG (request and body read) under a deadline. + * An exchange cut short by the deadline surfaces as LightragTimeoutError; + * anything else passes through untouched. + */ + private async bounded( + path: string, + timeoutMs: number, + run: (signal: AbortSignal) => Promise, + ): Promise { const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(), 2000); + const timer = setTimeout(() => controller.abort(), timeoutMs); try { + return await run(controller.signal); + } catch (err) { + if (controller.signal.aborted) { + throw new LightragTimeoutError(path, timeoutMs); + } + throw err; + } finally { + clearTimeout(timer); + } + } + + async health(): Promise { + const cfg = await this.requireEnabled(); + return this.bounded('/health', HEALTH_TIMEOUT_MS, async (signal) => { const res = await this.fetchImpl(`${cfg.baseUrl}/health`, { method: 'GET', headers: this.headers(cfg.apiKey), - signal: controller.signal, + signal, }); await this.ensureOk(res, '/health'); const body: unknown = await res.json(); return { ok: true, configuration: extractRuntimeConfig(body) }; - } finally { - clearTimeout(timer); - } + }); } async ingestText(input: IIngestTextInput): Promise { @@ -91,18 +130,21 @@ export class LightragHttpClient extends ILightragClient { knowledgeId: input.knowledgeId, intent: 'write', }); - const res = await this.fetchImpl(`${cfg.baseUrl}/documents/text`, { - method: 'POST', - headers: this.headers(cfg.apiKey, { - 'content-type': 'application/json', - }), - body: JSON.stringify({ - text: input.text, - file_source: input.fileSource, - }), + return this.bounded('/documents/text', WRITE_TIMEOUT_MS, async (signal) => { + const res = await this.fetchImpl(`${cfg.baseUrl}/documents/text`, { + method: 'POST', + headers: this.headers(cfg.apiKey, { + 'content-type': 'application/json', + }), + body: JSON.stringify({ + text: input.text, + file_source: input.fileSource, + }), + signal, + }); + await this.ensureOk(res, '/documents/text'); + return this.extractDocId(res, '/documents/text'); }); - await this.ensureOk(res, '/documents/text'); - return this.extractDocId(res, '/documents/text'); } async ingestUrl(input: IIngestUrlInput): Promise { @@ -121,18 +163,21 @@ export class LightragHttpClient extends ILightragClient { input.url, ); } - const res = await this.fetchImpl(`${cfg.baseUrl}/documents/text`, { - method: 'POST', - headers: this.headers(cfg.apiKey, { - 'content-type': 'application/json', - }), - body: JSON.stringify({ - text, - file_source: input.fileSource ?? input.url, - }), + return this.bounded('/documents/text', WRITE_TIMEOUT_MS, async (signal) => { + const res = await this.fetchImpl(`${cfg.baseUrl}/documents/text`, { + method: 'POST', + headers: this.headers(cfg.apiKey, { + 'content-type': 'application/json', + }), + body: JSON.stringify({ + text, + file_source: input.fileSource ?? input.url, + }), + signal, + }); + await this.ensureOk(res, '/documents/text'); + return this.extractDocId(res, '/documents/text'); }); - await this.ensureOk(res, '/documents/text'); - return this.extractDocId(res, '/documents/text'); } private async fetchAsCleanText(url: string): Promise { @@ -171,13 +216,20 @@ export class LightragHttpClient extends ILightragClient { // now 404s, same drift that killed /documents/url). Upload saves the // file to the input dir and processes it in the background, returning a // track_id like the text endpoints. - const res = await this.fetchImpl(`${cfg.baseUrl}/documents/upload`, { - method: 'POST', - headers: this.headers(cfg.apiKey), - body: form, - }); - await this.ensureOk(res, '/documents/upload'); - return this.extractDocId(res, '/documents/upload'); + return this.bounded( + '/documents/upload', + WRITE_TIMEOUT_MS, + async (signal) => { + const res = await this.fetchImpl(`${cfg.baseUrl}/documents/upload`, { + method: 'POST', + headers: this.headers(cfg.apiKey), + body: form, + signal, + }); + await this.ensureOk(res, '/documents/upload'); + return this.extractDocId(res, '/documents/upload'); + }, + ); } async query(input: IQueryInput): Promise { @@ -185,21 +237,24 @@ export class LightragHttpClient extends ILightragClient { knowledgeId: input.knowledgeId, intent: 'read', }); - const res = await this.fetchImpl(`${cfg.baseUrl}/query`, { - method: 'POST', - headers: this.headers(cfg.apiKey, { - 'content-type': 'application/json', - }), - body: JSON.stringify({ - query: input.query, - mode: input.mode ?? 'hybrid', - top_k: input.topK ?? 10, - include_references: true, - }), + return this.bounded('/query', QUERY_TIMEOUT_MS, async (signal) => { + const res = await this.fetchImpl(`${cfg.baseUrl}/query`, { + method: 'POST', + headers: this.headers(cfg.apiKey, { + 'content-type': 'application/json', + }), + body: JSON.stringify({ + query: input.query, + mode: input.mode ?? 'hybrid', + top_k: input.topK ?? 10, + include_references: true, + }), + signal, + }); + await this.ensureOk(res, '/query'); + const body: unknown = await res.json(); + return extractQueryResult(body); }); - await this.ensureOk(res, '/query'); - const body: unknown = await res.json(); - return extractQueryResult(body); } async deleteDocumentsByTrackIds( @@ -214,21 +269,28 @@ export class LightragHttpClient extends ILightragClient { docIds.push(...ids); } if (docIds.length === 0) return; - const res = await this.fetchImpl( - `${cfg.baseUrl}/documents/delete_document`, - { - method: 'DELETE', - headers: this.headers(cfg.apiKey, { - 'content-type': 'application/json', - }), - body: JSON.stringify({ - doc_ids: docIds, - delete_file: false, - delete_llm_cache: false, - }), + await this.bounded( + '/documents/delete_document', + DELETE_TIMEOUT_MS, + async (signal) => { + const res = await this.fetchImpl( + `${cfg.baseUrl}/documents/delete_document`, + { + method: 'DELETE', + headers: this.headers(cfg.apiKey, { + 'content-type': 'application/json', + }), + body: JSON.stringify({ + doc_ids: docIds, + delete_file: false, + delete_llm_cache: false, + }), + signal, + }, + ); + await this.ensureOk(res, '/documents/delete_document'); }, ); - await this.ensureOk(res, '/documents/delete_document'); } /** @@ -254,30 +316,37 @@ export class LightragHttpClient extends ILightragClient { */ async listDocuments(knowledgeId: string): Promise { const cfg = await this.requireEnabled({ knowledgeId, intent: 'write' }); - const res = await this.fetchImpl(`${cfg.baseUrl}/documents`, { - method: 'GET', - headers: this.headers(cfg.apiKey), + return this.bounded('/documents', READ_TIMEOUT_MS, async (signal) => { + const res = await this.fetchImpl(`${cfg.baseUrl}/documents`, { + method: 'GET', + headers: this.headers(cfg.apiKey), + signal, + }); + await this.ensureOk(res, '/documents'); + const body: unknown = await res.json(); + return extractDocuments(body); }); - await this.ensureOk(res, '/documents'); - const body: unknown = await res.json(); - return extractDocuments(body); } private async fetchTrackStatus( cfg: ResolvedRequestConfig, trackId: string, ): Promise { - const res = await this.fetchImpl( - `${cfg.baseUrl}/documents/track_status/${encodeURIComponent(trackId)}`, - { - method: 'GET', - headers: this.headers(cfg.apiKey), - }, - ); - if (res.status === 404) return { documents: [] }; - await this.ensureOk(res, `/documents/track_status/${trackId}`); - const body: unknown = await res.json(); - return extractTrackStatus(body); + const path = `/documents/track_status/${trackId}`; + return this.bounded(path, READ_TIMEOUT_MS, async (signal) => { + const res = await this.fetchImpl( + `${cfg.baseUrl}/documents/track_status/${encodeURIComponent(trackId)}`, + { + method: 'GET', + headers: this.headers(cfg.apiKey), + signal, + }, + ); + if (res.status === 404) return { documents: [] }; + await this.ensureOk(res, path); + const body: unknown = await res.json(); + return extractTrackStatus(body); + }); } private async resolveDocIdsByTrackId( @@ -292,13 +361,20 @@ export class LightragHttpClient extends ILightragClient { const cfg = await this.requireEnabled( knowledgeId ? { knowledgeId, intent: 'read' } : undefined, ); - const res = await this.fetchImpl(`${cfg.baseUrl}/graph/label/list`, { - method: 'GET', - headers: this.headers(cfg.apiKey), - }); - await this.ensureOk(res, '/graph/label/list'); - const body: unknown = await res.json(); - return extractLabels(body); + return this.bounded( + '/graph/label/list', + GRAPH_TIMEOUT_MS, + async (signal) => { + const res = await this.fetchImpl(`${cfg.baseUrl}/graph/label/list`, { + method: 'GET', + headers: this.headers(cfg.apiKey), + signal, + }); + await this.ensureOk(res, '/graph/label/list'); + const body: unknown = await res.json(); + return extractLabels(body); + }, + ); } async getGraph(input: IGetGraphInput): Promise { @@ -314,16 +390,19 @@ export class LightragHttpClient extends ILightragClient { if (input.maxNodes !== undefined) { params.set('max_nodes', String(input.maxNodes)); } - const res = await this.fetchImpl( - `${cfg.baseUrl}/graphs?${params.toString()}`, - { - method: 'GET', - headers: this.headers(cfg.apiKey), - }, - ); - await this.ensureOk(res, '/graphs'); - const body: unknown = await res.json(); - return extractGraph(body); + return this.bounded('/graphs', GRAPH_TIMEOUT_MS, async (signal) => { + const res = await this.fetchImpl( + `${cfg.baseUrl}/graphs?${params.toString()}`, + { + method: 'GET', + headers: this.headers(cfg.apiKey), + signal, + }, + ); + await this.ensureOk(res, '/graphs'); + const body: unknown = await res.json(); + return extractGraph(body); + }); } /** @@ -335,13 +414,19 @@ export class LightragHttpClient extends ILightragClient { const cfg = await this.requireEnabled( knowledgeId ? { knowledgeId, intent: 'write' } : undefined, ); - const res = await this.fetchImpl( - `${cfg.baseUrl}/documents/pipeline_status`, - { method: 'GET', headers: this.headers(cfg.apiKey) }, + return this.bounded( + '/documents/pipeline_status', + READ_TIMEOUT_MS, + async (signal) => { + const res = await this.fetchImpl( + `${cfg.baseUrl}/documents/pipeline_status`, + { method: 'GET', headers: this.headers(cfg.apiKey), signal }, + ); + await this.ensureOk(res, '/documents/pipeline_status'); + const body: unknown = await res.json(); + return extractPipelineStatus(body); + }, ); - await this.ensureOk(res, '/documents/pipeline_status'); - const body: unknown = await res.json(); - return extractPipelineStatus(body); } /** @@ -357,11 +442,17 @@ export class LightragHttpClient extends ILightragClient { const cfg = await this.requireEnabled( knowledgeId ? { knowledgeId, intent: 'write' } : undefined, ); - const res = await this.fetchImpl( - `${cfg.baseUrl}/documents/reprocess_failed`, - { method: 'POST', headers: this.headers(cfg.apiKey) }, + await this.bounded( + '/documents/reprocess_failed', + READ_TIMEOUT_MS, + async (signal) => { + const res = await this.fetchImpl( + `${cfg.baseUrl}/documents/reprocess_failed`, + { method: 'POST', headers: this.headers(cfg.apiKey), signal }, + ); + await this.ensureOk(res, '/documents/reprocess_failed'); + }, ); - await this.ensureOk(res, '/documents/reprocess_failed'); } private async requireEnabled( diff --git a/api/src/slices/reins/lightrag/domain/lightrag.types.ts b/api/src/slices/reins/lightrag/domain/lightrag.types.ts index 89a97756..533f721c 100644 --- a/api/src/slices/reins/lightrag/domain/lightrag.types.ts +++ b/api/src/slices/reins/lightrag/domain/lightrag.types.ts @@ -154,3 +154,20 @@ export class LightragClientError extends Error { this.name = 'LightragClientError'; } } + +/** + * LightRAG took the connection and never answered. Its own kind of failure + * because it is the signature of a LightRAG that is up but stuck: on + * 2026-09-18 its Postgres was moved to another node, /health kept answering + * in half a second, and every call that touched the database hung on a pool + * of dead connections until someone deleted the pod. + */ +export class LightragTimeoutError extends LightragClientError { + constructor( + path: string, + public readonly waitedMs: number, + ) { + super(`LightRAG ${path} timed out after ${waitedMs / 1000} s`, 504, path); + this.name = 'LightragTimeoutError'; + } +} From eae3b33329463cec2eedc551c35272787a937c6e Mon Sep 17 00:00:00 2001 From: ntorbinskiy Date: Fri, 18 Sep 2026 16:05:40 +0300 Subject: [PATCH 2/2] fix(reins): step aside from a hung LightRAG and tell the agent in words (CLEAN-100) --- api/src/slices/reins/README.md | 20 ++++ .../reins/knowledge/knowledge.tool.spec.ts | 41 +++++++- .../slices/reins/knowledge/knowledge.tool.ts | 12 ++- .../reins/source/data/source.gateway.spec.ts | 94 +++++++++++++++++++ .../reins/source/data/source.gateway.ts | 59 +++++++++++- 5 files changed, 223 insertions(+), 3 deletions(-) diff --git a/api/src/slices/reins/README.md b/api/src/slices/reins/README.md index 692f932c..64bb466f 100644 --- a/api/src/slices/reins/README.md +++ b/api/src/slices/reins/README.md @@ -118,6 +118,26 @@ LightRAG's `updated_at` on the failed verdict travels with the row when it is re-queued (`indexRequeuedOverAt`); seeing that same verdict again is "not reached yet", any other timestamp is a new failure. +## LightRAG can be up and answer nothing + +On 2026-09-18 LightRAG's Postgres pod was moved to another node on the Mazda +dev cluster. The data survived (PVC) and Postgres was back within a minute, but +LightRAG kept its pool of connections to the pod that was gone: `/health` and +`/documents/pipeline_status` answered in half a second, everything that touches +the database hung, and its log went silent. It does not recover by itself; +deleting the LightRAG pod is the fix. Agents saw only the MCP client's own +60 s limit (`MCP error -32001: Request timed out`), which sent people looking +at MCP. + +Every call in `LightragHttpClient` is therefore bounded and a cut-short one is +a `LightragTimeoutError`: 30 s for status reads, 50 s for a query (under the +MCP client's 60 s, so `query_knowledge` answers in words), two minutes for an +upload, five for a delete. When the document listing times out, the source +gateway stops talking to that instance for the run or pass (each upload would +only run into its own limit, one after another), leaves every row as it was, +and logs one error line that names the state: "answers /health but /documents +timed out ... restart the LightRAG pod". That line is the one to alert on. + ## Changing the extraction model requires clearing the LLM cache LightRAG caches every extraction and summary response in `lightrag_llm_cache`, diff --git a/api/src/slices/reins/knowledge/knowledge.tool.spec.ts b/api/src/slices/reins/knowledge/knowledge.tool.spec.ts index 3da1c649..8074584a 100644 --- a/api/src/slices/reins/knowledge/knowledge.tool.spec.ts +++ b/api/src/slices/reins/knowledge/knowledge.tool.spec.ts @@ -11,6 +11,7 @@ import { IKnowledgeGateway } from './domain/knowledge.gateway'; import { IAgentGateway } from '#/agent/agent/domain'; import { ITemplateGateway } from '#/agent/template/domain'; import { IAuthTokenPayload } from '#/user/auth/domain'; +import { LightragTimeoutError } from '../lightrag/domain/lightrag.types'; const K2_SECRET = 'the Grangemouth override code is 9944'; @@ -28,12 +29,13 @@ interface Harness { findExistingByIds: jest.Mock; } -function makeHarness(boundIds: string[]): Harness { +function makeHarness(boundIds: string[], failWith?: Error): Harness { const queriedIds: string[] = []; const knowledgeService = { query: jest.fn(async (knowledgeId: string) => { queriedIds.push(knowledgeId); + if (failWith) throw failWith; if (knowledgeId === 'k2') { // If the tool ever lets a query through to K2, the secret leaks // into the result and the assertions below catch it. @@ -338,3 +340,40 @@ describe('FR-006 — a multi-base answer attributes each part', () => { expect(parsed.answer).toBe('shared answer'); }); }); + +describe('query_knowledge when the knowledge base does not answer', () => { + test('says so in words and asks the agent to come back later', async () => { + // 2026-09-18: LightRAG hung on dead database connections and the agent + // only ever saw the MCP client's "-32001: Request timed out", which sent + // everyone looking at MCP. + const { tool } = makeHarness( + ['k1'], + new LightragTimeoutError('/query', 50_000), + ); + + const result = await tool.query( + { query: 'CX-5 advantages over RAV4' }, + null, + agentRequest('agent-1'), + ); + + expect(result.isError).toBeUndefined(); + expect(textOf(result)).toContain('Relays is not answering right now'); + expect(textOf(result)).toContain('LightRAG /query timed out after 50 s'); + expect(textOf(result)).toContain('try again in a few minutes'); + }); + + test('keeps the plain wording for any other failure', async () => { + const { tool } = makeHarness(['k1'], new Error('LightRAG /query failed: 502')); + + const result = await tool.query( + { query: 'anything' }, + null, + agentRequest('agent-1'), + ); + + expect(textOf(result)).toContain( + 'Relays could not be reached: LightRAG /query failed: 502', + ); + }); +}); diff --git a/api/src/slices/reins/knowledge/knowledge.tool.ts b/api/src/slices/reins/knowledge/knowledge.tool.ts index 4433a35a..6a4d5cba 100644 --- a/api/src/slices/reins/knowledge/knowledge.tool.ts +++ b/api/src/slices/reins/knowledge/knowledge.tool.ts @@ -8,6 +8,7 @@ import { ITemplateGateway } from '#/agent/template/domain'; import { IDynamicallyDescribedTool } from '#/mcp/interfaces/dynamic-description.interface'; import { KnowledgeService } from './domain/knowledge.service'; import { IKnowledgeGateway } from './domain/knowledge.gateway'; +import { LightragTimeoutError } from '../lightrag/domain/lightrag.types'; // The batching sentence is not stylistic advice: one call spends several // seconds inside the knowledge service composing an answer, and the service @@ -148,10 +149,19 @@ export class KnowledgeTool implements IDynamicallyDescribedTool { return { knowledge_id: id, knowledge_name, ...r }; } catch (e) { const message = e instanceof Error ? e.message : 'query failed'; + // Logged here because this failure never reaches the outer catch: + // it is folded into a normal result, and on 2026-09-18 that left + // an hour of timeouts with no line of ours in the API log. + this.logger.warn( + `query_knowledge failed for agent=${callerAgentId} knowledge=${id}: ${message}`, + ); return { knowledge_id: id, knowledge_name, - error: `Knowledge base ${knowledge_name ?? id} could not be reached: ${message}`, + error: + e instanceof LightragTimeoutError + ? `Knowledge base ${knowledge_name ?? id} is not answering right now (${message}). It is most likely restarting: say so to the user and try again in a few minutes instead of repeating the call straight away.` + : `Knowledge base ${knowledge_name ?? id} could not be reached: ${message}`, }; } }), 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 6c8168ea..effdb6b4 100644 --- a/api/src/slices/reins/source/data/source.gateway.spec.ts +++ b/api/src/slices/reins/source/data/source.gateway.spec.ts @@ -4,6 +4,7 @@ import { ISourceData } from '../domain/source.types'; import { ITrackStatus, IDocumentRecord, + LightragTimeoutError, } from '../../lightrag/domain/lightrag.types'; import { indexBudgetMs } from '../domain/indexBudget'; @@ -157,6 +158,7 @@ function makeLightragStub( Promise.resolve(queue.length > 1 ? queue.shift()! : queue[0]), ), listDocuments: jest.fn(() => Promise.resolve(documents)), + health: jest.fn(() => Promise.resolve({ ok: true, configuration: null })), }; } @@ -1204,3 +1206,95 @@ describe('SourceGateway.waitForSourceIndexed: a refusal naming a processed origi expect(result.id).toBe('src-1'); }); }); + +describe('SourceGateway: a LightRAG that takes the call and never answers', () => { + function hung() { + const lightrag = makeLightragStub([], []); + lightrag.listDocuments.mockRejectedValue( + new LightragTimeoutError('/documents', 30_000), + ); + return lightrag; + } + + it('does not upload into it: an index run reports every source and writes nothing', async () => { + // Each upload would hang to its own two-minute limit, one after another; + // on a base of a few hundred sources that is a day spent failing. + const prisma = makePrismaStub(); + const lightrag = hung(); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.indexSources([ + makeSource(), + makeSource({ id: 'src-2', name: 'b.txt' }), + ]); + + expect(outcomes.map((o) => o.status)).toEqual(['failed', 'failed']); + expect(outcomes[0].error).toContain('LightRAG is not answering'); + expect(outcomes[0].error).toContain('/documents timed out after 30 s'); + expect(lightrag.ingestText).not.toHaveBeenCalled(); + expect(lightrag.getTrackStatus).not.toHaveBeenCalled(); + expect(prisma.source.update).not.toHaveBeenCalled(); + }); + + it('leaves in-flight rows alone on a reconcile pass', async () => { + const prisma = makePrismaStub({ 'src-1': 'doc-1' }); + const lightrag = hung(); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.confirmProcessed([ + makeSource({ indexState: 'processing' }), + ]); + + // Not `pending`: a stalled-pipeline nudge would only hang the same way. + expect(outcomes[0].status).toBe('failed'); + expect(lightrag.getTrackStatus).not.toHaveBeenCalled(); + expect(prisma.source.update).not.toHaveBeenCalled(); + expect(prisma.docIds['src-1']).toBe('doc-1'); + }); + + it('keeps due retries due instead of spending them', async () => { + const prisma = makePrismaStub({ 'src-1': 'doc-1' }); + const lightrag = hung(); + const gateway = makeGateway(prisma, lightrag); + + const outcomes = await gateway.retryFailed([ + makeSource({ + indexState: 'failed', + indexError: 'RetryError[...]', + indexAttempts: 1, + indexRetryAt: new Date(0), + }), + ]); + + expect(outcomes[0].action).toBe('failed'); + expect(lightrag.ingestText).not.toHaveBeenCalled(); + expect(prisma.source.update).not.toHaveBeenCalled(); + }); + + it('checks /health to say which kind of outage it is', async () => { + const lightrag = hung(); + const gateway = makeGateway(makePrismaStub(), lightrag); + + await gateway.indexSources([makeSource()]); + + expect(lightrag.health).toHaveBeenCalledTimes(1); + }); + + it('still treats any other listing failure as non-fatal', async () => { + const prisma = makePrismaStub(); + const lightrag = makeLightragStub([processed()]); + lightrag.listDocuments.mockRejectedValue(new Error('LightRAG /documents failed: 502')); + const gateway = makeGateway(prisma, lightrag); + jest.useFakeTimers(); + try { + const run = gateway.indexSources([makeSource()]); + await jest.advanceTimersByTimeAsync(POLL_MS * 2); + const outcomes = await run; + + expect(outcomes[0].indexed).toBe(true); + expect(lightrag.ingestText).toHaveBeenCalledTimes(1); + } finally { + jest.useRealTimers(); + } + }); +}); diff --git a/api/src/slices/reins/source/data/source.gateway.ts b/api/src/slices/reins/source/data/source.gateway.ts index c59cebba..c52985fc 100644 --- a/api/src/slices/reins/source/data/source.gateway.ts +++ b/api/src/slices/reins/source/data/source.gateway.ts @@ -14,6 +14,7 @@ import { ILightragClient } from '../../lightrag/domain/lightrag.client'; import { IDocumentProcessingStatus, IDocumentRecord, + LightragTimeoutError, } from '../../lightrag/domain/lightrag.types'; import { ISourceGateway } from '../domain/source.gateway'; import { @@ -116,6 +117,14 @@ const ALREADY_STORED = /Document storage already contains ['"]([^'"]+)['"]/i; interface IDocumentSnapshot { byId: Map; byName: Map; + /** + * Set when LightRAG took the request and never answered: what to report for + * every source of that base instead of talking to it further. A LightRAG in + * that state answers nothing that touches its database, so each upload or + * status read would only run into its own timeout, one after another - a + * run over a few hundred sources would take a day to fail. + */ + hung: string | null; } function normalizeName(name: string): string { @@ -509,6 +518,11 @@ export class SourceGateway extends ISourceGateway { for (const source of sources) { const known = await snapshotFor(source.knowledgeId); + if (known.hung !== null) { + // Reported for this run only; the rows keep whatever they held. + outcomes.set(source.id, this.failed(source, known.hung)); + continue; + } const existing = await this.checkExistingIndex(source, known.byId); if (existing.kind === 'indexed') { @@ -655,6 +669,10 @@ export class SourceGateway extends ISourceGateway { known = await this.snapshotDocuments(source.knowledgeId); snapshots.set(source.knowledgeId, known); } + if (known.hung !== null) { + outcomes.push(this.failed(source, known.hung)); + continue; + } const existing = await this.checkExistingIndex(source, known.byId); if (existing.kind === 'indexed') { outcomes.push(await this.succeed(source, existing.docId)); @@ -766,6 +784,13 @@ export class SourceGateway extends ISourceGateway { } for (const [knowledgeId, rows] of byBase) { const known = await this.snapshotDocuments(knowledgeId); + if (known.hung !== null) { + // Nothing to retry against; the rows stay due for the next pass. + for (const source of rows) { + outcomes.push(this.retried(source, 'failed', known.hung)); + } + continue; + } const handles = await this.prisma.source.findMany({ where: { id: { in: rows.map((r) => r.id) } }, select: { id: true, lightragDocId: true }, @@ -1011,12 +1036,17 @@ export class SourceGateway extends ISourceGateway { private async snapshotDocuments( knowledgeId: string, ): Promise { - const empty: IDocumentSnapshot = { byId: new Map(), byName: new Map() }; + const empty: IDocumentSnapshot = { + byId: new Map(), + byName: new Map(), + hung: null, + }; try { const documents = await this.lightrag.listDocuments(knowledgeId); const snapshot: IDocumentSnapshot = { byId: new Map(), byName: new Map(), + hung: null, }; for (const doc of documents) { snapshot.byId.set(doc.id, doc); @@ -1028,11 +1058,38 @@ export class SourceGateway extends ISourceGateway { } catch (err) { // Not fatal: the per-source track lookups still work, the snapshot only // helps reconcile documents Ranch has lost the id for. + if (err instanceof LightragTimeoutError) { + await this.reportHung(knowledgeId, err); + return { ...empty, hung: `LightRAG is not answering (${err.message})` }; + } this.logger.warn(`listDocuments failed: ${errorMessage(err)}`); return empty; } } + /** + * The one log line to alert on. A timeout on a data call while /health still + * answers is the signature of LightRAG holding a pool of dead database + * connections (its Postgres was restarted or moved): it never recovers by + * itself, and until its pod is restarted every knowledge query hangs. + */ + private async reportHung( + knowledgeId: string, + err: LightragTimeoutError, + ): Promise { + let healthy = false; + try { + healthy = (await this.lightrag.health()).ok; + } catch { + healthy = false; + } + this.logger.error( + healthy + ? `LightRAG for ${knowledgeId} answers /health but ${err.path} timed out after ${err.waitedMs / 1000} s: its database connections are most likely stuck (a restarted Postgres leaves the pool dead). Restart the LightRAG pod; nothing recovers until then.` + : `LightRAG for ${knowledgeId} is not answering: ${err.message}`, + ); + } + /** * Pulls the filename out of LightRAG's refusal and returns the document it * refers to, status included. The caller decides what that status is worth: