diff --git a/README.md b/README.md index c84b6365e2..11c052833c 100644 --- a/README.md +++ b/README.md @@ -121,6 +121,9 @@ from `graphiti-api` over Render's private network. Ingestion is asynchronous — `/messages` queues the episode and returns immediately, then Graphiti extracts entities and facts in the background. Give it 10–30 seconds before searching. +`"success": true` means the episode was queued, not that it was ingested: extraction happens +after the response, so a failure there shows up in the `graphiti-api` logs rather than in the +reply. Step 3 is how you confirm it actually landed. Set your URL and key once: @@ -168,6 +171,12 @@ export GRAPHITI_API_KEY=... # graphiti-api → Environment in the Render Dashb -d '{"group_ids": ["demo"], "query": "who leads the payments team?", "max_facts": 10}' ``` + > `/search` embeds your query before it can search, so it's the first endpoint to fail when + > `OPENAI_API_KEY` is unusable — steps 1–3 touch OpenAI either not at all or only in the + > background, and keep answering normally. A `429` naming an exhausted quota is billing on + > the OpenAI account, not a bad key: a new key on the same account returns the same thing. + > A `502` means the key itself was rejected. + Among the results you'll find both sides of the handover, each stamped with when it became true: diff --git a/server/graph_service/main.py b/server/graph_service/main.py index 0041eb7b60..0b03c6bdca 100644 --- a/server/graph_service/main.py +++ b/server/graph_service/main.py @@ -1,13 +1,18 @@ +import logging from contextlib import asynccontextmanager -from fastapi import Depends, FastAPI +import openai +from fastapi import Depends, FastAPI, Request from fastapi.responses import JSONResponse from graph_service.auth import require_api_key from graph_service.config import get_settings +from graph_service.openai_errors import describe_failure from graph_service.routers import ingest, retrieve from graph_service.zep_graphiti import initialize_graphiti, shutdown_graphiti +logger = logging.getLogger(__name__) + @asynccontextmanager async def lifespan(_: FastAPI): @@ -34,3 +39,15 @@ async def lifespan(_: FastAPI): @app.get('/healthcheck') async def healthcheck(): return JSONResponse(content={'status': 'healthy'}, status_code=200) + + +# Registered on the base class, so a subclass the SDK adds later is still mapped rather than +# falling through to a 500. Starlette resolves handlers along the MRO; openai_errors.py decides +# which status each failure becomes. +@app.exception_handler(openai.APIError) +async def handle_openai_error(_: Request, exc: openai.APIError) -> JSONResponse: + status_code, detail = describe_failure(exc) + # Logged as well as returned: the response goes to whoever made the request, and the operator + # reading logs is usually someone else. + logger.error('OpenAI call failed, returning %s: %s', status_code, exc) + return JSONResponse(content={'detail': detail}, status_code=status_code) diff --git a/server/graph_service/openai_errors.py b/server/graph_service/openai_errors.py new file mode 100644 index 0000000000..7fc20ed298 --- /dev/null +++ b/server/graph_service/openai_errors.py @@ -0,0 +1,40 @@ +"""Maps OpenAI's failures onto HTTP statuses, so they don't reach the caller as a bare 500. + +Every retrieval endpoint embeds its query before it can search, so an unusable OPENAI_API_KEY +surfaces on `/search` rather than at ingestion time. Unmapped, it reached the client as a bare 500 +`Internal Server Error`, which reads as a bug in the graph query — the actual cause was only in +the service logs, and only as a traceback. + +main.py registers the handler against openai.APIError, the base class, so a subclass the SDK adds +later is still mapped. openai.OpenAIError siblings that aren't APIError (LengthFinishReasonError, +ContentFilterFinishReasonError) stay 500s deliberately: they describe a response that arrived, not +an upstream that refused us. +""" + +import openai + + +def describe_failure(exc: openai.APIError) -> tuple[int, str]: + """Map an OpenAI failure to a status and a message that names what to go and fix.""" + if isinstance(exc, openai.RateLimitError): + # 429 either way, because OpenAI returns 429 for a real rate limit and for + # insufficient_quota alike, and only the caller can tell whether retrying is worth it. + # The distinction that matters to the operator is in the message, so pass it through. + return 429, ( + 'OpenAI rejected the request: either a rate limit, or the quota on the account ' + f'behind OPENAI_API_KEY is exhausted. Upstream said: {exc}' + ) + + if isinstance(exc, openai.AuthenticationError | openai.PermissionDeniedError): + # 502, not 401/403: the caller's GRAPHITI_API_KEY was accepted, and echoing OpenAI's + # status would tell them to go fix a credential they do not hold. + return 502, ( + "OpenAI rejected this service's credentials. Check that OPENAI_API_KEY is set to a " + f'valid key with access to the configured model. Upstream said: {exc}' + ) + + if isinstance(exc, openai.APIStatusError): + return 502, f'OpenAI returned an error. Upstream said: {exc}' + + # Everything else is a connection failure or a timeout: no response ever arrived. + return 504, f'Could not reach OpenAI. Upstream said: {exc}' diff --git a/server/graph_service/routers/ingest.py b/server/graph_service/routers/ingest.py index d035631057..8e94be7ec7 100644 --- a/server/graph_service/routers/ingest.py +++ b/server/graph_service/routers/ingest.py @@ -1,4 +1,5 @@ import asyncio +import logging from contextlib import asynccontextmanager from functools import partial @@ -9,6 +10,8 @@ from graph_service.dto import AddEntityNodeRequest, AddMessagesRequest, Message, Result from graph_service.zep_graphiti import ZepGraphitiDep +logger = logging.getLogger(__name__) + class AsyncWorker: def __init__(self): @@ -18,11 +21,23 @@ def __init__(self): async def worker(self): while True: try: - print(f'Got a job: (size of remaining queue: {self.queue.qsize()})') job = await self.queue.get() + # print, not logger.info: uvicorn leaves the root logger at WARNING, so an + # info-level line here never reaches the Render log. After the get, so it reports + # a job that arrived rather than one being awaited. + print(f'Got a job: (size of remaining queue: {self.queue.qsize()})') await job() except asyncio.CancelledError: break + except Exception: + # Drop the episode and keep going. This loop runs once per process, so an escaping + # exception ended ingestion for the life of the service — silently, since nothing + # awaits self.task — while /messages carried on answering 202 into a queue with no + # reader. An exhausted OPENAI_API_KEY quota was enough to trigger it. + # + # No retry: add_episode is not idempotent, and a dead key would be retried + # forever. A caller that needs the episode re-posts it. + logger.exception('Episode ingestion failed, dropping the job and continuing') async def start(self): self.task = asyncio.create_task(self.worker()) diff --git a/server/pyproject.toml b/server/pyproject.toml index d8101db474..ed04b8ff7a 100644 --- a/server/pyproject.toml +++ b/server/pyproject.toml @@ -13,6 +13,9 @@ dependencies = [ "pydantic-settings>=2.4.0", "uvicorn>=0.44.0", "httpx>=0.28.1", + # Imported directly by main.py, which maps openai's exceptions onto HTTP statuses. A + # transitive dependency of graphiti-core too, but declared here so that stays incidental. + "openai>=1.92.2", ] [project.optional-dependencies] diff --git a/server/tests/test_upstream_failures.py b/server/tests/test_upstream_failures.py new file mode 100644 index 0000000000..8832476c7e --- /dev/null +++ b/server/tests/test_upstream_failures.py @@ -0,0 +1,177 @@ +"""What the service does when OpenAI refuses the request. + +The failure that motivated these: an exhausted OpenAI quota turned `POST /search` into a bare +500 `Internal Server Error` and killed the ingestion worker silently, so `POST /messages` kept +answering 202 into a queue nobody drained. Both looked like bugs in the graph query. + +Need no OpenAI key and no database: the Graphiti client is a stand-in that raises. +""" + +import asyncio + +import httpx +import openai +import pytest +from fastapi.testclient import TestClient + +from graph_service.config import get_settings +from graph_service.main import app +from graph_service.routers.ingest import AsyncWorker +from graph_service.zep_graphiti import get_graphiti + +API_KEY = 'test-api-key-6Yp2Qk' +AUTH = {'Authorization': f'Bearer {API_KEY}'} + + +def _openai_error( + cls: type[openai.APIStatusError], status: int, message: str +) -> openai.APIStatusError: + """Build a real openai error, since the handler dispatches on the exception type.""" + request = httpx.Request('POST', 'https://api.openai.com/v1/embeddings') + response = httpx.Response(status, request=request) + return cls(message, response=response, body={'error': {'message': message}}) + + +# Built per use, not shared as constants: raising one instance in several cases accumulates +# __traceback__ and __context__ on it, which couples the cases through it. +def quota_error() -> openai.APIStatusError: + return _openai_error( + openai.RateLimitError, + 429, + 'You exceeded your current quota, please check your plan and billing details.', + ) + + +def bad_key_error() -> openai.APIStatusError: + return _openai_error(openai.AuthenticationError, 401, 'Incorrect API key provided: sk-xxx.') + + +class RaisingGraphiti: + """Stands in for ZepGraphiti, failing the way an unusable OPENAI_API_KEY makes it fail.""" + + def __init__(self, error: Exception): + self.error = error + + async def search(self, **_kwargs): + raise self.error + + +@pytest.fixture +def build_client(monkeypatch): + """A TestClient over the real app, with auth satisfied and no lifespan. + + No lifespan, so no Graphiti client is built and no index build runs; the dependency override + below is what the routers get instead. + """ + monkeypatch.setenv('GRAPHITI_API_KEY', API_KEY) + monkeypatch.setenv('OPENAI_API_KEY', 'sk-not-used') + get_settings.cache_clear() + + def _build(error: Exception): + app.dependency_overrides[get_graphiti] = lambda: RaisingGraphiti(error) + return TestClient(app) + + yield _build + app.dependency_overrides.clear() + get_settings.cache_clear() + + +SEARCH_BODY = {'group_ids': ['demo'], 'query': 'who leads the payments team?', 'max_facts': 10} + + +def test_exhausted_quota_is_not_an_internal_server_error(build_client): + """A refused OpenAI call is upstream's fault, and the response should say which upstream.""" + response = build_client(quota_error()).post('/search', json=SEARCH_BODY, headers=AUTH) + + assert response.status_code == 429 + detail = response.json()['detail'] + assert 'OpenAI' in detail + assert 'quota' in detail.lower() + + +def test_rejected_openai_key_is_reported_as_a_bad_gateway(build_client): + """The caller's key was fine; OPENAI_API_KEY is the one the operator has to go fix.""" + response = build_client(bad_key_error()).post('/search', json=SEARCH_BODY, headers=AUTH) + + assert response.status_code == 502 + assert 'OPENAI_API_KEY' in response.json()['detail'] + + +@pytest.mark.parametrize( + ('error', 'expected_status'), + [ + pytest.param( + _openai_error(openai.InternalServerError, 500, 'The server had an error.'), + 502, + id='openai-broke', + ), + pytest.param( + openai.APIConnectionError(request=httpx.Request('POST', 'https://api.openai.com')), + 504, + id='never-reached-openai', + ), + ], +) +def test_other_openai_failures_are_gateway_errors(build_client, error, expected_status): + """The two fallback branches: an error OpenAI returned, and one where it never answered. + + Neither is the caller's fault, so neither may come back as a 5xx pointing at this service. + """ + response = build_client(error).post('/search', json=SEARCH_BODY, headers=AUTH) + + assert response.status_code == expected_status + assert 'OpenAI' in response.json()['detail'] + + +async def _drain(worker: AsyncWorker): + """Run the worker until the queue is empty, then let the job it last took finish. + + Polls rather than joins: the worker never calls queue.task_done(), so queue.join() would + block forever. + """ + await worker.start() + for _ in range(200): + if worker.queue.empty(): + break + await asyncio.sleep(0.01) + await asyncio.sleep(0.05) + await worker.stop() + + +@pytest.mark.asyncio +async def test_worker_survives_a_failing_job(): + """A job that raises must not take the worker down with it. + + The worker outlives every request, so a job that kills it leaves /messages answering 202 + for the rest of the process's life while nothing is ingested. + """ + worker = AsyncWorker() + ran: list[str] = [] + + async def failing(): + ran.append('failing') + raise quota_error() + + async def succeeding(): + ran.append('succeeding') + + await worker.queue.put(failing) + await worker.queue.put(succeeding) + await _drain(worker) + + assert ran == ['failing', 'succeeding'], 'the worker stopped after the failing job' + + +@pytest.mark.asyncio +async def test_failing_job_is_logged(caplog): + """A dropped episode has to leave a trace, or ingestion fails where nobody can see it.""" + worker = AsyncWorker() + + async def failing(): + raise quota_error() + + await worker.queue.put(failing) + with caplog.at_level('ERROR'): + await _drain(worker) + + assert any(r.exc_info for r in caplog.records), 'no traceback was logged' diff --git a/server/uv.lock b/server/uv.lock index 4e36ea483e..2a8de56783 100644 --- a/server/uv.lock +++ b/server/uv.lock @@ -262,6 +262,7 @@ dependencies = [ { name = "fastapi" }, { name = "graphiti-core" }, { name = "httpx" }, + { name = "openai" }, { name = "pydantic-settings" }, { name = "uvicorn" }, ] @@ -286,6 +287,7 @@ requires-dist = [ { name = "graphiti-core", specifier = ">=0.28.2" }, { name = "graphiti-core", extras = ["falkordb"], marker = "extra == 'dev'", specifier = ">=0.28.2" }, { name = "httpx", specifier = ">=0.28.1" }, + { name = "openai", specifier = ">=1.92.2" }, { name = "pydantic", marker = "extra == 'dev'", specifier = ">=2.8.2" }, { name = "pydantic-settings", specifier = ">=2.4.0" }, { name = "pyright", marker = "extra == 'dev'", specifier = ">=1.1.380" },