diff --git a/AGENTS.md b/AGENTS.md index f25e452..83d624a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,7 +31,7 @@ will **not** tell you: that — `tests/test_client_contract.py::test_delete_noop_on_token_mismatch` is the claim. - `client.py` and `testing.py` implement the same rules twice, in SQL and in Python, because one runs in the database and one in the process. They cannot share an implementation, so - `tests/test_client_contract.py` couples them by behaviour — a change to either adapter's fetch, + `tests/test_client_contract.py` couples them by behaviour — a change to either adapter's insert, fetch, terminal, or DLQ semantics belongs in that suite. - `schema.py` — the three partial indexes and the `_lease_ck` CHECK are load-bearing, not decoration. diff --git a/docs/usage/timers.md b/docs/usage/timers.md index e31cb8f..99a3a33 100644 --- a/docs/usage/timers.md +++ b/docs/usage/timers.md @@ -59,7 +59,7 @@ for user-supplied scheduling. ## Deduplication with `timer_id` `timer_id` flows into a `String(255)` column with a partial unique index -on `(queue, timer_id) WHERE timer_id IS NOT NULL`. The producer switches +on `(queue, timer_id) WHERE timer_id IS NOT NULL`. The insert switches to `pg_insert(...).on_conflict_do_nothing(...)` so re-publishing the same id is a silent no-op (returns `None`): diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index 29e56aa..f460c8f 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -187,6 +187,7 @@ def __init__( # noqa: PLR0913 # Serializer lives on fd_config — wire it onto the producer so encoded # bodies use the same path as the broker's own publish flow. producer.serializer = fd_config._serializer # noqa: SLF001 + producer.broker_config = broker_config specification = BrokerSpec( url=_spec_url(engine, outbox_table), protocol="postgresql", diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index 41497b3..636b828 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -17,8 +17,10 @@ import abc import asyncio import datetime as dt +import typing import uuid from typing import TYPE_CHECKING +from weakref import WeakKeyDictionary from sqlalchemy import ( ARRAY, @@ -30,15 +32,19 @@ delete, func, insert, + literal, select, text, true, union_all, update, ) +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from faststream_outbox import schema_validation +from faststream_outbox._scheduling import is_future_dated, resolve_next_attempt_client_side +from faststream_outbox._time import utcnow from faststream_outbox.message import OutboxInnerMessage from faststream_outbox.schema import ( _DLQ_INJECTED_COLUMNS, @@ -48,7 +54,6 @@ if TYPE_CHECKING: - import typing from collections.abc import Mapping, Sequence from sqlalchemy import CTE, ColumnElement, Insert, Table, Update @@ -82,6 +87,41 @@ def engine(self) -> "AsyncEngine | None": ... @abc.abstractmethod def table(self) -> "Table": ... + @abc.abstractmethod + async def insert( + self, + session: "AsyncSession", + *, + queue: str, + payload: bytes, + headers: dict[str, str] | None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, + timer_id: str | None = None, + ) -> int | None: + """Insert one encoded row on *session* and NOTIFY its queue if the row is eligible now. + + Returns the row id, or ``None`` when a live row already holds *timer_id* on *queue*. + Application code should call ``broker.publish`` instead: this skips the publish + middlewares and the ``published`` metric. + """ + + @abc.abstractmethod + async def insert_batch( + self, + session: "AsyncSession", + *, + queue: str, + rows: "Sequence[tuple[bytes, dict[str, str] | None]]", + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, + ) -> None: + """Insert every ``(payload, headers)`` row on *session* in one statement, or none of them. + + Application code should call ``broker.publish_batch`` instead, for the same reason as + :meth:`insert`. + """ + @abc.abstractmethod async def fetch( self, @@ -205,6 +245,137 @@ def _build_dlq_cte(outbox: "Table", dlq: "Table") -> "Insert": return insert(dlq).from_select([*(col for _, col in _DLQ_PROJECTION), *_DLQ_INJECTED_COLUMNS], archived) +def _build_insert_stmts(table: "Table", channel: str) -> "dict[tuple[bool, bool, bool], Insert]": + """Build the single-row INSERT once per ``(timer_id set, activate_in set, notify)``; values arrive as parameters. + + The notify variants call ``pg_notify`` in ``RETURNING``, so the NOTIFY costs no extra round trip and + fires only when a row actually landed (an ``ON CONFLICT DO NOTHING`` skip returns no row). + """ + # Server-side compute keeps timing immune to worker/DB clock skew (mirrors + # client.mark_pending_with_lease). + activate_in = func.now() + func.make_interval(0, 0, 0, 0, 0, 0, bindparam("activate_in_seconds", type_=Float)) + plain = insert(table) + timer = pg_insert(table).on_conflict_do_nothing( + index_elements=[table.c.queue, table.c.timer_id], + index_where=table.c.timer_id.is_not(None), + ) + notify = func.pg_notify(literal(channel), table.c.queue) + stmts: dict[tuple[bool, bool, bool], Insert] = {} + for has_timer, base in ((False, plain), (True, timer)): + for has_activate_in, stmt in ((False, base), (True, base.values(next_attempt_at=activate_in))): + stmts[has_timer, has_activate_in, False] = stmt.returning(table.c.id) + stmts[has_timer, has_activate_in, True] = stmt.returning(table.c.id, notify) + return stmts + + +class _OutboxInserter: + """The outbox INSERT + NOTIFY on a caller's session; needs no engine, so an engine-less broker can publish.""" + + def __init__(self, table: "Table") -> None: + self._table = table + self._channel = f"outbox_{table.name}" + # Per-transaction NOTIFY dedup memo (see _notify). Keyed on the innermost active + # transaction object, weak-referenced so entries GC when the transaction ends -- no + # explicit reset, and a rolled-back savepoint cannot suppress a later real NOTIFY. + self._notified: WeakKeyDictionary[typing.Any, set[str]] = WeakKeyDictionary() + self._insert_stmts = _build_insert_stmts(table, self._channel) + + async def insert( + self, + session: "AsyncSession", + *, + queue: str, + payload: bytes, + headers: dict[str, str] | None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, + timer_id: str | None = None, + ) -> int | None: + params: dict[str, typing.Any] = {"queue": queue, "payload": payload, "headers": headers} + if activate_in is not None: + params["activate_in_seconds"] = activate_in.total_seconds() + elif activate_at is not None: + params["next_attempt_at"] = activate_at + if timer_id is not None: + params["timer_id"] = timer_id + # Skip NOTIFY only when the row is genuinely future-dated. A past activate_at + # (e.g. a recovered idempotency token) is immediately eligible — fire NOTIFY. + # Future-dated rows skip NOTIFY: listeners can't act before the gate opens, and + # polling fires them at the next tick. + emitted = self._notified_queues(session) + notify = not is_future_dated(activate_in, activate_at, utcnow()) and (emitted is None or queue not in emitted) + stmt = self._insert_stmts[timer_id is not None, activate_in is not None, notify] + result = await session.execute(stmt, params) + row_id: int | None = result.scalar() + if row_id is not None and notify: + # Re-read: under autobegin the transaction only exists once the INSERT has run. + emitted = self._notified_queues(session) + if emitted is not None: + emitted.add(queue) + return row_id + + async def insert_batch( + self, + session: "AsyncSession", + *, + queue: str, + rows: "Sequence[tuple[bytes, dict[str, str] | None]]", + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, + ) -> None: + # Client-side time for batch: executemany doesn't compose cleanly with + # column-level SQL expressions, and a few-ms drift versus the DB is + # harmless for user-supplied scheduling. Retries still use server time. + now = utcnow() + next_at = resolve_next_attempt_client_side(activate_in, activate_at, now) + values: list[dict[str, typing.Any]] = [] + for payload, headers in rows: + row: dict[str, typing.Any] = {"queue": queue, "payload": payload, "headers": headers} + if next_at is not None: + row["next_attempt_at"] = next_at + values.append(row) + await session.execute(insert(self._table), values) + # Skip NOTIFY only when genuinely future-dated; past times are eligible. + if not is_future_dated(activate_in, activate_at, now): + await self._notify(session, queue) + + def _notified_queues(self, session: "AsyncSession") -> set[str] | None: + """Return the queues already notified in *session*'s innermost transaction, or None outside one.""" + sync_session = session.sync_session + txn = sync_session.get_nested_transaction() or sync_session.get_transaction() + if txn is None: + return None + return self._notified.setdefault(txn, set()) + + async def _notify(self, session: "AsyncSession", queue: str) -> None: + # ``pg_notify(:channel, :payload)`` -- parameterized so channel and payload + # bind cleanly (raw NOTIFY accepts only literals -- injection-prone). Runs on + # the caller's session so NOTIFY commits with the row insert; rollback silently + # discards it. Non-Postgres dialects ignore it. + # + # Dedup: Postgres already coalesces identical (channel, payload) NOTIFYs per + # transaction at delivery, so a second pg_notify for the same queue in the same + # transaction only wastes a round-trip. Memoize emitted queues per innermost + # active transaction (get_nested_transaction() during a savepoint, else the outer + # transaction) -- each a distinct, weak-referenceable object that GCs when the + # transaction ends, so a rolled-back savepoint cannot suppress a later real NOTIFY. + # Key on the *sync* SessionTransaction, not the async AsyncSessionTransaction proxy: + # SQLAlchemy regenerates that proxy on every call and only weak-references it, so + # under autobegin (no explicit `session.begin()`) it would GC between publishes and + # defeat the memo. The Session holds the sync transaction strongly for its lifetime, + # so the key survives across publishes whether the caller uses `session.begin()` or + # autobegin. No active transaction -> emit unconditionally. + emitted = self._notified_queues(session) + if emitted is not None: + if queue in emitted: + return + emitted.add(queue) + await session.execute( + text("SELECT pg_notify(:channel, :payload)"), + {"channel": self._channel, "payload": queue}, + ) + + class OutboxClient(AbstractOutboxClient): def __init__( self, @@ -221,6 +392,7 @@ def __init__( self._table = outbox_table self._dlq_table = dlq_table self._fetch_stmt = _build_fetch_stmt(outbox_table) + self._inserter = _OutboxInserter(outbox_table) self._dlq_cte = None if dlq_table is None else _build_dlq_cte(outbox_table, dlq_table) @property @@ -236,6 +408,40 @@ def engine(self) -> "AsyncEngine": """ return self._engine + async def insert( + self, + session: "AsyncSession", + *, + queue: str, + payload: bytes, + headers: dict[str, str] | None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, + timer_id: str | None = None, + ) -> int | None: + return await self._inserter.insert( + session, + queue=queue, + payload=payload, + headers=headers, + activate_in=activate_in, + activate_at=activate_at, + timer_id=timer_id, + ) + + async def insert_batch( + self, + session: "AsyncSession", + *, + queue: str, + rows: "Sequence[tuple[bytes, dict[str, str] | None]]", + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, + ) -> None: + await self._inserter.insert_batch( + session, queue=queue, rows=rows, activate_in=activate_in, activate_at=activate_at + ) + async def fetch( self, conn: "AsyncConnection | None", diff --git a/faststream_outbox/publisher/producer.py b/faststream_outbox/publisher/producer.py index 212bb0d..382f02c 100644 --- a/faststream_outbox/publisher/producer.py +++ b/faststream_outbox/publisher/producer.py @@ -1,23 +1,21 @@ -"""``OutboxProducer`` — the canonical insert path for outbox rows. +"""``OutboxProducer``: the canonical publish path for outbox rows. Both ``OutboxBroker.publish`` / ``publish_batch`` and ``OutboxPublisher.publish`` route through this producer via FastStream's ``_basic_publish(cmd, producer=...)`` -middleware stack, so encode → insert → NOTIFY semantics live in one place. The -producer never opens its own session — every command carries the caller's -``AsyncSession`` so the row commits atomically with the caller's domain writes. +middleware stack, so encode → insert → ``published`` semantics live in one place. The +insert itself (and its NOTIFY) belongs to the broker's client. The producer never opens +its own session: every command carries the caller's ``AsyncSession`` so the row commits +atomically with the caller's domain writes. """ import time import typing -from weakref import WeakKeyDictionary from faststream._internal.endpoint.utils import ParserComposition from faststream._internal.parser import DefaultCodec -from sqlalchemy import Float, Table, bindparam, func, insert, literal, text -from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy import Table -from faststream_outbox._scheduling import is_future_dated, resolve_next_attempt_client_side -from faststream_outbox._time import utcnow +from faststream_outbox.client import _OutboxInserter from faststream_outbox.envelope import _encode_payload from faststream_outbox.metrics import MetricsRecorder, _noop_recorder, _safe_emit from faststream_outbox.parser.parser import OutboxParser @@ -30,34 +28,13 @@ from fast_depends.library.serializer import SerializerProto from faststream._internal.parser import CodecProto from faststream._internal.types import AsyncCallable, CustomCallable - from sqlalchemy import Insert - - -def _build_insert_stmts(table: Table, channel: str) -> "dict[tuple[bool, bool, bool], Insert]": - """Build the single-row INSERT once per ``(timer_id set, activate_in set, notify)``; values arrive as parameters. - - The notify variants call ``pg_notify`` in ``RETURNING``, so the NOTIFY costs no extra round trip and - fires only when a row actually landed (an ``ON CONFLICT DO NOTHING`` skip returns no row). - """ - # Server-side compute keeps timing immune to worker/DB clock skew (mirrors - # client.mark_pending_with_lease). - activate_in = func.now() + func.make_interval(0, 0, 0, 0, 0, 0, bindparam("activate_in_seconds", type_=Float)) - plain = insert(table) - timer = pg_insert(table).on_conflict_do_nothing( - index_elements=[table.c.queue, table.c.timer_id], - index_where=table.c.timer_id.is_not(None), - ) - notify = func.pg_notify(literal(channel), table.c.queue) - stmts: dict[tuple[bool, bool, bool], Insert] = {} - for has_timer, base in ((False, plain), (True, timer)): - for has_activate_in, stmt in ((False, base), (True, base.values(next_attempt_at=activate_in))): - stmts[has_timer, has_activate_in, False] = stmt.returning(table.c.id) - stmts[has_timer, has_activate_in, True] = stmt.returning(table.c.id, notify) - return stmts + + from faststream_outbox.client import AbstractOutboxClient + from faststream_outbox.configs import OutboxBrokerConfig class OutboxProducer: - """``ProducerProto[OutboxPublishCommand]`` — runs encode + insert + NOTIFY on caller's session.""" + """``ProducerProto[OutboxPublishCommand]``: encodes, inserts through the broker's client, emits ``published``.""" _parser: "AsyncCallable" _decoder: "AsyncCallable" @@ -70,13 +47,8 @@ def __init__( decoder: typing.Optional["CustomCallable"], metrics_recorder: MetricsRecorder = _noop_recorder, ) -> None: - self._table = table - self._channel = f"outbox_{table.name}" - # Per-transaction NOTIFY dedup memo (see _notify). Keyed on the innermost active - # transaction object, weak-referenced so entries GC when the transaction ends -- no - # explicit reset, and a rolled-back savepoint cannot suppress a later real NOTIFY. - self._notified: WeakKeyDictionary[typing.Any, set[str]] = WeakKeyDictionary() self.serializer: SerializerProto | None = None + self.broker_config: OutboxBrokerConfig | None = None # ProducerProto[0.7] requires a `codec` attribute. The outbox owns its # own encoding pipeline (_encode_payload) and never reads this attribute # at runtime — it exists solely to satisfy the protocol. @@ -85,7 +57,11 @@ def __init__( self._parser = ParserComposition(parser, default.parse_message) self._decoder = ParserComposition(decoder, default.decode_message) self._metrics_recorder = metrics_recorder - self._insert_stmts = _build_insert_stmts(table, self._channel) + self._engineless_inserter = _OutboxInserter(table) + + def _client(self) -> "AbstractOutboxClient | _OutboxInserter": + client = None if self.broker_config is None else self.broker_config.client + return self._engineless_inserter if client is None else client def _emit_metric(self, event: str, tags: "Mapping[str, typing.Any]") -> None: # Delegate to the shared swallow-and-DEBUG-log helper so the recorder-isolation @@ -106,7 +82,15 @@ async def publish(self, cmd: OutboxPublishCommand) -> int | None: serializer=self.serializer, ) size_bytes = len(payload) - row_id = await self._do_publish(cmd, payload, hdrs) + row_id = await self._client().insert( + cmd.session, + queue=cmd.queue, + payload=payload, + headers=hdrs, + activate_in=cmd.activate_in, + activate_at=cmd.activate_at, + timer_id=cmd.timer_id, + ) except Exception as exc: self._emit_metric( "published", @@ -133,61 +117,24 @@ async def publish(self, cmd: OutboxPublishCommand) -> int | None: ) return row_id - async def _do_publish( - self, - cmd: OutboxPublishCommand, - payload: bytes, - hdrs: dict[str, str] | None, - ) -> int | None: - params: dict[str, typing.Any] = {"queue": cmd.queue, "payload": payload, "headers": hdrs} - if cmd.activate_in is not None: - params["activate_in_seconds"] = cmd.activate_in.total_seconds() - elif cmd.activate_at is not None: - params["next_attempt_at"] = cmd.activate_at - if cmd.timer_id is not None: - params["timer_id"] = cmd.timer_id - # Skip NOTIFY only when the row is genuinely future-dated. A past activate_at - # (e.g. a recovered idempotency token) is immediately eligible — fire NOTIFY. - # Future-dated rows skip NOTIFY: listeners can't act before the gate opens, and - # polling fires them at the next tick. - emitted = self._notified_queues(cmd.session) - notify = not is_future_dated(cmd.activate_in, cmd.activate_at, utcnow()) and ( - emitted is None or cmd.queue not in emitted - ) - stmt = self._insert_stmts[cmd.timer_id is not None, cmd.activate_in is not None, notify] - result = await cmd.session.execute(stmt, params) - row_id: int | None = result.scalar() - if row_id is not None and notify: - # Re-read: under autobegin the transaction only exists once the INSERT has run. - emitted = self._notified_queues(cmd.session) - if emitted is not None: - emitted.add(cmd.queue) - return row_id - async def publish_batch(self, cmd: OutboxPublishCommand) -> None: - bodies = cmd.batch_bodies - # Client-side time for batch: executemany doesn't compose cleanly with - # column-level SQL expressions, and a few-ms drift versus the DB is - # harmless for user-supplied scheduling. Retries still use server time. - now = utcnow() - next_at = resolve_next_attempt_client_side(cmd.activate_in, cmd.activate_at, now) - rows: list[dict[str, typing.Any]] = [] + rows: list[tuple[bytes, dict[str, str] | None]] = [] total_size = 0 start_perf = time.perf_counter() try: # P3: encode inside the try so a serialization failure on any body still # emits the ``published`` error metric. - for body in bodies: + for body in cmd.batch_bodies: payload, hdrs = _encode_payload(body, headers=cmd.headers, serializer=self.serializer) total_size += len(payload) - row: dict[str, typing.Any] = {"queue": cmd.queue, "payload": payload, "headers": hdrs} - if next_at is not None: - row["next_attempt_at"] = next_at - rows.append(row) - await cmd.session.execute(insert(self._table), rows) - # Skip NOTIFY only when genuinely future-dated; past times are eligible. - if not is_future_dated(cmd.activate_in, cmd.activate_at, now): - await self._notify(cmd.session, cmd.queue) + rows.append((payload, hdrs)) + await self._client().insert_batch( + cmd.session, + queue=cmd.queue, + rows=rows, + activate_in=cmd.activate_in, + activate_at=cmd.activate_at, + ) except Exception as exc: self._emit_metric( "published", @@ -214,39 +161,3 @@ async def publish_batch(self, cmd: OutboxPublishCommand) -> None: async def request(self, cmd: OutboxPublishCommand) -> typing.NoReturn: raise NotImplementedError(_REQUEST_UNSUPPORTED_MSG) - - def _notified_queues(self, session: typing.Any) -> set[str] | None: - """Return the queues already notified in *session*'s innermost transaction, or None outside one.""" - sync_session = session.sync_session - txn = sync_session.get_nested_transaction() or sync_session.get_transaction() - if txn is None: - return None - return self._notified.setdefault(txn, set()) - - async def _notify(self, session: typing.Any, queue: str) -> None: - # ``pg_notify(:channel, :payload)`` -- parameterized so channel and payload - # bind cleanly (raw NOTIFY accepts only literals -- injection-prone). Runs on - # the caller's session so NOTIFY commits with the row insert; rollback silently - # discards it. Non-Postgres dialects ignore it. - # - # Dedup: Postgres already coalesces identical (channel, payload) NOTIFYs per - # transaction at delivery, so a second pg_notify for the same queue in the same - # transaction only wastes a round-trip. Memoize emitted queues per innermost - # active transaction (get_nested_transaction() during a savepoint, else the outer - # transaction) -- each a distinct, weak-referenceable object that GCs when the - # transaction ends, so a rolled-back savepoint cannot suppress a later real NOTIFY. - # Key on the *sync* SessionTransaction, not the async AsyncSessionTransaction proxy: - # SQLAlchemy regenerates that proxy on every call and only weak-references it, so - # under autobegin (no explicit `session.begin()`) it would GC between publishes and - # defeat the memo. The Session holds the sync transaction strongly for its lifetime, - # so the key survives across publishes whether the caller uses `session.begin()` or - # autobegin. No active transaction -> emit unconditionally. - emitted = self._notified_queues(session) - if emitted is not None: - if queue in emitted: - return - emitted.add(queue) - await session.execute( - text("SELECT pg_notify(:channel, :payload)"), - {"channel": self._channel, "payload": queue}, - ) diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index 4bba0ac..df4d6e8 100644 --- a/faststream_outbox/testing.py +++ b/faststream_outbox/testing.py @@ -131,6 +131,58 @@ def feed( msg = "feed() requires next_attempt_at to be timezone-aware" raise ValueError(msg) _validate_like_postgres(queue=queue, timer_id=timer_id, headers=headers) + row = self._store( + queue=queue, payload=payload, headers=headers, next_attempt_at=next_attempt_at, timer_id=timer_id + ) + return None if row is None else row.id + + async def insert( + self, + session: "AsyncSession", + *, + queue: str, + payload: bytes, + headers: dict[str, str] | None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, + timer_id: str | None = None, + ) -> int | None: + del session # ignored (no real DB) + _validate_like_postgres(queue=queue, timer_id=timer_id, headers=headers) + row = self._store( + queue=queue, + payload=payload, + headers=headers, + next_attempt_at=resolve_next_attempt_client_side(activate_in, activate_at, utcnow()), + timer_id=timer_id, + ) + return None if row is None else row.id + + async def insert_batch( + self, + session: "AsyncSession", + *, + queue: str, + rows: "Sequence[tuple[bytes, dict[str, str] | None]]", + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, + ) -> None: + del session # ignored (no real DB) + for _, headers in rows: + _validate_like_postgres(queue=queue, timer_id=None, headers=headers) + next_at = resolve_next_attempt_client_side(activate_in, activate_at, utcnow()) + for payload, headers in rows: + self._store(queue=queue, payload=payload, headers=headers, next_attempt_at=next_at, timer_id=None) + + def _store( + self, + *, + queue: str, + payload: bytes, + headers: dict[str, str] | None, + next_attempt_at: dt.datetime | None, + timer_id: str | None, + ) -> _FakeRow | None: # 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): @@ -148,7 +200,7 @@ def feed( ) self._rows.append(row) self._next_id += 1 - return row.id + return row @property def rows(self) -> list[_FakeRow]: diff --git a/tests/test_client_contract.py b/tests/test_client_contract.py index 093c921..c7b8c62 100644 --- a/tests/test_client_contract.py +++ b/tests/test_client_contract.py @@ -7,10 +7,10 @@ them to one behavioural contract instead. Every scenario runs against **both** the fake (everywhere) and real Postgres (auto-skipped when unreachable). -Scope is exactly the shared ``AbstractOutboxClient`` surface: ``fetch``, +Scope is exactly the shared ``AbstractOutboxClient`` surface: ``insert`` / +``insert_batch`` (+ ``timer_id`` dedup and headers JSONB rejects), ``fetch``, ``delete_with_lease``, ``mark_pending_with_lease`` (+ the DLQ side-effect). -``cancel_timer`` and ``timer_id`` insert-dedup live on the broker / producer, not on -the client interface, so they are covered by ``test_integration.py`` / +``cancel_timer`` lives on the broker, so it is covered by ``test_integration.py`` / ``test_fake.py``, not here. A per-adapter harness hides the substrate differences (how a row is seeded, which @@ -29,11 +29,13 @@ import datetime as dt import os import uuid -from collections.abc import AsyncGenerator, AsyncIterator +from collections.abc import AsyncGenerator, AsyncIterator, Sequence +from unittest.mock import AsyncMock import pytest from sqlalchemy import MetaData, Table, insert, select, text -from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine +from sqlalchemy.exc import StatementError +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine from faststream_outbox import make_dlq_table, make_outbox_table from faststream_outbox._time import utcnow @@ -51,10 +53,43 @@ class _FakeHarness: """Adapter-agnostic harness backed by an in-memory ``FakeOutboxClient``.""" kind = "fake" + rejects: type[Exception] = ValueError def __init__(self, *, with_dlq: bool = True) -> None: self._client = FakeOutboxClient(dlq_enabled=with_dlq) + async def insert( + self, + *, + queue: str = "q", + payload: bytes = b"x", + headers: dict[str, str] | None = None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, + timer_id: str | None = None, + ) -> int | None: + return await self._client.insert( + AsyncMock(spec=AsyncSession), + queue=queue, + payload=payload, + headers=headers, + activate_in=activate_in, + activate_at=activate_at, + timer_id=timer_id, + ) + + async def insert_batch( + self, + rows: Sequence[tuple[bytes, dict[str, str] | None]], + *, + queue: str = "q", + activate_in: dt.timedelta | None = None, + ) -> None: + await self._client.insert_batch(AsyncMock(spec=AsyncSession), queue=queue, rows=rows, activate_in=activate_in) + + async def queue_ids(self, queue: str) -> list[int]: + return sorted(r.id for r in self._client.rows if r.queue == queue) + async def seed( self, *, @@ -139,12 +174,50 @@ class _RealHarness: """Adapter-agnostic harness backed by a real ``OutboxClient`` against Postgres.""" kind = "real" + rejects: type[Exception] = StatementError 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 if with_dlq else None) + self._sessions = async_sessionmaker(engine, expire_on_commit=False) + + async def insert( + self, + *, + queue: str = "q", + payload: bytes = b"x", + headers: dict[str, str] | None = None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, + timer_id: str | None = None, + ) -> int | None: + async with self._sessions() as session, session.begin(): + return await self._client.insert( + session, + queue=queue, + payload=payload, + headers=headers, + activate_in=activate_in, + activate_at=activate_at, + timer_id=timer_id, + ) + + async def insert_batch( + self, + rows: Sequence[tuple[bytes, dict[str, str] | None]], + *, + queue: str = "q", + activate_in: dt.timedelta | None = None, + ) -> None: + async with self._sessions() as session, session.begin(): + await self._client.insert_batch(session, queue=queue, rows=rows, activate_in=activate_in) + + async def queue_ids(self, queue: str) -> list[int]: + async with self._engine.connect() as conn: + result = await conn.execute(select(self._table.c.id).where(self._table.c.queue == queue)) + return sorted(result.scalars().all()) async def seed( self, @@ -278,6 +351,94 @@ async def contract_without_dlq(request: pytest.FixtureRequest) -> AsyncIterator[ yield harness +# --- insert ---------------------------------------------------------------- + + +async def test_insert_stores_an_eligible_row(contract: _Harness) -> None: + rid = await contract.insert(queue="orders", payload=b"a", headers={"k": "v"}) + assert rid is not None + row = await contract.get_row(rid) + assert row is not None + assert (row["queue"], row["payload"], row["timer_id"]) == ("orders", b"a", None) + msgs = await contract.fetch(["orders"], limit=10, lease_ttl_seconds=60.0) + assert [(m.id, m.headers) for m in msgs] == [(rid, {"k": "v"})] + + +async def test_insert_returns_none_on_live_timer_conflict(contract: _Harness) -> None: + first = await contract.insert(queue="orders", timer_id="t1") + assert first is not None + assert await contract.insert(queue="orders", timer_id="t1") is None + assert await contract.queue_ids("orders") == [first] + + +async def test_insert_same_timer_id_on_another_queue_lands(contract: _Harness) -> None: + assert await contract.insert(queue="orders", timer_id="t1") is not None + assert await contract.insert(queue="audit", timer_id="t1") is not None + + +async def test_insert_reuses_timer_id_after_delivery(contract: _Harness) -> None: + first = await contract.insert(queue="orders", timer_id="t1") + assert first is not None + [msg] = await contract.fetch(["orders"], limit=10, lease_ttl_seconds=60.0) + assert msg.acquired_token is not None + assert await contract.delete(msg.id, msg.acquired_token) + second = await contract.insert(queue="orders", timer_id="t1") + assert second is not None + assert second != first + + +async def test_insert_activate_in_gates_the_row(contract: _Harness) -> None: + before = utcnow() + rid = await contract.insert(queue="orders", activate_in=dt.timedelta(hours=1)) + assert rid is not None + assert await contract.fetch(["orders"], limit=10, lease_ttl_seconds=60.0) == [] + row = await contract.get_row(rid) + assert row is not None + next_at = row["next_attempt_at"] + assert isinstance(next_at, dt.datetime) + assert before + dt.timedelta(minutes=59) < next_at < before + dt.timedelta(minutes=61) + + +async def test_insert_activate_at_stores_that_time(contract: _Harness) -> None: + at = utcnow() + dt.timedelta(hours=2) + rid = await contract.insert(queue="orders", activate_at=at) + assert rid is not None + row = await contract.get_row(rid) + assert row is not None + assert row["next_attempt_at"] == at + + +async def test_insert_past_activate_at_is_eligible(contract: _Harness) -> None: + rid = await contract.insert(queue="orders", activate_at=utcnow() - dt.timedelta(minutes=5)) + msgs = await contract.fetch(["orders"], limit=10, lease_ttl_seconds=60.0) + assert [m.id for m in msgs] == [rid] + + +async def test_insert_rejects_headers_jsonb_cannot_hold(contract: _Harness) -> None: + with pytest.raises(contract.rejects): + await contract.insert(queue="orders", headers={"k": object()}) # ty: ignore[invalid-argument-type] + assert await contract.queue_ids("orders") == [] + + +async def test_insert_batch_stores_every_row_in_order(contract: _Harness) -> None: + await contract.insert_batch([(b"a", None), (b"b", {"k": "v"})], queue="orders") + msgs = await contract.fetch(["orders"], limit=10, lease_ttl_seconds=60.0) + assert [(m.payload, m.headers) for m in sorted(msgs, key=lambda m: m.id)] == [(b"a", None), (b"b", {"k": "v"})] + + +async def test_insert_batch_activate_in_gates_every_row(contract: _Harness) -> None: + await contract.insert_batch([(b"a", None), (b"b", None)], queue="orders", activate_in=dt.timedelta(hours=1)) + assert len(await contract.queue_ids("orders")) == 2 + assert await contract.fetch(["orders"], limit=10, lease_ttl_seconds=60.0) == [] + + +async def test_insert_batch_with_one_rejected_row_stores_nothing(contract: _Harness) -> None: + rows = [(b"a", None), (b"b", {"k": object()})] + with pytest.raises(contract.rejects): + await contract.insert_batch(rows, queue="orders") # ty: ignore[invalid-argument-type] + assert await contract.queue_ids("orders") == [] + + # --- fetch ----------------------------------------------------------------- diff --git a/tests/test_integration.py b/tests/test_integration.py index 6e425ab..24d96aa 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -2466,6 +2466,22 @@ def _listener(conn, cursor, statement, parameters, context, executemany) -> None event.remove(engine.sync_engine, "before_cursor_execute", _listener) +async def test_broker_without_engine_publishes_on_the_callers_session( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(outbox_table=outbox_table) + session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) + with _count_pg_notify(pg_engine) as count: + async with session_factory() as session, session.begin(): + rid = await broker.publish({"n": 1}, queue="orders", session=session) + await broker.publish_batch({"n": 2}, {"n": 3}, queue="orders", session=session) + assert rid is not None + assert count[0] == 1 + async with pg_engine.connect() as conn: + result = await conn.execute(select(outbox_table.c.id).where(outbox_table.c.queue == "orders")) + assert len(result.scalars().all()) == 3 + + async def test_notify_deduped_within_transaction_same_queue(pg_engine, outbox_table) -> None: broker = OutboxBroker(pg_engine, outbox_table=outbox_table) session_factory = async_sessionmaker(pg_engine, expire_on_commit=False)