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

@@ -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

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,
+        ),
     ]

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}

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