fix: align the fake client with the real one and return fetched rows oldest first

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.

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,

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

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.