diff --git a/README.md b/README.md index fc02918..bf97b28 100644 --- a/README.md +++ b/README.md @@ -59,7 +59,7 @@ flowchart TD - **Distributed Rate Limiting**: Atomic Redis sliding-window counter per API key with configurable quotas and documented failure policies (`fail_open` vs `fail_closed`). - **Resilient Retries**: `tenacity`-powered exponential backoff for transient failures (timeouts, connection resets, 500, 502, 503, 504) while immediately failing on client errors (400, 422). - **Correlation ID Tracing**: Request IDs generated or accepted from `X-Request-ID` headers, injected into `structlog` contextvars, and traced across DB logs and responses. -- **HMAC-Signed Webhooks**: Background delivery of generation results with HMAC-SHA256 signatures (`X-Webhook-Signature`) and delivery retry tracking in PostgreSQL. +- **HMAC-Signed Webhooks**: Optional delivery of generation results with HMAC-SHA256 signatures (`X-Webhook-Signature`) and delivery retry tracking in PostgreSQL. - **Audit Logging**: Full audit trail of every request, token count, latency, and status in PostgreSQL. - **Production Observability**: Health checks (`/health`, `/ready`) and operational metrics (`/metrics`). - **OpenAPI 3.1 Documentation**: Clean interactive Swagger docs (`/docs`) and schema (`/openapi.json`). @@ -159,7 +159,6 @@ All errors return a standardized JSON envelope with a machine-readable `code`, h | `UPSTREAM_TIMEOUT` | 504 | Upstream provider timed out after max retries | | `UPSTREAM_UNAVAILABLE` | 503 | Upstream provider returned 5xx or connection failed | | `PROVIDER_CONFIGURATION_ERROR`| 500 | Gateway missing required upstream API credentials | -| `WEBHOOK_DELIVERY_FAILED` | 500 | Webhook callback failed after all retries | | `INTERNAL_ERROR` | 500 | Unhandled internal exception | --- @@ -382,7 +381,7 @@ The service is packaged as a standard container ready for deployment on any cont - **Streaming (SSE)**: Currently optimized for atomic request/response completions. Streaming SSE token responses are planned for a future release. - **Provider Count**: Currently includes OpenAI and Anthropic adapters. Additional providers (Google Gemini, Mistral, Ollama) can be added by implementing the `LLMProvider` interface. -- **Background Worker**: Webhook deliveries are currently executed via background async tasks. For high-volume enterprise deployments, offloading deliveries to Celery, ARQ, or BullMQ is recommended. +- **Background Worker**: Webhook deliveries currently run synchronously after generation auditing. For high-volume enterprise deployments, offloading deliveries to Celery, ARQ, or BullMQ is recommended. --- @@ -410,3 +409,14 @@ The service is packaged as a standard container ready for deployment on any cont ## License MIT License - see [LICENSE](LICENSE) for details. + +## Generation and webhook failure handling + +Generation audits record the upstream provider outcome independently of optional +webhook notifications. A webhook dispatch or persistence error is logged and its +transaction is rolled back; it does not replace a successful generation response +or mask the original provider exception. Webhook delivery remains synchronous and +uses the existing retry policy. + +Editable installation includes only the `app` package; migration scripts remain +available in the repository for Alembic commands. diff --git a/app/services/generation.py b/app/services/generation.py index 7494162..3ccc3d1 100644 --- a/app/services/generation.py +++ b/app/services/generation.py @@ -49,31 +49,6 @@ async def execute( try: response = await provider_adapter.generate(request, request_id) - audit_record = GenerationRequest( - id=request_id, - client_id=client_api_key.id, - provider=provider_name, - model=request.model, - status="success", - input_tokens=response.input_tokens, - output_tokens=response.output_tokens, - total_tokens=response.total_tokens, - latency_ms=response.latency_ms, - ) - self.db.add(audit_record) - await self.db.commit() - - if request.webhook_url: - await self.webhook_service.dispatch( - webhook_url=str(request.webhook_url), - request_id=request_id, - event_type="generation.completed", - payload=response.model_dump(), - db_session=self.db, - ) - - return response - except Exception as exc: status = "failed" error_code = "INTERNAL_ERROR" @@ -96,17 +71,57 @@ async def execute( self.db.add(audit_record) await self.db.commit() - if request.webhook_url: - await self.webhook_service.dispatch( - webhook_url=str(request.webhook_url), - request_id=request_id, - event_type="generation.failed", - payload={ - "code": error_code, - "message": error_message, - "request_id": request_id, - }, - db_session=self.db, - ) + await self._dispatch_webhook( + request, + request_id, + "generation.failed", + {"code": error_code, "message": error_message, "request_id": request_id}, + ) raise + + audit_record = GenerationRequest( + id=request_id, + client_id=client_api_key.id, + provider=provider_name, + model=request.model, + status="success", + input_tokens=response.input_tokens, + output_tokens=response.output_tokens, + total_tokens=response.total_tokens, + latency_ms=response.latency_ms, + ) + self.db.add(audit_record) + await self.db.commit() + + await self._dispatch_webhook( + request, + request_id, + "generation.completed", + response.model_dump(), + ) + return response + + async def _dispatch_webhook( + self, + request: ChatRequest, + request_id: str, + event_type: str, + payload: dict, + ) -> None: + """Keep optional notification failures separate from the generation outcome.""" + if not request.webhook_url: + return + try: + await self.webhook_service.dispatch( + webhook_url=str(request.webhook_url), + request_id=request_id, + event_type=event_type, + payload=payload, + db_session=self.db, + ) + except Exception: + await self.db.rollback() + logger.exception( + "Webhook dispatch failed", request_id=request_id, event_type=event_type + ) diff --git a/pyproject.toml b/pyproject.toml index 64ea7b5..f92a2b2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ dependencies = [ "pydantic>=2.7.0", "pydantic-settings>=2.2.0", "httpx>=0.27.0", - "sqlalchemy>=2.0.28", + "sqlalchemy[asyncio]>=2.0.28", "alembic>=1.13.1", "asyncpg>=0.29.0", "aiosqlite>=0.20.0", @@ -64,3 +64,6 @@ python_version = "3.12" warn_return_any = true warn_unused_configs = true ignore_missing_imports = true + +[tool.setuptools.packages.find] +include = ["app*"] diff --git a/tests/unit/test_generation.py b/tests/unit/test_generation.py new file mode 100644 index 0000000..cc76b39 --- /dev/null +++ b/tests/unit/test_generation.py @@ -0,0 +1,65 @@ +"""Generation outcomes must not depend on optional webhook persistence.""" + +from unittest.mock import AsyncMock, Mock + +import pytest +from sqlalchemy import select + +from app.models.generation import GenerationRequest +from app.schemas.generate import ChatRequest +from app.schemas.response import LLMResponse +from app.services.generation import GenerationService + + +@pytest.mark.asyncio +async def test_webhook_error_preserves_success(test_session, test_api_key): + api_key, _ = test_api_key + response = LLMResponse( + request_id="webhook-success", + provider="openai", + model="gpt-4o", + content="Hello", + input_tokens=1, + output_tokens=1, + total_tokens=2, + finish_reason="stop", + latency_ms=1, + ) + provider = Mock(provider_name="openai", generate=AsyncMock(return_value=response)) + service = GenerationService( + test_session, provider_factory=Mock(resolve=Mock(return_value=provider)) + ) + service.webhook_service.dispatch = AsyncMock(side_effect=RuntimeError("webhook DB failure")) + request = ChatRequest( + model="gpt-4o", + messages=[{"role": "user", "content": "Hi"}], + webhook_url="https://example.com/webhook", + ) + + assert await service.execute(request, api_key, response.request_id) == response + records = (await test_session.execute(select(GenerationRequest))).scalars().all() + assert len(records) == 1 + assert records[0].status == "success" + + +@pytest.mark.asyncio +async def test_webhook_error_preserves_provider_exception(test_session, test_api_key): + api_key, _ = test_api_key + failure = RuntimeError("provider unavailable") + provider = Mock(provider_name="openai", generate=AsyncMock(side_effect=failure)) + service = GenerationService( + test_session, provider_factory=Mock(resolve=Mock(return_value=provider)) + ) + service.webhook_service.dispatch = AsyncMock(side_effect=RuntimeError("webhook DB failure")) + request = ChatRequest( + model="gpt-4o", + messages=[{"role": "user", "content": "Hi"}], + webhook_url="https://example.com/webhook", + ) + + with pytest.raises(RuntimeError) as raised: + await service.execute(request, api_key, "webhook-failure") + assert raised.value is failure + record = (await test_session.execute(select(GenerationRequest))).scalar_one() + assert record.status == "failed" + assert record.error_message == "provider unavailable"