diff --git a/docs/capabilities.md b/docs/capabilities.md index 7179857..482ffca 100644 --- a/docs/capabilities.md +++ b/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). diff --git a/docs/execution-model.md b/docs/execution-model.md index 9f76dc4..98ac5e5 100644 --- a/docs/execution-model.md +++ b/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 diff --git a/docs/invariants.md b/docs/invariants.md index b6df6b4..88f561e 100644 --- a/docs/invariants.md +++ b/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 diff --git a/docs/progress-report.md b/docs/progress-report.md index 4d4efe0..63cbc43 100644 --- a/docs/progress-report.md +++ b/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,6 +51,7 @@ 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 @@ -57,7 +59,8 @@ 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 } } } diff --git a/docs/tcb-model.md b/docs/tcb-model.md index 0daa056..aeb8000 100644 --- a/docs/tcb-model.md +++ b/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 diff --git a/pkg/dbconn/dbconn.go b/pkg/dbconn/dbconn.go index 6dd1508..97fca53 100644 --- a/pkg/dbconn/dbconn.go +++ b/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, ¤tLockerPID) 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 } diff --git a/pkg/executor/native.go b/pkg/executor/native.go index 7cc4d1c..2d1c71f 100644 --- a/pkg/executor/native.go +++ b/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} } diff --git a/pkg/executor/native_integration_test.go b/pkg/executor/native_integration_test.go index f0b645c..7751cd6 100644 --- a/pkg/executor/native_integration_test.go +++ b/pkg/executor/native_integration_test.go @@ -85,6 +85,45 @@ func TestBuildIndexConcurrentlyBuildsValidIndex(t *testing.T) { assert.True(t, valid, "the index must be valid") } +func TestBuildIndexConcurrentlyCallerOwnedBuildsValidIndex(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.caller_owned_t (id int PRIMARY KEY, c int)", schema)) + require.NoError(t, err) + blocker, err := pool.BeginTx(t.Context(), pgx.TxOptions{IsoLevel: pgx.RepeatableRead}) + require.NoError(t, err) + var count int + require.NoError(t, blocker.QueryRow(t.Context(), fmt.Sprintf("SELECT count(*) FROM %s.caller_owned_t", schema)).Scan(&count)) + + tracker, err := progress.NewTracker(progress.WallClock{}) + require.NoError(t, err) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + type result struct { + rep executor.IndexBuildReport + err error + } + done := make(chan result, 1) + go func() { + rep, buildErr := executor.BuildIndexConcurrentlyWithProgress(ctx, pool, + fmt.Sprintf("CREATE INDEX CONCURRENTLY caller_owned_idx ON %s.caller_owned_t (c)", schema), + executor.ConcurrentBudget{CallerOwned: true}, tracker) + done <- result{rep: rep, err: buildErr} + }() + require.Eventually(t, func() bool { + snapshot, progressErr := tracker.Progress(t.Context()) + return progressErr == nil && snapshot.Detail.Work != nil && snapshot.Detail.Work.LockersTotal >= 1 + }, 30*time.Second, 20*time.Millisecond, "the build must wait for the old snapshot") + require.NoError(t, blocker.Rollback(t.Context())) + buildResult := <-done + require.NoError(t, buildResult.err) + rep := buildResult.rep + assert.Equal(t, "caller_owned_idx", rep.Index) + assert.NotZero(t, rep.IndexOID) + exists, valid := indexState(t, pool, schema, "caller_owned_idx") + assert.True(t, exists) + assert.True(t, valid) +} + func TestBuildIndexConcurrentlyReportsServerProgressAndFinishes(t *testing.T) { pool, schema := newPool(t) _, err := pool.Exec(t.Context(), fmt.Sprintf(`CREATE TABLE %s.progress_t AS @@ -239,6 +278,83 @@ func TestBuildIndexConcurrentlyOperatorCancelIsNotBudgetExhaustion(t *testing.T) } } +func TestBuildIndexConcurrentlyCallerOwnedCancelViaTrackerPID(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.pid_t (id int PRIMARY KEY, c int)", schema)) + require.NoError(t, err) + blocker, err := pool.BeginTx(t.Context(), pgx.TxOptions{IsoLevel: pgx.RepeatableRead}) + require.NoError(t, err) + var count int + require.NoError(t, blocker.QueryRow(t.Context(), fmt.Sprintf("SELECT count(*) FROM %s.pid_t", schema)).Scan(&count)) + t.Cleanup(func() { require.NoError(t, blocker.Rollback(context.WithoutCancel(t.Context()))) }) + + tracker, err := progress.NewTracker(progress.WallClock{}) + require.NoError(t, err) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { + _, buildErr := executor.BuildIndexConcurrentlyWithProgress(ctx, pool, + fmt.Sprintf("CREATE INDEX CONCURRENTLY pid_idx ON %s.pid_t (c)", schema), + executor.ConcurrentBudget{CallerOwned: true}, tracker) + done <- buildErr + }() + + var pid uint32 + require.Eventually(t, func() bool { + pid = tracker.BuildPID() + if pid == 0 { + return false + } + var active bool + err := pool.QueryRow(t.Context(), "SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE pid = $1 AND state = 'active')", pid).Scan(&active) + return err == nil && active + }, 30*time.Second, 20*time.Millisecond, "the tracked build backend must become active") + var cancelled bool + require.NoError(t, pool.QueryRow(t.Context(), "SELECT pg_cancel_backend($1)", pid).Scan(&cancelled)) + require.True(t, cancelled) + + buildErr := <-done + require.ErrorIs(t, buildErr, executor.ErrCancelledExternally) + var budgetErr *executor.BudgetError + assert.False(t, errors.As(buildErr, &budgetErr)) + var invalidErr *executor.InvalidIndexError + require.ErrorAs(t, buildErr, &invalidErr) + require.ErrorIs(t, invalidErr.Build, executor.ErrCancelledExternally) +} + +func TestBuildIndexConcurrentlyReportsLockers(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.lockers_t (id int PRIMARY KEY, c int)", schema)) + require.NoError(t, err) + blocker, err := pool.BeginTx(t.Context(), pgx.TxOptions{IsoLevel: pgx.RepeatableRead}) + require.NoError(t, err) + var blockerPID uint32 + require.NoError(t, blocker.QueryRow(t.Context(), "SELECT pg_backend_pid()").Scan(&blockerPID)) + var count int + require.NoError(t, blocker.QueryRow(t.Context(), fmt.Sprintf("SELECT count(*) FROM %s.lockers_t", schema)).Scan(&count)) + + tracker, err := progress.NewTracker(progress.WallClock{}) + require.NoError(t, err) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { + _, buildErr := executor.BuildIndexConcurrentlyWithProgress(ctx, pool, + fmt.Sprintf("CREATE INDEX CONCURRENTLY lockers_idx ON %s.lockers_t (c)", schema), + executor.ConcurrentBudget{CallerOwned: true}, tracker) + done <- buildErr + }() + + require.Eventually(t, func() bool { + snapshot, progressErr := tracker.Progress(t.Context()) + return progressErr == nil && snapshot.Detail.Work != nil && + snapshot.Detail.Work.LockersTotal >= 1 && snapshot.Detail.CurrentLockerPID == blockerPID + }, 30*time.Second, 20*time.Millisecond, "the server must publish the blocking snapshot PID") + require.NoError(t, blocker.Rollback(t.Context())) + require.NoError(t, <-done) +} + func TestBuildIndexConcurrentlyMissingTable(t *testing.T) { pool, schema := newPool(t) _, err := executor.BuildIndexConcurrently(t.Context(), pool, diff --git a/pkg/executor/native_internal_test.go b/pkg/executor/native_internal_test.go index 2185103..0c3f16c 100644 --- a/pkg/executor/native_internal_test.go +++ b/pkg/executor/native_internal_test.go @@ -70,6 +70,13 @@ func TestAsConcurrentBudgetError(t *testing.T) { assert.Equal(t, sqlstateQueryCanceled, pgErr.Code) }) + t.Run("57014 under a caller-owned deadline is an external cancellation", func(t *testing.T) { + err := asConcurrentBudgetError(cancelled, ConcurrentBudget{CallerOwned: true}, 2*time.Second) + require.ErrorIs(t, err, ErrCancelledExternally) + var budgetErr *BudgetError + assert.False(t, errors.As(err, &budgetErr)) + }) + t.Run("any other failure is neither", func(t *testing.T) { err := asConcurrentBudgetError(&pgconn.PgError{Code: "42P07"}, budget, 2*time.Second) assert.NotErrorIs(t, err, ErrCancelledExternally) diff --git a/pkg/executor/native_test.go b/pkg/executor/native_test.go index 497d029..16ac4b6 100644 --- a/pkg/executor/native_test.go +++ b/pkg/executor/native_test.go @@ -1,6 +1,7 @@ package executor_test import ( + "context" "math" "testing" "time" @@ -35,6 +36,20 @@ func TestBuildIndexConcurrentlyRejectsUnboundedBudget(t *testing.T) { } } +func TestBuildIndexConcurrentlyRejectsCallerOwnedOverallBudget(t *testing.T) { + _, err := executor.BuildIndexConcurrently(t.Context(), nil, + "CREATE INDEX CONCURRENTLY idx ON public.t (c)", + executor.ConcurrentBudget{Overall: time.Second, CallerOwned: true}) + require.Error(t, err) +} + +func TestBuildIndexConcurrentlyCallerOwnedNeedsCancellableContext(t *testing.T) { + _, err := executor.BuildIndexConcurrently(context.WithoutCancel(t.Context()), nil, + "CREATE INDEX CONCURRENTLY idx ON public.t (c)", + executor.ConcurrentBudget{CallerOwned: true}) + require.ErrorIs(t, err, executor.ErrCallerOwnedNeedsCancellableContext) +} + func TestBuildIndexConcurrentlyAdmission(t *testing.T) { tests := []struct { name string diff --git a/pkg/progress/progress.go b/pkg/progress/progress.go index 7c5624d..e009f91 100644 --- a/pkg/progress/progress.go +++ b/pkg/progress/progress.go @@ -42,7 +42,7 @@ const ( // field semantics. Adding a phase or operation value is a contract change // and bumps this version, even when no field is added or renamed. Adding a // field also bumps this version so strict consumers can detect the new shape. -const FormatVersion = 2 +const FormatVersion = 3 // Operation is the current operation's execution class. type Operation string @@ -67,14 +67,16 @@ const ( // guess at. Rows and bytes are reserved for copy-and-swap; native operations // do not fabricate them. type Work struct { - RowsCopied uint64 `json:"rows_copied"` - RowsTotal uint64 `json:"rows_total"` - BytesCopied uint64 `json:"bytes_copied"` - BytesTotal uint64 `json:"bytes_total"` - BlocksDone uint64 `json:"blocks_done"` - BlocksTotal uint64 `json:"blocks_total"` - TuplesDone uint64 `json:"tuples_done"` - TuplesTotal uint64 `json:"tuples_total"` + RowsCopied uint64 `json:"rows_copied"` + RowsTotal uint64 `json:"rows_total"` + BytesCopied uint64 `json:"bytes_copied"` + BytesTotal uint64 `json:"bytes_total"` + BlocksDone uint64 `json:"blocks_done"` + BlocksTotal uint64 `json:"blocks_total"` + TuplesDone uint64 `json:"tuples_done"` + TuplesTotal uint64 `json:"tuples_total"` + LockersTotal uint64 `json:"lockers_total"` + LockersDone uint64 `json:"lockers_done"` } // Detail describes the operation currently executing. @@ -83,11 +85,12 @@ type Detail struct { // Statement is the canonical, qualified SQL the executor is running for // the current step, never a rendered or prettified form. Terminal snapshots // retain it so observers can identify the statement that produced the outcome. - Statement string `json:"statement,omitempty"` - ServerPhase string `json:"server_phase,omitempty"` - Active bool `json:"active"` - Attempt int `json:"attempt,omitempty"` - Work *Work `json:"work,omitempty"` + Statement string `json:"statement,omitempty"` + ServerPhase string `json:"server_phase,omitempty"` + Active bool `json:"active"` + Attempt int `json:"attempt,omitempty"` + Work *Work `json:"work,omitempty"` + CurrentLockerPID uint32 `json:"current_locker_pid,omitempty"` } // Snapshot is one immutable progress observation. For a terminal phase the @@ -177,6 +180,15 @@ func (t *Tracker) SetConcurrentBuild(session dbconn.RowQuerier, pid uint32) { t.session, t.buildPID = session, pid } +// BuildPID returns the active concurrent build's backend PID. An +// orchestrator passes this PID to pg_cancel_backend from a second connection +// to stop the build. It returns zero when no build is active. +func (t *Tracker) BuildPID() uint32 { + t.mu.RLock() + defer t.mu.RUnlock() + return t.buildPID +} + // StopConcurrentBuild waits for an in-flight observation and releases the // reserved session back to the executor before its catalog verdict. func (t *Tracker) StopConcurrentBuild() { @@ -238,7 +250,9 @@ func (t *Tracker) Progress(ctx context.Context) (Snapshot, error) { work := Work{ BlocksDone: p.BlocksDone, BlocksTotal: p.BlocksTotal, TuplesDone: p.TuplesDone, TuplesTotal: p.TuplesTotal, + LockersTotal: p.LockersTotal, LockersDone: p.LockersDone, } s.Detail.Active, s.Detail.ServerPhase, s.Detail.Work = true, p.Phase, &work + s.Detail.CurrentLockerPID = p.CurrentLockerPID return s, nil } diff --git a/pkg/progress/progress_test.go b/pkg/progress/progress_test.go index f8c4973..d054362 100644 --- a/pkg/progress/progress_test.go +++ b/pkg/progress/progress_test.go @@ -176,7 +176,7 @@ func TestStartResetsPriorRunState(t *testing.T) { // The JSON shape is the adapter-facing contract: exact keys, exact // omissions, driven through a real poll so the test pins what a consumer -// actually receives. A consumer pins format_version 2 against this test. +// actually receives. A consumer pins format_version 3 against this test. func TestSnapshotJSONShape(t *testing.T) { session := fakeSession{query: func(context.Context, string, ...any) pgx.Row { return fakeRow{scan: func(dest ...any) error { @@ -185,6 +185,10 @@ func TestSnapshotJSONShape(t *testing.T) { *(dest[2].(*uint64)) = 40 // blocks_total *(dest[3].(*uint64)) = 7 // tuples_done *(dest[4].(*uint64)) = 21 // tuples_total + *(dest[5].(*uint64)) = 3 // lockers_total + *(dest[6].(*uint64)) = 1 // lockers_done + pid := int32(31337) + *(dest[7].(**int32)) = &pid return nil }} }} @@ -203,7 +207,7 @@ func TestSnapshotJSONShape(t *testing.T) { raw, err := json.Marshal(snapshot) require.NoError(t, err) assert.JSONEq(t, `{ - "format_version": 2, + "format_version": 3, "phase": "running", "step": 2, "total_steps": 3, @@ -215,6 +219,7 @@ func TestSnapshotJSONShape(t *testing.T) { "server_phase": "building index", "active": true, "attempt": 2, + "current_locker_pid": 31337, "work": { "rows_copied": 0, "rows_total": 0, @@ -223,7 +228,9 @@ func TestSnapshotJSONShape(t *testing.T) { "blocks_done": 11, "blocks_total": 40, "tuples_done": 7, - "tuples_total": 21 + "tuples_total": 21, + "lockers_total": 3, + "lockers_done": 1 } } }`, string(raw)) @@ -259,7 +266,7 @@ func TestSnapshotJSONOmitsUnsetOptionalFields(t *testing.T) { raw, err := json.Marshal(snapshot) require.NoError(t, err) assert.JSONEq(t, `{ - "format_version": 2, + "format_version": 3, "phase": "pending", "elapsed_ns": 0, "step_elapsed_ns": 0, @@ -277,6 +284,10 @@ func TestProgressMergesServerIndexBuildWork(t *testing.T) { *(dest[2].(*uint64)) = 40 // blocks_total *(dest[3].(*uint64)) = 7 // tuples_done *(dest[4].(*uint64)) = 21 // tuples_total + *(dest[5].(*uint64)) = 3 // lockers_total + *(dest[6].(*uint64)) = 1 // lockers_done + pid := int32(31337) + *(dest[7].(**int32)) = &pid return nil }} }} @@ -291,9 +302,29 @@ func TestProgressMergesServerIndexBuildWork(t *testing.T) { assert.Equal(t, uint64(40), s.Detail.Work.BlocksTotal) assert.Equal(t, uint64(7), s.Detail.Work.TuplesDone) assert.Equal(t, uint64(21), s.Detail.Work.TuplesTotal) + assert.Equal(t, uint64(3), s.Detail.Work.LockersTotal) + assert.Equal(t, uint64(1), s.Detail.Work.LockersDone) + assert.Equal(t, uint32(31337), s.Detail.CurrentLockerPID) assert.Zero(t, s.Detail.Work.RowsCopied, "native progress must not fabricate copy counters") } +func TestBuildPIDLifecycle(t *testing.T) { + tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)}) + require.NoError(t, err) + tracker.Start(1, progress.OperationConcurrentIndex) + tracker.SetConcurrentBuild(fakeSession{}, 4242) + assert.Equal(t, uint32(4242), tracker.BuildPID()) + + tracker.StopConcurrentBuild() + assert.Zero(t, tracker.BuildPID()) + tracker.SetConcurrentBuild(fakeSession{}, 4242) + tracker.Finish(nil) + assert.Zero(t, tracker.BuildPID()) + tracker.SetConcurrentBuild(fakeSession{}, 4242) + tracker.Start(1, progress.OperationConcurrentIndex) + assert.Zero(t, tracker.BuildPID()) +} + // A build that has left the progress view is reported inactive, with no // stale server detail attached. func TestProgressClearsActiveWhenServerRowIsGone(t *testing.T) {