From 2184c6d848f421b4e453b24591db77bf3d283847 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 20:21:11 +0300 Subject: [PATCH 1/9] docs(planning): design for per-table autovacuum tuning Exports outbox_autovacuum_ddl (importable ALTER TABLE statement for an Alembic migration) + a separate warn-level check_outbox_autovacuum probe. Recommend-not-require: package applies nothing, never raises. Records the load-bearing constraint (SQLAlchemy Table can't carry reloptions, so autogenerate can't emit them), the scale_factor=0 fix, fillfactor excluded on evidence, and benefit documented not benchmarked. --- .../2026-07-15.02-autovacuum-tuning.md | 153 ++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100644 planning/changes/2026-07-15.02-autovacuum-tuning.md diff --git a/planning/changes/2026-07-15.02-autovacuum-tuning.md b/planning/changes/2026-07-15.02-autovacuum-tuning.md new file mode 100644 index 0000000..5909e23 --- /dev/null +++ b/planning/changes/2026-07-15.02-autovacuum-tuning.md @@ -0,0 +1,153 @@ +--- +summary: Add `outbox_autovacuum_ddl(table_name, ...)` (an importable `ALTER TABLE … SET (autovacuum_*)` statement users drop into an Alembic migration) and a separate warn-level `check_outbox_autovacuum(engine, table)` probe that reports when a table lacks the aggressive settings. Recommend-not-require: the package applies nothing and never raises for this. Benefit documented (the scale_factor death-spiral), not benchmarked. +--- + +# Design: Per-table autovacuum tuning + +## Summary + +The outbox is a high-churn queue table: every message is one `INSERT` + one lease +`UPDATE` + one terminal `DELETE`, so the benchmark measured `dead_tup = 2 × +messages`. Postgres' default `autovacuum_vacuum_scale_factor = 0.2` fires vacuum +only after a *fraction of the table* becomes dead, which on a queue table lets +bloat accumulate — the heap and indexes grow physically even though the live row +count stays flat, so every fetch scans more pages over time. + +This change ships two small public helpers, both **opt-in** and **user-applied**: + +- `outbox_autovacuum_ddl(table_name="outbox", *, vacuum_threshold=1000, insert_threshold=1000) -> str` + — renders the recommended `ALTER TABLE … SET (autovacuum_*)` statement for the + user to `op.execute(...)` in an Alembic migration (or run via psql). +- `async def check_outbox_autovacuum(engine, table) -> list[str]` — a warn-level + probe that reads `pg_class.reloptions` and reports when a table lacks the + aggressive settings. + +The package **applies nothing itself and never raises** for autovacuum — it is a +performance recommendation, not a correctness requirement. The benefit (bloat +control) is documented, not benchmarked. + +## Motivation + +The performance review placed the library in Tsvettsikh's **Variant 2**, whose +central cost is row mutation churn driving autovacuum. The benchmark harness +quantified the churn: `dead_tup = 2 × messages` (the lease `UPDATE` leaves one +dead tuple, the terminal `DELETE` another). Under Postgres' default +`autovacuum_vacuum_scale_factor = 0.2`, a queue table with a small live set but +enormous churn triggers vacuum rarely — and if it ever bloats, the fraction is *of +the bloated size*, so vacuum fires ever less often: a self-reinforcing bloat +spiral that a small live-row count hides until fetch latency degrades. + +## Design + +### The constraint that shapes everything + +Autovacuum settings **cannot** ride the schema-declaration path the columns and +indexes use. SQLAlchemy 2.0's Postgres dialect accepts no reloptions kwarg on +`Table` (only `on_commit`, `partition_by`, `tablespace`, `using`, `with_oids`) — +verified against the installed 2.0.50. So the metadata `Table` cannot carry +`autovacuum_*`, which means Alembic autogenerate can *never* diff or emit them. +This is stricter than the `postgresql_where` blindness the codebase already +documents: there the `Table` could express the predicate and only autogenerate +was blind; here the `Table` cannot express it at all. There is therefore no +"declare it and the user's migration brings it up" option — the settings must be +applied by an explicit statement the user runs. + +### The recommended settings + +``` +autovacuum_vacuum_scale_factor = 0 -- disable the fraction-of-table trigger +autovacuum_vacuum_threshold = 1000 -- vacuum every N dead tuples, size-independent +autovacuum_vacuum_insert_scale_factor = 0 -- same, for insert-triggered vacuum (PG13+) +autovacuum_vacuum_insert_threshold = 1000 +``` + +`scale_factor = 0` is the structural fix — it removes the death-spiral trigger so +vacuum frequency tracks *churn*, not table size. The constant thresholds are +tunable per workload. The outbox inserts every message, so the insert-triggered +pair (PG13+) matters too. **`fillfactor` is deliberately excluded**: the harness +measured HOT as structurally impossible on this table (the claim `UPDATE` mutates +`acquired_token`/`acquired_at`, the columns both partial indexes key on), so +`fillfactor` buys almost nothing — excluded on evidence, not omission. + +### The DDL helper + +`outbox_autovacuum_ddl(table_name="outbox", *, vacuum_threshold=1000, insert_threshold=1000) -> str` +returns the `ALTER TABLE "" SET (…)` statement (identifier quoted). Used as: + +```python +from faststream_outbox import outbox_autovacuum_ddl + +def upgrade() -> None: + op.execute(outbox_autovacuum_ddl("outbox")) +``` + +The recommended reloptions live in **one module-level constant** that both this +renderer and the probe read, so the two cannot drift. + +### The validation probe + +`async def check_outbox_autovacuum(engine, table) -> list[str]` reads +`pg_class.reloptions` for the table and returns human-readable warnings when the +aggressive settings are absent; `[]` means OK. Opt-in (call from a startup hook or +`/health`), **warn-level, never raises** — so it does not behave like the +load-bearing index/CHECK probes in `validate_schema`. It asserts the *structural* +property (`autovacuum_vacuum_scale_factor = 0`) and that a threshold is set, **not** +the exact threshold value, so a user's legitimate threshold tuning does not trip a +false warning. It is deliberately **not** folded into `validate_schema()`, which +keeps its raise-on-mismatch contract. + +### Where it lives + +A new `faststream_outbox/autovacuum.py` holds the reloptions constant, the DDL +renderer, and the probe (rather than swelling `schema.py`, which is focused on the +table factory, or `client.py`). Both functions are exported from `__init__.py` as +public API. + +## Non-goals + +- **The package does not apply the settings.** No `after_create` DDL event, no + auto-`ALTER`. The schema is user-owned; the user runs the statement in their + migration. This keeps the user-owned-schema invariant intact. +- **Not in `validate_schema()`.** That method raises on mismatch; autovacuum is a + recommendation, so its probe is a separate warn-level function. +- **No `fillfactor`** (see above — measured near-useless here). +- **No churn benchmark.** The benefit is bloat-over-time, which autovacuum's async + ~1-minute schedule makes noisy and un-gateable; it needs minutes to diverge and + is well-established Postgres behavior. Documented, not measured in CI. +- **No default-on.** The helper is inherently opt-in (the user chooses to add it); + there is no default to flip and no silent schema change on upgrade. + +## Testing + +Deterministic — the correctness of both helpers is cleanly gateable, unlike +batching's timing-sensitive counter. + +- **Unit (no DB):** `outbox_autovacuum_ddl` renders the exact expected SQL, + including identifier quoting and `vacuum_threshold` / `insert_threshold` + overrides. +- **Integration (real Postgres, deterministic):** `check_outbox_autovacuum` + returns `[]` on a table with the reloptions applied (via the helper's own SQL), + returns warnings on a table without them, and does **not** false-warn on a table + with `scale_factor = 0` but a custom threshold (e.g. `5000`). + +## Benefit (documented, not benchmarked) + +`docs/operations/alembic.md` gains the migration recipe +(`op.execute(outbox_autovacuum_ddl("outbox"))`) and the mechanism explanation: why +`scale_factor = 0` breaks the death-spiral, how to tune the thresholds, and the +`check_outbox_autovacuum` startup/health usage — anchored to the measured +`dead_tup = 2 × messages` churn rate. `architecture/schema.md` (the truth-home) +notes that aggressive autovacuum is recommended-not-required and points at the +helper + probe. + +## Risk + +- **Users may never apply it** (silent bloat). Mitigated by the opt-in probe: a + user who wires `check_outbox_autovacuum` into `/health` is warned. Those who + wire up neither the migration nor the probe are no worse off than today. +- **Threshold defaults are workload-dependent.** `1000` is a sane starting point, + not a universal optimum; the helper's override params and the docs make tuning + explicit, and the probe checks the structural `scale_factor = 0`, not the value. +- **PG13+ for the insert-vacuum reloptions.** The project is Postgres-only and + tests against 17; the insert-triggered pair requires PG13+, which is well within + support. Documented. From 4a372a54cd56e982d737eb525e28007a8c814a93 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 20:34:23 +0300 Subject: [PATCH 2/9] feat: add outbox_autovacuum_ddl migration-statement helper Co-Authored-By: Claude Opus 4.8 (1M context) --- faststream_outbox/__init__.py | 2 ++ faststream_outbox/autovacuum.py | 58 +++++++++++++++++++++++++++++++++ tests/test_unit.py | 27 +++++++++++++++ 3 files changed, 87 insertions(+) create mode 100644 faststream_outbox/autovacuum.py diff --git a/faststream_outbox/__init__.py b/faststream_outbox/__init__.py index 85d034c..b5b2c8e 100644 --- a/faststream_outbox/__init__.py +++ b/faststream_outbox/__init__.py @@ -4,6 +4,7 @@ from faststream._internal.broker import BrokerUsecase from faststream._internal.testing.broker import TestBroker +from faststream_outbox.autovacuum import outbox_autovacuum_ddl from faststream_outbox.broker import OutboxBroker from faststream_outbox.message import OutboxMessage from faststream_outbox.metrics import MetricsRecorder @@ -36,6 +37,7 @@ "TestOutboxBroker", "make_dlq_table", "make_outbox_table", + "outbox_autovacuum_ddl", ] try: diff --git a/faststream_outbox/autovacuum.py b/faststream_outbox/autovacuum.py new file mode 100644 index 0000000..dcda33c --- /dev/null +++ b/faststream_outbox/autovacuum.py @@ -0,0 +1,58 @@ +"""Per-table autovacuum tuning for the outbox table. + +The outbox is a high-churn queue table: every message is one INSERT + one lease +UPDATE + one terminal DELETE, so dead tuples accumulate at ~2x the message rate. +Postgres' default ``autovacuum_vacuum_scale_factor = 0.2`` fires vacuum only after a +*fraction of the table* is dead -- on a queue table that lets bloat accumulate, and +if the table ever bloats the fraction is of the bloated size, so vacuum fires ever +less often (a death spiral). Setting the scale factor to 0 disables that trigger so +vacuum frequency tracks *churn* via a constant threshold, not table size. + +SQLAlchemy's ``Table`` cannot carry reloptions (the PG dialect accepts no such +kwarg), so Alembic autogenerate can never emit them. These settings must be applied +by an explicit statement the user runs -- :func:`outbox_autovacuum_ddl` renders it +for an Alembic migration; :func:`check_outbox_autovacuum` reports when a table lacks +it. The package applies nothing itself and never raises for autovacuum: it is a +performance recommendation, not a correctness requirement. +""" + +from sqlalchemy.dialects import postgresql + + +# Single source of truth for the reloption keys, shared by the renderer and the probe. +# Scale-factor keys are STRUCTURAL: they must be 0 to break the fraction-of-table +# death-spiral. Threshold keys are TUNABLE: they must merely be present (any value). +_SCALE_FACTOR_KEYS: tuple[str, str] = ( + "autovacuum_vacuum_scale_factor", + "autovacuum_vacuum_insert_scale_factor", +) +_VACUUM_THRESHOLD_KEY = "autovacuum_vacuum_threshold" +_INSERT_THRESHOLD_KEY = "autovacuum_vacuum_insert_threshold" + +# Standalone PG identifier preparer (no engine needed) -- quotes reserved words / odd +# characters exactly as SQLAlchemy's own DDL does; leaves simple names unquoted. +_IDENTIFIER_PREPARER = postgresql.dialect().identifier_preparer + + +def outbox_autovacuum_ddl( + table_name: str = "outbox", + *, + vacuum_threshold: int = 1000, + insert_threshold: int = 1000, +) -> str: + """Render the recommended ``ALTER TABLE … SET (autovacuum_*)`` statement. + + Drop it into an Alembic migration (``op.execute(outbox_autovacuum_ddl("outbox"))``) + or run it via psql. ``vacuum_threshold`` / ``insert_threshold`` tune how many dead + (resp. inserted) tuples trigger autovacuum; the scale factors are fixed at 0 -- that + is the structural fix, not a knob. The insert-triggered reloptions require Postgres 13+. + """ + quoted = _IDENTIFIER_PREPARER.quote(table_name) + options = ( + (_SCALE_FACTOR_KEYS[0], "0"), + (_VACUUM_THRESHOLD_KEY, str(vacuum_threshold)), + (_SCALE_FACTOR_KEYS[1], "0"), + (_INSERT_THRESHOLD_KEY, str(insert_threshold)), + ) + settings = ", ".join(f"{key} = {value}" for key, value in options) + return f"ALTER TABLE {quoted} SET ({settings})" diff --git a/tests/test_unit.py b/tests/test_unit.py index 8425d9e..4abf5d6 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -38,6 +38,7 @@ TestOutboxBroker, make_dlq_table, make_outbox_table, + outbox_autovacuum_ddl, ) from faststream_outbox.annotations import OutboxMessage as AnnotatedOutboxMessage from faststream_outbox.broker import OutboxParamsStorage @@ -4105,3 +4106,29 @@ async def test_spec_url_uses_password_masked_engine_dsn_when_engine_present() -> assert urls, "spec url must be non-empty when an engine is present" assert "supersecret" not in urls[0], "password must be masked in the AsyncAPI server url" assert "db.example" in urls[0] + + +def test_outbox_autovacuum_ddl_default() -> None: + sql = outbox_autovacuum_ddl("outbox") + assert sql == ( + "ALTER TABLE outbox SET (" + "autovacuum_vacuum_scale_factor = 0, " + "autovacuum_vacuum_threshold = 1000, " + "autovacuum_vacuum_insert_scale_factor = 0, " + "autovacuum_vacuum_insert_threshold = 1000)" + ) + + +def test_outbox_autovacuum_ddl_threshold_overrides() -> None: + sql = outbox_autovacuum_ddl("outbox", vacuum_threshold=5000, insert_threshold=2000) + assert "autovacuum_vacuum_threshold = 5000" in sql + assert "autovacuum_vacuum_insert_threshold = 2000" in sql + # scale factors stay structural regardless of threshold overrides + assert "autovacuum_vacuum_scale_factor = 0" in sql + assert "autovacuum_vacuum_insert_scale_factor = 0" in sql + + +def test_outbox_autovacuum_ddl_quotes_reserved_table_name() -> None: + # A reserved word must be quoted so the ALTER TABLE parses. + sql = outbox_autovacuum_ddl("user") + assert sql.startswith('ALTER TABLE "user" SET (') From c63ea3a5381c993cbc4d376bedb6c8ed112bfda0 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 20:43:14 +0300 Subject: [PATCH 3/9] feat: add check_outbox_autovacuum warn-level reloptions probe --- faststream_outbox/__init__.py | 3 +- faststream_outbox/autovacuum.py | 62 +++++++++++++++++++++++++++++++++ tests/test_integration.py | 34 ++++++++++++++++++ 3 files changed, 98 insertions(+), 1 deletion(-) diff --git a/faststream_outbox/__init__.py b/faststream_outbox/__init__.py index b5b2c8e..13efa75 100644 --- a/faststream_outbox/__init__.py +++ b/faststream_outbox/__init__.py @@ -4,7 +4,7 @@ from faststream._internal.broker import BrokerUsecase from faststream._internal.testing.broker import TestBroker -from faststream_outbox.autovacuum import outbox_autovacuum_ddl +from faststream_outbox.autovacuum import check_outbox_autovacuum, outbox_autovacuum_ddl from faststream_outbox.broker import OutboxBroker from faststream_outbox.message import OutboxMessage from faststream_outbox.metrics import MetricsRecorder @@ -35,6 +35,7 @@ "OutboxRouter", "RetryStrategyProto", "TestOutboxBroker", + "check_outbox_autovacuum", "make_dlq_table", "make_outbox_table", "outbox_autovacuum_ddl", diff --git a/faststream_outbox/autovacuum.py b/faststream_outbox/autovacuum.py index dcda33c..e59a1eb 100644 --- a/faststream_outbox/autovacuum.py +++ b/faststream_outbox/autovacuum.py @@ -16,9 +16,17 @@ performance recommendation, not a correctness requirement. """ +from typing import TYPE_CHECKING + +from sqlalchemy import text from sqlalchemy.dialects import postgresql +if TYPE_CHECKING: + from sqlalchemy import Table + from sqlalchemy.ext.asyncio import AsyncEngine + + # Single source of truth for the reloption keys, shared by the renderer and the probe. # Scale-factor keys are STRUCTURAL: they must be 0 to break the fraction-of-table # death-spiral. Threshold keys are TUNABLE: they must merely be present (any value). @@ -56,3 +64,57 @@ def outbox_autovacuum_ddl( ) settings = ", ".join(f"{key} = {value}" for key, value in options) return f"ALTER TABLE {quoted} SET ({settings})" + + +# reloptions come back from asyncpg as a ``list[str]`` of ``"key=value"`` items, or +# None when the table has no options set (or does not exist). NULL nspname match uses +# current_schema() so a search_path-relative table resolves the same way the app does. +_RELOPTIONS_QUERY = text( + "SELECT c.reloptions FROM pg_class c " + "JOIN pg_namespace n ON n.oid = c.relnamespace " + "WHERE c.relname = :table AND n.nspname = COALESCE(:schema, current_schema())", +) + +_SEE_DOCS = "apply outbox_autovacuum_ddl() in a migration -- see docs/operations/alembic.md" + + +def _parse_reloptions(reloptions: "list[str] | None") -> dict[str, str]: + """Turn asyncpg's ``["k=v", …]`` reloptions array (or None) into a dict.""" + if not reloptions: + return {} + parsed: dict[str, str] = {} + for item in reloptions: + key, _, value = item.partition("=") + parsed[key] = value + return parsed + + +async def check_outbox_autovacuum(engine: "AsyncEngine", table: "Table") -> list[str]: + """Report (do not raise) when *table* lacks the aggressive autovacuum settings. + + Reads ``pg_class.reloptions`` and returns a human-readable warning per missing + requirement; ``[]`` means the recommended settings are present. Opt-in -- call it + from a startup hook or ``/health``. Warn-level by design: autovacuum tuning is a + performance recommendation, not a correctness requirement, so this is deliberately + NOT part of ``validate_schema()`` (which raises on mismatch). It checks the + structural ``scale_factor = 0`` and that a threshold is set, not the threshold + value, so a user's legitimate threshold tuning does not trip a false warning. + """ + async with engine.connect() as conn: + reloptions = ( + await conn.execute(_RELOPTIONS_QUERY, {"table": table.name, "schema": table.schema}) + ).scalar_one_or_none() + options = _parse_reloptions(reloptions) + warnings: list[str] = [] + for key in _SCALE_FACTOR_KEYS: + value = options.get(key) + if value is None: + warnings.append(f"{table.name}: {key} is unset (want 0) -- bloat accumulates under churn; {_SEE_DOCS}.") + elif float(value) != 0.0: + warnings.append(f"{table.name}: {key} is {value}, not 0 -- bloat accumulates under churn; {_SEE_DOCS}.") + warnings.extend( + f"{table.name}: {key} is unset -- {_SEE_DOCS}." + for key in (_VACUUM_THRESHOLD_KEY, _INSERT_THRESHOLD_KEY) + if key not in options + ) + return warnings diff --git a/tests/test_integration.py b/tests/test_integration.py index 8994605..3b29509 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -19,8 +19,10 @@ NoRetry, OutboxBroker, OutboxResponse, + check_outbox_autovacuum, make_dlq_table, make_outbox_table, + outbox_autovacuum_ddl, ) from faststream_outbox.client import OutboxClient from faststream_outbox.envelope import _encode_payload as encode_payload @@ -2125,3 +2127,35 @@ async def test_validate_schema_passes_under_ck_convention_with_literally_named_c finally: async with pg_engine.begin() as conn: await conn.run_sync(metadata.drop_all) + + +async def test_check_autovacuum_ok_when_applied(pg_engine: AsyncEngine, outbox_table: Table) -> None: + async with pg_engine.begin() as conn: + await conn.execute(text(outbox_autovacuum_ddl(outbox_table.name))) + assert await check_outbox_autovacuum(pg_engine, outbox_table) == [] + + +async def test_check_autovacuum_warns_when_unset(pg_engine: AsyncEngine, outbox_table: Table) -> None: + # Fresh table (the fixture applies no reloptions) -> a warning per required key. + warnings = await check_outbox_autovacuum(pg_engine, outbox_table) + assert warnings # non-empty + joined = " ".join(warnings) + assert "autovacuum_vacuum_scale_factor" in joined + assert "autovacuum_vacuum_threshold" in joined + + +async def test_check_autovacuum_ok_with_custom_threshold(pg_engine: AsyncEngine, outbox_table: Table) -> None: + # scale_factor still 0, but a deliberately different threshold -> no false warning. + async with pg_engine.begin() as conn: + await conn.execute(text(outbox_autovacuum_ddl(outbox_table.name, vacuum_threshold=5000))) + assert await check_outbox_autovacuum(pg_engine, outbox_table) == [] + + +async def test_check_autovacuum_warns_when_scale_factor_nonzero(pg_engine: AsyncEngine, outbox_table: Table) -> None: + # A user who set a nonzero scale factor is still exposed to the death-spiral. + async with pg_engine.begin() as conn: + await conn.execute( + text(f'ALTER TABLE "{outbox_table.name}" SET (autovacuum_vacuum_scale_factor = 0.1)'), + ) + warnings = await check_outbox_autovacuum(pg_engine, outbox_table) + assert any("autovacuum_vacuum_scale_factor" in w for w in warnings) From 6c00956434447d3dfcab5f319ec657e78bc07753 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 20:47:42 +0300 Subject: [PATCH 4/9] docs: document autovacuum tuning helper, probe, and rationale --- architecture/schema.md | 13 ++++++++++++ docs/operations/alembic.md | 42 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+) diff --git a/architecture/schema.md b/architecture/schema.md index 55c6c10..cfb3ca3 100644 --- a/architecture/schema.md +++ b/architecture/schema.md @@ -74,3 +74,16 @@ composition lives in `_compose_schema_mismatch_message` (`client.py`), gated on Alembic is optional (`faststream-outbox[validate]`); without it `validate_schema()` raises `ImportError`, but every other path works. + +## Autovacuum (recommended, not enforced) + +The outbox is high-churn (`dead_tup ≈ 2 × messages`: the lease `UPDATE` + terminal +`DELETE` each leave a dead tuple). Aggressive autovacuum +(`autovacuum_vacuum_scale_factor = 0` + a constant threshold, for both the vacuum and +insert-triggered pairs) is **recommended but not enforced**: SQLAlchemy's `Table` +cannot carry reloptions, so autogenerate can't emit them, and the package applies +nothing itself. `outbox_autovacuum_ddl()` (in `autovacuum.py`) renders the migration +statement; `check_outbox_autovacuum()` is a warn-level `pg_class.reloptions` probe, +kept out of `validate_schema()` so that method's raise-on-mismatch contract is +unchanged. `fillfactor` is excluded on evidence (HOT is impossible — the claim +`UPDATE` mutates both partial indexes' key columns). diff --git a/docs/operations/alembic.md b/docs/operations/alembic.md index 70b9a1b..e2e1fe9 100644 --- a/docs/operations/alembic.md +++ b/docs/operations/alembic.md @@ -167,6 +167,48 @@ consequence that bites is a missing `server_default=now()` on `next_attempt_at`; see the server-defaults caveat in [Schema validation](../usage/schema-validation.md). +## Autovacuum tuning (recommended) + +The outbox is a high-churn queue table: every message is one `INSERT`, one lease +`UPDATE`, and one terminal `DELETE`, so dead tuples accumulate at roughly **twice +the message rate**. Postgres' default `autovacuum_vacuum_scale_factor = 0.2` fires +vacuum only after a *fraction of the table* is dead — on a queue table that lets +bloat grow, and if the table bloats the fraction is *of the bloated size*, so vacuum +fires ever less often. Set the scale factor to `0` so vacuum triggers on a constant +dead-tuple count instead, tracking churn rather than table size. + +SQLAlchemy's `Table` cannot carry these reloptions, so `alembic revision +--autogenerate` will never emit them — apply them with an explicit statement. +`faststream_outbox` renders it for you: + +```python +from alembic import op +from faststream_outbox import outbox_autovacuum_ddl + + +def upgrade() -> None: + op.execute(outbox_autovacuum_ddl("outbox")) +``` + +This sets `autovacuum_vacuum_scale_factor = 0` and `autovacuum_vacuum_threshold = +1000` (plus the insert-triggered pair, Postgres 13+). Tune the thresholds for your +message rate — `outbox_autovacuum_ddl("outbox", vacuum_threshold=5000, +insert_threshold=5000)` — a higher threshold vacuums less often. `fillfactor` is +intentionally not set: the lease `UPDATE` touches indexed columns, so HOT updates +are impossible and `fillfactor` buys almost nothing here. + +To catch a table that never had the settings applied, call `check_outbox_autovacuum` +from a startup hook or `/health`. It returns warnings (never raises) and is separate +from `validate_schema()`: + +```python +from faststream_outbox import check_outbox_autovacuum + +warnings = await check_outbox_autovacuum(engine, outbox_table) +if warnings: + logger.warning("outbox autovacuum not tuned: %s", "; ".join(warnings)) +``` + ## Fixing drift autogenerate can't see { #fixing-drift-autogenerate-cant-see } Two kinds of drift that From 69675f183468ab0ae509dae3548cdf672bc9c73c Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 21:24:10 +0300 Subject: [PATCH 5/9] fix(autovacuum): add schema param to outbox_autovacuum_ddl check_outbox_autovacuum is schema-aware but outbox_autovacuum_ddl always rendered an unqualified ALTER TABLE, so a user with the outbox in a named Postgres schema had no way to target the same table the probe checks. Add an optional keyword-only schema param (default None, byte-identical to current output); when set, quote schema and table independently and join with '.' rather than passing a dotted string to a single quote() call. --- docs/operations/alembic.md | 4 ++++ faststream_outbox/autovacuum.py | 12 ++++++++++-- tests/test_integration.py | 24 ++++++++++++++++++++++++ tests/test_unit.py | 13 +++++++++++++ 4 files changed, 51 insertions(+), 2 deletions(-) diff --git a/docs/operations/alembic.md b/docs/operations/alembic.md index e2e1fe9..bab7201 100644 --- a/docs/operations/alembic.md +++ b/docs/operations/alembic.md @@ -197,6 +197,10 @@ insert_threshold=5000)` — a higher threshold vacuums less often. `fillfactor` intentionally not set: the lease `UPDATE` touches indexed columns, so HOT updates are impossible and `fillfactor` buys almost nothing here. +If your outbox table lives in a non-default `MetaData(schema=...)`, pass the same +schema — `outbox_autovacuum_ddl("outbox", schema="app")` — so the `ALTER TABLE` +targets that table rather than an unqualified name resolved via `search_path`. + To catch a table that never had the settings applied, call `check_outbox_autovacuum` from a startup hook or `/health`. It returns warnings (never raises) and is separate from `validate_schema()`: diff --git a/faststream_outbox/autovacuum.py b/faststream_outbox/autovacuum.py index e59a1eb..17cbd06 100644 --- a/faststream_outbox/autovacuum.py +++ b/faststream_outbox/autovacuum.py @@ -45,6 +45,7 @@ def outbox_autovacuum_ddl( table_name: str = "outbox", *, + schema: str | None = None, vacuum_threshold: int = 1000, insert_threshold: int = 1000, ) -> str: @@ -54,8 +55,15 @@ def outbox_autovacuum_ddl( or run it via psql. ``vacuum_threshold`` / ``insert_threshold`` tune how many dead (resp. inserted) tuples trigger autovacuum; the scale factors are fixed at 0 -- that is the structural fix, not a knob. The insert-triggered reloptions require Postgres 13+. + + ``schema`` defaults to ``None``, which renders an unqualified table name that resolves + via the connection's ``search_path`` -- matching both ``Table.schema=None`` and + :func:`check_outbox_autovacuum`'s ``COALESCE(:schema, current_schema())`` lookup. Pass + the same ``schema`` as the outbox ``Table`` (e.g. ``table.schema``) when it lives in a + named schema, so this DDL targets the same table the probe checks. """ - quoted = _IDENTIFIER_PREPARER.quote(table_name) + quoted_table = _IDENTIFIER_PREPARER.quote(table_name) + quoted_name = quoted_table if schema is None else f"{_IDENTIFIER_PREPARER.quote(schema)}.{quoted_table}" options = ( (_SCALE_FACTOR_KEYS[0], "0"), (_VACUUM_THRESHOLD_KEY, str(vacuum_threshold)), @@ -63,7 +71,7 @@ def outbox_autovacuum_ddl( (_INSERT_THRESHOLD_KEY, str(insert_threshold)), ) settings = ", ".join(f"{key} = {value}" for key, value in options) - return f"ALTER TABLE {quoted} SET ({settings})" + return f"ALTER TABLE {quoted_name} SET ({settings})" # reloptions come back from asyncpg as a ``list[str]`` of ``"key=value"`` items, or diff --git a/tests/test_integration.py b/tests/test_integration.py index 3b29509..cf4bb81 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -2159,3 +2159,27 @@ async def test_check_autovacuum_warns_when_scale_factor_nonzero(pg_engine: Async ) warnings = await check_outbox_autovacuum(pg_engine, outbox_table) assert any("autovacuum_vacuum_scale_factor" in w for w in warnings) + + +async def test_check_autovacuum_ok_when_applied_in_named_schema(pg_engine: AsyncEngine) -> None: + """The DDL helper's ``schema=`` must target the same table the schema-aware probe checks. + + Before the fix, ``outbox_autovacuum_ddl`` had no ``schema`` param and always rendered an + unqualified ``ALTER TABLE outbox ...``, which resolves via search_path -- silently missing + (or mistargeting) a table that lives in a named schema, while the probe (which is already + schema-aware) kept warning the real table was untuned. + """ + schema = f"sch_av_{uuid.uuid4().hex[:8]}" + metadata = MetaData(schema=schema) + table = make_outbox_table(metadata, table_name="outbox") + async with pg_engine.begin() as conn: + await conn.execute(text(f'CREATE SCHEMA "{schema}"')) + await conn.run_sync(metadata.create_all) + try: + async with pg_engine.begin() as conn: + await conn.execute(text(outbox_autovacuum_ddl(table.name, schema=table.schema))) + assert await check_outbox_autovacuum(pg_engine, table) == [] + finally: + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.drop_all) + await conn.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) diff --git a/tests/test_unit.py b/tests/test_unit.py index 4abf5d6..d38d71b 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -4132,3 +4132,16 @@ def test_outbox_autovacuum_ddl_quotes_reserved_table_name() -> None: # A reserved word must be quoted so the ALTER TABLE parses. sql = outbox_autovacuum_ddl("user") assert sql.startswith('ALTER TABLE "user" SET (') + + +def test_outbox_autovacuum_ddl_schema_qualified() -> None: + # A named schema must qualify the table so the DDL targets the same table the + # schema-aware probe checks, rather than an unqualified search_path-relative one. + sql = outbox_autovacuum_ddl("outbox", schema="app") + assert sql == ( + "ALTER TABLE app.outbox SET (" + "autovacuum_vacuum_scale_factor = 0, " + "autovacuum_vacuum_threshold = 1000, " + "autovacuum_vacuum_insert_scale_factor = 0, " + "autovacuum_vacuum_insert_threshold = 1000)" + ) From f6e2fe987233a41406714f7ac023d63107cd18cb Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 21:48:42 +0300 Subject: [PATCH 6/9] docs(planning): revise autovacuum design to validate_schema flag Fold the standalone warn-level check_outbox_autovacuum into an opt-in validate_schema(check_autovacuum=True) that enforces (raises), giving one schema-health entry point. Records the accepted consequences: autovacuum becomes enforceable rather than advisory, and the check couples to the [validate] (Alembic) extra. --- .../2026-07-15.02-autovacuum-tuning.md | 121 +++++++++++------- 1 file changed, 78 insertions(+), 43 deletions(-) diff --git a/planning/changes/2026-07-15.02-autovacuum-tuning.md b/planning/changes/2026-07-15.02-autovacuum-tuning.md index 5909e23..d5ab871 100644 --- a/planning/changes/2026-07-15.02-autovacuum-tuning.md +++ b/planning/changes/2026-07-15.02-autovacuum-tuning.md @@ -1,5 +1,5 @@ --- -summary: Add `outbox_autovacuum_ddl(table_name, ...)` (an importable `ALTER TABLE … SET (autovacuum_*)` statement users drop into an Alembic migration) and a separate warn-level `check_outbox_autovacuum(engine, table)` probe that reports when a table lacks the aggressive settings. Recommend-not-require: the package applies nothing and never raises for this. Benefit documented (the scale_factor death-spiral), not benchmarked. +summary: Add `outbox_autovacuum_ddl(table_name, *, schema, ...)` (an importable `ALTER TABLE … SET (autovacuum_*)` statement users drop into an Alembic migration) and an opt-in `validate_schema(check_autovacuum=True)` flag that enforces the aggressive settings (raises like any other schema check). The package applies nothing itself; autovacuum is a recommendation by default and enforced only when you opt in. Benefit documented (the scale_factor death-spiral), not benchmarked. [Revised: the check moved from a standalone warn-level probe into validate_schema — see Design.] --- # Design: Per-table autovacuum tuning @@ -13,18 +13,31 @@ only after a *fraction of the table* becomes dead, which on a queue table lets bloat accumulate — the heap and indexes grow physically even though the live row count stays flat, so every fetch scans more pages over time. -This change ships two small public helpers, both **opt-in** and **user-applied**: +This change ships: -- `outbox_autovacuum_ddl(table_name="outbox", *, vacuum_threshold=1000, insert_threshold=1000) -> str` +- `outbox_autovacuum_ddl(table_name="outbox", *, schema=None, vacuum_threshold=1000, insert_threshold=1000) -> str` — renders the recommended `ALTER TABLE … SET (autovacuum_*)` statement for the - user to `op.execute(...)` in an Alembic migration (or run via psql). -- `async def check_outbox_autovacuum(engine, table) -> list[str]` — a warn-level - probe that reads `pg_class.reloptions` and reports when a table lacks the - aggressive settings. - -The package **applies nothing itself and never raises** for autovacuum — it is a -performance recommendation, not a correctness requirement. The benefit (bloat -control) is documented, not benchmarked. + user to `op.execute(...)` in an Alembic migration (or run via psql). Schema-aware. +- An opt-in `check_autovacuum: bool = False` parameter on `validate_schema(...)`: + when `True`, `validate_schema` also probes `pg_class.reloptions` and **raises** + (labeled distinctly from a schema mismatch) if the table lacks the aggressive + settings, exactly as it raises for any other schema requirement. + +The package **applies nothing itself**. Autovacuum is a recommendation by default +(untuned is not an error); it becomes an enforced requirement only for the caller +who opts into `validate_schema(check_autovacuum=True)`. The benefit (bloat control) +is documented, not benchmarked. + +> **Design revision (post-review).** The check originally shipped as a standalone +> warn-level `check_outbox_autovacuum(engine, table) -> list[str]` free function. +> On review it was folded into `validate_schema` behind the `check_autovacuum` flag +> so there is a single schema-health entry point. The consequences, accepted +> deliberately: (1) autovacuum becomes **enforceable** (raises when opted-in) rather +> than purely advisory — the raise-vs-advise contract of `validate_schema` is +> preserved because everything it reports is now fatal; (2) autovacuum checking is +> **coupled to the `[validate]` extra** (Alembic), since `validate_schema` requires +> it — the standalone probe did not. Serious deployments that tune autovacuum +> typically carry `[validate]` for migrations anyway. ## Motivation @@ -81,35 +94,49 @@ def upgrade() -> None: op.execute(outbox_autovacuum_ddl("outbox")) ``` -The recommended reloptions live in **one module-level constant** that both this -renderer and the probe read, so the two cannot drift. +The recommended reloption keys live in **module-level constants** that both this +renderer and the probe logic read, so the two cannot drift. + +### The `validate_schema(check_autovacuum=…)` flag -### The validation probe +`validate_schema` gains `check_autovacuum: bool = False` (threaded through +`OutboxClient.validate_schema` and `OutboxBroker.validate_schema`). Default `False` +is today's behavior. When `True`, `validate_schema` also probes +`pg_class.reloptions` and appends autovacuum findings to the errors it raises — +**distinctly labeled** ("autovacuum not tuned: …"), not lumped under the +"Outbox schema mismatch" prefix, so an operator can tell an untuned-but-working +table from a genuinely wrong schema. The check is *structural*: it requires +`autovacuum_vacuum_scale_factor = 0` and that a threshold is set, **not** the exact +threshold value, so a user's legitimate threshold tuning does not trip it. -`async def check_outbox_autovacuum(engine, table) -> list[str]` reads -`pg_class.reloptions` for the table and returns human-readable warnings when the -aggressive settings are absent; `[]` means OK. Opt-in (call from a startup hook or -`/health`), **warn-level, never raises** — so it does not behave like the -load-bearing index/CHECK probes in `validate_schema`. It asserts the *structural* -property (`autovacuum_vacuum_scale_factor = 0`) and that a threshold is set, **not** -the exact threshold value, so a user's legitimate threshold tuning does not trip a -false warning. It is deliberately **not** folded into `validate_schema()`, which -keeps its raise-on-mismatch contract. +Rationale for folding it in rather than a standalone warn-level function: +`validate_schema` is the single schema-health entry point and already raises on +every requirement, so gating autovacuum behind an opt-in flag keeps one call site +and one contract (everything it reports is fatal). The alternative — a separate +`check_outbox_autovacuum` returning `list[str]` — was implemented first and removed +on review (see the Summary's design-revision note). ### Where it lives -A new `faststream_outbox/autovacuum.py` holds the reloptions constant, the DDL -renderer, and the probe (rather than swelling `schema.py`, which is focused on the -table factory, or `client.py`). Both functions are exported from `__init__.py` as -public API. +`faststream_outbox/autovacuum.py` holds the reloption-key constants, the DDL +renderer (`outbox_autovacuum_ddl`, exported), and the reloptions-checking helper +(`pg_class.reloptions` query + parse + the structural check) as an internal +function. `OutboxClient.validate_schema` (in `client.py`) calls that helper when +`check_autovacuum=True`. Only `outbox_autovacuum_ddl` is public in `__init__.py`; +the check is reached through `validate_schema`. ## Non-goals - **The package does not apply the settings.** No `after_create` DDL event, no auto-`ALTER`. The schema is user-owned; the user runs the statement in their migration. This keeps the user-owned-schema invariant intact. -- **Not in `validate_schema()`.** That method raises on mismatch; autovacuum is a - recommendation, so its probe is a separate warn-level function. +- **No always-on enforcement.** `check_autovacuum` defaults to `False`, so a caller + who does not opt in never sees autovacuum raise. Untuned autovacuum is not a + correctness failure; enforcement is the opting-in caller's explicit choice. +- **No standalone warn-level probe / no `list[str]` return.** Folded into + `validate_schema` (raises), so there is no alembic-free `check_outbox_autovacuum` + and no advisory-list return for a `/health` JSON payload — a `/health` check calls + `validate_schema(check_autovacuum=True)` and reports unhealthy on the raise. - **No `fillfactor`** (see above — measured near-useless here). - **No churn benchmark.** The benefit is bloat-over-time, which autovacuum's async ~1-minute schedule makes noisy and un-gateable; it needs minutes to diverge and @@ -123,31 +150,39 @@ Deterministic — the correctness of both helpers is cleanly gateable, unlike batching's timing-sensitive counter. - **Unit (no DB):** `outbox_autovacuum_ddl` renders the exact expected SQL, - including identifier quoting and `vacuum_threshold` / `insert_threshold` - overrides. -- **Integration (real Postgres, deterministic):** `check_outbox_autovacuum` - returns `[]` on a table with the reloptions applied (via the helper's own SQL), - returns warnings on a table without them, and does **not** false-warn on a table - with `scale_factor = 0` but a custom threshold (e.g. `5000`). + including identifier quoting, schema-qualification, and `vacuum_threshold` / + `insert_threshold` overrides. +- **Integration (real Postgres, deterministic):** `validate_schema(check_autovacuum=True)` + does **not** raise (for the autovacuum reason) on a table with the reloptions + applied (via the helper's own SQL, including a named schema), **raises** a + distinctly-labeled error on a table without them, does **not** raise on a table + with `scale_factor = 0` but a custom threshold (e.g. `5000`), and + `check_autovacuum=False` never checks autovacuum. Round-trip: the helper's output + satisfies the check. ## Benefit (documented, not benchmarked) `docs/operations/alembic.md` gains the migration recipe (`op.execute(outbox_autovacuum_ddl("outbox"))`) and the mechanism explanation: why `scale_factor = 0` breaks the death-spiral, how to tune the thresholds, and the -`check_outbox_autovacuum` startup/health usage — anchored to the measured -`dead_tup = 2 × messages` churn rate. `architecture/schema.md` (the truth-home) -notes that aggressive autovacuum is recommended-not-required and points at the -helper + probe. +`validate_schema(check_autovacuum=True)` startup/health usage — anchored to the +measured `dead_tup = 2 × messages` churn rate. `architecture/schema.md` (the +truth-home) notes that aggressive autovacuum is recommended by default and enforced +via the `validate_schema` flag. ## Risk -- **Users may never apply it** (silent bloat). Mitigated by the opt-in probe: a - user who wires `check_outbox_autovacuum` into `/health` is warned. Those who - wire up neither the migration nor the probe are no worse off than today. +- **Users may never apply it** (silent bloat). Mitigated by the opt-in enforcement: + a user who calls `validate_schema(check_autovacuum=True)` in a startup gate is told + loudly. Those who opt into neither the migration nor the flag are no worse off than + today. +- **Enforcement can halt a working deployment.** `check_autovacuum=True` raises on an + untuned-but-otherwise-fine table. This is the opting-in caller's explicit choice; + the default `False` never does this, and the error is labeled distinctly from a + real schema mismatch so the cause is unambiguous. - **Threshold defaults are workload-dependent.** `1000` is a sane starting point, not a universal optimum; the helper's override params and the docs make tuning - explicit, and the probe checks the structural `scale_factor = 0`, not the value. + explicit, and the check verifies the structural `scale_factor = 0`, not the value. - **PG13+ for the insert-vacuum reloptions.** The project is Postgres-only and tests against 17; the insert-triggered pair requires PG13+, which is well within support. Documented. From 3668f6d5f5ff5e6857077527c81314dc6a0bcb7e Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 23:08:29 +0300 Subject: [PATCH 7/9] refactor(autovacuum): fold check into validate_schema(check_autovacuum=) Remove the standalone warn-level check_outbox_autovacuum(engine, table) free function and fold the reloptions check into validate_schema behind an opt-in check_autovacuum flag that enforces (raises) rather than advises. autovacuum.py keeps a pure autovacuum_findings(table_name, reloptions) helper (the old probe minus the DB query) that OutboxClient.validate_schema calls after running _RELOPTIONS_QUERY on its existing connection. Autovacuum errors are labeled distinctly ("Outbox autovacuum not tuned: ") from a schema mismatch so an operator can tell an untuned-but-working table from a wrong schema. With dlq_table=None and check_autovacuum=False (the default) every path is bit-for-bit identical to before. Co-Authored-By: Claude Opus 4.8 (1M context) --- architecture/schema.md | 19 +++++---- docs/operations/alembic.md | 13 +++--- faststream_outbox/__init__.py | 3 +- faststream_outbox/autovacuum.py | 51 +++++++++-------------- faststream_outbox/broker.py | 10 +++-- faststream_outbox/client.py | 27 ++++++++++--- faststream_outbox/testing.py | 2 +- tests/test_integration.py | 71 ++++++++++++++++++++++----------- 8 files changed, 114 insertions(+), 82 deletions(-) diff --git a/architecture/schema.md b/architecture/schema.md index cfb3ca3..885b5b8 100644 --- a/architecture/schema.md +++ b/architecture/schema.md @@ -75,15 +75,18 @@ composition lives in `_compose_schema_mismatch_message` (`client.py`), gated on Alembic is optional (`faststream-outbox[validate]`); without it `validate_schema()` raises `ImportError`, but every other path works. -## Autovacuum (recommended, not enforced) +## Autovacuum (recommended by default, enforced via a flag) The outbox is high-churn (`dead_tup ≈ 2 × messages`: the lease `UPDATE` + terminal `DELETE` each leave a dead tuple). Aggressive autovacuum (`autovacuum_vacuum_scale_factor = 0` + a constant threshold, for both the vacuum and -insert-triggered pairs) is **recommended but not enforced**: SQLAlchemy's `Table` -cannot carry reloptions, so autogenerate can't emit them, and the package applies -nothing itself. `outbox_autovacuum_ddl()` (in `autovacuum.py`) renders the migration -statement; `check_outbox_autovacuum()` is a warn-level `pg_class.reloptions` probe, -kept out of `validate_schema()` so that method's raise-on-mismatch contract is -unchanged. `fillfactor` is excluded on evidence (HOT is impossible — the claim -`UPDATE` mutates both partial indexes' key columns). +insert-triggered pairs) is **recommended by default**: SQLAlchemy's `Table` cannot +carry reloptions, so autogenerate can't emit them, and the package applies nothing +itself. `outbox_autovacuum_ddl()` (in `autovacuum.py`) renders the migration +statement the user runs. Enforcement is opt-in: `validate_schema(check_autovacuum=True)` +(threaded through `OutboxClient`/`OutboxBroker`) reads `pg_class.reloptions` and +**raises** a distinctly-labeled "Outbox autovacuum not tuned: " error when the +table lacks the settings — separate from the "Outbox schema mismatch: " prefix, so +an operator can tell the two apart. Because it rides `validate_schema()`, the check +is coupled to the `[validate]` (Alembic) extra. `fillfactor` is excluded on evidence +(HOT is impossible — the claim `UPDATE` mutates both partial indexes' key columns). diff --git a/docs/operations/alembic.md b/docs/operations/alembic.md index bab7201..84b03b0 100644 --- a/docs/operations/alembic.md +++ b/docs/operations/alembic.md @@ -201,16 +201,13 @@ If your outbox table lives in a non-default `MetaData(schema=...)`, pass the sam schema — `outbox_autovacuum_ddl("outbox", schema="app")` — so the `ALTER TABLE` targets that table rather than an unqualified name resolved via `search_path`. -To catch a table that never had the settings applied, call `check_outbox_autovacuum` -from a startup hook or `/health`. It returns warnings (never raises) and is separate -from `validate_schema()`: +To catch a table that never had the settings applied, pass `check_autovacuum=True` +to `validate_schema()`. Requires the `[validate]` (Alembic) extra: ```python -from faststream_outbox import check_outbox_autovacuum - -warnings = await check_outbox_autovacuum(engine, outbox_table) -if warnings: - logger.warning("outbox autovacuum not tuned: %s", "; ".join(warnings)) +# In a startup hook or /health check -- raises if the outbox table is not tuned +# (and if the schema itself has drifted). Requires the [validate] extra. +await broker.validate_schema(check_autovacuum=True) ``` ## Fixing drift autogenerate can't see { #fixing-drift-autogenerate-cant-see } diff --git a/faststream_outbox/__init__.py b/faststream_outbox/__init__.py index 13efa75..b5b2c8e 100644 --- a/faststream_outbox/__init__.py +++ b/faststream_outbox/__init__.py @@ -4,7 +4,7 @@ from faststream._internal.broker import BrokerUsecase from faststream._internal.testing.broker import TestBroker -from faststream_outbox.autovacuum import check_outbox_autovacuum, outbox_autovacuum_ddl +from faststream_outbox.autovacuum import outbox_autovacuum_ddl from faststream_outbox.broker import OutboxBroker from faststream_outbox.message import OutboxMessage from faststream_outbox.metrics import MetricsRecorder @@ -35,7 +35,6 @@ "OutboxRouter", "RetryStrategyProto", "TestOutboxBroker", - "check_outbox_autovacuum", "make_dlq_table", "make_outbox_table", "outbox_autovacuum_ddl", diff --git a/faststream_outbox/autovacuum.py b/faststream_outbox/autovacuum.py index 17cbd06..9d46392 100644 --- a/faststream_outbox/autovacuum.py +++ b/faststream_outbox/autovacuum.py @@ -11,22 +11,15 @@ SQLAlchemy's ``Table`` cannot carry reloptions (the PG dialect accepts no such kwarg), so Alembic autogenerate can never emit them. These settings must be applied by an explicit statement the user runs -- :func:`outbox_autovacuum_ddl` renders it -for an Alembic migration; :func:`check_outbox_autovacuum` reports when a table lacks -it. The package applies nothing itself and never raises for autovacuum: it is a -performance recommendation, not a correctness requirement. +for an Alembic migration; ``validate_schema(check_autovacuum=True)`` enforces it +(raises) when the table lacks it. The package applies nothing itself; enforcement +is opt-in via that flag, not a default. """ -from typing import TYPE_CHECKING - from sqlalchemy import text from sqlalchemy.dialects import postgresql -if TYPE_CHECKING: - from sqlalchemy import Table - from sqlalchemy.ext.asyncio import AsyncEngine - - # Single source of truth for the reloption keys, shared by the renderer and the probe. # Scale-factor keys are STRUCTURAL: they must be 0 to break the fraction-of-table # death-spiral. Threshold keys are TUNABLE: they must merely be present (any value). @@ -58,9 +51,10 @@ def outbox_autovacuum_ddl( ``schema`` defaults to ``None``, which renders an unqualified table name that resolves via the connection's ``search_path`` -- matching both ``Table.schema=None`` and - :func:`check_outbox_autovacuum`'s ``COALESCE(:schema, current_schema())`` lookup. Pass - the same ``schema`` as the outbox ``Table`` (e.g. ``table.schema``) when it lives in a - named schema, so this DDL targets the same table the probe checks. + the ``validate_schema(check_autovacuum=True)`` reloptions lookup's + ``COALESCE(:schema, current_schema())``. Pass the same ``schema`` as the outbox + ``Table`` (e.g. ``table.schema``) when it lives in a named schema, so this DDL + targets the same table the check reads. """ quoted_table = _IDENTIFIER_PREPARER.quote(table_name) quoted_name = quoted_table if schema is None else f"{_IDENTIFIER_PREPARER.quote(schema)}.{quoted_table}" @@ -97,32 +91,25 @@ def _parse_reloptions(reloptions: "list[str] | None") -> dict[str, str]: return parsed -async def check_outbox_autovacuum(engine: "AsyncEngine", table: "Table") -> list[str]: - """Report (do not raise) when *table* lacks the aggressive autovacuum settings. +def autovacuum_findings(table_name: str, reloptions: "list[str] | None") -> list[str]: + """Return one finding string per missing/wrong autovacuum setting; ``[]`` means OK. - Reads ``pg_class.reloptions`` and returns a human-readable warning per missing - requirement; ``[]`` means the recommended settings are present. Opt-in -- call it - from a startup hook or ``/health``. Warn-level by design: autovacuum tuning is a - performance recommendation, not a correctness requirement, so this is deliberately - NOT part of ``validate_schema()`` (which raises on mismatch). It checks the - structural ``scale_factor = 0`` and that a threshold is set, not the threshold - value, so a user's legitimate threshold tuning does not trip a false warning. + Pure -- takes already-fetched ``pg_class.reloptions`` (see :data:`_RELOPTIONS_QUERY`) + so the caller (``OutboxClient.validate_schema``) owns the query and the connection. + Structural check: each scale-factor key must be present AND 0; each threshold key + must be present (any value -- a user's custom threshold must not be flagged). """ - async with engine.connect() as conn: - reloptions = ( - await conn.execute(_RELOPTIONS_QUERY, {"table": table.name, "schema": table.schema}) - ).scalar_one_or_none() options = _parse_reloptions(reloptions) - warnings: list[str] = [] + findings: list[str] = [] for key in _SCALE_FACTOR_KEYS: value = options.get(key) if value is None: - warnings.append(f"{table.name}: {key} is unset (want 0) -- bloat accumulates under churn; {_SEE_DOCS}.") + findings.append(f"{table_name}: {key} is unset (want 0) -- bloat accumulates under churn; {_SEE_DOCS}.") elif float(value) != 0.0: - warnings.append(f"{table.name}: {key} is {value}, not 0 -- bloat accumulates under churn; {_SEE_DOCS}.") - warnings.extend( - f"{table.name}: {key} is unset -- {_SEE_DOCS}." + findings.append(f"{table_name}: {key} is {value}, not 0 -- bloat accumulates under churn; {_SEE_DOCS}.") + findings.extend( + f"{table_name}: {key} is unset -- {_SEE_DOCS}." for key in (_VACUUM_THRESHOLD_KEY, _INSERT_THRESHOLD_KEY) if key not in options ) - return warnings + return findings diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index 4578537..5e08622 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -358,9 +358,13 @@ async def ping(self, timeout: float | None = None) -> bool: # Only reached when the timeout scope cancelled the probe. return False - async def validate_schema(self) -> None: - """Validate the user's table matches what the package expects. Opt-in.""" - await self.client.validate_schema() + async def validate_schema(self, *, check_autovacuum: bool = False) -> None: + """Validate the user's table matches what the package expects. Opt-in. + + Pass ``check_autovacuum=True`` to also enforce the recommended autovacuum + reloptions; see :meth:`OutboxClient.validate_schema`. + """ + await self.client.validate_schema(check_autovacuum=check_autovacuum) async def publish( # ty: ignore[invalid-method-override] self, diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index b9c8d96..49a7ca4 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -41,6 +41,7 @@ # ``_import_checker`` so every optional-extra site uses the same shape. Users who # don't call validate_schema() never trigger the runtime import path. from faststream_outbox._import_checker import is_alembic_installed +from faststream_outbox.autovacuum import _RELOPTIONS_QUERY, autovacuum_findings from faststream_outbox.message import OutboxInnerMessage from faststream_outbox.schema import ( _DLQ_INJECTED_COLUMNS, @@ -137,7 +138,7 @@ async def mark_pending_with_lease( ) -> bool: ... @abc.abstractmethod - async def validate_schema(self) -> None: ... + async def validate_schema(self, *, check_autovacuum: bool = False) -> None: ... @abc.abstractmethod async def ping(self) -> bool: ... @@ -435,13 +436,20 @@ async def mark_pending_with_lease( result = await conn.execute(stmt, {"delay": max(0.0, delay_seconds)}) return (result.rowcount or 0) > 0 - async def validate_schema(self) -> None: + async def validate_schema(self, *, check_autovacuum: bool = False) -> None: """Validate that the database table(s) match the package's expected columns. Raises ``RuntimeError`` listing every mismatch across the outbox table and, when configured, the DLQ table. Opt-in: call from your startup hook or ``/health`` endpoint, not from ``broker.start()`` (so Alembic can run migrations against the same DB without blocking startup). + + Pass ``check_autovacuum=True`` to also enforce the recommended + ``autovacuum_vacuum_scale_factor = 0`` + threshold reloptions (see + :func:`faststream_outbox.autovacuum.outbox_autovacuum_ddl`); an untuned table + raises a distinctly-labeled "Outbox autovacuum not tuned: " error, separate + from a schema mismatch, so an operator can tell the two apart. Default + ``False`` never checks it. """ async with self._engine.connect() as conn: errors = await conn.run_sync(_validate_schema_sync, self._table) @@ -456,10 +464,19 @@ async def validate_schema(self) -> None: errors.extend(blind_errors) if self._dlq_table is not None: errors.extend(await conn.run_sync(_validate_dlq_schema_sync, self._dlq_table)) + autovacuum_errors: list[str] = [] + if check_autovacuum: + reloptions = ( + await conn.execute(_RELOPTIONS_QUERY, {"table": self._table.name, "schema": self._table.schema}) + ).scalar_one_or_none() + autovacuum_errors = autovacuum_findings(self._table.name, reloptions) + message_parts: list[str] = [] if errors: - raise RuntimeError( - _compose_schema_mismatch_message(errors, has_blind_drift=bool(blind_errors)), - ) + message_parts.append(_compose_schema_mismatch_message(errors, has_blind_drift=bool(blind_errors))) + if autovacuum_errors: + message_parts.append("Outbox autovacuum not tuned: " + "; ".join(autovacuum_errors)) + if message_parts: + raise RuntimeError("\n\n".join(message_parts)) async def ping(self) -> bool: # Bound the probe: an unwrapped connect+SELECT 1 against a half-dead TCP socket diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index 01f07af..fdaf361 100644 --- a/faststream_outbox/testing.py +++ b/faststream_outbox/testing.py @@ -235,7 +235,7 @@ async def cancel_timer(self, *, queue: str, timer_id: str) -> bool: return True return False - async def validate_schema(self) -> None: + async def validate_schema(self, *, check_autovacuum: bool = False) -> None: # Silently passing here would give tests false confidence — a user calling # ``broker.validate_schema()`` against ``TestOutboxBroker`` would see a green # test regardless of whether the real DB schema matches the canonical one. diff --git a/tests/test_integration.py b/tests/test_integration.py index cf4bb81..71f10e2 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -19,11 +19,11 @@ NoRetry, OutboxBroker, OutboxResponse, - check_outbox_autovacuum, make_dlq_table, make_outbox_table, outbox_autovacuum_ddl, ) +from faststream_outbox.autovacuum import _RELOPTIONS_QUERY, autovacuum_findings from faststream_outbox.client import OutboxClient from faststream_outbox.envelope import _encode_payload as encode_payload from faststream_outbox.publisher.fake import OutboxFakePublisher @@ -2129,45 +2129,66 @@ async def test_validate_schema_passes_under_ck_convention_with_literally_named_c await conn.run_sync(metadata.drop_all) -async def test_check_autovacuum_ok_when_applied(pg_engine: AsyncEngine, outbox_table: Table) -> None: - async with pg_engine.begin() as conn: - await conn.execute(text(outbox_autovacuum_ddl(outbox_table.name))) - assert await check_outbox_autovacuum(pg_engine, outbox_table) == [] +async def test_validate_schema_autovacuum_raises_when_untuned(pg_engine: AsyncEngine, outbox_table: Table) -> None: + """Fresh table (the fixture applies no reloptions) -> distinctly-labeled raise, not a schema mismatch.""" + client = OutboxClient(pg_engine, outbox_table) + with pytest.raises(RuntimeError, match="autovacuum not tuned") as excinfo: + await client.validate_schema(check_autovacuum=True) + assert "schema mismatch" not in str(excinfo.value) -async def test_check_autovacuum_warns_when_unset(pg_engine: AsyncEngine, outbox_table: Table) -> None: - # Fresh table (the fixture applies no reloptions) -> a warning per required key. - warnings = await check_outbox_autovacuum(pg_engine, outbox_table) - assert warnings # non-empty - joined = " ".join(warnings) - assert "autovacuum_vacuum_scale_factor" in joined - assert "autovacuum_vacuum_threshold" in joined +async def test_validate_schema_autovacuum_ok_when_applied(pg_engine: AsyncEngine, outbox_table: Table) -> None: + async with pg_engine.begin() as conn: + await conn.execute(text(outbox_autovacuum_ddl(outbox_table.name))) + client = OutboxClient(pg_engine, outbox_table) + await client.validate_schema(check_autovacuum=True) # must NOT raise -async def test_check_autovacuum_ok_with_custom_threshold(pg_engine: AsyncEngine, outbox_table: Table) -> None: - # scale_factor still 0, but a deliberately different threshold -> no false warning. +async def test_validate_schema_autovacuum_ok_with_custom_threshold( + pg_engine: AsyncEngine, + outbox_table: Table, +) -> None: + # scale_factor still 0, but a deliberately different threshold -> no false raise. async with pg_engine.begin() as conn: await conn.execute(text(outbox_autovacuum_ddl(outbox_table.name, vacuum_threshold=5000))) - assert await check_outbox_autovacuum(pg_engine, outbox_table) == [] + client = OutboxClient(pg_engine, outbox_table) + await client.validate_schema(check_autovacuum=True) # must NOT raise -async def test_check_autovacuum_warns_when_scale_factor_nonzero(pg_engine: AsyncEngine, outbox_table: Table) -> None: +async def test_validate_schema_autovacuum_raises_when_scale_factor_nonzero( + pg_engine: AsyncEngine, + outbox_table: Table, +) -> None: # A user who set a nonzero scale factor is still exposed to the death-spiral. async with pg_engine.begin() as conn: await conn.execute( text(f'ALTER TABLE "{outbox_table.name}" SET (autovacuum_vacuum_scale_factor = 0.1)'), ) - warnings = await check_outbox_autovacuum(pg_engine, outbox_table) - assert any("autovacuum_vacuum_scale_factor" in w for w in warnings) + client = OutboxClient(pg_engine, outbox_table) + with pytest.raises(RuntimeError, match="autovacuum not tuned"): + await client.validate_schema(check_autovacuum=True) -async def test_check_autovacuum_ok_when_applied_in_named_schema(pg_engine: AsyncEngine) -> None: - """The DDL helper's ``schema=`` must target the same table the schema-aware probe checks. +async def test_validate_schema_autovacuum_false_does_not_check(pg_engine: AsyncEngine, outbox_table: Table) -> None: + """``check_autovacuum=False`` (the default) never raises for autovacuum, even on an untuned table.""" + client = OutboxClient(pg_engine, outbox_table) + await client.validate_schema(check_autovacuum=False) # must NOT raise + await client.validate_schema() # default also must NOT raise + + +async def test_autovacuum_findings_ok_when_applied_in_named_schema(pg_engine: AsyncEngine) -> None: + """The DDL helper's ``schema=`` must target the same table the schema-aware reloptions query reads. Before the fix, ``outbox_autovacuum_ddl`` had no ``schema`` param and always rendered an unqualified ``ALTER TABLE outbox ...``, which resolves via search_path -- silently missing - (or mistargeting) a table that lives in a named schema, while the probe (which is already - schema-aware) kept warning the real table was untuned. + (or mistargeting) a table that lives in a named schema, while the reloptions query (already + schema-aware via ``COALESCE(:schema, current_schema())``) kept reporting the real table as + untuned. + + NB: this exercises ``_RELOPTIONS_QUERY`` + ``autovacuum_findings`` directly rather than through + ``OutboxClient.validate_schema(check_autovacuum=True)`` — the latter's Alembic-diff schema + probe has a pre-existing, unrelated limitation with non-default Postgres schemas (see the + autovacuum-refactor report), which is orthogonal to the autovacuum check itself. """ schema = f"sch_av_{uuid.uuid4().hex[:8]}" metadata = MetaData(schema=schema) @@ -2178,7 +2199,11 @@ async def test_check_autovacuum_ok_when_applied_in_named_schema(pg_engine: Async try: async with pg_engine.begin() as conn: await conn.execute(text(outbox_autovacuum_ddl(table.name, schema=table.schema))) - assert await check_outbox_autovacuum(pg_engine, table) == [] + async with pg_engine.connect() as conn: + reloptions = ( + await conn.execute(_RELOPTIONS_QUERY, {"table": table.name, "schema": table.schema}) + ).scalar_one_or_none() + assert autovacuum_findings(table.name, reloptions) == [] finally: async with pg_engine.begin() as conn: await conn.run_sync(metadata.drop_all) From 311512ab5ea4ebe7d4c486f240607fcf70ed92d4 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 23:12:07 +0300 Subject: [PATCH 8/9] fix(schema): validate_schema reflects named-schema tables (include_schemas) _run_validate configured Alembic's MigrationContext without include_schemas=True, so compare_metadata never reflected tables outside the connection's default schema. Any outbox table in a non-default MetaData(schema=...) therefore failed validate_schema() with "table 'outbox' does not exist", regardless of whether the schema was actually correct. Add include_schemas=True and narrow _include_name's schema branch to the target schema (name == table.schema; Alembic reports the default schema as None, so this matches None-vs-None for a default-schema table and the literal name otherwise) so unrelated schemas never reflect into the diff as false drift. Unblocks the named-schema autovacuum round-trip: the autovacuum check now routes through validate_schema(check_autovacuum=True) on a named-schema table. Co-Authored-By: Claude Opus 4.8 (1M context) --- architecture/schema.md | 7 ++++++ faststream_outbox/client.py | 11 ++++++++- tests/test_integration.py | 49 +++++++++++++++++++++++-------------- 3 files changed, 48 insertions(+), 19 deletions(-) diff --git a/architecture/schema.md b/architecture/schema.md index 885b5b8..f786513 100644 --- a/architecture/schema.md +++ b/architecture/schema.md @@ -72,6 +72,13 @@ Autogenerate-fixable drift (columns, plain indexes, DLQ) gets no pointer. Messag composition lives in `_compose_schema_mismatch_message` (`client.py`), gated on `has_blind_drift`. +The alembic diff runs with `include_schemas=True` so a table in a non-default +`MetaData(schema=...)` is reflected and compared (without it, `compare_metadata` +only sees the default schema and a named-schema table falsely reads as "table does +not exist"). `_include_name` narrows schema reflection to the target schema +(`name == table.schema`, where the default schema is reported as `None`) so +unrelated schemas never surface as false drift. + Alembic is optional (`faststream-outbox[validate]`); without it `validate_schema()` raises `ImportError`, but every other path works. diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index 49a7ca4..cffd6bb 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -702,8 +702,13 @@ def _run_validate( canonical_factory(canonical_metadata, table.name) def _include_name(name: str | None, type_: str, parent_names: "Mapping[str, str | None]") -> bool: + # ``include_schemas=True`` makes Alembic enumerate EVERY schema and call this with + # ``type_ == "schema"`` per schema — restrict to the target schema so unrelated + # schemas' tables never reflect into the diff (which would surface as false drift). + # Alembic reports the default schema as ``None`` here, so ``name == table.schema`` + # matches None-vs-None for a default-schema table and the literal name otherwise. if type_ == "schema": - return True + return name == table.schema if type_ == "table": return name == table.name return parent_names.get("table_name") == table.name @@ -713,6 +718,10 @@ def _include_name(name: str | None, type_: str, parent_names: "Mapping[str, str opts={ "compare_type": True, "compare_server_default": False, + # Reflect beyond the default schema so a table in a non-default + # ``MetaData(schema=...)`` is visible to ``compare_metadata`` (else it reads as + # "table does not exist"); ``_include_name`` narrows reflection to the target schema. + "include_schemas": True, "include_name": _include_name, "target_metadata": canonical_metadata, }, diff --git a/tests/test_integration.py b/tests/test_integration.py index 71f10e2..9275c95 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -23,7 +23,6 @@ make_outbox_table, outbox_autovacuum_ddl, ) -from faststream_outbox.autovacuum import _RELOPTIONS_QUERY, autovacuum_findings from faststream_outbox.client import OutboxClient from faststream_outbox.envelope import _encode_payload as encode_payload from faststream_outbox.publisher.fake import OutboxFakePublisher @@ -107,6 +106,28 @@ async def test_validate_schema_passes_for_correct_table(pg_engine, outbox_table) await client.validate_schema() # should not raise +async def test_validate_schema_passes_for_table_in_named_schema(pg_engine: AsyncEngine) -> None: + """A correct outbox table in a non-default ``MetaData(schema=...)`` must validate. + + ``_run_validate`` configures Alembic with ``include_schemas=True`` so its reflection + reaches beyond the default schema; without it, a named-schema table is invisible to + ``compare_metadata`` and falsely reads as ``table 'outbox' does not exist``. + """ + schema = f"sch_vs_{uuid.uuid4().hex[:8]}" + metadata = MetaData(schema=schema) + table = make_outbox_table(metadata, table_name="outbox") + async with pg_engine.begin() as conn: + await conn.execute(text(f'CREATE SCHEMA "{schema}"')) + await conn.run_sync(metadata.create_all) + try: + client = OutboxClient(pg_engine, table) + await client.validate_schema() # must NOT raise + finally: + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.drop_all) + await conn.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + + async def test_validate_schema_fails_for_missing_table(pg_engine) -> None: metadata = MetaData() table = make_outbox_table(metadata, table_name="does_not_exist_xyz") @@ -2176,19 +2197,14 @@ async def test_validate_schema_autovacuum_false_does_not_check(pg_engine: AsyncE await client.validate_schema() # default also must NOT raise -async def test_autovacuum_findings_ok_when_applied_in_named_schema(pg_engine: AsyncEngine) -> None: - """The DDL helper's ``schema=`` must target the same table the schema-aware reloptions query reads. - - Before the fix, ``outbox_autovacuum_ddl`` had no ``schema`` param and always rendered an - unqualified ``ALTER TABLE outbox ...``, which resolves via search_path -- silently missing - (or mistargeting) a table that lives in a named schema, while the reloptions query (already - schema-aware via ``COALESCE(:schema, current_schema())``) kept reporting the real table as - untuned. +async def test_validate_schema_autovacuum_ok_when_applied_in_named_schema(pg_engine: AsyncEngine) -> None: + """Full round-trip: ``validate_schema(check_autovacuum=True)`` on a named-schema table does not raise. - NB: this exercises ``_RELOPTIONS_QUERY`` + ``autovacuum_findings`` directly rather than through - ``OutboxClient.validate_schema(check_autovacuum=True)`` — the latter's Alembic-diff schema - probe has a pre-existing, unrelated limitation with non-default Postgres schemas (see the - autovacuum-refactor report), which is orthogonal to the autovacuum check itself. + The DDL helper's ``schema=`` must target the same table the schema-aware reloptions query + reads: ``outbox_autovacuum_ddl(name, schema=table.schema)`` applies the reloptions, and the + check (schema-aware via ``COALESCE(:schema, current_schema())``, plus the schema-reflecting + ``validate_schema`` diff) confirms them — neither the schema probe nor the autovacuum probe + falsely fires for a correct table in a non-default ``MetaData(schema=...)``. """ schema = f"sch_av_{uuid.uuid4().hex[:8]}" metadata = MetaData(schema=schema) @@ -2199,11 +2215,8 @@ async def test_autovacuum_findings_ok_when_applied_in_named_schema(pg_engine: As try: async with pg_engine.begin() as conn: await conn.execute(text(outbox_autovacuum_ddl(table.name, schema=table.schema))) - async with pg_engine.connect() as conn: - reloptions = ( - await conn.execute(_RELOPTIONS_QUERY, {"table": table.name, "schema": table.schema}) - ).scalar_one_or_none() - assert autovacuum_findings(table.name, reloptions) == [] + client = OutboxClient(pg_engine, table) + await client.validate_schema(check_autovacuum=True) # must NOT raise finally: async with pg_engine.begin() as conn: await conn.run_sync(metadata.drop_all) From 5c69454609e909974a0fe0be3f9e24fc281b04cf Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 23:27:09 +0300 Subject: [PATCH 9/9] fix(schema): normalize explicitly-named default schema in validate_schema MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With include_schemas=True, Alembic reports the connection's default schema to the include_name hook as None. The prior `name == table.schema` check therefore excluded a table that explicitly names the default schema (MetaData(schema="public"), or a named schema on the connection's search_path): the table never reflected and a CORRECT table falsely raised "table 'outbox' does not exist" — a false positive that would hard-fail a correct deployment's startup gate. Normalize table.schema against connection.dialect.default_schema_name: when the table names the default schema, treat it as None so it matches Alembic's None. Non-default named schemas are unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- architecture/schema.md | 10 +++++++--- faststream_outbox/client.py | 16 +++++++++++----- tests/test_integration.py | 21 +++++++++++++++++++++ 3 files changed, 39 insertions(+), 8 deletions(-) diff --git a/architecture/schema.md b/architecture/schema.md index f786513..70f8581 100644 --- a/architecture/schema.md +++ b/architecture/schema.md @@ -75,9 +75,13 @@ composition lives in `_compose_schema_mismatch_message` (`client.py`), gated on The alembic diff runs with `include_schemas=True` so a table in a non-default `MetaData(schema=...)` is reflected and compared (without it, `compare_metadata` only sees the default schema and a named-schema table falsely reads as "table does -not exist"). `_include_name` narrows schema reflection to the target schema -(`name == table.schema`, where the default schema is reported as `None`) so -unrelated schemas never surface as false drift. +not exist"). `_include_name` narrows schema reflection to the target schema so +unrelated schemas never surface as false drift. Because Alembic reports the +connection's default schema to the hook as `None`, `table.schema` is first +normalized against `connection.dialect.default_schema_name` — a table that +explicitly names the default schema (`MetaData(schema="public")`, or a named schema +that is on the connection's `search_path`) becomes `None` so it still matches, +avoiding a false "table does not exist" on a correct table. Alembic is optional (`faststream-outbox[validate]`); without it `validate_schema()` raises `ImportError`, but every other path works. diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index cffd6bb..ec33555 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -701,14 +701,20 @@ def _run_validate( canonical_metadata = MetaData(schema=table.schema) canonical_factory(canonical_metadata, table.name) + # Alembic reports the connection's DEFAULT schema to ``include_name`` as ``None`` (with + # ``include_schemas=True``). So a table that explicitly names the default schema + # (``MetaData(schema="public")``, or a named schema that happens to be on the + # search_path) must be normalized to ``None`` before comparing, else ``name == "public"`` + # never matches Alembic's ``None`` and a CORRECT table falsely reads as "does not exist". + default_schema_name = connection.dialect.default_schema_name + target_schema = None if table.schema == default_schema_name else table.schema + def _include_name(name: str | None, type_: str, parent_names: "Mapping[str, str | None]") -> bool: # ``include_schemas=True`` makes Alembic enumerate EVERY schema and call this with - # ``type_ == "schema"`` per schema — restrict to the target schema so unrelated - # schemas' tables never reflect into the diff (which would surface as false drift). - # Alembic reports the default schema as ``None`` here, so ``name == table.schema`` - # matches None-vs-None for a default-schema table and the literal name otherwise. + # ``type_ == "schema"`` per schema — restrict to the (normalized) target schema so + # unrelated schemas' tables never reflect into the diff (which would be false drift). if type_ == "schema": - return name == table.schema + return name == target_schema if type_ == "table": return name == table.name return parent_names.get("table_name") == table.name diff --git a/tests/test_integration.py b/tests/test_integration.py index 9275c95..8606159 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -128,6 +128,27 @@ async def test_validate_schema_passes_for_table_in_named_schema(pg_engine: Async await conn.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) +async def test_validate_schema_passes_for_explicitly_named_default_schema(pg_engine: AsyncEngine) -> None: + """A table whose ``MetaData(schema=...)`` explicitly names the connection's DEFAULT schema validates. + + With ``include_schemas=True`` Alembic reports the default schema to ``include_name`` as + ``None``, so a naive ``name == table.schema`` excludes a table declared with the literal + default-schema name (e.g. ``MetaData(schema="public")``) — the table never reflects and a + CORRECT table falsely raises "table 'outbox' does not exist". ``_run_validate`` normalizes + the explicitly-named default schema to ``None`` before comparing. + """ + metadata = MetaData(schema="public") + table = make_outbox_table(metadata, table_name=f"outbox_pub_{uuid.uuid4().hex[:8]}") + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.create_all) + try: + client = OutboxClient(pg_engine, table) + await client.validate_schema() # must NOT raise + finally: + async with pg_engine.begin() as conn: + await conn.run_sync(metadata.drop_all) + + async def test_validate_schema_fails_for_missing_table(pg_engine) -> None: metadata = MetaData() table = make_outbox_table(metadata, table_name="does_not_exist_xyz")