fix(server): use dedicated client for /messages background ingest#1622
fix(server): use dedicated client for /messages background ingest#1622Algorhythmicss wants to merge 1 commit into
Conversation
The /messages endpoint queues ingestion onto a background AsyncWorker but the
queued task closed over the request-scoped `graphiti` client from the
`get_graphiti` FastAPI dependency. That dependency's `finally: await
client.close()` runs when the HTTP request returns (202), closing the Neo4j
driver before the queued job executes. The job then raised
`DriverError('Driver closed')`, and because the worker loop only caught
`CancelledError`, the exception silently killed the worker and every ingest
became a no-op (0 nodes persisted, /search always empty).
- Extract client construction into `build_graphiti(settings)`.
- `add_messages_task` now builds and closes its own client.
- Worker loop catches and logs generic exceptions so one bad job no longer
kills ingestion.
Verified on the Neo4j backend: entities and bi-temporal RELATES_TO edges are
created and /search returns facts.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
I have read the CLA Document and I hereby sign the CLA behalf on myself, e-mail: example@example.com or I have read the CLA Document and I hereby sign the CLA behalf of my company, e-mail: example@example.com Signature is valid for 6 months. This bot will be retriggered when the Contributor License Agreement comment has been provided. Posted by the CLA Assistant Lite bot. |
|
@Algorhythmicss Reproduced both issues: queued /messages jobs ran on a client FastAPI had already closed ('Driver closed'), and one failing job killed the worker loop permanently. With your change ingestion gets its own client and the worker survives failing jobs. Verified with live requests. |
Problem
POST /messagesreturns202but never persists anything —MATCH (n) RETURN count(n)stays0and/searchalways returns{"facts": []}.The endpoint queues
add_messages_taskonto the module-levelAsyncWorker, but the task closes over the request-scopedgraphiticlient from theget_graphitiFastAPI dependency. That dependency'sfinally: await client.close()runs as soon as the HTTP request returns (202) — before the queued job executes. When the worker later runs the job, the Neo4j driver is already closed →DriverError('Driver closed').Because the worker loop only catches
asyncio.CancelledError, that exception propagates out ofwhile Trueand silently kills the worker, so all subsequent ingests are dropped too with no log output.Fix
build_graphiti(settings)inzep_graphiti.py.add_messages_tasknow builds and closes its own client (independent of request scope).Testing
Verified on the Neo4j 5.26.2 backend (
docker compose up): after the fix, ingesting messages createsEntitynodes and bi-temporalRELATES_TOedges, andPOST /searchreturns the extracted facts withvalid_at/invalid_atmetadata.🤖 Generated with Claude Code