diff --git a/docs/usage/observability.md b/docs/usage/observability.md index bd78503..99e3472 100644 --- a/docs/usage/observability.md +++ b/docs/usage/observability.md @@ -158,9 +158,9 @@ publish-scope metrics do not fire in test mode. Middleware consume-scope metrics still fire, because handlers are still consumed through the normal middleware stack. -The recorder-seam `published` event provides synthetic publish-side -coverage in test mode via `FakeOutboxProducer`. The synthetic events use -`duration_seconds=0.0` since the in-memory client has no real write to -time. +The recorder-seam `published` event still fires in test mode, with the +same tags as in production: `status="error"` and `exception_type` when +encoding fails, and a measured `duration_seconds` for the in-memory +insert. `TestKafkaBroker` and `TestRabbitBroker` behave the same way, for the same reason. diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index 790e782..41497b3 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -141,6 +141,11 @@ async def validate_schema(self, *, check_autovacuum: bool = False) -> None: ... async def ping(self) -> bool: ... +_NO_DLQ_TABLE_MSG = ( + "delete_with_lease received a dlq_payload but the client has no dlq_table; refusing to drop the audit row silently" +) + + def _build_fetch_stmt(outbox: "Table") -> "Update": """Build the claim statement once; queues, limit, lease TTL and the new token are bind parameters. @@ -261,7 +266,8 @@ async def fetch( async with conn.begin(): result = await conn.execute(self._fetch_stmt, params) rows = result.mappings().all() - return [_row_to_message(dict(row)) for row in rows] + # UPDATE ... RETURNING has no ORDER BY, so the claimed rows come back in arbitrary order. + return sorted((_row_to_message(dict(row)) for row in rows), key=lambda m: (m.next_attempt_at, m.id)) async def delete_with_lease( self, @@ -312,10 +318,7 @@ async def delete_with_lease( # P10: a dlq_payload with no dlq_table would silently degrade to a plain # DELETE — losing the audit row. If the broker/client wiring ever desyncs, # fail loudly rather than drop forensics. - msg = ( - "delete_with_lease received a dlq_payload but the client has no dlq_table; " - "refusing to drop the audit row silently" - ) + msg = _NO_DLQ_TABLE_MSG raise RuntimeError(msg) params = { "message_id": message_id, diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index 8870798..4bba0ac 100644 --- a/faststream_outbox/testing.py +++ b/faststream_outbox/testing.py @@ -12,6 +12,8 @@ """ import datetime as dt +import json +import time import typing import uuid from contextlib import contextmanager @@ -23,11 +25,11 @@ from faststream_outbox._scheduling import resolve_next_attempt_client_side, validate_activate_args from faststream_outbox._time import utcnow from faststream_outbox.broker import OutboxBroker -from faststream_outbox.client import AbstractOutboxClient +from faststream_outbox.client import _NO_DLQ_TABLE_MSG, AbstractOutboxClient from faststream_outbox.envelope import _encode_payload from faststream_outbox.message import OutboxInnerMessage from faststream_outbox.metrics import _safe_emit -from faststream_outbox.response import _REQUEST_UNSUPPORTED_MSG, OutboxPublishCommand +from faststream_outbox.response import _MAX_QUEUE_LENGTH, _REQUEST_UNSUPPORTED_MSG, OutboxPublishCommand from faststream_outbox.schema import _DLQ_INJECTED_COLUMNS, _DLQ_PROJECTION @@ -82,10 +84,30 @@ def _lease_token_matches(row_token: "uuid.UUID | None", claim_token: "uuid.UUID return claim_token is not None and row_token == claim_token +def _validate_like_postgres(*, queue: str, timer_id: str | None, headers: dict[str, str] | None) -> None: + """Reject what the INSERT would fail on: over-long ``String(255)`` values and headers JSONB can't hold.""" + if len(queue) > _MAX_QUEUE_LENGTH: + msg = f"queue must be at most {_MAX_QUEUE_LENGTH} characters (got {len(queue)})" + raise ValueError(msg) + if timer_id is not None and len(timer_id) > _MAX_QUEUE_LENGTH: + msg = f"timer_id must be at most {_MAX_QUEUE_LENGTH} characters (got {len(timer_id)})" + raise ValueError(msg) + try: + json.dumps(headers) + except (TypeError, ValueError) as exc: + msg = f"headers must be JSON-serializable: {exc}" + raise ValueError(msg) from exc + + class FakeOutboxClient(AbstractOutboxClient): - """In-memory ``OutboxClient`` substitute. Same surface, list-of-rows storage.""" + """In-memory ``OutboxClient`` substitute. Same surface, list-of-rows storage. - def __init__(self) -> None: + ``dlq_enabled=False`` mirrors an ``OutboxClient`` built without a ``dlq_table``: a + ``dlq_payload`` is refused instead of producing an audit row. + """ + + def __init__(self, *, dlq_enabled: bool = True) -> None: + self._dlq_enabled = dlq_enabled self._rows: list[_FakeRow] = [] self._next_id = 1 # Populated when ``delete_with_lease`` receives a ``dlq_payload``. Mirrors @@ -108,6 +130,7 @@ def feed( if next_attempt_at is not None and next_attempt_at.tzinfo is None: msg = "feed() requires next_attempt_at to be timezone-aware" raise ValueError(msg) + _validate_like_postgres(queue=queue, timer_id=timer_id, headers=headers) # Mirror the real client's partial-unique-on-(queue, timer_id) behavior: # re-feeding a timer that already exists is a no-op. if timer_id is not None and any(r.queue == queue and r.timer_id == timer_id for r in self._rows): @@ -185,6 +208,8 @@ async def delete_with_lease( for i, row in enumerate(self._rows): if row.id == message_id and _lease_token_matches(row.acquired_token, acquired_token): if dlq_payload is not None: + if not self._dlq_enabled: + raise RuntimeError(_NO_DLQ_TABLE_MSG) # Mirror the real CTE side-effect: DLQ row materializes in the same # call as the DELETE, before the row is removed. Built from the shared # _DLQ_PROJECTION so the fake can't drift from the real CTE; failed_at @@ -336,19 +361,26 @@ async def _sync_dispatch(fake_client: FakeOutboxClient, broker: OutboxBroker, qu await subscriber.dispatch_one(_to_inner(fake_row)) -def _emit_published(broker: OutboxBroker, queue: str, *, count: int, size_bytes: int) -> None: - """Emit the producer-side ``published`` metric on the broker's recorder (test-broker parity).""" - _safe_emit( - broker.config.broker_config.metrics_recorder, - "published", - { - "queue": queue, - "status": "success", - "count": count, - "size_bytes": size_bytes, - "duration_seconds": 0.0, - }, - ) +def _emit_published( + broker: OutboxBroker, + queue: str, + *, + count: int, + size_bytes: int, + start_perf: float, + exc: Exception | None = None, +) -> None: + """Emit the producer-side ``published`` metric with the same tags as ``OutboxProducer``.""" + tags: dict[str, typing.Any] = { + "queue": queue, + "status": "success" if exc is None else "error", + "count": count, + "size_bytes": size_bytes, + "duration_seconds": time.perf_counter() - start_perf, + } + if exc is not None: + tags["exception_type"] = type(exc).__name__ + _safe_emit(broker.config.broker_config.metrics_recorder, "published", tags) def _notify_subscriber(broker: OutboxBroker, queue: str) -> None: @@ -376,9 +408,18 @@ async def _fake_publish_one( run_loops: bool, ) -> int | None: """Shared single-row insert for ``FakeOutboxProducer.publish`` and the ``broker.publish`` patch (P29).""" - payload, hdrs = _encode_payload(body, headers=headers, correlation_id=correlation_id, serializer=serializer) - row_id = fake_client.feed(queue=queue, payload=payload, headers=hdrs, next_attempt_at=next_at, timer_id=timer_id) - _emit_published(broker, queue, count=0 if row_id is None else 1, size_bytes=len(payload)) + start_perf = time.perf_counter() + size_bytes = 0 + try: + payload, hdrs = _encode_payload(body, headers=headers, correlation_id=correlation_id, serializer=serializer) + size_bytes = len(payload) + row_id = fake_client.feed( + queue=queue, payload=payload, headers=hdrs, next_attempt_at=next_at, timer_id=timer_id + ) + except Exception as exc: + _emit_published(broker, queue, count=0, size_bytes=size_bytes, start_perf=start_perf, exc=exc) + raise + _emit_published(broker, queue, count=0 if row_id is None else 1, size_bytes=size_bytes, start_perf=start_perf) if run_loops: _notify_subscriber(broker, queue) # P30: the loops own dispatch; just wake them elif row_id is not None: @@ -405,15 +446,24 @@ async def _fake_publish_many( (atomic batch INSERT -> published -> subscriber fetch), so a handler never observes a half-inserted batch and the event order isn't inverted. """ + start_perf = time.perf_counter() total_size = 0 + encoded: list[tuple[bytes, dict[str, str] | None]] = [] + try: + for body in bodies: + payload, hdrs = _encode_payload(body, headers=headers, serializer=serializer) + total_size += len(payload) + _validate_like_postgres(queue=queue, timer_id=None, headers=hdrs) + encoded.append((payload, hdrs)) + except Exception as exc: + _emit_published(broker, queue, count=0, size_bytes=total_size, start_perf=start_perf, exc=exc) + raise landed_ids: list[int] = [] - for body in bodies: - payload, hdrs = _encode_payload(body, headers=headers, serializer=serializer) - total_size += len(payload) + for payload, hdrs in encoded: row_id = fake_client.feed(queue=queue, payload=payload, headers=hdrs, next_attempt_at=next_at) if row_id is not None: landed_ids.append(row_id) - _emit_published(broker, queue, count=len(landed_ids), size_bytes=total_size) + _emit_published(broker, queue, count=len(landed_ids), size_bytes=total_size, start_perf=start_perf) if run_loops: _notify_subscriber(broker, queue) else: @@ -583,7 +633,7 @@ class TestOutboxBroker(TestBroker[OutboxBroker, OutboxBroker], broker=OutboxBrok def __init__(self, broker: OutboxBroker, *, run_loops: bool = False, **kwargs: typing.Any) -> None: super().__init__(broker, **kwargs) - self.fake_client = FakeOutboxClient() + self.fake_client = FakeOutboxClient(dlq_enabled=broker.config.broker_config.dlq_table is not None) self.fake_producer = None # set when the producer slot is swapped (see _patch_producer) self.run_loops = run_loops self._outbox_broker = broker # for feed()'s P30 wakeup diff --git a/tests/test_client_contract.py b/tests/test_client_contract.py index 1c57060..093c921 100644 --- a/tests/test_client_contract.py +++ b/tests/test_client_contract.py @@ -25,10 +25,11 @@ authority stays a documented invariant (AGENTS.md), not an assertion here. """ +import contextlib import datetime as dt import os import uuid -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator, AsyncIterator import pytest from sqlalchemy import MetaData, Table, insert, select, text @@ -51,8 +52,8 @@ class _FakeHarness: kind = "fake" - def __init__(self) -> None: - self._client = FakeOutboxClient() + def __init__(self, *, with_dlq: bool = True) -> None: + self._client = FakeOutboxClient(dlq_enabled=with_dlq) async def seed( self, @@ -139,11 +140,11 @@ class _RealHarness: kind = "real" - def __init__(self, engine: AsyncEngine, table: Table, dlq: Table) -> None: + def __init__(self, engine: AsyncEngine, table: Table, dlq: Table, *, with_dlq: bool = True) -> None: self._engine = engine self._table = table self._dlq = dlq - self._client = OutboxClient(engine, table, dlq_table=dlq) + self._client = OutboxClient(engine, table, dlq_table=dlq if with_dlq else None) async def seed( self, @@ -237,11 +238,10 @@ async def dlq_rows(self) -> list[dict[str, object]]: _Harness = _FakeHarness | _RealHarness -@pytest.fixture(params=["fake", "real"]) -async def contract(request: pytest.FixtureRequest) -> AsyncIterator[_Harness]: - """Yield a harness over each adapter. The ``real`` param auto-skips without Postgres.""" - if request.param == "fake": - yield _FakeHarness() +@contextlib.asynccontextmanager +async def _harness(kind: str, *, with_dlq: bool) -> AsyncGenerator[_Harness]: + if kind == "fake": + yield _FakeHarness(with_dlq=with_dlq) return engine = create_async_engine(PG_DSN, future=True) try: @@ -257,13 +257,27 @@ async def contract(request: pytest.FixtureRequest) -> AsyncIterator[_Harness]: async with engine.begin() as conn: await conn.run_sync(metadata.create_all) try: - yield _RealHarness(engine, table, dlq) + yield _RealHarness(engine, table, dlq, with_dlq=with_dlq) finally: async with engine.begin() as conn: await conn.run_sync(metadata.drop_all) await engine.dispose() +@pytest.fixture(params=["fake", "real"]) +async def contract(request: pytest.FixtureRequest) -> AsyncIterator[_Harness]: + """Yield a harness over each adapter. The ``real`` param auto-skips without Postgres.""" + async with _harness(request.param, with_dlq=True) as harness: + yield harness + + +@pytest.fixture(params=["fake", "real"]) +async def contract_without_dlq(request: pytest.FixtureRequest) -> AsyncIterator[_Harness]: + """Like ``contract``, but the client has no DLQ table.""" + async with _harness(request.param, with_dlq=False) as harness: + yield harness + + # --- fetch ----------------------------------------------------------------- @@ -304,10 +318,20 @@ async def test_fetch_skips_fresh_lease(contract: _Harness) -> None: assert msgs == [] +async def test_fetch_returns_rows_oldest_first(contract: _Harness) -> None: + rids = [await contract.seed(queue="orders", next_attempt_offset=-float(i)) for i in range(1, 21)] + rids += [await contract.seed(queue="audit", next_attempt_offset=-float(i) - 0.5) for i in range(1, 21)] + rids.append( + await contract.seed( + queue="orders", next_attempt_offset=-0.25, leased_token=uuid.uuid4(), acquired_age=120.0, deliveries_count=1 + ) + ) + msgs = await contract.fetch(["orders", "audit"], limit=41, lease_ttl_seconds=60.0) + assert sorted(m.id for m in msgs) == sorted(rids) + assert msgs == sorted(msgs, key=lambda m: (m.next_attempt_at, m.id)) + + async def test_fetch_selects_oldest_under_limit(contract: _Harness) -> None: - # FIFO *selection* under contention is the contract; within-batch *return order* - # is unspecified for the real client (F2-09 — the outer UPDATE…RETURNING is - # unordered), so assert on the chosen set, not its order. rid_a = await contract.seed(queue="orders", next_attempt_offset=-10.0) rid_b = await contract.seed(queue="orders", next_attempt_offset=-20.0) rid_c = await contract.seed(queue="orders", next_attempt_offset=-30.0) @@ -386,6 +410,15 @@ async def test_delete_noop_on_unleased_row(contract: _Harness) -> None: assert await contract.get_row(rid) is not None +async def test_delete_with_dlq_payload_but_no_dlq_table_refuses(contract_without_dlq: _Harness) -> None: + token = uuid.uuid4() + rid = await contract_without_dlq.seed(queue="orders", leased_token=token, acquired_age=1.0, deliveries_count=1) + with pytest.raises(RuntimeError, match="no dlq_table"): + await contract_without_dlq.delete(rid, token, dlq_payload={"failure_reason": "boom", "last_exception": None}) + assert await contract_without_dlq.get_row(rid) is not None + assert await contract_without_dlq.dlq_rows() == [] + + async def test_delete_with_dlq_materializes_audit_row(contract: _Harness) -> None: """INVARIANT: the DLQ archive is written atomically with the delete, in both adapters. diff --git a/tests/test_fake.py b/tests/test_fake.py index 800aeaf..e9602fa 100644 --- a/tests/test_fake.py +++ b/tests/test_fake.py @@ -1282,6 +1282,86 @@ def test_fake_client_feed_rejects_naive_next_attempt_at() -> None: fake.feed(queue="q", payload=b"x", next_attempt_at=dt.datetime.now()) # noqa: DTZ005 # naive on purpose +@pytest.mark.parametrize( + ("kwargs", "error"), + [ + ({"queue": "q" * 256}, "queue"), + ({"queue": "q", "timer_id": "t" * 256}, "timer_id"), + ({"queue": "q", "headers": {"k": object()}}, "headers"), + ], +) +def test_fake_client_feed_rejects_rows_postgres_would_reject(kwargs: dict[str, typing.Any], error: str) -> None: + fake = FakeOutboxClient() + with pytest.raises(ValueError, match=error): + fake.feed(payload=b"x", **kwargs) + assert fake.rows == [] + + +@pytest.mark.parametrize("batch", [False, True]) +async def test_fake_publish_emits_published_error_on_encode_failure(batch: bool) -> None: + events: list[tuple[str, dict[str, typing.Any]]] = [] + + def recorder(event: str, tags: Mapping[str, typing.Any]) -> None: + events.append((event, dict(tags))) + + broker = OutboxBroker(outbox_table=make_outbox_table(MetaData()), metrics_recorder=recorder) + unencodable = object() + async with TestOutboxBroker(broker): + publish = ( + broker.publish_batch("ok", unencodable, queue="orders") # ty: ignore[missing-argument] + if batch + else broker.publish(unencodable, queue="orders") # ty: ignore[missing-argument] + ) + with pytest.raises(Exception): # noqa: B017, PT011 # whatever the serializer raises, the metric must record it + await publish + + published = [tags for event, tags in events if event == "published"] + assert len(published) == 1 + assert published[0]["status"] == "error" + assert published[0]["count"] == 0 + assert "exception_type" in published[0] + + +async def test_fake_publish_reports_a_measured_duration() -> None: + durations: list[float] = [] + + def recorder(event: str, tags: Mapping[str, typing.Any]) -> None: + if event == "published": + durations.append(tags["duration_seconds"]) + + broker = OutboxBroker(outbox_table=make_outbox_table(MetaData()), metrics_recorder=recorder) + async with TestOutboxBroker(broker): + await broker.publish("one", queue="orders") # ty: ignore[missing-argument] + await broker.publish_batch("a", "b", queue="orders") # ty: ignore[missing-argument] + + assert len(durations) == 2 + assert all(d > 0.0 for d in durations) + + +async def test_test_broker_fake_client_matches_the_broker_dlq_config() -> None: + without_dlq = OutboxBroker(outbox_table=make_outbox_table(MetaData())) + with_dlq = OutboxBroker( + outbox_table=make_outbox_table(MetaData()), + dlq_table=make_dlq_table(MetaData()), + ) + token = uuid.uuid4() + payload = {"failure_reason": "boom", "last_exception": None} + + no_dlq_client = TestOutboxBroker(without_dlq).fake_client + rid = no_dlq_client.feed(queue="q", payload=b"x") + assert rid is not None + no_dlq_client.rows[0].acquired_token = token + with pytest.raises(RuntimeError, match="no dlq_table"): + await no_dlq_client.delete_with_lease(None, rid, token, dlq_payload=payload) + + dlq_client = TestOutboxBroker(with_dlq).fake_client + rid = dlq_client.feed(queue="q", payload=b"x") + assert rid is not None + dlq_client.rows[0].acquired_token = token + assert await dlq_client.delete_with_lease(None, rid, token, dlq_payload=payload) + assert len(dlq_client.dlq_rows) == 1 + + async def test_fake_headers_not_shared_by_reference() -> None: """P32: the fake must copy headers at its boundaries so handler/caller mutation can't corrupt rows.""" fake = FakeOutboxClient()