Agent Zero
245 lines
8.0 KiB
Python
245 lines
8.0 KiB
Python
"""Transactional outbox for reliable domain event delivery.
|
|
|
|
Instead of publishing events directly to an in-process bus (which is lost
|
|
on crash/restart), domain events are written to the ``event_outbox`` table
|
|
**within the same database transaction** as the business operation. A
|
|
background worker then polls the outbox and publishes events to the
|
|
in-process event bus.
|
|
|
|
Usage in services::
|
|
|
|
from app.core.outbox import enqueue_outbox_event
|
|
|
|
await enqueue_outbox_event(db, tenant_id, "contact.created", {
|
|
"contact_id": str(contact.id),
|
|
"tenant_id": str(tenant_id),
|
|
})
|
|
# ... later, the transaction commits and the event is durable.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import redis.asyncio as aioredis
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── SQL statements (raw text for FOR UPDATE SKIP LOCKED) ────────────────────
|
|
|
|
_INSERT_SQL = text(
|
|
"""
|
|
INSERT INTO event_outbox (tenant_id, event_name, payload)
|
|
VALUES (:tenant_id, :event_name, CAST(:payload AS JSONB))
|
|
"""
|
|
)
|
|
|
|
_CLAIM_SQL = text(
|
|
"""
|
|
UPDATE event_outbox
|
|
SET status = 'processing',
|
|
updated_at = now()
|
|
WHERE id IN (
|
|
SELECT id FROM event_outbox
|
|
WHERE status = 'pending'
|
|
AND (next_retry_at IS NULL OR next_retry_at <= now())
|
|
ORDER BY created_at
|
|
LIMIT :batch_size
|
|
FOR UPDATE SKIP LOCKED
|
|
)
|
|
RETURNING id, tenant_id, event_name, payload, attempts, max_attempts
|
|
"""
|
|
)
|
|
|
|
_MARK_PUBLISHED_SQL = text(
|
|
"""
|
|
UPDATE event_outbox
|
|
SET status = 'published',
|
|
published_at = now(),
|
|
updated_at = now()
|
|
WHERE id = :id
|
|
"""
|
|
)
|
|
|
|
_FAIL_SQL = text(
|
|
"""
|
|
UPDATE event_outbox
|
|
SET status = 'failed',
|
|
updated_at = now()
|
|
WHERE id = :id
|
|
"""
|
|
)
|
|
|
|
_RETRY_SQL = text(
|
|
"""
|
|
UPDATE event_outbox
|
|
SET status = 'pending',
|
|
attempts = :attempts,
|
|
next_retry_at = :next_retry_at,
|
|
updated_at = now()
|
|
WHERE id = :id
|
|
"""
|
|
)
|
|
|
|
|
|
def _json_payload(payload: dict[str, Any]) -> str:
|
|
"""Serialise payload to a JSON string suitable for JSONB cast."""
|
|
import json
|
|
|
|
return json.dumps(payload, default=str)
|
|
|
|
|
|
async def enqueue_outbox_event(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
event_name: str,
|
|
payload: dict[str, Any],
|
|
) -> None:
|
|
"""Insert an event into the outbox table within the current transaction.
|
|
|
|
The event is only persisted when the surrounding transaction commits.
|
|
This guarantees at-least-once delivery — no event is lost even if the
|
|
process crashes after the business operation but before the event is
|
|
published.
|
|
|
|
Args:
|
|
db: Active async SQLAlchemy session (part of the business transaction).
|
|
tenant_id: Tenant scope for the event.
|
|
event_name: Logical event name (e.g. ``"contact.created"``).
|
|
payload: Event payload dict (will be stored as JSONB).
|
|
"""
|
|
await db.execute(
|
|
_INSERT_SQL,
|
|
{
|
|
"tenant_id": str(tenant_id),
|
|
"event_name": event_name,
|
|
"payload": _json_payload(payload),
|
|
},
|
|
)
|
|
|
|
|
|
async def process_outbox_batch(
|
|
db: AsyncSession,
|
|
redis: aioredis.Redis | None = None,
|
|
batch_size: int = 50,
|
|
) -> int:
|
|
"""Process one batch of pending outbox events.
|
|
|
|
1. Claim up to *batch_size* pending events using ``FOR UPDATE SKIP LOCKED``
|
|
so multiple workers don't interfere.
|
|
2. Publish each event to the in-process event bus (for local handlers).
|
|
3. On success: mark as ``published``.
|
|
4. On failure: increment attempts, schedule retry with exponential
|
|
backoff, or mark as ``failed`` if max attempts exceeded.
|
|
|
|
Args:
|
|
db: Async SQLAlchemy session for this batch.
|
|
redis: Optional Redis client (unused for now, reserved for future
|
|
cross-process pub/sub).
|
|
batch_size: Maximum events to process in one batch.
|
|
|
|
Returns:
|
|
Number of events successfully published.
|
|
"""
|
|
from app.core.event_bus import get_event_bus
|
|
|
|
event_bus = get_event_bus()
|
|
published_count = 0
|
|
|
|
# Claim a batch of pending events
|
|
rows = (
|
|
await db.execute(_CLAIM_SQL, {"batch_size": batch_size})
|
|
).fetchall()
|
|
|
|
if not rows:
|
|
return 0
|
|
|
|
for row in rows:
|
|
event_id = row[0]
|
|
event_name = row[2]
|
|
payload = row[3]
|
|
attempts = row[4]
|
|
max_attempts = row[5]
|
|
|
|
# payload comes back as a dict from JSONB
|
|
if isinstance(payload, str):
|
|
import json
|
|
payload_dict = json.loads(payload)
|
|
else:
|
|
payload_dict = payload
|
|
|
|
try:
|
|
# Enrich payload with event metadata for idempotency
|
|
payload_dict.setdefault("_event_id", str(event_id))
|
|
payload_dict.setdefault("_event_name", event_name)
|
|
payload_dict.setdefault("_event_timestamp", datetime.now(timezone.utc).isoformat())
|
|
|
|
# Idempotency check: has this event already been processed? (P1.5 fix)
|
|
already_processed = await db.execute(
|
|
text("SELECT 1 FROM consumer_inbox WHERE event_id = :eid AND status = 'processed' LIMIT 1"),
|
|
{"eid": str(event_id)},
|
|
)
|
|
if already_processed.first():
|
|
# Event was already processed by all consumers — mark as published
|
|
await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
|
|
published_count += 1
|
|
logger.debug("Outbox event %s already processed, marking as published", event_id)
|
|
continue
|
|
|
|
results = await event_bus.publish_with_results(event_name, payload_dict)
|
|
|
|
# Check if any handlers were registered at all
|
|
handler_count = len(results)
|
|
# If any handler raised, treat as failure
|
|
handler_errors = [r for r in results if r is not None]
|
|
if handler_errors:
|
|
raise handler_errors[0]
|
|
|
|
if handler_count == 0:
|
|
# No handlers registered — mark as 'no_handlers' not 'published'
|
|
await db.execute(
|
|
text("UPDATE event_outbox SET status = 'no_handlers', published_at = now() WHERE id = :id"),
|
|
{"id": str(event_id)},
|
|
)
|
|
logger.warning("Outbox event %s (%s) had no handlers registered", event_id, event_name)
|
|
else:
|
|
# Record in consumer_inbox for idempotency (P1.5 fix)
|
|
await db.execute(
|
|
text("INSERT INTO consumer_inbox (event_id, consumer_name, status, processed_at) VALUES (:eid, :name, 'processed', now()) ON CONFLICT DO NOTHING"),
|
|
{"eid": str(event_id), "name": event_name},
|
|
)
|
|
await db.execute(_MARK_PUBLISHED_SQL, {"id": str(event_id)})
|
|
published_count += 1
|
|
except Exception as exc:
|
|
logger.error(
|
|
"Failed to publish outbox event %s (%s): %s",
|
|
event_id, event_name, exc,
|
|
exc_info=True,
|
|
)
|
|
new_attempts = attempts + 1
|
|
if new_attempts >= max_attempts:
|
|
await db.execute(_FAIL_SQL, {"id": str(event_id)})
|
|
logger.warning(
|
|
"Outbox event %s marked as failed after %d attempts",
|
|
event_id, new_attempts,
|
|
)
|
|
else:
|
|
backoff = timedelta(seconds=(2 ** new_attempts) * 10)
|
|
next_retry = datetime.now(timezone.utc) + backoff
|
|
await db.execute(
|
|
_RETRY_SQL,
|
|
{
|
|
"id": str(event_id),
|
|
"attempts": new_attempts,
|
|
"next_retry_at": next_retry,
|
|
},
|
|
)
|
|
|
|
await db.commit()
|
|
return published_count
|