pr7 replay

README.md

@@ -14,7 +14,7 @@ A handful of decisions shape the whole thing. They're all on purpose.
 
 **At-least-once, and honest about it.** Duplicates are caught at ingest with a unique constraint on `(source, idempotency_key)`. Each delivery is claimed atomically before any work starts, so two workers can't run the same one. Claims expire on a lease and carry a fencing token, so a worker that stalls, gets replaced, and wakes up later writes nothing. Delivery can still duplicate when things go wrong. That's expected, and destinations should handle it. The system doesn't claim exactly-once, because it can't.
 
-**Retries back off, and they end.** A failed delivery is rescheduled with exponential backoff and jitter, so a struggling destination gets breathing room instead of a stampede. Attempts are capped. Whatever runs out goes to dead-letter and waits to be replayed, instead of retrying forever. And a destination you've switched off doesn't burn attempts at all. Its deliveries are held, then flow again the moment it's back on.
+**Retries back off, and they end.** A failed delivery is rescheduled with exponential backoff and jitter, so a struggling destination gets breathing room instead of a stampede. Attempts are capped. Whatever runs out goes to dead-letter and waits to be replayed, instead of retrying forever. Replay buys exactly one new attempt: succeed and it's delivered, fail and it's back in the inbox right away — not off in the background running another backoff ladder. The operator stays in the loop. And a destination you've switched off doesn't burn attempts at all. Its deliveries are held, then flow again the moment it's back on.
 
 **Destinations are locked in at ingest.** When an event arrives, its list of destinations is frozen. Replay re-runs that same list. The delivery history stays an honest record of what happened.
 
@@ -31,14 +31,14 @@ cp backend/.env.example backend/.env   # set POSTGRES_* and the DSNs
 docker compose up --build
 ```
 
-That starts Postgres, Redis, the API on `:8000`, and the worker. From there you can configure sources, destinations, and routes. Send webhooks to `POST /ingest/{source}`. Read back the event feed, the full detail for any event including its delivery attempts, and the dead-letter inbox.
+That starts Postgres, Redis, the API on `:8000`, and the worker. From there you can configure sources, destinations, and routes. Send webhooks to `POST /ingest/{source}`. Read back the event feed, the full detail for any event including its delivery attempts, and the dead-letter inbox — and send anything in it back through the pipeline with `POST /deliveries/{id}/replay`.
 
 ## Status
 
-Still being built, but the whole backend delivery story now runs end to end. An event comes in, gets verified and stored, fans out to every matched destination, and gets delivered with real retry semantics: exponential backoff with jitter, a cap on attempts, and dead-letter at the end of the line. The claim and finalize path is fenced, so even a worker that loses its lease mid-flight can't corrupt the record. A sweeper recovers anything a lost enqueue or a crash leaves stranded. Every tunable, from the lease to the backoff curve to the attempt cap, is a validated setting instead of a constant buried in the worker. The state machine is tested end to end.
+Still being built, but the backend is feature-complete: ingest with signature checks and dedupe, routing with fan-out, delivery with backoff, a cap, and dead-lettering — and now replay. `POST /deliveries/{id}/replay` sends a dead-lettered delivery back through the exact same path: same claim, same worker, same ledger. Nothing about replay is a special case, which is the point. The state machine and the replay contract are tested end to end.
 
-Next up is replay: one click to send anything in dead-letter back through the same path.
+Next up is the React dashboard.
 
 ## Planned
 
-Failed deliveries become replayable in one click, reusing the same delivery path. 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.
+A React dashboard will sit on top of the read API for inspecting payloads and replaying failures in one click. 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/db.py

@@ -1,9 +1,4 @@
-from collections.abc import AsyncIterator
-from typing import Annotated
-
-from fastapi import Depends
 from sqlalchemy.ext.asyncio import (
-    AsyncSession,
     async_sessionmaker,
     create_async_engine,
 )
@@ -18,11 +13,3 @@
 
 class Base(DeclarativeBase):
     pass
-
-
-async def get_session() -> AsyncIterator[AsyncSession]:
-    async with AsyncSessionLocal() as session:
-        yield session
-
-
-SessionDep = Annotated[AsyncSession, Depends(get_session)]

backend/app/deps.py

@@ -0,0 +1,23 @@
+from collections.abc import AsyncIterator
+from typing import Annotated
+
+from fastapi import Depends, Request
+from saq import Queue
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db import AsyncSessionLocal
+
+
+def get_queue(request: Request) -> Queue:
+    return request.app.state.queue
+
+
+QueueDep = Annotated[Queue, Depends(get_queue)]
+
+
+async def get_session() -> AsyncIterator[AsyncSession]:
+    async with AsyncSessionLocal() as session:
+        yield session
+
+
+SessionDep = Annotated[AsyncSession, Depends(get_session)]

backend/app/routers/config.py

@@ -2,7 +2,7 @@
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 
-from app.db import SessionDep
+from app.deps import SessionDep
 from app.models import Source
 from app.schemas import SourceCreate, SourceRead
 

backend/app/routers/deliveries.py

@@ -1,17 +1,62 @@
+import logging
+import uuid
 from typing import Annotated
 
-from fastapi import APIRouter, Query
-from sqlalchemy import select
+from fastapi import APIRouter, HTTPException, Query, status
+from sqlalchemy import and_, select, update
 from sqlalchemy.orm import selectinload
 
-from app.db import SessionDep
+from app.deps import QueueDep, SessionDep
 from app.models import Delivery, DeliveryStatus
 from app.schemas import DeliveryInboxItem, DeliveryRead, EventRead
 
-router = APIRouter(prefix="/deliveries/dead_letter", tags=["deliveries"])
+logger = logging.getLogger(__name__)
 
+router = APIRouter(prefix="/deliveries", tags=["deliveries"])
 
-@router.get("", response_model=list[DeliveryInboxItem])
+
+@router.post("/{delivery_id}/replay", status_code=status.HTTP_202_ACCEPTED)
+async def replay(delivery_id: uuid.UUID, session: SessionDep, queue: QueueDep):
+    owner = (
+        await session.execute(
+            update(Delivery)
+            .where(
+                and_(
+                    Delivery.id == delivery_id,
+                    Delivery.status == DeliveryStatus.dead_letter,
+                )
+            )
+            .values(
+                status=DeliveryStatus.pending,
+                next_attempt_at=None,
+                locked_by=None,
+                locked_until=None,
+            )
+            .returning(Delivery.id)
+        )
+    ).one_or_none()
+
+    if owner is not None:
+        await session.commit()
+        try:
+            await queue.enqueue("deliver", delivery_id=str(delivery_id))
+        except Exception:
+            logger.warning(
+                "replay enqueue failed for %s; sweeper will recover", delivery_id
+            )
+        return
+
+    delivery = (
+        await session.execute(select(Delivery).where(Delivery.id == delivery_id))
+    ).scalar_one_or_none()
+
+    if delivery is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="unknown delivery")
+
+    raise HTTPException(status.HTTP_409_CONFLICT, detail=f"{delivery.status.value}")
+
+
+@router.get("/dead_letter", response_model=list[DeliveryInboxItem])
 async def inbox(
     session: SessionDep,
     limit: Annotated[int, Query(le=200)] = 50,

backend/app/routers/destinations.py

@@ -3,7 +3,7 @@
 from fastapi import APIRouter, HTTPException, status
 from sqlalchemy import select
 
-from app.db import SessionDep
+from app.deps import SessionDep
 from app.models import Destination
 from app.schemas import DestinationCreate, DestinationRead, DestinationUpdate
 

backend/app/routers/events.py

@@ -5,7 +5,7 @@
 from sqlalchemy import select, tuple_
 from sqlalchemy.orm import selectinload
 
-from app.db import SessionDep
+from app.deps import SessionDep
 from app.models import Delivery, DeliveryStatus, Event, Source
 from app.pagination import CursorError, decode_cursor, encode_cursor
 from app.queries import event_rollups

backend/app/routers/ingest.py

@@ -8,7 +8,7 @@
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 
-from app.db import SessionDep
+from app.deps import QueueDep, SessionDep
 from app.models import Delivery, Event, Route, Source
 from app.schemas import IngestAck
 from app.security import verify
@@ -27,6 +27,7 @@ async def ingest(
     source_name: str,
     request: Request,
     session: SessionDep,
+    queue: QueueDep,
     response: Response,
     x_webhook_signature: Annotated[str | None, Header()] = None,
     idempotency_key: Annotated[str | None, Header()] = None,
@@ -98,9 +99,7 @@ async def ingest(
 
     try:
         for delivery in deliveries:
-            await request.app.state.queue.enqueue(
-                "deliver", delivery_id=str(delivery.id)
-            )
+            await 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/routers/routes.py

@@ -4,7 +4,7 @@
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 
-from app.db import SessionDep
+from app.deps import SessionDep
 from app.models import Destination, Route, Source
 from app.schemas import RouteCreate, RouteRead
 

backend/app/tasks.py

@@ -282,9 +282,7 @@ async def sweep(ctx: WorkerContext) -> None:
         queue = ctx["worker"].queue
         for did in redispatch:
             try:
-                await queue.enqueue(
-                    "deliver", delivery_id=str(did), key=f"deliver:{did}"
-                )
+                await queue.enqueue("deliver", delivery_id=str(did))
             except Exception:
                 logger.exception("failed to enqueue delivery %s", did)
 

backend/tests/conftest.py

@@ -7,7 +7,8 @@
 from httpx import ASGITransport, AsyncClient
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
-from app.db import Base, get_session
+from app.db import Base
+from app.deps import get_queue, get_session
 from app.main import app
 from app.models import (
     Delivery,
@@ -18,6 +19,7 @@
     Source,
 )
 from app.security import sign
+from tests.fakes import FakeQueue
 
 TEST_DATABASE_URL = os.environ["TEST_DATABASE_URL"]
 
@@ -55,7 +57,10 @@ async def override_get_session():
         async with maker() as session:
             yield session
 
+    override_get_queue = FakeQueue()
+
     app.dependency_overrides[get_session] = override_get_session
+    app.dependency_overrides[get_queue] = lambda: override_get_queue
     transport = ASGITransport(app=app)
     async with AsyncClient(transport=transport, base_url="http://test") as ac:
         yield ac

backend/tests/fakes.py

@@ -1,8 +1,11 @@
-from typing import override
+from datetime import UTC, datetime, timedelta
+from typing import cast, override
 
 import httpx
+from sqlalchemy import update
 
-from app.tasks import DeliveryResult, DeliverySnapshot, SendFn
+from app.models import Delivery
+from app.tasks import DeliveryResult, DeliverySnapshot, SendFn, WorkerContext, deliver
 
 
 class FakeQueue:
@@ -39,3 +42,61 @@ async def _send_fn(
 class FakeWorker:
     def __init__(self, queue) -> None:
         self.queue = queue
+
+
+def ctx(*, queue=None, client=None, sessionmaker=None) -> WorkerContext:
+    worker_context = {}
+    if queue is not None:
+        worker_context["worker"] = FakeWorker(queue)
+    if client is not None:
+        worker_context["client"] = client
+    if sessionmaker is not None:
+        worker_context["sessionmaker"] = sessionmaker
+    return cast(WorkerContext, worker_context)
+
+
+def fail_result() -> DeliveryResult:
+    return DeliveryResult(
+        success=False,
+        response_status=500,
+        error=None,
+        duration_ms=10,
+    )
+
+
+def ok_result() -> DeliveryResult:
+    return DeliveryResult(
+        success=True,
+        response_status=200,
+        response_body="",
+        error=None,
+        duration_ms=10,
+    )
+
+
+def reclaiming_send(sessionmaker_factory, delivery, a_calls, b_calls):
+
+    async def a_send(
+        client: httpx.AsyncClient, snapshot: DeliverySnapshot
+    ) -> DeliveryResult:
+        a_calls.append(snapshot)
+        async with sessionmaker_factory() as s:
+            await s.execute(
+                update(Delivery)
+                .where(Delivery.id == delivery.id)
+                .values(locked_until=datetime.now(UTC) - timedelta(seconds=1))
+            )
+            await s.commit()
+        await deliver(
+            ctx(client=client, sessionmaker=sessionmaker_factory),
+            delivery_id=str(delivery.id),
+            send_fn=send_fn([ok_result()], b_calls),
+        )
+
+        return fail_result()
+
+    return a_send
+
+
+def fake_rng(a: float, b: float) -> float:
+    return b - a