feat(executor): caller-owned deadline, lockers, and backend PID for concurrent builds (#76)

`BuildIndexConcurrently` gains a caller-owned cancellation mode, the
progress tracker becomes the operator's stop path for a running build,
and progress snapshots report the lockers a concurrent build is waiting
on.

## Why
A concurrent index build on a large table can run for hours. Today the
only bound the executor accepts is a fixed server-side
`statement_timeout` (`ConcurrentBudget.Overall`), which fits a
synchronous attempt but not an orchestrator that keeps a build alive for
as long as it can renew a lease. That orchestrator's decision tree has
three branches — *my lease lapsed* (retry later), *an operator stopped
it* (do not retry), *the server's budget killed it* (retry with a
different bound) — so the executor must report three distinguishable
outcomes. Its operators also need to stop a running build without being
handed a backend PID they could misuse after the build returns, and to
see why a build in the "waiting for old snapshots" phase is not moving,
which requires the lockers columns of `pg_stat_progress_create_index`.

## What
- `ConcurrentBudget.CallerOwned`: the session runs with
`statement_timeout = 0` and the caller's cancellable context is the
statement's only bound. `Overall` must be zero
(`ErrCallerOwnedOverallBudget`), and a context that cannot be cancelled
is refused with `ErrCallerOwnedNeedsCancellableContext` before any
session is acquired, so the statement remains bounded by construction
(LK-2). The bounded mode and every existing caller are unchanged.
- Cancellation is a three-way partition in both modes: the caller's own
context ending is `ErrCancelledByCaller` (`cancelled-by-caller`)
whichever of the server's 57014 or the client's context error arrives
first; a 57014 under a live context is `ErrCancelledExternally`
(`cancelled-externally`); a 57014 at the bounded mode's deadline is
`*BudgetError`.
- `progress.Tracker.CancelBuild(ctx)` signals the active build's backend
over the tracker's reserved session, under the same lock that guards the
build's lifecycle and only while the build is active (`ErrNoActiveBuild`
otherwise; `ErrBuildNotRunning` when the backend had no statement to
cancel). The tracker never exposes the PID itself.
- `dbconn.ConcurrentIndexProgress` reads `lockers_total`,
`lockers_done`, `current_locker_pid`; the snapshot carries them as
`work.lockers_total` / `work.lockers_done` and
`detail.current_locker_pid`. `format_version` bumps from 2 to 3 and
`docs/progress-report.md` documents the new fields and the stop path.
- LK-2 in `docs/invariants.md` records the caller-owned exception;
capability, execution-model and TCB docs describe the caller-owned
context as a bound different in kind, not an absence of one.
- Integration tests on a real server: a caller-owned build completes
once its blocker releases; a build stopped via `tracker.CancelBuild`
returns `ErrCancelledExternally` with its invalid leftover reported and
the tracker then refuses a second cancel; a build whose caller cancels
returns `ErrCancelledByCaller`, never `ErrCancelledExternally` or a
`*BudgetError`; a blocked build publishes `lockers_total ≥ 1` and the
blocker's PID as `current_locker_pid`.

## Before / after
```
Before
  caller ── ConcurrentBudget{Overall: 30m} ──▶ SET statement_timeout = 30m ──▶ CREATE INDEX CONCURRENTLY
           (no way to say "for as long as my lease holds")
  57014 before the deadline ──▶ ErrCancelledExternally   (operator? caller? indistinguishable)
  tracker.Progress() ──▶ phase, blocks, tuples            (no lockers, no stop path)

After
  caller ── ConcurrentBudget{CallerOwned: true} + cancellable ctx
        ├─ ctx.Done() == nil ──▶ ErrCallerOwnedNeedsCancellableContext (refused)
        └─ SET statement_timeout = 0 ──▶ CREATE INDEX CONCURRENTLY
             ├─ caller's ctx ended    ──▶ ErrCancelledByCaller    (+ catalog verdict)
             ├─ tracker.CancelBuild   ──▶ ErrCancelledExternally  (+ catalog verdict)
             └─ (bounded mode only) deadline ──▶ *BudgetError
  tracker.CancelBuild(ctx) ──▶ pg_cancel_backend on the active build only; PID never leaves the tracker
  tracker.Progress()       ──▶ phase, blocks, tuples, lockers_total/done, current_locker_pid
```

Author: Kiran01bm

docs/capabilities.md

@@ -32,8 +32,8 @@ refused form would take, what an operator who accepts a maintenance window can d
 
 pg-sprite is an **online schema-change engine** for PostgreSQL: it takes one table-shape
 change, classifies it against the live database, and either executes it through the
-safest known online pattern — bounded `lock_timeout`/`statement_timeout` on every
-session — or refuses with a typed reason. The measure of the tool is not how many object
+safest known online pattern — bounded server timeouts, or an explicit caller-owned
+cancellable context for concurrent index builds — or refuses with a typed reason. The measure of the tool is not how many object
 types it models but whether a change it accepts can hurt a production workload. The full
 positioning is [vision.md](vision.md); how it differs from planners and imperative
 copy tools by *problem class* is [architecture.md](architecture.md).
@@ -96,8 +96,9 @@ each matrix table answers *how* — the route the change takes (or will take) th
 engine:
 
 - **native, as-is** — the statement is already online-safe (metadata-only, or already
-  the online idiom); executed directly under bounded
-  `lock_timeout`/`statement_timeout` sessions.
+  the online idiom); executed directly under bounded sessions. Concurrent index builds
+  may use a caller-owned cancellable context instead of `statement_timeout`; other native
+  work uses `lock_timeout`/`statement_timeout`.
 - **native, safer sequence** — the blocking form is substituted with the equivalent
   online sequence before execution; the rewrites are catalogued in
   [safer-sequences.md](safer-sequences.md).

docs/execution-model.md

@@ -68,7 +68,9 @@ autocommit-each-step has two shapes in the executor:
 - **`CREATE INDEX CONCURRENTLY` (step kind `concurrent-index-build`)** is
   true autocommit on a dedicated budgeted session: it refuses to run inside
   any transaction block and internally manages multiple transactions of its
-  own.
+  own. Its bound is either the session's overall `statement_timeout` or, in
+  explicit caller-owned mode, the caller's cancellable context while
+  `statement_timeout` is disabled.
 
 Each step's class is the `kind` field of its step report in the JSON
 verdict — the field retry logic branches on. A failed `brief` step means

docs/invariants.md

@@ -141,8 +141,10 @@ strong-lock acquisition (swap, catalog flips, trigger install in fallback mode)
 ([mysql-vs-postgresql § the lock queue](mysql-vs-postgresql.md#why-ddl-is-dangerous-the-lock-queue)).
 **Exception policy required:** `CREATE INDEX CONCURRENTLY` and `REINDEX CONCURRENTLY` wait on
 other transactions via lock waits that a naive `lock_timeout` cancels — leaving an `INVALID`
-index — so they get their own wait policy (no per-lock timeout, one overall statement deadline)
-rather than the blanket timeout. `VALIDATE CONSTRAINT` is different in kind: its cancellation is
+index — so they get their own wait policy rather than the blanket timeout: no per-lock timeout,
+with either one overall server statement deadline or a caller-owned cancellable context as the
+statement's only bound. The executor refuses a non-cancellable context in caller-owned mode, so
+the statement remains bounded by construction. `VALIDATE CONSTRAINT` is different in kind: its cancellation is
 transactionally clean (the constraint simply stays `NOT VALID`; no debris), so the sequence
 executor's validate class deliberately keeps a bounded per-lock timeout — queueing behind a
 conflicting lock holder must not stall a sequence for the whole scan budget — while the scan

docs/progress-report.md

@@ -16,7 +16,8 @@ phase or operation value is a contract change and bumps `format_version`, even i
 is added or renamed.
 
 Adding a field bumps `format_version` so a strict consumer can detect the new shape from the
-version. The current version is **2**: version 2 added `detail.statement`.
+version. The current version is **3**: version 2 added `detail.statement`; version 3 added
+`detail.current_locker_pid`, `work.lockers_total`, and `work.lockers_done`.
 
 The [plan report](plan-report.md), [lint report](lint-report.md), and
 [suggest report](suggest-report.md) are separate contracts with their own `format_version`;
@@ -50,14 +51,16 @@ licenses a consumer to intervene in the change itself.
 | `server_phase` | string | active concurrent build only | PostgreSQL's own phase string from `pg_stat_progress_create_index`, verbatim. |
 | `active` | bool | always | Whether an operation is executing now. `false` with `phase: "running"` means a concurrent build's progress row has left the server view. |
 | `attempt` | int | bounded retries only | The current attempt number when the executor is inside its bounded retry loop. |
+| `current_locker_pid` | int | while waiting on a locker | PostgreSQL backend PID currently blocking the concurrent build; omitted when none is published. |
 | `work` | object | server-observed work only | Present exactly when the server published a progress row; then **every** counter below is present, so a fresh build reports honest zeros rather than an empty object. |
 
 `statement` is the submitter's statement after qualification and canonicalization, so a
 consumer rendering it into a shared surface must clamp and escape it.
 
 ### Work counters
 
-`blocks_done` / `blocks_total` and `tuples_done` / `tuples_total` come from
+`blocks_done` / `blocks_total`, `tuples_done` / `tuples_total`, and
+`lockers_done` / `lockers_total` come from
 `pg_stat_progress_create_index` during a concurrent index build. `rows_copied` /
 `rows_total` and `bytes_copied` / `bytes_total` are reserved for copy-and-swap and are `0`
 on every native operation — the engine never fabricates copy counters.
@@ -98,7 +101,7 @@ A poll during step 2 of a 3-step sequence, mid concurrent index build:
 
 ```json
 {
-  "format_version": 2,
+  "format_version": 3,
   "phase": "running",
   "step": 2,
   "total_steps": 3,
@@ -110,6 +113,7 @@ A poll during step 2 of a 3-step sequence, mid concurrent index build:
     "server_phase": "building index",
     "active": true,
     "attempt": 2,
+    "current_locker_pid": 31337,
     "work": {
       "rows_copied": 0,
       "rows_total": 0,
@@ -118,7 +122,9 @@ A poll during step 2 of a 3-step sequence, mid concurrent index build:
       "blocks_done": 11,
       "blocks_total": 40,
       "tuples_done": 7,
-      "tuples_total": 21
+      "tuples_total": 21,
+      "lockers_total": 3,
+      "lockers_done": 1
     }
   }
 }

docs/tcb-model.md

@@ -114,7 +114,8 @@ Performance. When a trade-off is hard, the higher priority wins. This is
 
 **Put a limit on everything (TIGER_STYLE).** Every loop bounded, every queue bounded, every
 retry counted, every wait deadlined. Existing instances include bounded native attempts, retry
-budgets, and `lock_timeout` / `statement_timeout` on every session. The future copy-and-swap
+budgets, bounded session timeouts, and caller-owned cancellable contexts where a server
+statement timeout is explicitly disabled. The future copy-and-swap
 path will also bound its change buffer, chunk target time, and slot-lag ceiling. The rule makes
 limits the *default*: an unbounded anything in a TCB package is a review-blocking defect. Where a
 loop is intentionally endless (the applier's consume loop), that must be stated and its exit

pkg/dbconn/dbconn.go

@@ -119,11 +119,14 @@ func ServerMajor(ctx context.Context, pool *pgxpool.Pool) (int, error) {
 
 // IndexBuildProgress is one server observation of a concurrent index build.
 type IndexBuildProgress struct {
-	Phase       string
-	BlocksDone  uint64
-	BlocksTotal uint64
-	TuplesDone  uint64
-	TuplesTotal uint64
+	Phase            string
+	BlocksDone       uint64
+	BlocksTotal      uint64
+	TuplesDone       uint64
+	TuplesTotal      uint64
+	LockersTotal     uint64
+	LockersDone      uint64
+	CurrentLockerPID uint32
 }
 
 // RowQuerier is the session capability needed for a progress observation.
@@ -136,15 +139,21 @@ type RowQuerier interface {
 // has already left the progress view.
 func ConcurrentIndexProgress(ctx context.Context, session RowQuerier, backendPID uint32) (IndexBuildProgress, bool, error) {
 	var p IndexBuildProgress
-	err := session.QueryRow(ctx, `SELECT phase, blocks_done, blocks_total, tuples_done, tuples_total
+	var currentLockerPID *int32
+	err := session.QueryRow(ctx, `SELECT phase, blocks_done, blocks_total, tuples_done, tuples_total,
+		lockers_total, lockers_done, current_locker_pid
 		FROM pg_catalog.pg_stat_progress_create_index WHERE pid = $1`, backendPID).
-		Scan(&p.Phase, &p.BlocksDone, &p.BlocksTotal, &p.TuplesDone, &p.TuplesTotal)
+		Scan(&p.Phase, &p.BlocksDone, &p.BlocksTotal, &p.TuplesDone, &p.TuplesTotal,
+			&p.LockersTotal, &p.LockersDone, &currentLockerPID)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return p, false, nil
 	}
 	if err != nil {
 		return p, false, fmt.Errorf("read concurrent index progress for backend %d: %w", backendPID, err)
 	}
+	if currentLockerPID != nil {
+		p.CurrentLockerPID = uint32(*currentLockerPID)
+	}
 	return p, true, nil
 }
 

pkg/executor/native.go

@@ -88,6 +88,9 @@ var (
 	// connection, every failed build would resolve indeterminate. Like an
 	// unbounded budget, an unusable verdict is refused by construction.
 	ErrPoolTooSmall = errors.New("concurrent index build needs a pool of at least two connections: one for the build session, one reserved for the catalog verdict")
+	// ErrCallerOwnedNeedsCancellableContext is returned when caller-owned
+	// mode has no cancellation signal to bound the statement.
+	ErrCallerOwnedNeedsCancellableContext = errors.New("a caller-owned build needs a cancellable context: with statement_timeout disabled the context is the statement's only bound")
 	// ErrCancelledExternally is returned when the build's statement was
 	// cancelled (SQLSTATE 57014) before its overall budget elapsed: the
 	// executor's statement_timeout cannot have fired yet, so the
@@ -121,15 +124,21 @@ const verdictTimeout = 30 * time.Second
 // waits by implementation, so a session lock_timeout would cancel a healthy
 // build mid-wait — and that cancellation is exactly what creates the invalid
 // index this executor exists to prevent. The statement therefore runs with
-// lock_timeout disabled and one overall deadline. That is safe with respect
-// to the lock queue: the SHARE UPDATE EXCLUSIVE lock a concurrent build
-// waits for does not block normal reads or writes queued behind it.
+// lock_timeout disabled and one bound on the whole statement: a server
+// deadline, or in caller-owned mode the caller's cancellable context. That
+// is safe with respect to the lock queue: the SHARE UPDATE EXCLUSIVE lock a
+// concurrent build waits for does not block normal reads or writes queued
+// behind it.
 type ConcurrentBudget struct {
 	// Overall bounds the whole statement, waits included, via
 	// statement_timeout. It must be at least one millisecond (PostgreSQL's
 	// granularity); expect index builds on large tables to need a generous
 	// value.
 	Overall time.Duration
+	// CallerOwned makes the caller's cancellable context the statement's only
+	// bound: the session runs with statement_timeout disabled and Overall must
+	// be zero. The executor refuses a context that cannot be cancelled.
+	CallerOwned bool
 }
 
 // maxOverallBudget is PostgreSQL's ceiling for statement_timeout (the
@@ -140,6 +149,14 @@ const maxOverallBudget = time.Duration(math.MaxInt32) * time.Millisecond
 
 // validate rejects budgets that would leave the statement unbounded.
 func (b ConcurrentBudget) validate() error {
+	if b.CallerOwned {
+		// INV: LK-2 — the bound moves from the server timer to the caller's
+		// cancellable context, checked before a session is acquired.
+		if b.Overall != 0 {
+			return fmt.Errorf("caller-owned budget requires Overall to be zero, got %s", b.Overall)
+		}
+		return nil
+	}
 	// INV: LK-2 — the build is bounded by construction; below one
 	// millisecond the setting would round to zero, which disables
 	// statement_timeout entirely.
@@ -252,9 +269,12 @@ func (e *InvalidIndexError) Unwrap() []error {
 //   - a failed build that provably left nothing returns its failure alone
 //     — a retry can start immediately.
 //
-// Cancellation by the overall budget surfaces as a *BudgetError; a
+// Cancellation by the server-owned overall budget surfaces as a *BudgetError; a
 // cancellation arriving before the budget elapsed cannot be the budget's
 // own statement_timeout and surfaces as ErrCancelledExternally instead.
+// In caller-owned mode the cancellable context is the only bound and
+// statement_timeout is disabled; SQLSTATE 57014 always surfaces as
+// ErrCancelledExternally.
 // Caller cancellation is a race: the client returns while the cancel signal
 // travels to the server, so a build cancelled at the finish line may still
 // complete. The guarantee is about the catalog, not the race: after this
@@ -288,6 +308,11 @@ func buildIndexConcurrently(ctx context.Context, pool *pgxpool.Pool, sql string,
 	if err := b.validate(); err != nil {
 		return rep, err
 	}
+	// INV: LK-2 — caller-owned mode is bounded by the caller's cancellation
+	// signal, so a context without one is refused before any session use.
+	if b.CallerOwned && ctx.Done() == nil {
+		return rep, ErrCallerOwnedNeedsCancellableContext
+	}
 	build, err := admitConcurrentIndexBuild(sql)
 	if err != nil {
 		return rep, err
@@ -693,12 +718,15 @@ func acquireBudgetedSession(ctx context.Context, pool *pgxpool.Pool, b Concurren
 
 	// INV: LK-2 — the CONCURRENTLY exception policy: no per-lock timeout
 	// (a lock_timeout would cancel the statement's snapshot waits and leave
-	// the invalid index this executor exists to prevent); one overall
-	// statement deadline bounds every statement instead. A bare integer is
+	// the invalid index this executor exists to prevent); either one overall
+	// statement deadline or caller cancellation bounds the build instead. A bare integer is
 	// milliseconds to PostgreSQL; the settings are applied here regardless
 	// of the pool's defaults.
-	budgets := "SET lock_timeout = 0; SET statement_timeout = " +
-		strconv.FormatInt(b.Overall.Milliseconds(), 10)
+	statementTimeout := b.Overall.Milliseconds()
+	if b.CallerOwned {
+		statementTimeout = 0
+	}
+	budgets := "SET lock_timeout = 0; SET statement_timeout = " + strconv.FormatInt(statementTimeout, 10)
 	if _, err := conn.Exec(ctx, budgets); err != nil {
 		// The two SETs may have partially applied — PostgreSQL runs a
 		// simple-query batch statement by statement — so the session must
@@ -742,6 +770,10 @@ func acquireBudgetedSession(ctx context.Context, pool *pgxpool.Pool, b Concurren
 func asConcurrentBudgetError(err error, b ConcurrentBudget, elapsed time.Duration) error {
 	var pgErr *pgconn.PgError
 	if errors.As(err, &pgErr) && pgErr.Code == sqlstateQueryCanceled {
+		if b.CallerOwned {
+			return fmt.Errorf("%w (after %s, caller-owned deadline): %w",
+				ErrCancelledExternally, elapsed.Round(time.Millisecond), err)
+		}
 		if elapsed >= b.Overall {
 			return &BudgetError{Cause: CauseStatement, Budget: b.Overall}
 		}