From 74936b3972e18eee7fbcd3704cbe3b486fc82b26 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Sun, 2 Aug 2026 23:47:29 +0200 Subject: [PATCH] Phase 5 (v2): Processing-Recovery, Retention-Cleanup, Replay-Delivery-Reset - 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) --- app/core/outbox.py | 115 +++++++++++++++++++++++- app/core/worker.py | 43 +++++++++ app/routes/outbox.py | 32 +++++++ tests/test_outbox_phase5.py | 168 ++++++++++++++++++++++++++++++++++++ 4 files changed, 356 insertions(+), 2 deletions(-) diff --git a/app/core/outbox.py b/app/core/outbox.py index 2aef2ce..c5e27fd 100644 --- a/app/core/outbox.py +++ b/app/core/outbox.py @@ -96,6 +96,19 @@ _RETRY_SQL = text( """ ) +# ── 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( @@ -139,6 +152,29 @@ _REPLAY_ALL_SQL = text( """ ) +# ── 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( @@ -405,6 +441,9 @@ async def process_outbox_batch( {"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}) @@ -444,7 +483,15 @@ async def replay_failed_event(db: AsyncSession, event_id: uuid.UUID) -> bool: {"event_id": str(event_id)}, ) row = result.first() - return row is not None + 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: @@ -464,7 +511,15 @@ async def replay_all_failed_events(db: AsyncSession, tenant_id: uuid.UUID) -> in _REPLAY_ALL_SQL, {"tenant_id": str(tenant_id)}, ) - return len(result.fetchall()) + 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 ─────────────────────────────────────────── @@ -553,3 +608,59 @@ def get_consumer_registry() -> dict[str, list[str]]: 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 diff --git a/app/core/worker.py b/app/core/worker.py index a5e7a4f..8d42a86 100644 --- a/app/core/worker.py +++ b/app/core/worker.py @@ -256,6 +256,44 @@ async def process_outbox_job(ctx: dict[str, Any]) -> None: register_job("process_outbox", process_outbox_job) +# ── Outbox retention cleanup job ───────────────────────────────────────────── + +async def cleanup_outbox_job(ctx: dict[str, Any]) -> None: + """Delete published outbox events older than 30 days. + + Runs hourly to prevent the outbox table from growing indefinitely. + Iterates per-tenant for RLS compliance. + """ + from app.core.db import get_worker_session_factory + from app.core.outbox import cleanup_published_events + from sqlalchemy import text as sa_text + + factory = get_worker_session_factory() + async with factory() as db: + try: + tenant_result = await db.execute(sa_text("SELECT id FROM tenants")) + tenant_ids = [row[0] for row in tenant_result] + + total_deleted = 0 + for tenant_id in tenant_ids: + await db.execute( + sa_text("SELECT set_config('app.current_tenant_id', :tid, true)"), + {"tid": str(tenant_id)}, + ) + deleted = await cleanup_published_events(db, retention_days=30) + total_deleted += deleted + await db.commit() + + if total_deleted: + logger.info("Outbox retention: cleaned up %d published events", total_deleted) + except Exception: + logger.error("Outbox retention cleanup failed", exc_info=True) + await db.rollback() + + +register_job("cleanup_outbox", cleanup_outbox_job) + + class WorkerSettings: """ARQ worker settings.""" functions = get_all_jobs() @@ -280,4 +318,9 @@ class WorkerSettings: _wrap_cron_with_lock("process_outbox", process_outbox_job, ttl_seconds=30), second={0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55}, ), + # Outbox retention cleanup — hourly + cron( + _wrap_cron_with_lock("cleanup_outbox", cleanup_outbox_job, ttl_seconds=300), + minute=0, + ), ] diff --git a/app/routes/outbox.py b/app/routes/outbox.py index 35405cc..6c98f11 100644 --- a/app/routes/outbox.py +++ b/app/routes/outbox.py @@ -18,9 +18,11 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.core.db import get_db from app.core.outbox import ( + cleanup_published_events, get_consumer_registry, get_failed_events, get_outbox_stats, + recover_stuck_events, replay_all_failed_events, replay_failed_event, ) @@ -128,3 +130,33 @@ async def outbox_consumer_registry( This is read from ``event_bus._handlers`` at request time. """ return {"registry": get_consumer_registry()} + + +@router.post("/recover-stuck") +async def outbox_recover_stuck( + timeout_seconds: int = Query(120, ge=10, le=3600), + db: AsyncSession = Depends(get_db), + current_user: dict[str, Any] = Depends(require_admin), +): + """Reset events stuck in 'processing' status back to 'pending'. + + If a worker crashes mid-processing, events remain in 'processing' forever. + This endpoint resets events that have been in 'processing' longer than + *timeout_seconds* back to 'pending' so they can be retried. + """ + count = await recover_stuck_events(db, timeout_seconds=timeout_seconds) + return {"recovered_count": count, "timeout_seconds": timeout_seconds} + + +@router.post("/cleanup-published") +async def outbox_cleanup_published( + retention_days: int = Query(30, ge=1, le=365), + db: AsyncSession = Depends(get_db), + current_user: dict[str, Any] = Depends(require_admin), +): + """Delete published events older than *retention_days*. + + Prevents the outbox table from growing indefinitely. + """ + count = await cleanup_published_events(db, retention_days=retention_days) + return {"deleted_count": count, "retention_days": retention_days} diff --git a/tests/test_outbox_phase5.py b/tests/test_outbox_phase5.py index 3cb9052..9e0b96d 100644 --- a/tests/test_outbox_phase5.py +++ b/tests/test_outbox_phase5.py @@ -420,3 +420,171 @@ def test_route_import(): assert "/api/v1/outbox/replay/{event_id}" in paths assert "/api/v1/outbox/replay-all" in paths assert "/api/v1/outbox/consumer-registry" in paths + + +# ── Processing recovery (stuck events) ───────────────────────────────────────── + +@pytest.mark.asyncio +async def test_recover_stuck_events(db_session: AsyncSession): + """recover_stuck_events resets events stuck in 'processing' back to 'pending'.""" + from app.core.outbox import recover_stuck_events + + tenant_id = uuid.uuid4() + + # Insert a processing event with old updated_at (simulating crash) + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status, updated_at) " + "VALUES (:tid, 'test.stuck', CAST(:payload AS JSONB), 'processing', now() - interval '5 minutes')" + ), + {"tid": str(tenant_id), "payload": '{"k": "v"}'}, + ) + await db_session.flush() + await db_session.commit() + + await set_tenant_context(db_session, tenant_id) + count = await recover_stuck_events(db_session, timeout_seconds=60) + await db_session.commit() + + assert count == 1 + + rows = ( + await db_session.execute( + text("SELECT status FROM event_outbox WHERE tenant_id = :tid"), + {"tid": str(tenant_id)}, + ) + ).fetchall() + assert rows[0][0] == "pending" + + +@pytest.mark.asyncio +async def test_recover_stuck_events_skips_recent(db_session: AsyncSession): + """recover_stuck_events does not reset recently-claimed processing events.""" + from app.core.outbox import recover_stuck_events + + tenant_id = uuid.uuid4() + + # Insert a processing event with recent updated_at (should not be recovered) + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status, updated_at) " + "VALUES (:tid, 'test.stuck.recent', CAST(:payload AS JSONB), 'processing', now())" + ), + {"tid": str(tenant_id), "payload": '{"k": "v"}'}, + ) + await db_session.flush() + await db_session.commit() + + await set_tenant_context(db_session, tenant_id) + count = await recover_stuck_events(db_session, timeout_seconds=60) + await db_session.commit() + + assert count == 0 # should not recover recent events + + +# ── Retention cleanup ────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_cleanup_published_events(db_session: AsyncSession): + """cleanup_published_events deletes old published events.""" + from app.core.outbox import cleanup_published_events + + tenant_id = uuid.uuid4() + + # Insert an old published event (31 days ago) + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status, published_at) " + "VALUES (:tid, 'test.old', CAST(:payload AS JSONB), 'published', now() - interval '31 days')" + ), + {"tid": str(tenant_id), "payload": '{"k": "v"}'}, + ) + # Insert a recent published event (1 day ago) + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status, published_at) " + "VALUES (:tid, 'test.recent', CAST(:payload AS JSONB), 'published', now() - interval '1 day')" + ), + {"tid": str(tenant_id), "payload": '{"k": "v"}'}, + ) + await db_session.flush() + await db_session.commit() + + await set_tenant_context(db_session, tenant_id) + deleted = await cleanup_published_events(db_session, retention_days=30) + await db_session.commit() + + assert deleted == 1 # only the old one + + rows = ( + await db_session.execute( + text("SELECT event_name FROM event_outbox WHERE tenant_id = :tid"), + {"tid": str(tenant_id)}, + ) + ).fetchall() + assert len(rows) == 1 + assert rows[0][0] == "test.recent" + + +# ── Replay resets outbox_deliveries ───────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_replay_resets_deliveries(db_session: AsyncSession): + """replay_failed_event also deletes old outbox_deliveries for the event.""" + tenant_id = uuid.uuid4() + + # Insert a failed event + await db_session.execute( + text( + "INSERT INTO event_outbox (tenant_id, event_name, payload, status, attempts, max_attempts, error_message, failed_at) " + "VALUES (:tid, 'test.replay.deliveries', CAST(:payload AS JSONB), 'failed', 5, 5, 'err', now())" + ), + {"tid": str(tenant_id), "payload": '{"k": "v"}'}, + ) + await db_session.flush() + await db_session.commit() + + row = ( + await db_session.execute( + text("SELECT id FROM event_outbox WHERE tenant_id = :tid"), + {"tid": str(tenant_id)}, + ) + ).first() + event_id = row[0] + + # Insert a delivery record for this event + await db_session.execute( + text( + "INSERT INTO outbox_deliveries (event_id, consumer_name, status, attempt_count, last_error) " + "VALUES (:eid, 'test_consumer', 'failed', 5, 'err')" + ), + {"eid": str(event_id)}, + ) + await db_session.flush() + await db_session.commit() + + await set_tenant_context(db_session, tenant_id) + replayed = await replay_failed_event(db_session, event_id) + await db_session.commit() + + assert replayed is True + + # Check deliveries were deleted + delivery_rows = ( + await db_session.execute( + text("SELECT count(*) FROM outbox_deliveries WHERE event_id = :eid"), + {"eid": str(event_id)}, + ) + ).first() + assert delivery_rows[0] == 0 + + +# ── New endpoints registered ──────────────────────────────────────────────────── + +def test_new_endpoints_registered(): + """Verify that recover-stuck and cleanup-published endpoints are registered.""" + from app.routes.outbox import router + + paths = {route.path for route in router.routes} + assert "/api/v1/outbox/recover-stuck" in paths + assert "/api/v1/outbox/cleanup-published" in paths