diff --git a/.gitignore b/.gitignore index c289cc3..9b523bb 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ __pycache__/ .venv/ *.egg-info/ .pytest_cache/ +build/ diff --git a/README.md b/README.md index d694745..ca33ffa 100644 --- a/README.md +++ b/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. diff --git a/backend/app/main.py b/backend/app/main.py index 8b107de..756d85c 100644 --- a/backend/app/main.py +++ b/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) diff --git a/backend/app/routers/ingest.py b/backend/app/routers/ingest.py index b8f6a73..54530fe 100644 --- a/backend/app/routers/ingest.py +++ b/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) diff --git a/backend/app/tasks.py b/backend/app/tasks.py new file mode 100644 index 0000000..2463802 --- /dev/null +++ b/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, +) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index f29e60f..35f86e5 100644 --- a/backend/pyproject.toml +++ b/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", diff --git a/backend/tests/__init__.py b/backend/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 379f4b1..1a607fc 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -1,5 +1,6 @@ import json import os +from datetime import UTC, datetime import pytest import pytest_asyncio @@ -8,7 +9,14 @@ from app.db import Base, get_session from app.main import app -from app.models import Delivery, Destination, Event, Source +from app.models import ( + Delivery, + DeliveryAttempt, + DeliveryStatus, + Destination, + Event, + Source, +) from app.security import sign TEST_DATABASE_URL = os.environ["TEST_DATABASE_URL"] @@ -34,6 +42,11 @@ async def db_session(db_engine): yield session +@pytest_asyncio.fixture +async def sessionmaker_factory(db_engine): + return async_sessionmaker(db_engine, expire_on_commit=False) + + @pytest_asyncio.fixture async def client(db_engine): maker = async_sessionmaker(db_engine, expire_on_commit=False) @@ -80,7 +93,7 @@ async def source(client) -> str: @pytest_asyncio.fixture async def make_event(db_session): src = Source(name="seed-src", signing_secret="x") - dst = Destination(name="seed-dst", url="http://t.test") + dst = Destination(name="seed-dst", url="http://t.test/") db_session.add_all([src, dst]) await db_session.flush() n = 0 @@ -110,3 +123,53 @@ async def _make_event(*, source_id=None, received_at, statuses=()): return event return _make_event + + +@pytest_asyncio.fixture +async def make_delivery(db_session): + src = Source(name="mk-seed-src", signing_secret="x") + dst = Destination(name="mk-seed-dst", url="http://t.test/") + db_session.add_all([src, dst]) + await db_session.flush() + + n = 0 + + async def _make_delivery( + *, status, next_attempt_at, attempt_count, updated_at=None + ): + nonlocal n + n += 1 + event = Event( + source_id=src.id, + idempotency_key=f"k{n}", + payload={}, + headers={}, + ) + db_session.add(event) + await db_session.flush() + delivery = Delivery( + event_id=event.id, + destination_id=dst.id, + status=status, + next_attempt_at=next_attempt_at, + attempt_count=attempt_count, + updated_at=updated_at or datetime.now(UTC), + ) + db_session.add(delivery) + await db_session.flush() + + if status is not DeliveryStatus.pending: + attempt = DeliveryAttempt( + delivery_id=delivery.id, + attempt_number=1, + response_status=None, + response_body=None, + error=None, + duration_ms=0, + ) + db_session.add(attempt) + + await db_session.commit() + return delivery + + return _make_delivery diff --git a/backend/tests/fakes.py b/backend/tests/fakes.py new file mode 100644 index 0000000..f904379 --- /dev/null +++ b/backend/tests/fakes.py @@ -0,0 +1,41 @@ +from typing import override + +import httpx + +from app.tasks import DeliveryResult, DeliverySnapshot, SendFn + + +class FakeQueue: + def __init__(self) -> None: + self.enqueued = [] + + async def enqueue(self, name, **kwargs): + self.enqueued.append(kwargs) + + +class FakeRaisingQueue(FakeQueue): + def __init__(self, fail_for: set[str], fail_all: bool = False) -> None: + super().__init__() + self.fail_for = fail_for + self.fail_all = fail_all + + @override + async def enqueue(self, name, **kwargs): + if self.fail_all or kwargs["delivery_id"] in self.fail_for: + raise RuntimeError + await super().enqueue(name, **kwargs) + + +def send_fn(results: list[DeliveryResult], calls: list[DeliverySnapshot]) -> SendFn: + async def _send_fn( + client: httpx.AsyncClient, snapshot: DeliverySnapshot + ) -> DeliveryResult: + calls.append(snapshot) + return results.pop(0) + + return _send_fn + + +class FakeWorker: + def __init__(self, queue) -> None: + self.queue = queue diff --git a/backend/tests/test_claim.py b/backend/tests/test_claim.py new file mode 100644 index 0000000..247c5f2 --- /dev/null +++ b/backend/tests/test_claim.py @@ -0,0 +1,278 @@ +import asyncio +from datetime import UTC, datetime, timedelta +from typing import cast + +import httpx +from sqlalchemy import select +from sqlalchemy.orm import selectinload + +from app.models import Delivery, DeliveryStatus +from app.tasks import LEASE, DeliveryResult, DeliverySnapshot, WorkerContext, deliver +from tests.fakes import send_fn + + +def _ctx(*, client, sessionmaker) -> WorkerContext: + return cast(WorkerContext, {"client": client, "sessionmaker": sessionmaker}) + + +def _ok_result() -> DeliveryResult: + return DeliveryResult( + success=True, response_status=200, response_body="", error=None, duration_ms=10 + ) + + +async def test_claims_pending_row(make_delivery, sessionmaker_factory): + delivery = await make_delivery( + status=DeliveryStatus.pending, + attempt_count=0, + next_attempt_at=datetime.now(UTC) - timedelta(seconds=1), + ) + + results: list[DeliveryResult] = [_ok_result()] + calls: list[DeliverySnapshot] = [] + + async with httpx.AsyncClient() as worker_client: + await deliver( + _ctx(client=worker_client, sessionmaker=sessionmaker_factory), + delivery_id=str(delivery.id), + send_fn=send_fn(results, calls), + ) + + async with sessionmaker_factory() as check: + updated_delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.id == delivery.id) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert updated_delivery.next_attempt_at is None + assert updated_delivery.status == DeliveryStatus.succeeded + assert updated_delivery.attempt_count == 1 + assert len(updated_delivery.attempts) == 1 + assert updated_delivery.attempts[0].response_status == 200 + assert len(calls) == 1 + assert len(results) == 0 + + +async def test_claims_nextattemptat_null_pending_row( + make_delivery, sessionmaker_factory +): + delivery = await make_delivery( + status=DeliveryStatus.pending, + attempt_count=0, + next_attempt_at=None, + ) + + results: list[DeliveryResult] = [_ok_result()] + calls: list[DeliverySnapshot] = [] + + async with httpx.AsyncClient() as worker_client: + await deliver( + _ctx(client=worker_client, sessionmaker=sessionmaker_factory), + delivery_id=str(delivery.id), + send_fn=send_fn(results, calls), + ) + + async with sessionmaker_factory() as check: + updated_delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.id == delivery.id) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert updated_delivery.next_attempt_at is None + assert updated_delivery.status == DeliveryStatus.succeeded + assert updated_delivery.attempt_count == 1 + assert len(updated_delivery.attempts) == 1 + assert updated_delivery.attempts[0].response_status == 200 + assert len(calls) == 1 + assert len(results) == 0 + + +async def test_does_not_claim_future_scheduled_row(make_delivery, sessionmaker_factory): + next_attempt = datetime.now(UTC) + timedelta(minutes=1) + delivery = await make_delivery( + status=DeliveryStatus.pending, + attempt_count=0, + next_attempt_at=next_attempt, + ) + + results: list[DeliveryResult] = [_ok_result()] + calls: list[DeliverySnapshot] = [] + + async with httpx.AsyncClient() as worker_client: + await deliver( + _ctx(client=worker_client, sessionmaker=sessionmaker_factory), + delivery_id=str(delivery.id), + send_fn=send_fn(results, calls), + ) + + async with sessionmaker_factory() as check: + updated_delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.id == delivery.id) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert updated_delivery.next_attempt_at == next_attempt + assert updated_delivery.status == DeliveryStatus.pending + assert updated_delivery.attempt_count == 0 + assert len(updated_delivery.attempts) == 0 + assert len(calls) == 0 + assert len(results) == 1 + + +async def test_does_not_claim_already_claimed(make_delivery, sessionmaker_factory): + half_lease = timedelta(seconds=LEASE / 2) + delivery = await make_delivery( + status=DeliveryStatus.delivering, + attempt_count=1, + next_attempt_at=datetime.now(UTC) + half_lease, + updated_at=datetime.now(UTC) - half_lease, + ) + + results: list[DeliveryResult] = [_ok_result()] + calls: list[DeliverySnapshot] = [] + + async with httpx.AsyncClient() as worker_client: + await deliver( + _ctx(client=worker_client, sessionmaker=sessionmaker_factory), + delivery_id=str(delivery.id), + send_fn=send_fn(results, calls), + ) + + async with sessionmaker_factory() as check: + updated_delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.id == delivery.id) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert updated_delivery.next_attempt_at is not None + assert updated_delivery.status == DeliveryStatus.delivering + assert updated_delivery.attempt_count == 1 + assert len(updated_delivery.attempts) == 1 + assert len(calls) == 0 + assert len(results) == 1 + + +async def test_claims_orphaned_row(make_delivery, sessionmaker_factory): + delivery = await make_delivery( + status=DeliveryStatus.delivering, + attempt_count=1, + next_attempt_at=datetime.now(UTC), + updated_at=datetime.now(UTC) - timedelta(seconds=LEASE * 5), + ) + + results: list[DeliveryResult] = [_ok_result()] + calls: list[DeliverySnapshot] = [] + + async with httpx.AsyncClient() as worker_client: + await deliver( + _ctx(client=worker_client, sessionmaker=sessionmaker_factory), + delivery_id=str(delivery.id), + send_fn=send_fn(results, calls), + ) + + async with sessionmaker_factory() as check: + updated_delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.id == delivery.id) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert updated_delivery.next_attempt_at is None + assert updated_delivery.status == DeliveryStatus.succeeded + assert updated_delivery.attempt_count == 2 + assert len(updated_delivery.attempts) == 2 + assert len(calls) == 1 + assert len(results) == 0 + + +async def test_atomic_claim_under_concurrency(make_delivery, sessionmaker_factory): + delivery = await make_delivery( + status=DeliveryStatus.pending, + attempt_count=0, + next_attempt_at=datetime.now(UTC), + ) + + results: list[DeliveryResult] = [_ok_result(), _ok_result()] + calls: list[DeliverySnapshot] = [] + + async with httpx.AsyncClient() as worker_client: + ctx = _ctx(client=worker_client, sessionmaker=sessionmaker_factory) + + await asyncio.gather( + deliver( + ctx, + delivery_id=str(delivery.id), + send_fn=send_fn(results, calls), + ), + deliver( + ctx, + delivery_id=str(delivery.id), + send_fn=send_fn(results, calls), + ), + ) + + async with sessionmaker_factory() as check: + updated_delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.id == delivery.id) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert updated_delivery.next_attempt_at is None + assert updated_delivery.status == DeliveryStatus.succeeded + assert updated_delivery.attempt_count == 1 + assert len(updated_delivery.attempts) == 1 + assert len(calls) == 1 + assert len(results) == 1 + + +async def test_claims_failed_row(make_delivery, sessionmaker_factory): + past_lease = timedelta(seconds=LEASE * 2) + delivery = await make_delivery( + status=DeliveryStatus.failed, + attempt_count=1, + next_attempt_at=datetime.now(UTC) - past_lease, + updated_at=datetime.now(UTC) - past_lease, + ) + + results: list[DeliveryResult] = [_ok_result()] + calls: list[DeliverySnapshot] = [] + + async with httpx.AsyncClient() as worker_client: + await deliver( + _ctx(client=worker_client, sessionmaker=sessionmaker_factory), + delivery_id=str(delivery.id), + send_fn=send_fn(results, calls), + ) + + async with sessionmaker_factory() as check: + updated_delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.id == delivery.id) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert updated_delivery.next_attempt_at is None + assert updated_delivery.status == DeliveryStatus.succeeded + assert updated_delivery.attempt_count == 2 + assert len(updated_delivery.attempts) == 2 + assert len(calls) == 1 + assert len(results) == 0 diff --git a/backend/tests/test_deliver.py b/backend/tests/test_deliver.py new file mode 100644 index 0000000..d153815 --- /dev/null +++ b/backend/tests/test_deliver.py @@ -0,0 +1,147 @@ +import json +from datetime import UTC, datetime +from typing import cast + +import httpx +from sqlalchemy import select +from sqlalchemy.orm import selectinload + +from app.models import Delivery, DeliveryStatus +from app.tasks import WorkerContext, deliver + + +async def test_successful_delivery( + make_event, respx_mock, db_session, sessionmaker_factory +): + + event = await make_event( + received_at=datetime.fromisoformat("2011-11-04T00:05:23+04:00"), + statuses=(DeliveryStatus.pending,), + ) + + delivery_id = ( + await db_session.execute( + select(Delivery.id).where(Delivery.event_id == event.id) + ) + ).scalar_one() + + respx_mock.route(method="POST", host="t.test").mock( + return_value=httpx.Response(200, text="ok") + ) + + async with httpx.AsyncClient() as worker_client: + await deliver( + cast( + WorkerContext, + {"client": worker_client, "sessionmaker": sessionmaker_factory}, + ), + delivery_id=str(delivery_id), + ) + + async with sessionmaker_factory() as check: + delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.id == delivery_id) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert delivery.status == DeliveryStatus.succeeded + assert delivery.attempt_count == 1 + assert len(delivery.attempts) == 1 + assert delivery.attempts[0].response_status == 200 + + +async def test_failed_delivery( + make_event, respx_mock, db_session, sessionmaker_factory +): + + event = await make_event( + received_at=datetime.fromisoformat("2011-11-04T00:05:23+04:00"), + statuses=(DeliveryStatus.pending,), + ) + + delivery_id = ( + await db_session.execute( + select(Delivery.id).where(Delivery.event_id == event.id) + ) + ).scalar_one() + + respx_mock.route(method="POST", host="t.test").mock( + return_value=httpx.Response(500, text="err") + ) + + async with httpx.AsyncClient() as worker_client: + await deliver( + cast( + WorkerContext, + {"client": worker_client, "sessionmaker": sessionmaker_factory}, + ), + delivery_id=str(delivery_id), + ) + + async with sessionmaker_factory() as check: + delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.event_id == event.id) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert delivery.next_attempt_at > datetime.now(UTC) + assert delivery.status == DeliveryStatus.failed + assert delivery.attempt_count == 1 + assert len(delivery.attempts) == 1 + assert delivery.attempts[0].response_status == 500 + + +async def test_timeout_delivery( + make_event, respx_mock, db_session, sessionmaker_factory +): + + event = await make_event( + received_at=datetime.fromisoformat("2011-11-04T00:05:23+04:00"), + statuses=(DeliveryStatus.pending,), + ) + + delivery_id = ( + await db_session.execute( + select(Delivery.id).where(Delivery.event_id == event.id) + ) + ).scalar_one() + + respx_mock.route(method="POST", host="t.test").mock( + side_effect=httpx.TimeoutException + ) + + async with httpx.AsyncClient() as worker_client: + await deliver( + cast( + WorkerContext, + {"client": worker_client, "sessionmaker": sessionmaker_factory}, + ), + delivery_id=str(delivery_id), + ) + + async with sessionmaker_factory() as check: + delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.event_id == event.id) + .options( + selectinload(Delivery.attempts), selectinload(Delivery.destination) + ) + ) + ).scalar_one() + + assert delivery.next_attempt_at > datetime.now(UTC) + assert delivery.status == DeliveryStatus.failed + assert delivery.attempt_count == 1 + assert len(delivery.attempts) == 1 + assert delivery.attempts[0].error is not None + + call = respx_mock.calls[0] + assert str(call.request.url) == delivery.destination.url + assert json.loads(call.request.content) == event.payload diff --git a/backend/tests/test_ingest.py b/backend/tests/test_ingest.py index b715ad7..bae2431 100644 --- a/backend/tests/test_ingest.py +++ b/backend/tests/test_ingest.py @@ -1,3 +1,8 @@ +from app.main import app +from app.models import DeliveryStatus +from tests.fakes import FakeQueue, FakeRaisingQueue + + async def test_create_source_hides_secret(client): r = await client.post( "/sources", json={"name": "stripe-test", "signing_secret": "secret"} @@ -28,3 +33,110 @@ async def test_ingest_unknown_source(client): ) assert r.status_code == 404 + + +async def _setup_routes(client, source): + src_id = (await client.get("/sources")).json()[0]["id"] + dst = ( + await client.post( + "/destinations", + json={ + "name": "test_dst", + "url": "http://localhost:8000/docs/", + "signing_secret": "secret", + }, + ) + ).json()["id"] + dst2 = ( + await client.post( + "/destinations", + json={ + "name": "test_dst_2", + "url": "http://localhost:8000/", + "signing_secret": "secret_2", + }, + ) + ).json()["id"] + + await client.post( + "/routes", json={"source_id": f"{src_id}", "destination_id": f"{dst}"} + ) + await client.post( + "/routes", json={"source_id": f"{src_id}", "destination_id": f"{dst2}"} + ) + + +async def test_ingest_enqueue(client, source, signed): + await _setup_routes(client, source) + raw, headers = signed({"type": "payment.succeeded"}, key="evt_1") + + app.state.queue = FakeQueue() + + try: + r = await client.post(f"/ingest/{source}", content=raw, headers=headers) + assert r.status_code == 202 + event_id = r.json()["event_id"] + + r2 = await client.get(f"/events/{event_id}") + assert r2.status_code == 200 + + event_details = r2.json() + assert event_details["id"] == event_id + assert {e["delivery_id"] for e in app.state.queue.enqueued} == { + event_details["deliveries"][0]["id"], + event_details["deliveries"][1]["id"], + } + finally: + del app.state.queue + + +async def test_ingest_failed_enqueue(client, source, signed): + await _setup_routes(client, source) + raw, headers = signed({"type": "payment.succeeded"}, key="evt_1") + + app.state.queue = FakeRaisingQueue(fail_for=set(), fail_all=True) + + try: + r = await client.post(f"/ingest/{source}", content=raw, headers=headers) + assert r.status_code == 202 + event_id = r.json()["event_id"] + + r2 = await client.get(f"/events/{event_id}") + assert r2.status_code == 200 + + event_details = r2.json() + assert event_details["id"] == event_id + assert event_details["deliveries"][0]["status"] == DeliveryStatus.pending + assert event_details["deliveries"][1]["status"] == DeliveryStatus.pending + assert app.state.queue.enqueued == [] + finally: + del app.state.queue + + +async def test_ingest_no_enqueue_on_duplicate(client, source, signed): + await _setup_routes(client, source) + raw, headers = signed({"type": "payment.succeeded"}, key="evt_1") + + app.state.queue = FakeQueue() + + try: + r = await client.post(f"/ingest/{source}", content=raw, headers=headers) + assert r.status_code == 202 + event_id = r.json()["event_id"] + + r2 = await client.post(f"/ingest/{source}", content=raw, headers=headers) + assert r2.status_code == 200 + assert event_id == r2.json()["event_id"] + + r3 = await client.get(f"/events/{event_id}") + assert r3.status_code == 200 + + event_details = r3.json() + assert event_details["id"] == event_id + assert {e["delivery_id"] for e in app.state.queue.enqueued} == { + event_details["deliveries"][0]["id"], + event_details["deliveries"][1]["id"], + } + assert len(app.state.queue.enqueued) == 2 + finally: + del app.state.queue diff --git a/backend/tests/test_retry_loop.py b/backend/tests/test_retry_loop.py new file mode 100644 index 0000000..1fdf87d --- /dev/null +++ b/backend/tests/test_retry_loop.py @@ -0,0 +1,140 @@ +from datetime import UTC, datetime, timedelta +from typing import cast + +import httpx +from sqlalchemy import select, update +from sqlalchemy.orm import selectinload + +from app.main import app +from app.models import Delivery, DeliveryStatus +from app.tasks import ( + LEASE, + DeliveryResult, + DeliverySnapshot, + WorkerContext, + deliver, + sweep, +) +from tests.fakes import FakeQueue, FakeWorker, send_fn + + +async def test_retry_loop(client, source, sessionmaker_factory, signed): + src_id = (await client.get("/sources")).json()[0]["id"] + dst = ( + await client.post( + "/destinations", + json={ + "name": "test_dst", + "url": "http://localhost:8000/docs/", + "signing_secret": "secret", + }, + ) + ).json()["id"] + + await client.post( + "/routes", json={"source_id": f"{src_id}", "destination_id": f"{dst}"} + ) + + app.state.queue = FakeQueue() + + try: + raw, headers = signed({"type": "payment.succeeded"}, key="evt_1") + r = await client.post(f"/ingest/{source}", content=raw, headers=headers) + assert r.status_code == 202 + event_id = r.json()["event_id"] + r2 = await client.get(f"/events/{event_id}") + assert r2.status_code == 200 + event_details = r2.json() + + results: list[DeliveryResult] = [ + DeliveryResult( + success=False, + response_status=401, + response_body="", + error="error", + duration_ms=10, + ), + DeliveryResult( + success=True, + response_status=200, + response_body="", + error=None, + duration_ms=10, + ), + ] + calls: list[DeliverySnapshot] = [] + + async with httpx.AsyncClient() as worker_client: + await deliver( + cast( + WorkerContext, + {"client": worker_client, "sessionmaker": sessionmaker_factory}, + ), + delivery_id=str(event_details["deliveries"][0]["id"]), + send_fn=send_fn(results, calls), + ) + + async with sessionmaker_factory() as check: + updated_delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.id == event_details["deliveries"][0]["id"]) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert updated_delivery.status == DeliveryStatus.failed + assert updated_delivery.attempt_count == 1 + assert len(updated_delivery.attempts) == 1 + assert len(calls) == 1 + assert len(results) == 1 + + async with sessionmaker_factory() as check: + await check.execute( + update(Delivery) + .values( + next_attempt_at=datetime.now(UTC) - timedelta(seconds=LEASE * 2) + ) + .where(Delivery.id == event_details["deliveries"][0]["id"]) + ) + await check.commit() + + ctx = cast( + WorkerContext, + { + "worker": FakeWorker(app.state.queue), + "sessionmaker": sessionmaker_factory, + }, + ) + await sweep(ctx) + + assert event_details["deliveries"][0]["id"] in { + e["delivery_id"] for e in app.state.queue.enqueued + } + + async with httpx.AsyncClient() as worker_client: + await deliver( + cast( + WorkerContext, + {"client": worker_client, "sessionmaker": sessionmaker_factory}, + ), + delivery_id=str(event_details["deliveries"][0]["id"]), + send_fn=send_fn(results, calls), + ) + + async with sessionmaker_factory() as check: + updated_delivery = ( + await check.execute( + select(Delivery) + .where(Delivery.id == event_details["deliveries"][0]["id"]) + .options(selectinload(Delivery.attempts)) + ) + ).scalar_one() + + assert updated_delivery.status == DeliveryStatus.succeeded + assert updated_delivery.attempt_count == 2 + assert len(updated_delivery.attempts) == 2 + assert len(calls) == 2 + assert len(results) == 0 + finally: + del app.state.queue diff --git a/backend/tests/test_sweeper.py b/backend/tests/test_sweeper.py new file mode 100644 index 0000000..d012892 --- /dev/null +++ b/backend/tests/test_sweeper.py @@ -0,0 +1,110 @@ +from datetime import UTC, datetime, timedelta +from typing import cast + +from app.models import DeliveryStatus +from app.tasks import LEASE, REDISPATCH_LIMIT, WorkerContext, sweep +from tests.fakes import FakeQueue, FakeRaisingQueue, FakeWorker + + +def _ctx(*, queue, sessionmaker) -> WorkerContext: + return cast( + WorkerContext, {"worker": FakeWorker(queue), "sessionmaker": sessionmaker} + ) + + +async def test_sweep_mix_of_rows(make_delivery, sessionmaker_factory): + # succedeed + await make_delivery( + status=DeliveryStatus.succeeded, + attempt_count=1, + next_attempt_at=datetime.now(UTC), + ) + + pending = await make_delivery( + status=DeliveryStatus.pending, + attempt_count=0, + next_attempt_at=datetime.now(UTC) - timedelta(seconds=LEASE / 2), + ) + + failed = await make_delivery( + status=DeliveryStatus.failed, + attempt_count=1, + next_attempt_at=datetime.now(UTC) - timedelta(seconds=LEASE * 2), + updated_at=datetime.now(UTC) - timedelta(seconds=LEASE * 2), + ) + + pending_null = await make_delivery( + status=DeliveryStatus.pending, + attempt_count=0, + next_attempt_at=None, + ) + + orphaned = await make_delivery( + status=DeliveryStatus.delivering, + attempt_count=1, + next_attempt_at=datetime.now(UTC), + updated_at=datetime.now(UTC) - timedelta(seconds=LEASE * 5), + ) + + # scheduled + await make_delivery( + status=DeliveryStatus.pending, + attempt_count=0, + next_attempt_at=datetime.now(UTC) + timedelta(minutes=1), + ) + + # delivering + await make_delivery( + status=DeliveryStatus.delivering, + attempt_count=1, + next_attempt_at=datetime.now(UTC) + timedelta(seconds=LEASE / 2), + updated_at=datetime.now(UTC) - timedelta(seconds=LEASE / 2), + ) + + fake_queue = FakeQueue() + ctx = _ctx(queue=fake_queue, sessionmaker=sessionmaker_factory) + + await sweep(ctx) + + assert {e["delivery_id"] for e in fake_queue.enqueued} == { + str(pending.id), + str(failed.id), + str(pending_null.id), + str(orphaned.id), + } + + +async def test_bounded_batch(make_delivery, sessionmaker_factory): + for _ in range(REDISPATCH_LIMIT + 1): + await make_delivery( + status=DeliveryStatus.pending, + attempt_count=0, + next_attempt_at=datetime.now(UTC) - timedelta(seconds=LEASE / 2), + ) + + fake_queue = FakeQueue() + ctx = _ctx(queue=fake_queue, sessionmaker=sessionmaker_factory) + + await sweep(ctx) + + assert len(fake_queue.enqueued) == REDISPATCH_LIMIT + + +async def test_failure_is_skipped(make_delivery, sessionmaker_factory): + first = await make_delivery( + status=DeliveryStatus.pending, + attempt_count=0, + next_attempt_at=datetime.now(UTC) - timedelta(seconds=LEASE / 2), + ) + second = await make_delivery( + status=DeliveryStatus.pending, + attempt_count=0, + next_attempt_at=datetime.now(UTC) - timedelta(seconds=LEASE / 2), + ) + + fake_raising_queue = FakeRaisingQueue(fail_for={str(first.id)}) + ctx = _ctx(queue=fake_raising_queue, sessionmaker=sessionmaker_factory) + await sweep(ctx) + + assert len(fake_raising_queue.enqueued) == 1 + assert fake_raising_queue.enqueued[0]["delivery_id"] == str(second.id) diff --git a/compose.yml b/compose.yml index 68c1478..a6d56d9 100644 --- a/compose.yml +++ b/compose.yml @@ -26,6 +26,7 @@ services: retries: 10 api: + image: app build: ./backend environment: DATABASE_URL: postgresql+asyncpg://${POSTGRES_USER}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB} @@ -40,5 +41,20 @@ services: redis: condition: service_healthy + worker: + image: app + build: ./backend + command: watchfiles "saq app.tasks.settings" /app/app + environment: + DATABASE_URL: postgresql+asyncpg://${POSTGRES_USER}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB} + REDIS_URL: redis://redis:6379/0 + volumes: + - ./backend/app:/app/app + depends_on: + db: + condition: service_healthy + redis: + condition: service_healthy + volumes: pgdata: