fix: resolve the outbox table through search_path in validate_schema

faststream_outbox/autovacuum.py

@@ -60,10 +60,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
-    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.
+    the ``validate_schema(check_autovacuum=True)`` reloptions lookup, which resolves the
+    table the same way. 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}"
@@ -82,8 +82,7 @@ def outbox_autovacuum_ddl(
 
 
 # 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.
+# None when the table has no options set (or does not exist).
 _RELOPTIONS_QUERY = text(
     "SELECT c.reloptions FROM pg_class c "
     "JOIN pg_namespace n ON n.oid = c.relnamespace "

faststream_outbox/schema_validation.py

@@ -66,22 +66,24 @@ async def validate_schema(
     ``False`` never checks it.
     """
     async with engine.connect() as conn:
-        errors = await conn.run_sync(_validate_schema_sync, table)
+        schema = await conn.run_sync(_resolve_schema, table)
+        errors = await conn.run_sync(_validate_schema_sync, table, schema)
         # S2 / lease_ck: these two probes catch drift that `alembic revision
         # --autogenerate` cannot remediate — its index comparator ignores
         # postgresql_where and it has no check-constraint comparator at all.
         # Collect them separately so the raised error can point operators at
         # the hand-written-migration recipe (_AUTOGEN_BLIND_HINT) only when
         # one of them actually fired.
-        blind_errors = await conn.run_sync(_validate_index_predicates_sync, table)
-        blind_errors.extend(await conn.run_sync(_validate_check_constraints_sync, table))
+        blind_errors = await conn.run_sync(_validate_index_predicates_sync, table, schema)
+        blind_errors.extend(await conn.run_sync(_validate_check_constraints_sync, table, schema))
         errors.extend(blind_errors)
         if dlq_table is not None:
-            errors.extend(await conn.run_sync(_validate_dlq_schema_sync, dlq_table))
+            dlq_schema = await conn.run_sync(_resolve_schema, dlq_table)
+            errors.extend(await conn.run_sync(_validate_dlq_schema_sync, dlq_table, dlq_schema))
         autovacuum_errors: list[str] = []
         if check_autovacuum:
             reloptions = (
-                await conn.execute(_RELOPTIONS_QUERY, {"table": table.name, "schema": table.schema})
+                await conn.execute(_RELOPTIONS_QUERY, {"table": table.name, "schema": schema})
             ).scalar_one_or_none()
             autovacuum_errors = autovacuum_findings(table.name, reloptions)
     message_parts: list[str] = []
@@ -93,6 +95,24 @@ async def validate_schema(
         raise RuntimeError("\n\n".join(message_parts))
 
 
+_RESOLVE_SCHEMA_QUERY = text(
+    "SELECT n.nspname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace "
+    "WHERE c.oid = to_regclass(format('%I', CAST(:table AS text)))",
+)
+
+
+def _resolve_schema(connection: "Connection", table: "Table") -> str | None:
+    """Return the schema the app's queries reach *table* in.
+
+    An unqualified table resolves through the whole ``search_path``, not just ``current_schema()``,
+    so ask Postgres where the name lands. ``None`` means it doesn't resolve; the probes then fall
+    back to ``current_schema()`` and report the table missing.
+    """
+    if table.schema is not None:
+        return table.schema
+    return connection.execute(_RESOLVE_SCHEMA_QUERY, {"table": table.name}).scalar_one_or_none()
+
+
 def _normalize_predicate(predicate: str) -> str:
     """Canonicalize a Postgres partial-index predicate for comparison (lowercase, drop parens/space)."""
     return " ".join(predicate.lower().replace("(", " ").replace(")", " ").split())
@@ -123,7 +143,7 @@ def _normalize_predicate(predicate: str) -> str:
 )
 
 
-def _validate_index_predicates_sync(connection: "Connection", table: "Table") -> list[str]:
+def _validate_index_predicates_sync(connection: "Connection", table: "Table", schema: str | None) -> list[str]:
     """Compare the live partial-index WHERE predicates against what the package expects (S2).
 
     Alembic's index diff ignores ``postgresql_where``, so the alembic autogenerate pass
@@ -140,7 +160,7 @@ def _validate_index_predicates_sync(connection: "Connection", table: "Table") ->
     rows = (
         connection.execute(
             _INDEX_PREDICATE_QUERY,
-            {"table": table.name, "schema": table.schema},
+            {"table": table.name, "schema": schema},
         )
         .mappings()
         .all()
@@ -180,7 +200,7 @@ def _validate_index_predicates_sync(connection: "Connection", table: "Table") ->
 }
 
 
-def _validate_check_constraints_sync(connection: "Connection", table: "Table") -> list[str]:
+def _validate_check_constraints_sync(connection: "Connection", table: "Table", schema: str | None) -> list[str]:
     """Verify the live DB carries a CHECK enforcing each invariant the package needs.
 
     Alembic's ``compare_metadata`` registers no check-constraint comparator, so a missing
@@ -200,7 +220,7 @@ def _validate_check_constraints_sync(connection: "Connection", table: "Table") -
     rows = (
         connection.execute(
             _CHECK_CONSTRAINT_QUERY,
-            {"table": table.name, "schema": table.schema},
+            {"table": table.name, "schema": schema},
         )
         .mappings()
         .all()
@@ -241,19 +261,20 @@ def _compose_schema_mismatch_message(errors: list[str], *, has_blind_drift: bool
     return msg
 
 
-def _validate_schema_sync(connection: "Connection", table: "Table") -> list[str]:
+def _validate_schema_sync(connection: "Connection", table: "Table", schema: str | None) -> list[str]:
     """Run the outbox-table validation pass; see :func:`_run_validate` for the diff machinery."""
-    return _run_validate(connection, table, make_outbox_table)
+    return _run_validate(connection, table, schema, make_outbox_table)
 
 
-def _validate_dlq_schema_sync(connection: "Connection", table: "Table") -> list[str]:
+def _validate_dlq_schema_sync(connection: "Connection", table: "Table", schema: str | None) -> list[str]:
     """Run the DLQ-table validation pass; see :func:`_run_validate` for the diff machinery."""
-    return _run_validate(connection, table, make_dlq_table)
+    return _run_validate(connection, table, schema, make_dlq_table)
 
 
 def _run_validate(
     connection: "Connection",
     table: "Table",
+    schema: str | None,
     canonical_factory: "Callable[[MetaData, str], Table]",
 ) -> list[str]:
     """Run Alembic's autogenerate diff against the live DB and surface any "missing schema" drift.
@@ -286,7 +307,7 @@ def _run_validate(
     # user's schema onto the canonical copy so the autogenerate diff compares
     # ``app.outbox`` against ``app.outbox`` rather than the default search_path —
     # matching the schema-qualified DLQ CTE in ``client._build_dlq_cte`` (B10).
-    canonical_metadata = MetaData(schema=table.schema)
+    canonical_metadata = MetaData(schema=schema)
     canonical_factory(canonical_metadata, table.name)
 
     # Alembic reports the connection's DEFAULT schema to ``include_name`` as ``None`` (with
@@ -295,7 +316,7 @@ def _run_validate(
     # 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
+    target_schema = None if schema == default_schema_name else 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

tests/test_integration.py

@@ -167,6 +167,36 @@ async def test_validate_schema_passes_for_explicitly_named_default_schema(pg_eng
             await conn.run_sync(metadata.drop_all)
 
 
+async def test_validate_schema_finds_a_table_through_a_later_search_path_entry(pg_engine: AsyncEngine) -> None:
+    """An unqualified table that resolves through a later ``search_path`` entry validates.
+
+    With ``search_path = <empty schema>, public`` the app reads and writes ``public.<table>``,
+    so validation must look there too, not only in ``current_schema()``.
+    """
+    schema = f"sch_sp_{uuid.uuid4().hex[:8]}"
+    suffix = uuid.uuid4().hex[:8]
+    metadata = MetaData()
+    table = make_outbox_table(metadata, table_name=f"outbox_sp_{suffix}")
+    dlq_table = make_dlq_table(metadata, table_name=f"outbox_sp_dlq_{suffix}")
+    async with pg_engine.begin() as conn:
+        await conn.execute(text(f'CREATE SCHEMA "{schema}"'))
+        await conn.run_sync(metadata.create_all)
+        await conn.execute(text(outbox_autovacuum_ddl(table.name)))
+    dsn = pg_engine.url.render_as_string(hide_password=False)
+    search_path_engine = create_async_engine(
+        dsn,
+        connect_args={"server_settings": {"search_path": f'"{schema}", public'}},
+    )
+    try:
+        client = OutboxClient(search_path_engine, table, dlq_table=dlq_table)
+        await client.validate_schema(check_autovacuum=True)  # must NOT raise
+    finally:
+        await search_path_engine.dispose()
+        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")

tests/test_unit.py

@@ -1411,7 +1411,7 @@ def test_validate_schema_sync_raises_when_alembic_missing() -> None:
         patch("faststream_outbox.schema_validation.is_alembic_installed", new=False),
         pytest.raises(ImportError, match=r"pip install 'faststream-outbox\[validate\]'"),
     ):
-        _validate_schema_sync(MagicMock(), t)
+        _validate_schema_sync(MagicMock(), t, None)
 
 
 def test_compose_schema_mismatch_message_appends_hint_on_blind_drift() -> None:
@@ -1461,7 +1461,7 @@ def test_validate_check_constraints_passes_under_convention_resolved_name() -> N
     connection = _mock_check_constraint_connection(
         [{"name": "ck_outbox_faststream_outbox_faststream_lease_ck", "definition": _LEASE_CK_DEFINITION}],
     )
-    assert _validate_check_constraints_sync(connection, table) == []
+    assert _validate_check_constraints_sync(connection, table, None) == []
 
 
 def test_validate_check_constraints_passes_under_literal_name() -> None:
@@ -1470,14 +1470,14 @@ def test_validate_check_constraints_passes_under_literal_name() -> None:
     connection = _mock_check_constraint_connection(
         [{"name": "outbox_faststream_lease_ck", "definition": _LEASE_CK_DEFINITION}],
     )
-    assert _validate_check_constraints_sync(connection, table) == []
+    assert _validate_check_constraints_sync(connection, table, None) == []
 
 
 def test_validate_check_constraints_missing_describes_predicate() -> None:
     """When no live CHECK enforces the predicate, the error names the invariant, not a guessed name."""
     table = make_outbox_table(MetaData(), table_name="outbox_faststream")
     connection = _mock_check_constraint_connection([])  # constraint absent from the live DB
-    errors = _validate_check_constraints_sync(connection, table)
+    errors = _validate_check_constraints_sync(connection, table, None)
     assert errors == [
         (
             f"missing CHECK constraint enforcing '{_LEASE_CK_PREDICATE}' "
@@ -1493,7 +1493,7 @@ def test_validate_check_constraints_drifted_predicate_reported_as_missing() -> N
     connection = _mock_check_constraint_connection(
         [{"name": "outbox_faststream_lease_ck", "definition": drifted}],
     )
-    errors = _validate_check_constraints_sync(connection, table)
+    errors = _validate_check_constraints_sync(connection, table, None)
     assert errors == [
         (
             f"missing CHECK constraint enforcing '{_LEASE_CK_PREDICATE}' "