From c841fd9829b35b9b2d1fb1744d10666bac3c7ed9 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sun, 5 Jul 2026 10:28:35 +0300 Subject: [PATCH] docs: fix drift, broken samples, and inconsistencies across the docs set Audit of all 23 doc pages cross-referenced against src/, architecture/, and pyproject. Fixes grouped by severity: High (correctness): - how-it-works: a foreign-publish outage nacks via retry_strategy, not lease expiry (lease-expiry recovery is only for a mis-composed publisher chain). - testing: publisher example now passes the required session= (the publisher path is not session-patched like broker.publish); rows assertion uses an unconsumed queue since a consumed row is acked/deleted in sync mode. - alembic: add the load-bearing outbox_lease_ck CHECK to the initial migration (placement matches Alembic's rendered autogenerate output). Medium (accuracy): - add-kafka-relay: default outage budget is ~8-9 min, not ~13-14 (10 attempts, 1+2+..+256s ~= 8.5 min). - drain_timeout sweep: instrumentation-seams six->seven events (+list+table), setup-prometheus-opentelemetry three->four OTel counters, observability counting reframed onto a consistent event basis. Judgment calls (maintainer ruling): - Python floor 3.13+ -> 3.11+ (installation, first-outbox-app, CLAUDE.md) to match requires-python and CI. - OutboxResponse reframed around session lifetime, not FastAPI-specific. Low: snippet-completeness fixes (undefined vars in schema-validation, messaging-service, fastapi, relay, testing fragments), relay missing await, subscriber validation-file attribution, timers past-dated example + activate_at skew note, checklist 63-byte-guard wording, messaging-service timer_id wording, publisher broker.request note, index table reordered to lead with the newcomer path, and the "Both seams" OTel example switched to spans-only to drop a dead unscraped meter path. Also corrects architecture/subscriber.md: an undersized pool does not block start() (the user doc was already correct). Co-Authored-By: Claude Opus 4.8 (1M context) --- CLAUDE.md | 2 +- architecture/subscriber.md | 2 +- docs/concepts/instrumentation-seams.md | 11 +++-- docs/index.md | 5 ++- docs/introduction/how-it-works.md | 11 +++-- docs/introduction/installation.md | 2 +- docs/operations/alembic.md | 14 +++++- docs/operations/checklist.md | 13 +++--- docs/tutorials/add-kafka-relay.md | 4 +- docs/tutorials/first-outbox-app.md | 2 +- docs/usage/fastapi.md | 2 + docs/usage/messaging-service.md | 10 ++++- docs/usage/observability.md | 7 ++- docs/usage/publisher.md | 22 ++++++---- docs/usage/relay.md | 4 +- docs/usage/schema-validation.md | 6 +++ docs/usage/setup-prometheus-opentelemetry.md | 33 +++++++------- docs/usage/subscriber.md | 9 ++-- docs/usage/testing.md | 45 +++++++++++++++----- docs/usage/timers.md | 16 ++++--- 20 files changed, 154 insertions(+), 66 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 7a308f6..6ceab30 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -40,7 +40,7 @@ The package wires a FastStream `Broker`/`Registrator`/`Subscriber` trio whose tr ## Conventions -- Python 3.13+. +- Python 3.11+. - **Never use local/inline imports.** All imports at module top — no `import` inside functions/methods/`if TYPE_CHECKING` exception aside. Tests included. If `# noqa: PLC0415` is the only way to keep an import inline, hoist it instead. - `ruff` runs `select = ["ALL"]` with documented ignores in `pyproject.toml`; many `# noqa` are intentional. - Type checker is `ty`. Use `# ty: ignore[]` for intentional escapes. diff --git a/architecture/subscriber.md b/architecture/subscriber.md index 763bc9c..7d3e5d5 100644 --- a/architecture/subscriber.md +++ b/architecture/subscriber.md @@ -28,7 +28,7 @@ A subscriber can hold up to `fetch_batch_size + max_workers` leases at once, not Each subscriber holds `max_workers + 1` SQLAlchemy pool connections steady-state, plus one raw asyncpg connection for `LISTEN`. -- Size the pool for `Σ subscribers × (max_workers + 1)` or startup blocks on checkout. The asyncpg `LISTEN` connection lives outside the pool, so it does not count toward pool sizing. +- Size the pool for `Σ subscribers × (max_workers + 1)`. Undersizing does **not** block `broker.start()` — `start()` only schedules the loop tasks via `add_task` and returns; instead the fetch/worker loops stall on pool checkout at runtime. The asyncpg `LISTEN` connection lives outside the pool, so it does not count toward pool sizing. - Per process, Postgres `max_connections` must cover `replicas × Σ subscribers × (max_workers + 2)`: the `max_workers + 1` pool connections plus the out-of-pool asyncpg `LISTEN` connection. Undersize it and rolling deploys hit `FATAL: too many connections`. ## NOTIFY semantics diff --git a/docs/concepts/instrumentation-seams.md b/docs/concepts/instrumentation-seams.md index c3a5b3a..2ab1082 100644 --- a/docs/concepts/instrumentation-seams.md +++ b/docs/concepts/instrumentation-seams.md @@ -61,12 +61,14 @@ Three events fire **outside** the handler invocation, with no ## What the recorder seam observes naturally The recorder is a `Callable[[str, Mapping[str, Any]], None]` invoked at -six core subscriber events (`fetched`, `dispatched`, `acked`, -`nacked_retried`, `nacked_terminal`, `lease_lost`), a conditional -`dlq_written` when the DLQ is configured, and one producer event +seven core subscriber events (`fetched`, `dispatched`, `acked`, +`nacked_retried`, `nacked_terminal`, `lease_lost`, `drain_timeout`), a +conditional `dlq_written` when the DLQ is configured, and one producer event (`published`). It fires whether or not a handler is in scope: -- All three bus-invisible events above. +- All four bus-invisible events above: `fetched`, `lease_lost`, the + pre-consume `nacked_terminal(reason="max_deliveries")`, and `drain_timeout` + (a shutdown-drain overrun, which has no handler scope at all). - Plus `acked` / `nacked_retried` / `nacked_terminal` / `dispatched` / `published` from inside the handler-execution paths, with explicit `subscriber` and `queue` tags. @@ -91,6 +93,7 @@ physically cannot observe. | `fetched` ticks (including empty) | ❌ (no `StreamMessage` at fetch time) | ✅ | | `lease_lost` after `consume_scope` exits | ❌ | ✅ | | `nacked_terminal(reason="max_deliveries")` before consume opens | ❌ | ✅ | +| `drain_timeout` during `stop()` shutdown drain | ❌ (no `consume_scope`) | ✅ | ## Operator implication diff --git a/docs/index.md b/docs/index.md index 4ca7aaf..f8f9dbc 100644 --- a/docs/index.md +++ b/docs/index.md @@ -43,12 +43,15 @@ process, no Kafka. | If you want to… | Start at | |---|---| +| Install and write the first publisher / subscriber | [Installation](introduction/installation.md) → [Tutorial: Your first outbox app](tutorials/first-outbox-app.md) | | See it work end-to-end on a FastAPI app | [FastAPI integration](usage/fastapi.md) | | Relay outbox rows to Kafka / RabbitMQ / NATS / Redis | [Relay to Kafka / RabbitMQ / NATS](usage/relay.md) | | Understand the architecture before adopting | [How it works](introduction/how-it-works.md) | | Compare against CDC / Kafka transactions / a hand-rolled outbox | [Comparison](concepts/comparison.md) | | Deploy to production safely | [Production checklist](operations/checklist.md) | -| Install and write the first publisher / subscriber | [Installation](introduction/installation.md) → [Tutorial: Your first outbox app](tutorials/first-outbox-app.md) | + +The categorized index below (also the left-hand nav) lists every page with a +one-line summary. ## Documentation diff --git a/docs/introduction/how-it-works.md b/docs/introduction/how-it-works.md index 58e3a77..4b49d7c 100644 --- a/docs/introduction/how-it-works.md +++ b/docs/introduction/how-it-works.md @@ -189,9 +189,14 @@ stack a foreign-broker publisher decorator on the subscriber (`@kafka_pub @broker_outbox.subscriber("q")`) and the handler's return value is forwarded to the real bus. The outbox row stays the durability boundary — the row commits with the domain write, and the relay carries -at-least-once end to end. If the foreign publish fails, the row is **not** -nacked through the `retry_strategy`; its lease simply expires and a later -fetch retries the relay, so a transient bus outage never loses the row. +at-least-once end to end. Recovery comes from two tiers, so a bus outage +never loses the row: a **transient blip** is absorbed by the client library +(e.g. `aiokafka`) — the in-handler publish blocks until the broker returns, +which the subscriber sees as one slow *successful* publish, no nack; a +**sustained outage** eventually raises into the handler, which nacks the row +and hands it to the configured `retry_strategy` to reschedule. (The one path +that recovers via lease expiry rather than `retry_strategy` is a +mis-composed publisher chain — see [relay guardrails](../usage/relay.md#what-not-to-do).) > **Worked end-to-end example → [Relay tutorial](../tutorials/add-kafka-relay.md).** diff --git a/docs/introduction/installation.md b/docs/introduction/installation.md index be6e134..5409ad7 100644 --- a/docs/introduction/installation.md +++ b/docs/introduction/installation.md @@ -22,7 +22,7 @@ ## Requirements -- Python 3.13+ +- Python 3.11+ - PostgreSQL 12+ — the features used (partial indexes, `FOR UPDATE SKIP LOCKED`, `make_interval`, `pg_notify`) all predate 12. The examples and CI run on 17; that is what's exercised, so 17 is the safest choice if diff --git a/docs/operations/alembic.md b/docs/operations/alembic.md index 2bd95a3..70b9a1b 100644 --- a/docs/operations/alembic.md +++ b/docs/operations/alembic.md @@ -30,6 +30,7 @@ op.create_table('outbox', sa.Column('acquired_at', sa.DateTime(timezone=True), nullable=True), sa.Column('acquired_token', sa.Uuid(), nullable=True), sa.Column('timer_id', sa.String(length=255), nullable=True), + sa.CheckConstraint('(acquired_token IS NULL) = (acquired_at IS NULL)', name='outbox_lease_ck'), sa.PrimaryKeyConstraint('id') ) op.create_index('outbox_lease_idx', 'outbox', ['queue', 'acquired_at'], unique=False, @@ -63,9 +64,18 @@ Columns the operator can mostly ignore: `attempts_count` and `last_attempt_at` (debugging aids on retry-heavy rows). All four are maintained by the broker; no application code touches them. +The `outbox_lease_ck` CHECK (`(acquired_token IS NULL) = (acquired_at +IS NULL)`) is equally load-bearing: it enforces that a row's lease token +and lease timestamp are always set or cleared together, so a half-written +lease can never exist. Autogenerate renders it here **only because this is +a fresh `create_table`** — on an incremental migration onto a pre-existing +table Alembic has no check-constraint comparator and would ship a missing +or drifted CHECK silently (that gap is what +[`validate_schema()`](../usage/schema-validation.md) backstops). + The `# please adjust!` comment from Alembic is misleading here — -**don't adjust**. The column types, predicates, and indexes are -exactly what the broker depends on. The +**don't adjust**. The column types, the CHECK, the predicates, and the +indexes are exactly what the broker depends on. The [`validate_schema()`](../usage/schema-validation.md) check — when you wire it into a `/health` probe or CI gate — fails when the live DB drifts from this declaration. (It is opt-in; it never runs at `broker.start()`.) diff --git a/docs/operations/checklist.md b/docs/operations/checklist.md index 7ae610b..5554459 100644 --- a/docs/operations/checklist.md +++ b/docs/operations/checklist.md @@ -64,11 +64,14 @@ story. `[validate]` extra. Do **not** call at `broker.start()` — that would crash-loop on a pending migration. See [Schema validation § Where to call it](../usage/schema-validation.md#where-to-call-it). -- [ ] **Outbox `table_name` short enough for the NOTIFY channel** — the - channel name is `outbox_`, and `make_outbox_table` raises - `ValueError` at table-build time when that exceeds Postgres' 63-**byte** - identifier limit. There is no silent truncation or polling fallback — the - guard makes an over-long name impossible to ship. +- [ ] **Outbox `table_name` short enough for every derived identifier** — + `make_outbox_table` raises `ValueError` at table-build time when the + *longest* identifier it derives exceeds Postgres' 63-**byte** limit. That + longest identifier is usually an index/constraint name + (`_pending_idx`, `_timer_id_uq`), which is longer + than the NOTIFY channel `outbox_` — so a name that fits the + channel can still overflow an index name. There is no silent truncation or + polling fallback; the guard makes an over-long name impossible to ship. ## Observability diff --git a/docs/tutorials/add-kafka-relay.md b/docs/tutorials/add-kafka-relay.md index ff5300d..4ac0bf4 100644 --- a/docs/tutorials/add-kafka-relay.md +++ b/docs/tutorials/add-kafka-relay.md @@ -241,7 +241,9 @@ the durability boundary, and it stays in the table for the duration of the retry budget (the default `ExponentialRetry` allows 10 attempts). Once the budget is exhausted the row is deleted — the default configures no DLQ — so configure a longer `retry_strategy` or a `dlq_table` to survive outages -beyond that (with the default schedule, ~13–14 minutes). +beyond that (the default schedule spans roughly 8-9 minutes: nine backoffs +of 1, 2, 4, … 256 seconds sum to ~8.5 minutes before the 10th attempt is +terminal). In practice, `aiokafka`'s producer has its own client-side reconnect and retry logic, so a short Kafka outage usually completes from the diff --git a/docs/tutorials/first-outbox-app.md b/docs/tutorials/first-outbox-app.md index a723a19..40ee452 100644 --- a/docs/tutorials/first-outbox-app.md +++ b/docs/tutorials/first-outbox-app.md @@ -9,7 +9,7 @@ print it. ## Before you start -- Python 3.13+ +- Python 3.11+ - Docker (for a one-line Postgres container) - [uv](https://docs.astral.sh/uv/) for project setup - Roughly ten minutes diff --git a/docs/usage/fastapi.md b/docs/usage/fastapi.md index 7847e71..229e976 100644 --- a/docs/usage/fastapi.md +++ b/docs/usage/fastapi.md @@ -151,6 +151,8 @@ The [DLQ](./dlq.md) and the [metrics-recorder seam](./observability.md) `OutboxBroker(...)` — they forward to the inner broker. ```python +from faststream_outbox import make_dlq_table # alongside make_outbox_table + outbox_router = OutboxRouter( engine, outbox_table=outbox_table, diff --git a/docs/usage/messaging-service.md b/docs/usage/messaging-service.md index 286ca55..9c391ae 100644 --- a/docs/usage/messaging-service.md +++ b/docs/usage/messaging-service.md @@ -55,12 +55,18 @@ The broker and routers are wired with a real DI container ```python title="ioc.py" from modern_di import Group, Scope, providers +from sqlalchemy.ext.asyncio import create_async_engine from faststream_outbox import OutboxBroker from app.tables import OUTBOX_TABLE class Resources(Group): + # The caller owns the engine; the broker never disposes it. + database_engine = providers.Factory( + scope=Scope.APP, + creator=lambda: create_async_engine("postgresql+asyncpg://app:app@localhost/app"), + ) outbox_broker = providers.Factory( scope=Scope.APP, creator=lambda engine: OutboxBroker(engine, outbox_table=OUTBOX_TABLE), @@ -167,7 +173,9 @@ Register the router on the broker (the `OutboxBroker` built in `ioc.py`) with ## Pattern 2 — Fire-unless-cancelled timer The unread notification is a **delayed** outbox row, armed in the same create -transaction. `timer_id` makes it idempotent; `activate_in` defers it: +transaction. `timer_id` deduplicates while a row is live (at most one live +row per `(queue, timer_id)`, not a global idempotency key); `activate_in` +defers it: These two methods live on the same `OutboxEventProducer` from Pattern 1 (shown here as a continuation of the class): diff --git a/docs/usage/observability.md b/docs/usage/observability.md index ab2b90b..a858096 100644 --- a/docs/usage/observability.md +++ b/docs/usage/observability.md @@ -17,8 +17,11 @@ MetricsRecorder = Callable[[str, Mapping[str, Any]], None] The default (`_noop_recorder`) lets instrumentation sites call unconditionally. The recorder threads through `OutboxBrokerConfig` to: -- The subscriber's seven emission points via `OutboxSubscriber._emit_metric` -- The producer's single emission point via `OutboxProducer._emit_metric` +- The subscriber's seven core events via `OutboxSubscriber._emit_metric` + (`fetched`, `dispatched`, `acked`, `nacked_retried`, `nacked_terminal`, + `lease_lost`, `drain_timeout`), plus a conditional `dlq_written` when a DLQ + is configured +- The producer's single event (`published`) via `OutboxProducer._emit_metric` ### Bare seam diff --git a/docs/usage/publisher.md b/docs/usage/publisher.md index 5a01c93..97db26f 100644 --- a/docs/usage/publisher.md +++ b/docs/usage/publisher.md @@ -15,6 +15,9 @@ For "consume from A → enqueue to B" relay flows, a fourth path is available: returning `OutboxResponse(...)` from a handler. See [Chained publishing](#chained-publishing) below. +There is no request/reply: the outbox is fire-and-forget, so `broker.request(...)` +raises `NotImplementedError`. + ## `broker.publish` ```python @@ -144,15 +147,18 @@ flow routes the returned value through the producer; the same transactional contract applies (you provide the session, the row commits with your domain writes): -!!! note "This pattern is FastAPI-specific" +!!! note "The `session` must outlive the handler return" The returned `OutboxResponse` is published **after** the handler - returns, so its `session` must outlive the handler call. FastAPI's - `Depends(get_session)` provides exactly that — a session torn down by - the dependency after the response flow. Opening your own `async with - session_factory() as session:` inside the handler does **not** work - here: the session closes on `return`, before the row is inserted. - Outside FastAPI, call `broker.publish(..., session=session)` directly - inside your handler instead (see [§ `broker.publisher`](#not-a-relay-decorator)). + returns, so its `session` must still be open at that point. The + requirement is about session *lifetime*, not any particular framework: + provide the session through a dependency that the framework tears down + *after* the response flow — FastAPI's `Depends(get_session)` or + FastStream's own `Depends` / `Context` session both do this. Opening + your own `async with session_factory() as session:` inside the handler + does **not** work here: that session closes on `return`, before the row + is inserted — in that case call `broker.publish(..., session=session)` + directly inside the `async with` instead (see + [§ Not a relay decorator](#not-a-relay-decorator)). ```python from fastapi import Depends diff --git a/docs/usage/relay.md b/docs/usage/relay.md index 0fd187b..d1ef3f6 100644 --- a/docs/usage/relay.md +++ b/docs/usage/relay.md @@ -43,7 +43,7 @@ async def relay(body: dict) -> dict: return body ``` -That's the whole thing. `broker_outbox.publish(body, queue="outbox_queue", session=session)` +That's the whole thing. `await broker_outbox.publish(body, queue="outbox_queue", session=session)` in your domain transaction writes a row; the subscriber dispatches it; the handler returns it; the Kafka publisher decorator picks it up and publishes to `kafka_topic`. Failure handling, retries, and DLQ are unchanged from @@ -161,6 +161,8 @@ construction (before lifespan), so it's automatic. For the standalone (non-FastAPI) lifecycle, the order is: ```python +# kafka_router / outbox_router are constructed exactly as in the FastAPI +# example above (KafkaRouter(...) / OutboxRouter(...)). broker_kafka.include_router(kafka_router) broker_outbox.include_router(outbox_router) # then start diff --git a/docs/usage/schema-validation.md b/docs/usage/schema-validation.md index 610c00b..d746954 100644 --- a/docs/usage/schema-validation.md +++ b/docs/usage/schema-validation.md @@ -83,10 +83,16 @@ Or as a one-shot CI check after running migrations: ```python import asyncio + +from sqlalchemy import MetaData +from sqlalchemy.ext.asyncio import create_async_engine + from faststream_outbox import OutboxBroker, make_outbox_table async def main() -> None: + engine = create_async_engine("postgresql+asyncpg://outbox:outbox@localhost/outbox") + outbox_table = make_outbox_table(MetaData(), table_name="outbox") broker = OutboxBroker(engine, outbox_table=outbox_table) await broker.validate_schema() print("schema OK") diff --git a/docs/usage/setup-prometheus-opentelemetry.md b/docs/usage/setup-prometheus-opentelemetry.md index 0948e32..0ff08b8 100644 --- a/docs/usage/setup-prometheus-opentelemetry.md +++ b/docs/usage/setup-prometheus-opentelemetry.md @@ -126,10 +126,11 @@ and drop the `/metrics` route. Instrument names match `faststream.opentelemetry.TelemetryMiddleware` for the bus-scope metrics — `messaging.process.duration`, `messaging.publish.duration`, and (when `include_messages_counters=True`) -`messaging.process.messages` / `messaging.publish.messages` — plus three +`messaging.process.messages` / `messaging.publish.messages` — plus four outbox-specific counters the middleware can't emit: -`messaging.outbox.fetch.batches`, `messaging.outbox.lease_lost`, and -`messaging.outbox.dlq_written`. Units and constructor args +`messaging.outbox.fetch.batches`, `messaging.outbox.lease_lost`, +`messaging.outbox.dlq_written`, and `messaging.outbox.drain_timeout`. Units +and constructor args (`meter_provider`, `meter`, `include_messages_counters`) follow upstream. The `messaging.system="outbox"` attribute disambiguates outbox traffic from Kafka / Rabbit data on the same instruments. @@ -154,17 +155,21 @@ observability stack: ```bash pip install 'faststream-outbox[opentelemetry,prometheus]' \ - opentelemetry-exporter-otlp opentelemetry-exporter-prometheus uvicorn + opentelemetry-exporter-otlp uvicorn ``` +Here OpenTelemetry supplies **spans** (exported to OTLP) and Prometheus +supplies **all metrics** (two registries scraped over HTTP). The OTel +middleware runs span-only — its meters would otherwise land on the global +Prometheus registry, which neither endpoint below exposes, so they are left +off to avoid a dead, unscraped meter path. + ```python # app.py — run with `OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 \ # uvicorn app:app --host 0.0.0.0 --port 8000` from faststream.asgi import AsgiFastStream -from opentelemetry import metrics, trace +from opentelemetry import trace from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter -from opentelemetry.exporter.prometheus import PrometheusMetricReader -from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor @@ -184,9 +189,6 @@ tracer_provider = TracerProvider(resource=resource) tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter())) trace.set_tracer_provider(tracer_provider) -meter_provider = MeterProvider(resource=resource, metric_readers=[PrometheusMetricReader()]) -metrics.set_meter_provider(meter_provider) - # Two registries: the middleware and the recorder both define the same # faststream_* consume/publish collectors, so sharing one registry raises # "Duplicated timeseries in CollectorRegistry" at broker construction. @@ -202,8 +204,8 @@ broker = OutboxBroker( engine, outbox_table=outbox_table, middlewares=[ - # Bus-scope spans + meters around consume_scope / publish_scope. - OutboxTelemetryMiddleware(tracer_provider=tracer_provider, meter_provider=meter_provider), + # Bus-scope spans around consume_scope / publish_scope (span-only; see note above). + OutboxTelemetryMiddleware(tracer_provider=tracer_provider), OutboxPrometheusMiddleware(registry=MIDDLEWARE_REGISTRY, app_name="my-outbox-service"), ], # Outbox-internal events (fetched, lease_lost, terminal reasons, dlq_written) @@ -226,9 +228,10 @@ app = AsgiFastStream( ``` Traces flow to OTLP (Jaeger / Tempo / Honeycomb / collector); the -middleware's meters land on `/metrics` and the recorder's outbox-internal -counters on `/metrics/outbox` for Prometheus to scrape — two scrape targets, -one process. +Prometheus **middleware's** consume/publish meters land on `/metrics` and the +**recorder's** outbox-internal counters on `/metrics/outbox` for Prometheus to +scrape — two scrape targets, one process. (The OTel middleware here +contributes spans only, per the note above.) **The two seams overlap on consume/publish series.** Both the middleware and the recorder emit the same `faststream_received_*` / `faststream_published_*` diff --git a/docs/usage/subscriber.md b/docs/usage/subscriber.md index 9ff98d1..c43ea7d 100644 --- a/docs/usage/subscriber.md +++ b/docs/usage/subscriber.md @@ -103,9 +103,12 @@ Per-subscriber knobs, passed to `@broker.subscriber("…", …)`: async def handle_urgent(body: dict) -> None: ... ``` -The factory in `subscriber/factory.py` warns or raises on likely-wrong -combinations (`lease_ttl_seconds <= max_fetch_interval`, `max_deliveries` -without retry, `min_fetch_interval > max_fetch_interval`, etc.). +`OutboxSubscriberConfig.__post_init__` (in `subscriber/config.py`) warns or +raises on likely-wrong combinations (`lease_ttl_seconds <= max_fetch_interval`, +`max_deliveries` without retry, `min_fetch_interval > max_fetch_interval`, +etc.). Validation lives on the config, not the factory, so **every** +construction path — `@broker.subscriber`, `@router.subscriber`, direct +construction — is checked. The table above lists the outbox-specific knobs. The standard FastStream subscriber kwargs pass through unchanged too: `dependencies`, `parser`, diff --git a/docs/usage/testing.md b/docs/usage/testing.md index 608b754..2859307 100644 --- a/docs/usage/testing.md +++ b/docs/usage/testing.md @@ -37,23 +37,38 @@ async def test_handler() -> None: assert received == [1] ``` -In sync mode, `session=` is optional — the test broker patches -`broker.publish` to ignore it. The fake client keeps an in-memory list of -rows you can inspect via `fake_client.rows` — but `fake_client` is an -attribute of the `TestOutboxBroker` harness, not the broker, so bind the -harness to a name: +On `broker.publish` (and `publish_batch`), `session=` is optional in tests — +the test broker patches those methods to ignore it. This does **not** extend +to `broker.publisher("q").publish(...)`, which still requires a `session` +(see [Testing publishers](#testing-publishers) below). + +The fake client keeps an in-memory list of rows you can inspect via +`fake_client.rows` — but `fake_client` is an attribute of the +`TestOutboxBroker` harness, not the broker, so bind the harness to a name: ```python tb = TestOutboxBroker(broker) async with tb: - await broker.publish(1, queue="orders") + # "orders" has a subscriber, so in sync mode the row is dispatched and + # acked (deleted) before publish returns. Publish to a queue with no + # subscriber to inspect a row that survives. + await broker.publish(1, queue="audit") assert len(tb.fake_client.rows) == 1 ``` +An unconsumed queue is used deliberately: a row published to a queue that +*has* a subscriber is dispatched and deleted on ack within `publish`, so +`fake_client.rows` would be empty (`[]`) for it, not length 1. + ## Testing publishers ```python +from unittest.mock import AsyncMock + +from sqlalchemy.ext.asyncio import AsyncSession + + async def test_publisher() -> None: metadata = MetaData() outbox_table = make_outbox_table(metadata, table_name="outbox") @@ -67,15 +82,21 @@ async def test_publisher() -> None: pub = broker.publisher("orders") async with TestOutboxBroker(broker): - await pub.publish({"order_id": 1}) + # The publisher path builds its publish command (which validates the + # session type) before the fake producer is reached, so session= is + # required and must be an AsyncSession. The fake never touches it — a + # spec'd mock satisfies the check. + await pub.publish({"order_id": 1}, session=AsyncMock(spec=AsyncSession)) assert received == [{"order_id": 1}] ``` -`broker.publisher("q").publish(...)` works identically to +`broker.publisher("q").publish(...)` lands rows in the same fake store as `broker.publish(queue="q", ...)` — the test broker swaps the producer slot -for a `FakeOutboxProducer` that lands rows in the same fake store via the -FastStream `_basic_publish` flow. +for a `FakeOutboxProducer` via the FastStream `_basic_publish` flow. The one +difference in tests is the `session`: `broker.publish` is patched to make it +optional, but the publisher path is not, so pass a (mock) `AsyncSession` as +shown. ## Loop-driven mode @@ -87,6 +108,10 @@ delays — opt in with `run_loops=True`: import asyncio import json +# `broker` is built exactly as in the Basic test above: +# metadata = MetaData() +# outbox_table = make_outbox_table(metadata, table_name="outbox") +# broker = OutboxBroker(None, outbox_table=outbox_table) received: list[dict] = [] diff --git a/docs/usage/timers.md b/docs/usage/timers.md index 7544bba..ee5f424 100644 --- a/docs/usage/timers.md +++ b/docs/usage/timers.md @@ -24,7 +24,7 @@ await broker.publish( # Fire at a specific UTC instant: await broker.publish( {"x": 1}, queue="orders", session=session, - activate_at=dt.datetime(2026, 6, 1, 9, tzinfo=dt.UTC), + activate_at=dt.datetime(2027, 6, 1, 9, tzinfo=dt.UTC), ) ``` @@ -44,11 +44,15 @@ explicit `ValueError` rather than guessing your intended zone. ### Server-side vs client-side scheduling -For `publish`, `next_attempt_at` is computed server-side via `now() + -make_interval(secs => :s)` to stay clock-skew-safe. For `publish_batch`, -it's client-side (`datetime.now(UTC) + activate_in`) because executemany -doesn't compose cleanly with column-level SQL expressions, and the few-ms -drift is harmless for user-supplied scheduling. +For `publish` with `activate_in`, `next_attempt_at` is computed server-side +via `now() + make_interval(secs => :s)` to stay clock-skew-safe. With +`activate_at`, you supply an absolute instant, so it is stored verbatim and +compared against the *worker's* clock at fetch time — only `activate_in` is +skew-safe; `activate_at` is as accurate as your producers' and workers' +clocks agree. For `publish_batch`, `activate_in` is also client-side +(`datetime.now(UTC) + activate_in`) because executemany doesn't compose +cleanly with column-level SQL expressions, and the few-ms drift is harmless +for user-supplied scheduling. ## Deduplication with `timer_id`