feat: 0.5.0 payload - CONCURRENTLY by default on both surfaces + statement_timeout

CHANGELOG.md

@@ -6,6 +6,80 @@ the project follows [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Added
+
+- `CREATE INDEX` / `CREATE UNIQUE INDEX` on EXISTING tables now render
+  `CONCURRENTLY` by default in `plan` / `push` / `revision` output,
+  including `revision`-generated migration SQL (API: `concurrently=True`
+  on `plan` / `push` / `revision`; CLI opt-out `--no-concurrently` on
+  `push` / `revision`): the plain form takes a SHARE lock that blocks
+  all writes for the whole build, while CONCURRENTLY builds without
+  blocking writers. Indexes on tables created in the same plan stay
+  plain — a brand-new table has no concurrent writers, and the create
+  stays inside the plan's atomic transaction. Concurrently-rendered ops
+  run on push's autocommit segment (a failure there is a recorded
+  partial failure, not a rollback) and are still classified `risky`.
+  The knob threads through the advisory-lock winner path: the lock
+  winner's re-plan renders exactly what the caller requested.
+- `push` can bound each statement's runtime with a `statement_timeout`
+  (API: `push(..., statement_timeout=...)`, seconds, `None` = set
+  nothing; CLI: `push --statement-timeout`): the transactional segment
+  applies it as `SET LOCAL statement_timeout`, the concurrent
+  (autocommit) segment as a session `statement_timeout` that is `RESET`
+  before the connection returns to the pool — a pooled borrower never
+  inherits the budget. It threads through the advisory-lock winner path
+  with `concurrently`, and negative values are rejected up front with
+  a typed `SqlpushError`.
+- The concurrent (autocommit) segment of `push` now runs under a
+  session `lock_timeout` with the same value as the transactional
+  segment's: previously `CREATE INDEX CONCURRENTLY` had no lock budget
+  at all and could queue indefinitely behind another transaction's
+  table lock. The session GUC is `RESET` (and the connection
+  invalidated if the reset fails) so it never leaks to the pool's next
+  borrower.
+- Every operation in the versioned plan JSON (`diff --json`,
+  `check --json`, `Plan.to_json_dict`) now carries a `"concurrent"`
+  boolean — additive to the v1 contract, no existing key or value
+  changed. It reports whether the operation's SQL was rendered with
+  `CREATE INDEX CONCURRENTLY` (see the rendering change above).
+- `migrate` now replays CONCURRENTLY-containing chain files per-op on
+  the op-label delimiters: the plain segment runs first in one
+  transaction (chain files can create→index within one file), then
+  concurrent ops run statement-by-statement on a dedicated autocommit
+  connection (session `lock_timeout` matching the per-file txn, RESET +
+  close when the walk ends), and the versions row is written only after
+  every concurrent op succeeds — a failed concurrent op blocks the file
+  (partial failure, no versions row, strict-order stop, the already
+  committed plain segment reported honestly in the notes). Concurrent-
+  free files keep the exact 0.4.2 whole-text single-transaction replay,
+  so existing chains are byte-identically unaffected, and
+  `revision`-generated files round-trip through the chain.
+- `migrate` gains `--statement-timeout` (API:
+  `migrate(..., statement_timeout=...)`, seconds, unset by default):
+  applied as `SET LOCAL` in every per-file transaction and as a session
+  `statement_timeout` on the CONCURRENTLY autocommit connection (RESET
+  before it closes). Negative values are rejected up front with a
+  typed `SqlpushError` — same contract as `push`.
+
+### Known limitations
+
+- Generated chain files replay per-op on their `-- op N [label]`
+  delimiters; hand-edits bypass that tokenization (pinned chain spec
+  §7). A label-less body containing CONCURRENTLY routes whole to the
+  autocommit lane statement-by-statement, and lines starting `--` are
+  stripped from per-op parsing — dollar-quoted bodies containing `--`
+  lines are only safe in the whole-text fast path.
+- A failed or timed-out `CREATE INDEX CONCURRENTLY` leaves an INVALID
+  index behind: recover with `DROP INDEX CONCURRENTLY <name>` and
+  re-push (optionally `--no-concurrently`). `statement_timeout` applies
+  inside index builds, and an aborted build still leaves the INVALID
+  index.
+- Mixed-file crash window: plain segment committed + concurrent segment
+  applied + no versions row → a re-run fails loud on the existing
+  objects rather than silently re-applying.
+- Re-running against an already-existing index fails loud: the rendered
+  `CREATE INDEX CONCURRENTLY` carries no `IF NOT EXISTS`.
+
 ## [0.4.2] - 2026-09-02
 
 ### Fixed

README.md

@@ -104,8 +104,11 @@ repeated `--schema` / `--exclude` options.
 | `push` | applied | destructive blocked | error (incl. partial failure) | |
 
 `push --safe-only` runs only safe operations and skips the rest
-informationally (exit `0`). A failed `CREATE INDEX CONCURRENTLY` marks the
-run as partial failure (exit `2`) instead of silently half-applying.
+informationally (exit `0`). Indexes on existing tables build
+`CONCURRENTLY` by default (opt out with `--no-concurrently`); a failed
+`CREATE INDEX CONCURRENTLY` marks the run as partial failure (exit `2`)
+instead of silently half-applying, and leaves an INVALID index — drop it
+(`DROP INDEX CONCURRENTLY`) and re-push.
 
 ## FastAPI / SQLModel: replace `create_all`
 
@@ -140,9 +143,11 @@ flowchart LR
   internals included) before reflection even starts.
 - **Classifier** maps each operation to a risk class; unknown operations
   are `risky`, never silently safe.
-- **Executor** splits the plan: `CONCURRENTLY` statements run one-per-
-  transaction on autocommit, everything else applies in a single atomic
-  transaction with a bounded `lock_timeout`.
+- **Executor** splits the plan: existing-table indexes render
+  `CREATE INDEX CONCURRENTLY` and run one-per-transaction on autocommit
+  (`--no-concurrently` opts out; indexes on tables the same plan creates
+  stay in the atomic transaction), everything else applies in a single
+  atomic transaction with a bounded `lock_timeout`.
 - **Typed errors**: only `SqlpushError` / `ConnectFailed` /
   `MetadataImportError` escape the API, never raw driver exceptions.
 
@@ -172,12 +177,11 @@ deprecated)? There is a [migration guide](docs/migrating-from-migra.md).
 - The advisory-lock key derives from the database OID: two DSN spellings
   of the same database contend for the same lock.
 - `--json` output is a versioned contract (`"version": 1`) meant for
-  tooling; additive changes only within a version.
+  tooling; additive changes only within a version (operations now carry
+  a `concurrent` boolean).
 
 ## Roadmap (0.1.x)
 
-- `CREATE INDEX CONCURRENTLY` by default for indexes on existing tables
-- asyncpg DSN translation in `ensure_schema(AsyncEngine)`
 - jsonschema-validated `--json` output
 
 ## License

src/sqlpush/api.py

@@ -48,8 +48,10 @@ def _raise_typed(exc: SQLAlchemyError) -> NoReturn:
     raise SqlpushError(f"database error: {exc}") from exc
 
 
-def _build_plan(metadata: MetaData, engine: Engine, schemas, exclude) -> Plan:
-    p = _engine.plan(metadata, engine, schemas=schemas, exclude=exclude)
+def _build_plan(
+    metadata: MetaData, engine: Engine, schemas, exclude, concurrently: bool = True
+) -> Plan:
+    p = _engine.plan(metadata, engine, schemas=schemas, exclude=exclude, concurrently=concurrently)
     return Plan(operations=p.operations + tuple(hypertable_operations(metadata, engine)))
 
 
@@ -65,13 +67,13 @@ class _PlannerWithDirectives(DiffEngine):
     def __init__(self, engine: DiffEngine) -> None:
         self._engine = engine
 
-    def plan(self, metadata, engine, *, schemas=None, exclude=()) -> Plan:
-        return _build_plan(metadata, engine, schemas, exclude)
+    def plan(self, metadata, engine, *, schemas=None, exclude=(), concurrently=True) -> Plan:
+        return _build_plan(metadata, engine, schemas, exclude, concurrently)
 
 
-def plan(metadata, engine, *, schemas=None, exclude=()) -> Plan:
+def plan(metadata, engine, *, schemas=None, exclude=(), concurrently: bool = True) -> Plan:
     try:
-        return _build_plan(metadata, engine, schemas, exclude)
+        return _build_plan(metadata, engine, schemas, exclude, concurrently)
     except SQLAlchemyError as exc:
         _raise_typed(exc)
 
@@ -84,9 +86,11 @@ def push(
     allow_destructive=False,
     lock=True,
     lock_timeout=5.0,
+    statement_timeout=None,
     advisory_wait=30.0,
     schemas=None,
     exclude=(),
+    concurrently=True,
 ) -> Report:
     try:
         if lock:
@@ -100,14 +104,21 @@ def push(
                 reverify=_PlannerWithDirectives(_engine),
                 schemas=schemas,
                 exclude=exclude,
+                # THREADING (pinned): the lock winner re-plans via
+                # reverify.plan(...) — both knobs must reach it and the
+                # final apply_plan, or the winner renders differently
+                # than requested
+                concurrently=concurrently,
+                statement_timeout=statement_timeout,
             )
-        p = _build_plan(metadata, engine, schemas, exclude)
+        p = _build_plan(metadata, engine, schemas, exclude, concurrently)
         return apply_plan(
             engine,
             p,
             allow_destructive=allow_destructive,
             safe_only=safe_only,
             lock_timeout=lock_timeout,
+            statement_timeout=statement_timeout,
         )
     except SQLAlchemyError as exc:
         _raise_typed(exc)
@@ -126,14 +137,23 @@ def check(metadata, engine, *, schemas=None, exclude=()) -> CheckResult:
 
 
 def revision(
-    metadata, ref_engine, *, out_dir="migrations/versions", message=None, schemas=None, exclude=()
+    metadata,
+    ref_engine,
+    *,
+    out_dir="migrations/versions",
+    message=None,
+    schemas=None,
+    exclude=(),
+    concurrently=True,
 ) -> Path:
     """Generate the next annotated-SQL migration file from models-vs-ref drift.
 
     The reference DB must sit at the chain head (caller-provided — sqlpush
     stays docker-free). Empty drift refuses loudly: no empty files.
+    ``concurrently`` follows :func:`plan` (default True: existing-table
+    indexes render CONCURRENTLY in the generated file).
     """
-    p = plan(metadata, ref_engine, schemas=schemas, exclude=exclude)
+    p = plan(metadata, ref_engine, schemas=schemas, exclude=exclude, concurrently=concurrently)
     if not p.operations:
         raise SqlpushError("no drift between models and reference DB — nothing to revise")
     risk = max((op.risk for op in p.operations), key=lambda r: RISK_RANK[r])
@@ -175,14 +195,18 @@ def migrate(
     allow_destructive=False,
     advisory_wait=30.0,
     lock_timeout=5.0,
+    statement_timeout=None,
 ) -> 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).
     ``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
+    immediately — same contract as ``push``). ``statement_timeout``
+    (seconds, ``None`` = set nothing) bounds each statement's runtime —
+    ``SET LOCAL`` in per-file transactions, session-level on the
+    CONCURRENTLY autocommit connection. See
     ``chain.migrate.run_migrate`` for the execution contract.
     """
     engine, dispose = _sync_engine_from(target)
@@ -193,6 +217,7 @@ def migrate(
             allow_destructive=allow_destructive,
             advisory_wait=advisory_wait,
             lock_timeout=lock_timeout,
+            statement_timeout=statement_timeout,
         )
     except SQLAlchemyError as exc:
         # MigrationFileError/SqlpushError (typed) pass through untouched

src/sqlpush/apply/executor.py

@@ -3,6 +3,7 @@
 
 import contextlib
 import time
+from collections.abc import Iterator
 from typing import TYPE_CHECKING
 
 from sqlalchemy import text
@@ -26,7 +27,37 @@
 
 
 def _is_concurrent(op: PlannedOperation) -> bool:
-    return "CONCURRENTLY" in op.sql.upper()
+    # Union (A5): the `concurrent` flag is authoritative for generated
+    # plans (set by the diff's injection); the SQL substring keeps
+    # hand-built Plans splitting to the autocommit segment.
+    return op.concurrent or "CONCURRENTLY" in op.sql.upper()
+
+
+@contextlib.contextmanager
+def _session_gucs(conn, **gucs: str | int) -> Iterator[None]:
+    """Session-level ``SET``/``RESET`` with pooled-connection hygiene.
+
+    The autocommit (CONCURRENTLY) segment cannot use ``SET LOCAL`` —
+    there is no surrounding transaction — so its GUCs are session-level,
+    and a session ``SET`` survives segment end (0.4.2 lesson). RESET runs
+    per GUC in ``finally``; a reset failure is suppressed (it must never
+    mask the segment's own outcome) and the connection is invalidated so
+    the pool discards it instead of handing the next borrower a session
+    still carrying the shrunken budget.
+    """
+    try:
+        for name, value in gucs.items():
+            # names come from sqlpush code, values are ints inlined in
+            # the same style as the txn segment's SET LOCAL (PostgreSQL
+            # does not accept bind parameters for SET)
+            conn.execute(text(f"SET {name} = '{value}'"))
+        yield
+    finally:
+        for name in gucs:
+            try:
+                conn.execute(text(f"RESET {name}"))
+            except Exception:  # noqa: BLE001  # never mask the segment's outcome
+                conn.invalidate()
 
 
 def apply_plan(
@@ -36,8 +67,11 @@ def apply_plan(
     allow_destructive: bool = False,
     safe_only: bool = False,
     lock_timeout: float = 5.0,
+    statement_timeout: float | None = None,
 ) -> Report:
     start = time.monotonic()
+    if statement_timeout is not None and statement_timeout < 0:
+        raise SqlpushError(f"statement_timeout must be >= 0, got {statement_timeout}")
 
     blocked = tuple(op for op in plan.operations if op.risk is RiskClass.DESTRUCTIVE)
     if blocked and not allow_destructive:
@@ -69,23 +103,35 @@ def apply_plan(
     concurrent = [op for op in runnable if _is_concurrent(op)]
     if concurrent:
         with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
-            for op in concurrent:
-                try:
-                    conn.execute(text(op.sql))
-                    applied.append(AppliedOperation(op.type, "applied"))
-                except Exception:  # noqa: BLE001  # concurrent ops record any DB failure and continue
-                    applied.append(AppliedOperation(op.type, "failed"))
-                    partial_failure = True
+            # A7: the concurrent segment gets the SAME lock budget as the
+            # txn segment (session-level: autocommit has no surrounding
+            # transaction for SET LOCAL), plus the optional statement
+            # budget; both RESET before the connection returns to the pool
+            session_gucs: dict[str, int] = {"lock_timeout": int(lock_timeout * 1000)}
+            if statement_timeout is not None:
+                session_gucs["statement_timeout"] = int(statement_timeout * 1000)
+            with _session_gucs(conn, **session_gucs):
+                for op in concurrent:
+                    try:
+                        conn.execute(text(op.sql))
+                        applied.append(AppliedOperation(op.type, "applied"))
+                    except Exception:  # noqa: BLE001  # concurrent ops record any DB failure and continue
+                        applied.append(AppliedOperation(op.type, "failed"))
+                        partial_failure = True
 
     # --- transactional segment: atomic ----------------------------------
     plain = [op for op in runnable if not _is_concurrent(op)]
     if plain:
         try:
             with engine.begin() as conn:
                 # 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.
+                # (utility statement), so the int is inlined; both timeouts
+                # are typed float parameters, not user input.
                 conn.execute(text(f"SET LOCAL lock_timeout = {int(lock_timeout * 1000)}"))
+                if statement_timeout is not None:
+                    conn.execute(
+                        text(f"SET LOCAL statement_timeout = {int(statement_timeout * 1000)}")
+                    )
                 for op in plain:
                     conn.execute(text(op.sql))
             applied.extend(AppliedOperation(op.type, "applied") for op in plain)
@@ -146,14 +192,18 @@ def with_advisory_lock(
     reverify: DiffEngine | None = None,
     schemas: Sequence[str] | None = None,
     exclude: Sequence[str] = (),
+    concurrently: bool = True,
+    statement_timeout: float | None = None,
 ) -> Report:
     """Winner migrates; losers block (bounded), then re-verify.
 
     Losers poll ``pg_try_advisory_lock`` every 0.5 s against a
     ``time.monotonic()`` deadline; once the lock is acquired the winner
     path re-plans (covering the case where the previous winner died
     mid-push). Raises :class:`SqlpushError` if the wait budget is
-    exhausted.
+    exhausted. ``concurrently`` and ``statement_timeout`` thread into
+    BOTH the winner's re-plan and its apply — the re-plan must render
+    exactly what the caller asked for.
     """
     if reverify is None:
         raise SqlpushError(
@@ -179,7 +229,13 @@ def with_advisory_lock(
         # (killed mid-push = lock silently released)
         conn.rollback()
         try:
-            plan = reverify.plan(metadata, engine, schemas=schemas, exclude=exclude)
+            plan = reverify.plan(
+                metadata,
+                engine,
+                schemas=schemas,
+                exclude=exclude,
+                concurrently=concurrently,
+            )
             if not plan.drift:
                 return Report()
             return apply_plan(
@@ -188,6 +244,7 @@ def with_advisory_lock(
                 allow_destructive=allow_destructive,
                 safe_only=safe_only,
                 lock_timeout=timeout,
+                statement_timeout=statement_timeout,
             )
         finally:
             # best-effort unlock: a secondary failure here (e.g. the