Agent Zero
74936b3972
- recover_stuck_events: Reset processing events stuck >120s back to pending - cleanup_published_events: Delete published events older than 30 days - Replay now resets outbox_deliveries for clean retry - Worker: hourly retention cleanup cron job - API: /recover-stuck and /cleanup-published endpoints - process_outbox_batch: auto-recovery at start of each tenant iteration - 23/23 tests passing (5 new tests)
667 lines
23 KiB
Python
667 lines
23 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.
|
|
|
|
Phase 5 additions:
|
|
- DLQ: ``error_message`` and ``failed_at`` columns on ``event_outbox``
|
|
- Replay: ``replay_failed_event`` and ``replay_all_failed_events``
|
|
- Monitoring: ``get_outbox_stats`` and ``get_failed_events``
|
|
- Consumer registry: ``get_consumer_registry`` and ``outbox_deliveries``
|
|
"""
|
|
|
|
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, aggregate_type, aggregate_id, occurred_at, correlation_id, schema_version)
|
|
VALUES (:tenant_id, :event_name, CAST(:payload AS JSONB), :aggregate_type, :aggregate_id, COALESCE(:occurred_at, now()), :correlation_id, COALESCE(:schema_version, 1))
|
|
RETURNING id
|
|
"""
|
|
)
|
|
|
|
_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,
|
|
aggregate_type, aggregate_id, occurred_at, correlation_id, schema_version
|
|
"""
|
|
)
|
|
|
|
_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',
|
|
error_message = :error_message,
|
|
failed_at = now(),
|
|
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
|
|
"""
|
|
)
|
|
|
|
# ── Phase 5: Processing recovery (stuck events) ──────────────────────────────
|
|
|
|
_RECOVER_STUCK_SQL = text(
|
|
"""
|
|
UPDATE event_outbox
|
|
SET status = 'pending',
|
|
updated_at = now()
|
|
WHERE status = 'processing'
|
|
AND updated_at < now() - make_interval(secs => :timeout_seconds)
|
|
RETURNING id
|
|
"""
|
|
)
|
|
|
|
# ── Phase 5: outbox_deliveries SQL ──────────────────────────────────────────
|
|
|
|
_INSERT_DELIVERY_SQL = text(
|
|
"""
|
|
INSERT INTO outbox_deliveries (event_id, consumer_name, status, attempt_count, last_error, processed_at)
|
|
VALUES (:event_id, :consumer_name, :status, :attempt_count, :last_error, :processed_at)
|
|
ON CONFLICT (event_id, consumer_name) DO UPDATE SET
|
|
status = EXCLUDED.status,
|
|
attempt_count = EXCLUDED.attempt_count,
|
|
last_error = EXCLUDED.last_error,
|
|
processed_at = EXCLUDED.processed_at,
|
|
updated_at = now()
|
|
"""
|
|
)
|
|
|
|
# ── Phase 5: Replay SQL ─────────────────────────────────────────────────────
|
|
|
|
_REPLAY_ONE_SQL = text(
|
|
"""
|
|
UPDATE event_outbox
|
|
SET status = 'pending',
|
|
attempts = 0,
|
|
error_message = NULL,
|
|
next_retry_at = NULL,
|
|
updated_at = now()
|
|
WHERE id = :event_id AND status = 'failed'
|
|
RETURNING id
|
|
"""
|
|
)
|
|
|
|
_REPLAY_ALL_SQL = text(
|
|
"""
|
|
UPDATE event_outbox
|
|
SET status = 'pending',
|
|
attempts = 0,
|
|
error_message = NULL,
|
|
next_retry_at = NULL,
|
|
updated_at = now()
|
|
WHERE status = 'failed' AND tenant_id = :tenant_id
|
|
RETURNING id
|
|
"""
|
|
)
|
|
|
|
# ── Phase 5: Reset deliveries on replay ─────────────────────────────────────
|
|
|
|
_RESET_DELIVERIES_FOR_EVENT_SQL = text(
|
|
"DELETE FROM outbox_deliveries WHERE event_id = :event_id"
|
|
)
|
|
|
|
_RESET_DELIVERIES_FOR_TENANT_SQL = text(
|
|
"""
|
|
DELETE FROM outbox_deliveries
|
|
WHERE event_id IN (SELECT id FROM event_outbox WHERE tenant_id = :tenant_id AND status = 'pending')
|
|
"""
|
|
)
|
|
|
|
# ── Phase 5: Retention SQL ──────────────────────────────────────────────────
|
|
|
|
_RETENTION_PUBLISHED_SQL = text(
|
|
"""
|
|
DELETE FROM event_outbox
|
|
WHERE status = 'published'
|
|
AND published_at < now() - make_interval(secs => :retention_seconds)
|
|
"""
|
|
)
|
|
|
|
# ── Phase 5: Stats SQL ──────────────────────────────────────────────────────
|
|
|
|
_STATS_COUNT_SQL = text(
|
|
"SELECT status, COUNT(*) as count FROM event_outbox GROUP BY status"
|
|
)
|
|
|
|
_STATS_OLDEST_PENDING_SQL = text(
|
|
"""
|
|
SELECT EXTRACT(EPOCH FROM (now() - created_at)) as age_seconds
|
|
FROM event_outbox
|
|
WHERE status = 'pending'
|
|
ORDER BY created_at ASC
|
|
LIMIT 1
|
|
"""
|
|
)
|
|
|
|
# ── Phase 5: Failed events SQL ──────────────────────────────────────────────
|
|
|
|
_FAILED_EVENTS_SQL = text(
|
|
"""
|
|
SELECT id, tenant_id, event_name, error_message, failed_at, attempts, created_at
|
|
FROM event_outbox
|
|
WHERE status = 'failed'
|
|
ORDER BY failed_at DESC
|
|
LIMIT :limit OFFSET :offset
|
|
"""
|
|
)
|
|
|
|
|
|
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)
|
|
|
|
|
|
def _get_handler_name(handler: Any) -> str:
|
|
"""Extract a human-readable name from a handler callable."""
|
|
name = getattr(handler, "__name__", None)
|
|
if name:
|
|
return name
|
|
name = getattr(handler, "__qualname__", None)
|
|
if name:
|
|
return name
|
|
return str(handler)
|
|
|
|
|
|
async def enqueue_outbox_event(
|
|
db: AsyncSession,
|
|
tenant_id: uuid.UUID,
|
|
event_name: str,
|
|
payload: dict[str, Any],
|
|
*,
|
|
aggregate_type: str | None = None,
|
|
aggregate_id: uuid.UUID | None = None,
|
|
correlation_id: uuid.UUID | None = None,
|
|
schema_version: int = 1,
|
|
) -> 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. ``"crm.contact.created.v1"``).
|
|
payload: Event payload dict (will be stored as JSONB).
|
|
aggregate_type: Type of the aggregate (e.g. 'contact', 'task').
|
|
aggregate_id: UUID of the aggregate entity.
|
|
correlation_id: Optional correlation UUID for tracing across services.
|
|
schema_version: Event schema version (default 1).
|
|
"""
|
|
await db.execute(
|
|
_INSERT_SQL,
|
|
{
|
|
"tenant_id": str(tenant_id),
|
|
"event_name": event_name,
|
|
"payload": _json_payload(payload),
|
|
"aggregate_type": aggregate_type,
|
|
"aggregate_id": str(aggregate_id) if aggregate_id else None,
|
|
"occurred_at": None, # DB defaults to NOW()
|
|
"correlation_id": str(correlation_id) if correlation_id else None,
|
|
"schema_version": schema_version,
|
|
},
|
|
)
|
|
|
|
|
|
async def _process_single_outbox_event(
|
|
db: AsyncSession,
|
|
event_bus,
|
|
row: tuple,
|
|
) -> bool:
|
|
"""Process a single outbox event. Returns True if published successfully."""
|
|
event_id = row[0]
|
|
tenant_id = row[1]
|
|
event_name = row[2]
|
|
payload = row[3]
|
|
attempts = row[4]
|
|
max_attempts = row[5]
|
|
aggregate_type = row[6] if len(row) > 6 else None
|
|
aggregate_id = row[7] if len(row) > 7 else None
|
|
occurred_at = row[8] if len(row) > 8 else None
|
|
correlation_id = row[9] if len(row) > 9 else None
|
|
schema_version = row[10] if len(row) > 10 else 1
|
|
|
|
# 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 standardized event envelope metadata
|
|
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())
|
|
payload_dict.setdefault("_tenant_id", str(tenant_id))
|
|
payload_dict.setdefault("_aggregate_type", aggregate_type)
|
|
payload_dict.setdefault("_aggregate_id", str(aggregate_id) if aggregate_id else None)
|
|
payload_dict.setdefault("_occurred_at", occurred_at.isoformat() if occurred_at else None)
|
|
payload_dict.setdefault("_correlation_id", str(correlation_id) if correlation_id else None)
|
|
payload_dict.setdefault("_schema_version", schema_version)
|
|
|
|
# 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)})
|
|
logger.debug("Outbox event %s already processed, marking as published", event_id)
|
|
return True
|
|
|
|
# Phase 5: Get handler names for consumer registry before publishing
|
|
handlers = list(event_bus._handlers.get(event_name, [])) + list(event_bus._handlers.get("*", []))
|
|
handler_names = [_get_handler_name(h) for h in handlers]
|
|
|
|
results = await event_bus.publish_with_results(event_name, payload_dict)
|
|
|
|
# Phase 5: Write outbox_deliveries for each handler
|
|
current_attempt = attempts + 1
|
|
for i, result in enumerate(results):
|
|
consumer_name = handler_names[i] if i < len(handler_names) else f"handler_{i}"
|
|
if result is None:
|
|
# Success
|
|
await db.execute(
|
|
_INSERT_DELIVERY_SQL,
|
|
{
|
|
"event_id": str(event_id),
|
|
"consumer_name": consumer_name,
|
|
"status": "delivered",
|
|
"attempt_count": current_attempt,
|
|
"last_error": None,
|
|
"processed_at": datetime.now(timezone.utc),
|
|
},
|
|
)
|
|
else:
|
|
# Failure
|
|
await db.execute(
|
|
_INSERT_DELIVERY_SQL,
|
|
{
|
|
"event_id": str(event_id),
|
|
"consumer_name": consumer_name,
|
|
"status": "failed",
|
|
"attempt_count": current_attempt,
|
|
"last_error": str(result),
|
|
"processed_at": None,
|
|
},
|
|
)
|
|
|
|
# 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)})
|
|
return True
|
|
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), "error_message": str(exc)},
|
|
)
|
|
logger.warning(
|
|
"Outbox event %s marked as failed after %d attempts: %s",
|
|
event_id, new_attempts, exc,
|
|
)
|
|
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,
|
|
},
|
|
)
|
|
return False
|
|
|
|
|
|
async def process_outbox_batch(
|
|
db: AsyncSession,
|
|
redis: aioredis.Redis | None = None,
|
|
batch_size: int = 50,
|
|
tenant_ids: list[uuid.UUID] | None = None,
|
|
) -> int:
|
|
"""Process one batch of pending outbox events.
|
|
|
|
Iterates over all tenants, setting tenant context for RLS before
|
|
claiming and processing events for each tenant.
|
|
|
|
Args:
|
|
db: Async SQLAlchemy session for this batch (crm_worker role).
|
|
redis: Optional Redis client (unused for now, reserved for future
|
|
cross-process pub/sub).
|
|
batch_size: Maximum events to process per tenant in one batch.
|
|
tenant_ids: Optional list of tenant IDs to process. If None,
|
|
all tenants are loaded from the database.
|
|
|
|
Returns:
|
|
Number of events successfully published.
|
|
"""
|
|
from app.core.event_bus import get_event_bus
|
|
|
|
event_bus = get_event_bus()
|
|
published_count = 0
|
|
|
|
# Load tenant IDs if not provided
|
|
if tenant_ids is None:
|
|
result = await db.execute(text("SELECT id FROM tenants"))
|
|
tenant_ids = [row[0] for row in result]
|
|
|
|
for tenant_id in tenant_ids:
|
|
# Set tenant context for RLS — required for event_outbox and consumer_inbox
|
|
await db.execute(
|
|
text("SELECT set_config('app.current_tenant_id', :tid, true)"),
|
|
{"tid": str(tenant_id)},
|
|
)
|
|
|
|
# Phase 5: Recover stuck processing events (worker crash recovery)
|
|
await recover_stuck_events(db, timeout_seconds=120)
|
|
|
|
# Claim a batch of pending events for this tenant
|
|
rows = (
|
|
await db.execute(_CLAIM_SQL, {"batch_size": batch_size})
|
|
).fetchall()
|
|
|
|
if not rows:
|
|
continue
|
|
|
|
for row in rows:
|
|
published = await _process_single_outbox_event(db, event_bus, row)
|
|
if published:
|
|
published_count += 1
|
|
|
|
# Commit after each tenant to release locks
|
|
await db.commit()
|
|
|
|
return published_count
|
|
|
|
|
|
# ── Phase 5: Replay functions ────────────────────────────────────────────────
|
|
|
|
async def replay_failed_event(db: AsyncSession, event_id: uuid.UUID) -> bool:
|
|
"""Replay a single failed outbox event.
|
|
|
|
Resets the event to ``pending`` status with attempts=0,
|
|
error_message=NULL, next_retry_at=NULL.
|
|
|
|
Args:
|
|
db: Active async SQLAlchemy session with tenant context set.
|
|
event_id: UUID of the event to replay.
|
|
|
|
Returns:
|
|
True if the event was replayed, False if not found or not in 'failed' status.
|
|
"""
|
|
result = await db.execute(
|
|
_REPLAY_ONE_SQL,
|
|
{"event_id": str(event_id)},
|
|
)
|
|
row = result.first()
|
|
if row is not None:
|
|
# Reset delivery records so consumers get a clean slate
|
|
await db.execute(
|
|
_RESET_DELIVERIES_FOR_EVENT_SQL,
|
|
{"event_id": str(event_id)},
|
|
)
|
|
await db.commit()
|
|
return True
|
|
return False
|
|
|
|
|
|
async def replay_all_failed_events(db: AsyncSession, tenant_id: uuid.UUID) -> int:
|
|
"""Replay all failed outbox events for a tenant.
|
|
|
|
Resets all failed events to ``pending`` status with attempts=0,
|
|
error_message=NULL, next_retry_at=NULL.
|
|
|
|
Args:
|
|
db: Active async SQLAlchemy session with tenant context set.
|
|
tenant_id: Tenant scope for replay.
|
|
|
|
Returns:
|
|
Number of events replayed.
|
|
"""
|
|
result = await db.execute(
|
|
_REPLAY_ALL_SQL,
|
|
{"tenant_id": str(tenant_id)},
|
|
)
|
|
replayed_ids = result.fetchall()
|
|
if replayed_ids:
|
|
# Reset delivery records for all replayed events
|
|
await db.execute(
|
|
_RESET_DELIVERIES_FOR_TENANT_SQL,
|
|
{"tenant_id": str(tenant_id)},
|
|
)
|
|
await db.commit()
|
|
return len(replayed_ids)
|
|
|
|
|
|
# ── Phase 5: Monitoring functions ───────────────────────────────────────────
|
|
|
|
async def get_outbox_stats(db: AsyncSession) -> dict[str, Any]:
|
|
"""Get outbox statistics for the current tenant.
|
|
|
|
Requires tenant context to be set (RLS filters automatically).
|
|
|
|
Returns:
|
|
Dict with:
|
|
- ``counts``: dict mapping each status to its count.
|
|
- ``total``: total number of events.
|
|
- ``oldest_pending_age_seconds``: age of oldest pending event in seconds,
|
|
or None if no pending events.
|
|
"""
|
|
result = await db.execute(_STATS_COUNT_SQL)
|
|
counts = {row[0]: row[1] for row in result.fetchall()}
|
|
|
|
total = sum(counts.values())
|
|
|
|
oldest_result = await db.execute(_STATS_OLDEST_PENDING_SQL)
|
|
oldest_row = oldest_result.first()
|
|
oldest_pending_age_seconds = float(oldest_row[0]) if oldest_row else None
|
|
|
|
return {
|
|
"counts": counts,
|
|
"total": total,
|
|
"oldest_pending_age_seconds": oldest_pending_age_seconds,
|
|
}
|
|
|
|
|
|
async def get_failed_events(
|
|
db: AsyncSession,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> list[dict[str, Any]]:
|
|
"""Get failed outbox events for the current tenant.
|
|
|
|
Requires tenant context to be set (RLS filters automatically).
|
|
|
|
Args:
|
|
db: Active async SQLAlchemy session with tenant context set.
|
|
limit: Maximum number of events to return.
|
|
offset: Number of events to skip.
|
|
|
|
Returns:
|
|
List of dicts with: id, tenant_id, event_name, error_message,
|
|
failed_at, attempts, created_at, status.
|
|
"""
|
|
result = await db.execute(
|
|
_FAILED_EVENTS_SQL,
|
|
{"limit": limit, "offset": offset},
|
|
)
|
|
return [
|
|
{
|
|
"id": str(row[0]),
|
|
"tenant_id": str(row[1]),
|
|
"event_name": row[2],
|
|
"error_message": row[3],
|
|
"failed_at": row[4].isoformat() if row[4] else None,
|
|
"attempts": row[5],
|
|
"created_at": row[6].isoformat() if row[6] else None,
|
|
"status": "failed",
|
|
}
|
|
for row in result.fetchall()
|
|
]
|
|
|
|
|
|
# ── Phase 5: Consumer registry ──────────────────────────────────────────────
|
|
|
|
def get_consumer_registry() -> dict[str, list[str]]:
|
|
"""Get the registered event handler registry from the in-process event bus.
|
|
|
|
Returns a mapping of ``event_name`` to a list of consumer (handler) names.
|
|
This is read from ``event_bus._handlers`` at call time.
|
|
|
|
Returns:
|
|
Dict mapping event names to lists of handler names.
|
|
"""
|
|
from app.core.event_bus import get_event_bus
|
|
|
|
event_bus = get_event_bus()
|
|
registry: dict[str, list[str]] = {}
|
|
for event_name, handlers in event_bus._handlers.items():
|
|
if handlers:
|
|
registry[event_name] = [_get_handler_name(h) for h in handlers]
|
|
return registry
|
|
|
|
|
|
# ── Phase 5: Processing recovery ──────────────────────────────────────────────
|
|
|
|
async def recover_stuck_events(db: AsyncSession, timeout_seconds: int = 120) -> int:
|
|
"""Reset events stuck in 'processing' status back to 'pending'.
|
|
|
|
If a worker crashes mid-processing, events remain in 'processing' forever.
|
|
This function resets events that have been in 'processing' longer than
|
|
*timeout_seconds* back to 'pending' so they can be retried.
|
|
|
|
Must be called with tenant context set (RLS filters automatically).
|
|
|
|
Args:
|
|
db: Active async SQLAlchemy session.
|
|
timeout_seconds: How long an event can stay in 'processing' before reset.
|
|
|
|
Returns:
|
|
Number of events reset to 'pending'.
|
|
"""
|
|
result = await db.execute(
|
|
_RECOVER_STUCK_SQL,
|
|
{"timeout_seconds": timeout_seconds},
|
|
)
|
|
count = len(result.fetchall()) if result.returns_rows else 0
|
|
if count:
|
|
logger.info("Recovered %d stuck processing events (timeout=%ds)", count, timeout_seconds)
|
|
return count
|
|
|
|
|
|
# ── Phase 5: Retention cleanup ────────────────────────────────────────────────
|
|
|
|
async def cleanup_published_events(db: AsyncSession, retention_days: int = 30) -> int:
|
|
"""Delete published events older than *retention_days*.
|
|
|
|
Prevents the outbox table from growing indefinitely. Published events
|
|
are no longer needed after retention period.
|
|
|
|
Must be called with tenant context set (RLS filters automatically).
|
|
|
|
Args:
|
|
db: Active async SQLAlchemy session.
|
|
retention_days: Delete published events older than this many days.
|
|
|
|
Returns:
|
|
Number of events deleted.
|
|
"""
|
|
retention_seconds = retention_days * 86400
|
|
result = await db.execute(
|
|
_RETENTION_PUBLISHED_SQL,
|
|
{"retention_seconds": retention_seconds},
|
|
)
|
|
count = result.rowcount if hasattr(result, 'rowcount') else 0
|
|
if count:
|
|
logger.info("Cleaned up %d published events older than %d days", count, retention_days)
|
|
return count
|