Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`).
Expand Down Expand Up @@ -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 |

---
Expand Down Expand Up @@ -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.

---

Expand Down Expand Up @@ -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.
89 changes: 52 additions & 37 deletions app/services/generation.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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
)
5 changes: 4 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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*"]
65 changes: 65 additions & 0 deletions tests/unit/test_generation.py
Original file line number Diff line number Diff line change
@@ -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"
Loading