pr5 async delivery and sweeper

.gitignore

@@ -4,3 +4,4 @@ __pycache__/
 .venv/
 *.egg-info/
 .pytest_cache/
+build/

README.md

@@ -33,8 +33,10 @@ That starts Postgres, Redis, the API on `:8000`, and the worker. From there you
 
 ## Status
 
-Still being built. Signature checks, idempotent dedupe, routing with fan-out, and the read API are done and tested. The delivery worker is the current focus. That covers the atomic claim, the outbound POST, attempt logging, and the sweeper.
+Still being built, but the core loop runs end to end. Signature checks, idempotent dedupe, routing with fan-out, the read API, and the async delivery worker are all done and tested. The worker claims each delivery atomically, POSTs it, records the attempt, and reschedules failures for another try. A sweeper recovers anything a lost enqueue or a crashed worker left stranded. So an event can come in, get verified and stored, and be delivered to every destination with retries, today.
+
+Next up is making those retries smart: backoff, an attempt cap, and dead-lettering.
 
 ## Planned
 
-The retry engine comes next. It'll do exponential backoff with jitter so retries don't stampede, cap the attempts, and dead-letter whatever runs out. Failed deliveries will be replayable in one click. A React dashboard will sit on top of the read API for inspecting payloads and replaying failures. After that, a one-command deploy to Fly.io or Railway. Further out, the hub will be able to reshape payloads per route and sign its own outbound requests.
+Retries already happen, just on a flat delay. The next step makes them smart: exponential backoff with jitter so they don't stampede, a cap on attempts, and dead-letter for whatever runs out. Then failed deliveries become replayable in one click. A React dashboard will sit on top of the read API for inspecting payloads and replaying failures. After that, a one-command deploy to Fly.io or Railway. Further out, the hub will reshape payloads per route and sign its own outbound requests.

backend/app/main.py

@@ -1,8 +1,22 @@
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
+from saq import Queue
 
+from app.config import settings
 from app.routers import config, deliveries, destinations, events, ingest, routes
 
-app = FastAPI(title="webhook-hub")
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    app.state.queue = Queue.from_url(str(settings.redis_dsn))
+    try:
+        yield
+    finally:
+        await app.state.queue.disconnect()
+
+
+app = FastAPI(title="webhook-hub", lifespan=lifespan)
 
 app.include_router(config.router)
 app.include_router(ingest.router)

backend/app/routers/ingest.py

@@ -1,5 +1,6 @@
 import hashlib
 import json
+import logging
 from datetime import UTC, datetime
 from typing import Annotated
 
@@ -14,6 +15,8 @@
 
 router = APIRouter(prefix="/ingest", tags=["ingest"])
 
+logger = logging.getLogger(__name__)
+
 
 @router.post(
     "/{source_name}",
@@ -83,10 +86,21 @@ async def ingest(
     )
 
     now = datetime.now(UTC)
+    deliveries = []
     for dest_id in destination_ids:
-        session.add(
-            Delivery(event_id=event.id, destination_id=dest_id, next_attempt_at=now)
+        delivery = Delivery(
+            event_id=event.id, destination_id=dest_id, next_attempt_at=now
         )
+        deliveries.append(delivery)
+        session.add(delivery)
 
     await session.commit()
+
+    try:
+        for delivery in deliveries:
+            await request.app.state.queue.enqueue(
+                "deliver", delivery_id=str(delivery.id)
+            )
+    except Exception:
+        logger.warning("enqueue failed for event %s; sweeper will recover", event.id)
     return IngestAck(event_id=event.id)

backend/app/tasks.py

@@ -0,0 +1,234 @@
+import datetime as dt
+import logging
+import uuid
+from dataclasses import dataclass
+from time import perf_counter
+from typing import Awaitable, Callable
+
+import httpx
+from saq import CronJob, Queue
+from saq.types import Context, SettingsDict
+from sqlalchemy import and_, func, or_, select, update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.config import settings as app_settings
+from app.db import AsyncSessionLocal
+from app.models import Delivery, DeliveryAttempt, DeliveryStatus, Destination, Event
+
+LEASE = 60
+FIXED_DELAY = 100
+BODY_CAP = 4096
+REDISPATCH_LIMIT = 100
+
+logger = logging.getLogger(__name__)
+
+
+class WorkerContext(Context):
+    client: httpx.AsyncClient
+    sessionmaker: async_sessionmaker[AsyncSession]
+
+
+@dataclass
+class DeliveryResult:
+    success: bool
+    response_status: int | None = None
+    response_body: str | None = None
+    error: str | None = None
+    duration_ms: int = 0
+
+
+@dataclass(frozen=True)
+class DeliverySnapshot:
+    attempt_count: int
+    destination_id: uuid.UUID
+    event_id: uuid.UUID
+    url: str
+    payload: dict
+
+
+SendFn = Callable[[httpx.AsyncClient, DeliverySnapshot], Awaitable[DeliveryResult]]
+
+
+async def _real_send(
+    client: httpx.AsyncClient, snapshot: DeliverySnapshot
+) -> DeliveryResult:
+    start = perf_counter()
+    try:
+        resp = await client.post(snapshot.url, json=snapshot.payload)
+    except httpx.RequestError as exc:
+        ms = int((perf_counter() - start) * 1000)
+        return DeliveryResult(success=False, error=str(exc), duration_ms=ms)
+    ms = int((perf_counter() - start) * 1000)
+    return DeliveryResult(
+        success=200 <= resp.status_code < 300,
+        response_status=resp.status_code,
+        response_body=resp.text[:BODY_CAP],
+        duration_ms=ms,
+    )
+
+
+async def deliver(
+    ctx: WorkerContext, *, delivery_id: str, send_fn: SendFn = _real_send
+) -> None:
+    event_delivery_id = uuid.UUID(delivery_id)
+
+    async with ctx["sessionmaker"]() as session:
+        row = (
+            await session.execute(
+                update(Delivery)
+                .where(
+                    and_(
+                        Delivery.id == event_delivery_id,
+                        or_(
+                            and_(
+                                Delivery.status.in_(
+                                    [DeliveryStatus.pending, DeliveryStatus.failed]
+                                ),
+                                or_(
+                                    Delivery.next_attempt_at.is_(None),
+                                    Delivery.next_attempt_at <= func.now(),
+                                ),
+                            ),
+                            and_(
+                                Delivery.status == DeliveryStatus.delivering,
+                                Delivery.updated_at
+                                <= func.now() - dt.timedelta(seconds=LEASE),
+                            ),
+                        ),
+                    )
+                )
+                .values(status=DeliveryStatus.delivering)
+                .returning(
+                    Delivery.attempt_count,
+                    Delivery.destination_id,
+                    Delivery.event_id,
+                )
+            )
+        ).one_or_none()
+
+        if row is None:
+            return
+
+        dst = await session.get(Destination, row.destination_id)
+        event = await session.get(Event, row.event_id)
+
+        if dst is None or event is None:
+            logger.error(
+                "delivery %s references missing dst=%s event=%s",
+                event_delivery_id,
+                row.destination_id,
+                row.event_id,
+            )
+            return
+
+        snapshot = DeliverySnapshot(
+            attempt_count=row.attempt_count,
+            destination_id=row.destination_id,
+            event_id=row.event_id,
+            url=dst.url,
+            payload=event.payload,
+        )
+
+        await session.commit()
+
+    result = await send_fn(ctx["client"], snapshot)
+
+    async with ctx["sessionmaker"]() as session:
+        attempt = DeliveryAttempt(
+            delivery_id=event_delivery_id,
+            attempt_number=snapshot.attempt_count + 1,
+            response_status=result.response_status,
+            response_body=result.response_body,
+            error=result.error,
+            duration_ms=result.duration_ms,
+        )
+
+        session.add(attempt)
+
+        stmt = update(Delivery).where(Delivery.id == event_delivery_id)
+
+        if result.success:
+            await session.execute(
+                stmt.values(
+                    status=DeliveryStatus.succeeded,
+                    next_attempt_at=None,
+                    attempt_count=snapshot.attempt_count + 1,
+                )
+            )
+
+        else:
+            await session.execute(
+                stmt.values(
+                    status=DeliveryStatus.failed,
+                    next_attempt_at=func.now() + dt.timedelta(seconds=FIXED_DELAY),
+                    attempt_count=snapshot.attempt_count + 1,
+                )
+            )
+
+        await session.commit()
+
+
+async def sweep(ctx: WorkerContext) -> None:
+    async with ctx["sessionmaker"]() as session:
+        redispatch = (
+            (
+                await session.execute(
+                    select(Delivery.id)
+                    .where(
+                        or_(
+                            and_(
+                                Delivery.status.in_(
+                                    [DeliveryStatus.pending, DeliveryStatus.failed]
+                                ),
+                                or_(
+                                    Delivery.next_attempt_at.is_(None),
+                                    Delivery.next_attempt_at <= func.now(),
+                                ),
+                            ),
+                            and_(
+                                Delivery.status == DeliveryStatus.delivering,
+                                Delivery.updated_at
+                                <= func.now() - dt.timedelta(seconds=LEASE),
+                            ),
+                        ),
+                    )
+                    .limit(REDISPATCH_LIMIT)
+                )
+            )
+            .scalars()
+            .all()
+        )
+
+        logger.debug("%d deliveries are about to be re-dispatched", len(redispatch))
+
+        queue = ctx["worker"].queue
+        for did in redispatch:
+            try:
+                await queue.enqueue(
+                    "deliver", delivery_id=str(did), key=f"deliver:{did}"
+                )
+            except Exception:
+                logger.exception("failed to enqueue delivery %s", did)
+
+
+async def startup(ctx: WorkerContext) -> None:
+    ctx["client"] = httpx.AsyncClient(
+        timeout=httpx.Timeout(10.0), follow_redirects=False
+    )
+    ctx["sessionmaker"] = AsyncSessionLocal
+
+
+async def shutdown(ctx: WorkerContext) -> None:
+    await ctx["client"].aclose()
+
+
+queue = Queue.from_url(str(app_settings.redis_dsn))
+
+settings: SettingsDict[WorkerContext] = SettingsDict(
+    queue=queue,
+    functions=[deliver],
+    concurrency=10,
+    cron_jobs=[CronJob(sweep, cron="* * * * * */5")],
+    startup=startup,
+    shutdown=shutdown,
+)

backend/pyproject.toml

@@ -11,13 +11,14 @@ dependencies = [
   "pydantic>=2.13",
   "pydantic-settings>=2.14",
   "alembic>=1.18",
+  "saq[redis]>=0.26",
+  "httpx>=0.28",
 ]
 
 [project.optional-dependencies]
 dev = [
   "pytest>=9.1",
   "pytest-asyncio>=1.4",
-  "httpx>=0.28",
   "respx>=0.23",
   "ruff==0.15.17",
   "basedpyright==1.39.8",