fix: validate_schema() honors ck naming_convention on lease CHECK

architecture/dlq.md

@@ -35,3 +35,5 @@ There is no built-in retention/pruning. Operators are responsible for archival 
 `validate_schema()` delegates to `alembic.autogenerate.compare_metadata` against a throwaway `MetaData` populated by `make_outbox_table(...)` — so the canonical `Table` is the single source of truth and the validator never duplicates the schema declaration. It only flags **missing** schema (`add_*` / `modify_*` ops); `remove_*` ops are intentionally ignored so users may attach extras (audit columns, their own indexes). Alembic is an **optional dependency** (`faststream-outbox[validate]`); without it, `validate_schema()` raises `ImportError`, but every other code path works (the import lives at the top of `client.py` inside a try/except, with module-level sentinels `_alembic_compare_metadata` / `_AlembicMigrationContext` set to `None` on failure).
 
 Alembic's diff is **blind to three things the producer's `ON CONFLICT` arbiter and the lease invariant depend on**, so `validate_schema()` runs extra `pg_catalog` probes alongside it: the partial-index **WHERE predicates** (alembic ignores `postgresql_where`), the **uniqueness** of `timer_id_uq` (`pg_index.indisunique` — a same-named non-unique index passes the predicate check yet breaks `ON CONFLICT` at publish time), and the **`<table>_lease_ck` CHECK** definition (alembic has no check-constraint comparator). Each surfaces a drifted/non-partial/non-unique index or a missing/altered CHECK that the diff alone would miss.
+
+The CHECK probe must not hard-code the `<table>_lease_ck` name: a `MetaData` carrying a SQLAlchemy `ck` `naming_convention` re-templates the explicitly-named `CheckConstraint` (the name becomes the `%(constraint_name)s` token, so the live constraint is e.g. `ck_<table>_<table>_lease_ck`). `_validate_check_constraints_sync` therefore reads the expected name **off the `Table` object** (`_resolve_check_constraint_name` identifies the lease constraint by its normalized predicate and returns its convention-resolved `.name`), falling back to the literal `<table>_lease_ck` only when the table carries no matching constraint. Without this, every deployment using the SQLAlchemy/Alembic-recommended convention falsely fails validation with "missing CHECK constraint". The explicitly-named indexes are **not** affected — the `ix`/`uq` convention keys only re-template auto-named indexes.

docs/operations/alembic.md

@@ -216,9 +216,34 @@ Substitute the columns / `unique` / predicate from the table above for
 `outbox_pending_idx` and `outbox_lease_idx`.
 
 The recipes pass literal names (`'outbox_lease_ck'`, `'outbox_timer_id_uq'`) —
-the exact names the package emits. If your `MetaData` carries a SQLAlchemy
-`naming_convention`, wrap each name in `op.f('outbox_lease_ck')` so Alembic
-treats it as final and does not re-template it.
+the exact names the package emits **with no `naming_convention`**.
+
+If your `MetaData` carries a SQLAlchemy `naming_convention` with a `ck` key, the
+lease `CheckConstraint` is re-templated — `outbox_lease_ck` becomes e.g.
+`ck_outbox_outbox_lease_ck` (the explicit name fills the `%(constraint_name)s`
+token). `validate_schema()` reads the expected CHECK name **off your `Table`
+object**, so it already honours your convention — but your migration must create
+the constraint under that **same** rendered name, or the probe won't find it.
+
+Introspect the rendered name from your own table and recreate the constraint
+under it (`op.f(...)` passes it through literally):
+
+```python
+from sqlalchemy import CheckConstraint
+
+# `outbox_table` is what `make_outbox_table(your_metadata)` returned.
+ck = next(c for c in outbox_table.constraints if isinstance(c, CheckConstraint))
+print(ck.name)  # the exact name to use, e.g. 'ck_outbox_outbox_lease_ck'
+
+op.create_check_constraint(
+    op.f(ck.name),
+    'outbox',
+    '(acquired_token IS NULL) = (acquired_at IS NULL)',
+)
+```
+
+The explicitly-named indexes (`outbox_pending_idx` etc.) are **not** re-templated
+by the `ix`/`uq` convention keys, so their literal names stay correct as-is.
 
 ## DLQ retention via partition drop { #dlq-retention-via-partition-drop }
 

faststream_outbox/client.py

@@ -23,6 +23,7 @@
 
 from sqlalchemy import (
     ARRAY,
+    CheckConstraint,
     Float,
     MetaData,
     String,
@@ -543,6 +544,26 @@ def _validate_index_predicates_sync(connection: "Connection", table: "Table") ->
 }
 
 
+def _resolve_check_constraint_name(table: "Table", suffix: str, want: str) -> str:
+    """
+    Return the name the lease CHECK constraint actually carries on *table*.
+
+    A ``MetaData`` with a SQLAlchemy ``ck`` ``naming_convention`` re-templates an
+    explicitly-named ``CheckConstraint`` — the package's ``<table>_lease_ck`` becomes e.g.
+    ``ck_<table>_<table>_lease_ck`` — so the live DB name is NOT ``f"{table.name}{suffix}"``.
+    The convention-resolved name is already carried on the constraint object, so identify
+    our constraint by its (normalized) predicate and read its ``.name``. Falls back to the
+    literal ``f"{table.name}{suffix}"`` when the table carries no matching constraint (a
+    reflected/hand-built ``Table``), preserving the original "missing" report.
+    """
+    for constraint in table.constraints:
+        if isinstance(constraint, CheckConstraint) and constraint.name is not None:
+            predicate = _normalize_predicate(str(constraint.sqltext)).removeprefix("check ").strip()
+            if predicate == want:
+                return str(constraint.name)
+    return f"{table.name}{suffix}"
+
+
 def _validate_check_constraints_sync(connection: "Connection", table: "Table") -> list[str]:
     """
     Compare the live CHECK constraint(s) against what the package expects.
@@ -565,7 +586,7 @@ def _validate_check_constraints_sync(connection: "Connection", table: "Table") -
     live = {row["name"]: row["definition"] for row in rows}
     errors: list[str] = []
     for suffix, want in _EXPECTED_CHECK_CONSTRAINTS.items():
-        name = f"{table.name}{suffix}"
+        name = _resolve_check_constraint_name(table, suffix, want)
         if name not in live:
             errors.append(f"missing CHECK constraint {name!r} (expected '{want}')")
             continue