fix(hardening): 0.4.2 payload - bounded migrate lock, stamp --force, asyncpg translation, per-file lock_timeout

CHANGELOG.md

@@ -6,6 +6,35 @@ the project follows [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Fixed
+
+- `migrate` no longer blocks forever on the advisory lock: the chain
+  session's wait is now bounded — `pg_try_advisory_lock` polled every
+  0.5s against a monotonic deadline (default 30s), mirroring `push` —
+  and `migrate` exposes `--advisory-wait` (API: `advisory_wait=`) to
+  tune or zero it. An exhausted budget raises a typed
+  `SqlpushError` instead of hanging on a stuck holder. `stamp` shares
+  the chain session and gets the same bounded default.
+- `stamp` no longer silently refreshes the checksum of a file that was
+  edited after it was applied/stamped: the recorded checksum is read
+  first, and a registered-but-different checksum now refuses with a
+  typed error (first mismatch stops the walk — nothing after it
+  registers) instead of overwriting the registry. `--force` (API:
+  `force=True`) accepts the new content, so the chain's edit-detection
+  integrity survives re-stamping.
+- `migrate` now sets a per-file transaction-scoped `lock_timeout`
+  (default 5s), mirroring `push`: a chain file whose DDL is blocked
+  behind another transaction's lock fails fast with a typed error
+  instead of queuing indefinitely. Tunable via `--lock-timeout`
+  (API: `lock_timeout=`; negative values are rejected up front).
+- The sync facade (`ensure_schema` / `migrate` / `stamp` — everything
+  resolved from a DSN or `AsyncEngine`) now accepts `postgresql+asyncpg`
+  URLs by translating them to the `postgresql+psycopg` driver
+  (host, database, credentials and query options preserved) instead of
+  failing on the async-only driver at connect time. asyncpg is never
+  required to be installed in the sqlpush process; plain psycopg
+  targets are untouched.
+
 ## [0.4.1] - 2026-09-02
 
 ### Added

src/sqlpush/api.py

@@ -139,7 +139,12 @@ def revision(
     risk = max((op.risk for op in p.operations), key=lambda r: RISK_RANK[r])
     ops = [(f"[{op.risk.name}] {op.type} {op.table or '?'}", op.sql) for op in p.operations]
     out = Path(out_dir)
-    out.mkdir(parents=True, exist_ok=True)
+    try:
+        out.mkdir(parents=True, exist_ok=True)
+    except OSError as exc:
+        # same typing as the write below: only SqlpushError subclasses
+        # escape the API surface (e.g. out_dir under a file)
+        raise SqlpushError(f"cannot create migrations directory {out}: {exc}") from exc
     rev_id = next_revision_id(out)
     slug = re.sub(r"[^a-z0-9_]+", "_", (message or "migration").lower())[:40]
     path = out / f"{rev_id}_{slug}.sql"
@@ -163,16 +168,32 @@ def revision(
     return path
 
 
-def migrate(target, *, chain_dir="migrations/versions", allow_destructive=False) -> MigrateReport:
+def migrate(
+    target,
+    *,
+    chain_dir="migrations/versions",
+    allow_destructive=False,
+    advisory_wait=30.0,
+    lock_timeout=5.0,
+) -> MigrateReport:
     """Replay annotated-SQL chain files with gates + same-txn bookkeeping.
 
     ``target`` is a DSN string, sync ``Engine`` or ``AsyncEngine`` (resolved
-    via ``_sync_engine_from``; engines created here are disposed). See
+    via ``_sync_engine_from``; engines created here are disposed).
+    ``advisory_wait`` bounds the advisory-lock wait and ``lock_timeout``
+    bounds each per-file transaction's lock wait (seconds; 0 = fail
+    immediately — same contract as ``push``). See
     ``chain.migrate.run_migrate`` for the execution contract.
     """
     engine, dispose = _sync_engine_from(target)
     try:
-        return run_migrate(engine, chain_dir=chain_dir, allow_destructive=allow_destructive)
+        return run_migrate(
+            engine,
+            chain_dir=chain_dir,
+            allow_destructive=allow_destructive,
+            advisory_wait=advisory_wait,
+            lock_timeout=lock_timeout,
+        )
     except SQLAlchemyError as exc:
         # MigrationFileError/SqlpushError (typed) pass through untouched
         _raise_typed(exc)
@@ -181,16 +202,19 @@ def migrate(target, *, chain_dir="migrations/versions", allow_destructive=False)
             engine.dispose()
 
 
-def stamp(target, *, chain_dir="migrations/versions") -> MigrateReport:
+def stamp(target, *, chain_dir="migrations/versions", force: bool = False) -> MigrateReport:
     """Bootstrap: register chain files as applied WITHOUT executing SQL.
 
-    For adopting a DB whose schema already reflects the chain. ``target``
-    resolution and error typing match :func:`migrate`. See
-    ``chain.migrate.run_stamp`` for the report convention.
+    For adopting a DB whose schema already reflects the chain. A file
+    already registered with a different checksum (edited after apply)
+    is refused unless ``force`` is set — edit detection survives
+    re-stamping. ``target`` resolution and error typing match
+    :func:`migrate`. See ``chain.migrate.run_stamp`` for the report
+    convention.
     """
     engine, dispose = _sync_engine_from(target)
     try:
-        return run_stamp(engine, chain_dir=chain_dir)
+        return run_stamp(engine, chain_dir=chain_dir, force=force)
     except SQLAlchemyError as exc:
         # MigrationFileError/SqlpushError (typed) pass through untouched
         _raise_typed(exc)
@@ -199,18 +223,35 @@ def stamp(target, *, chain_dir="migrations/versions") -> MigrateReport:
             engine.dispose()
 
 
+def _translate_asyncpg(url):
+    """Map a ``postgresql+asyncpg`` URL onto the sync psycopg driver.
+
+    An asyncpg URL cannot back a SYNC engine (the dialect is
+    async-only), but psycopg is a runtime dependency — the translated
+    engine actually connects. asyncpg itself is never required in this
+    process. Everything but the driver (host/db/credentials/query
+    options) is preserved via the URL API, no string surgery.
+    """
+    if url.drivername == "postgresql+asyncpg":
+        return url.set(drivername="postgresql+psycopg")
+    return url
+
+
 def _sync_engine_from(target):
     if isinstance(target, AsyncEngine):
         from sqlalchemy import create_engine
         from sqlalchemy.pool import NullPool
 
-        dsn = target.url.render_as_string(hide_password=False)
-        return create_engine(dsn, poolclass=NullPool), True
+        dsn = _translate_asyncpg(target.url)
+        return create_engine(dsn.render_as_string(hide_password=False), poolclass=NullPool), True
     if isinstance(target, str):
         from sqlalchemy import create_engine
+        from sqlalchemy.engine import make_url
         from sqlalchemy.pool import NullPool
 
-        return create_engine(target, poolclass=NullPool), True
+        # make_url raises the same ArgumentError a bad DSN would raise
+        # inside create_engine — error typing downstream is unchanged
+        return create_engine(_translate_asyncpg(make_url(target)), poolclass=NullPool), True
     return target, False
 
 

src/sqlpush/chain/migrate.py

@@ -11,13 +11,15 @@
 
 Fail-loud ordering: any blocked file (parse error, checksum mismatch,
 destructive gate, SQL failure) stops the chain — nothing later runs (R4).
-A hung migrate (stuck waiting on the advisory lock) is diagnosed via
-pg_locks / pg_stat_activity.
+The advisory lock wait is BOUNDED (--advisory-wait, default 30s): an
+exhausted budget raises a typed error instead of hanging on a stuck
+holder — diagnose the holder via pg_locks / pg_stat_activity.
 """
 
 from __future__ import annotations
 
 import contextlib
+import time
 from collections.abc import Iterator
 from pathlib import Path
 
@@ -26,7 +28,7 @@
 
 from sqlpush.apply.executor import advisory_key
 from sqlpush.chain.format import MigrationFileError, checksum, parse_migration_file
-from sqlpush.types import MigrateReport, RiskClass
+from sqlpush.types import MigrateReport, RiskClass, SqlpushError
 
 _VERSIONS_DDL = (
     "CREATE TABLE IF NOT EXISTS public.sqlpush_versions ("
@@ -45,18 +47,33 @@ def _chain_files(chain_dir: str | Path) -> list[Path]:
 
 
 @contextlib.contextmanager
-def _chain_session(engine: Engine) -> Iterator[Connection]:
+def _chain_session(engine: Engine, *, advisory_wait: float = 30.0) -> Iterator[Connection]:
     """Session-scoped advisory lock + versions table, shared by every verb.
 
     Same key derivation as push (fnv1a_32(b"sqlpush") ^ db oid): serializes
     concurrent chain workers and excludes push on the same database. The
-    txn opened by the key query is committed right away — session advisory
-    locks survive COMMIT/ROLLBACK, and an idle-in-transaction session is
-    exposed to idle_in_transaction_session_timeout (executor.py note).
+    wait is BOUNDED, mirroring executor.with_advisory_lock:
+    ``pg_try_advisory_lock`` polled every 0.5 s against a
+    ``time.monotonic()`` deadline; an exhausted budget raises
+    :class:`SqlpushError` — a blocking ``pg_advisory_lock`` would hang
+    forever on a stuck holder. The txn opened by the key/probe queries
+    is committed right away — session advisory locks survive
+    COMMIT/ROLLBACK, and an idle-in-transaction session is exposed to
+    idle_in_transaction_session_timeout (executor.py note).
     """
+    if advisory_wait < 0:
+        raise SqlpushError(f"advisory_wait must be >= 0, got {advisory_wait}")
     with engine.connect() as conn:
         key = advisory_key(conn)
-        conn.execute(text("SELECT pg_advisory_lock(:k)"), {"k": key})
+        deadline = time.monotonic() + advisory_wait
+        locked = conn.execute(text("SELECT pg_try_advisory_lock(:k)"), {"k": key}).scalar()
+        while not locked and time.monotonic() < deadline:
+            time.sleep(0.5)
+            locked = conn.execute(text("SELECT pg_try_advisory_lock(:k)"), {"k": key}).scalar()
+        if not locked:
+            raise SqlpushError(
+                f"another sqlpush worker holds the advisory lock after {advisory_wait}s"
+            )
         conn.commit()
         try:
             conn.execute(text(_VERSIONS_DDL))
@@ -73,14 +90,26 @@ def _chain_session(engine: Engine) -> Iterator[Connection]:
                     conn.commit()
 
 
-def run_migrate(engine: Engine, *, chain_dir: str | Path, allow_destructive: bool) -> MigrateReport:
+def run_migrate(
+    engine: Engine,
+    *,
+    chain_dir: str | Path,
+    allow_destructive: bool,
+    advisory_wait: float = 30.0,
+    lock_timeout: float = 5.0,
+) -> MigrateReport:
+    if lock_timeout < 0:
+        # same contract as push (executor.with_advisory_lock): budgets
+        # are typed floats, never user input, and a negative one must
+        # fail before any file or connection work
+        raise SqlpushError(f"lock_timeout must be >= 0, got {lock_timeout}")
     applied: list[str] = []
     skipped: list[str] = []
     blocked: list[str] = []
     notes: list[str] = []
     partial = False
     chain = _chain_files(chain_dir)
-    with _chain_session(engine) as conn:
+    with _chain_session(engine, advisory_wait=advisory_wait) as conn:
         recorded = {
             row[0]: row[1]
             for row in conn.execute(text("SELECT name, sha256 FROM public.sqlpush_versions"))
@@ -107,6 +136,14 @@ def run_migrate(engine: Engine, *, chain_dir: str | Path, allow_destructive: boo
                 break
             try:
                 with conn.begin():
+                    # txn-scoped: SET LOCAL dies with the per-file txn
+                    # (style: push's transactional segment, executor.py).
+                    # NOTE: PostgreSQL does not accept bind parameters for
+                    # SET (utility statement), so the int is inlined;
+                    # lock_timeout is a typed float parameter, not user
+                    # input. A chain file blocked behind another
+                    # transaction's lock fails fast instead of queuing.
+                    conn.execute(text(f"SET LOCAL lock_timeout = {int(lock_timeout * 1000)}"))
                     # whole-file replay: exec_driver_sql bypasses text()'s
                     # bind-param parsing entirely — ":casts" and ":=" in
                     # hand-edited SQL must reach the server verbatim
@@ -130,14 +167,18 @@ def run_migrate(engine: Engine, *, chain_dir: str | Path, allow_destructive: boo
         )
 
 
-def run_stamp(engine: Engine, *, chain_dir: str | Path) -> MigrateReport:
+def run_stamp(engine: Engine, *, chain_dir: str | Path, force: bool = False) -> MigrateReport:
     """Register every parseable chain file WITHOUT executing any SQL.
 
     Bootstrap seam (spec §5 R7): adopt a DB whose schema already reflects
     the chain. Only the header must parse (fail-loud on THAT) — invalid SQL
-    in a body still registers, because stamp never executes anything. Each
-    registration is an idempotent upsert (``ON CONFLICT (name) DO UPDATE``),
-    so re-stamping refreshes checksums instead of failing.
+    in a body still registers, because stamp never executes anything. A
+    file already recorded with a DIFFERENT checksum is refused — it was
+    edited after apply/stamp, and silently refreshing would wipe the
+    edit-detection integrity — unless ``force`` is set; the first mismatch
+    raises and nothing after it registers (strict order, same as migrate).
+    Unrecorded files, unchanged re-stamps and forced re-stamps upsert
+    idempotently (``ON CONFLICT (name) DO UPDATE``).
 
     Report convention: registered files are listed in ``skipped`` (stamp
     never populates ``applied`` and never sets ``partial_failure``); a
@@ -149,6 +190,11 @@ def run_stamp(engine: Engine, *, chain_dir: str | Path) -> MigrateReport:
     notes: list[str] = []
     chain = _chain_files(chain_dir)
     with _chain_session(engine) as conn:
+        recorded = {
+            row[0]: row[1]
+            for row in conn.execute(text("SELECT name, sha256 FROM public.sqlpush_versions"))
+        }
+        conn.commit()
         for f in chain:
             raw = f.read_text()
             try:
@@ -157,6 +203,11 @@ def run_stamp(engine: Engine, *, chain_dir: str | Path) -> MigrateReport:
                 blocked.append(f.name)
                 notes.append(f"{f.name}: {exc}")
                 break  # orden estricto: nada posterior se registra
+            if f.name in recorded and recorded[f.name] != checksum(raw) and not force:
+                raise SqlpushError(
+                    f"{f.name}: checksum mismatch (file edited after apply?); "
+                    "pass force=True/--force to accept the new content"
+                )
             with conn.begin():
                 conn.execute(
                     text(

src/sqlpush/cli.py

@@ -4,7 +4,8 @@
 Exit codes: diff always 0; check 0 clean / 2 drift / 3 destructive drift;
 push 0 applied / 1 destructive blocked / 2 error (incl. partial failure);
 revision 0 written / 1 error (empty drift refuses); migrate 0 clean /
-1 blocked or partial failure.
+1 blocked or partial failure; stamp 0 registered / 1 blocked or refused
+(an edited file without --force raises a typed error: exit 1).
 """
 
 from __future__ import annotations
@@ -250,12 +251,20 @@ def revision(
 def migrate(
     dsn: str | None = typer.Option(None),
     allow_destructive: bool = typer.Option(False, "--allow-destructive"),
+    advisory_wait: float = typer.Option(30.0, "--advisory-wait"),
+    lock_timeout: float = typer.Option(5.0, "--lock-timeout"),
     out_dir: DirOpt = Path("migrations/versions"),
 ):
     """Replay pending migration files (gates + checksum bookkeeping)."""
     engine = _engine(dsn)
     try:
-        report = api.migrate(engine, chain_dir=out_dir, allow_destructive=allow_destructive)
+        report = api.migrate(
+            engine,
+            chain_dir=out_dir,
+            allow_destructive=allow_destructive,
+            advisory_wait=advisory_wait,
+            lock_timeout=lock_timeout,
+        )
     finally:
         engine.dispose()
     typer.echo(
@@ -272,12 +281,13 @@ def migrate(
 @app.command()
 def stamp(
     dsn: str | None = typer.Option(None),
+    force: bool = typer.Option(False, "--force"),
     out_dir: DirOpt = Path("migrations/versions"),
 ):
     """Adopt an existing DB: register chain files without executing SQL."""
     engine = _engine(dsn)
     try:
-        report = api.stamp(engine, chain_dir=out_dir)
+        report = api.stamp(engine, chain_dir=out_dir, force=force)
     finally:
         engine.dispose()
     typer.echo(

src/sqlpush/core/diff.py

@@ -405,7 +405,9 @@ def _dedup_enum_types(ops: list[PlannedOperation]) -> list[PlannedOperation]:
     copy stays embedded in its original add_table render. A metadata
     that defines one type name two DIFFERENT ways keeps both statements
     and still fails loudly at apply time: silent first-win would mask a
-    genuine contradiction.
+    genuine contradiction. Ops whose statements all survive pass
+    through as the ORIGINAL object (identity) — a re-joined copy of
+    unchanged SQL buys nothing.
     """
     seen: set[str] = set()
 
@@ -420,10 +422,12 @@ def _is_type_stmt(norm: str) -> bool:
             continue
         stmts = _split_statements(op.sql)
         kept: list[str] = []
+        dropped = False
         for stmt in stmts:
             norm = " ".join(stmt.split())
             if _is_type_stmt(norm):
                 if norm in seen:
+                    dropped = True
                     continue
                 seen.add(norm)
             if norm:
@@ -433,6 +437,9 @@ def _is_type_stmt(norm: str) -> bool:
             # survives); guard anyway — an empty op.sql is worse than
             # the duplicate it replaced
             continue
+        if not dropped:
+            out.append(op)
+            continue
         out.append(
             PlannedOperation(
                 type=op.type,