feat(progress): report the statement each step is executing

docs/progress-report.md

@@ -15,6 +15,9 @@ field shape: the closed vocabularies below (phases, operations) are pinned to it
 phase or operation value is a contract change and bumps `format_version`, even if no field
 is added or renamed.
 
+Additive optional fields do not bump `format_version`; consumers must ignore optional fields
+they do not recognize. The current version is **1**.
+
 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`;
 all version independently.
@@ -43,6 +46,7 @@ licenses a consumer to intervene in the change itself.
 | Field | Type | Presence | Meaning |
 |---|---|---|---|
 | `operation` | string | once execution starts | The current operation's execution class (see Operations). |
+| `statement` | string | while a step is executing | The exact SQL string the executor executes for this step, after front-door qualification and canonicalization — not a display rendering. It is present only while the step is executing. On a terminal `failed` snapshot the typed `*SequenceStepError` returned to the in-process caller carries the SQL, while `step`, `operation`, and `attempt` remain so a poller can locate the step in the plan. |
 | `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. |
@@ -99,6 +103,7 @@ A poll during step 2 of a 3-step sequence, mid concurrent index build:
   "step_elapsed_ns": 750000000,
   "detail": {
     "operation": "concurrent-index-build",
+    "statement": "CREATE INDEX CONCURRENTLY idx ON public.t (id)",
     "server_phase": "building index",
     "active": true,
     "attempt": 2,

pkg/executor/create.go

@@ -147,7 +147,7 @@ func executeCreate(ctx context.Context, pool *pgxpool.Pool, at preflight.AbsentT
 	for i, step := range steps {
 		start := time.Now()
 		if tracker != nil {
-			tracker.StartStep(i+1, progress.OperationBrief)
+			tracker.StartStep(i+1, progress.OperationBrief, step.SQL())
 			start = tracker.Now()
 		}
 		err := executeWithLockRetryObserved(ctx, retry, func(ctx context.Context) error {

pkg/executor/create_integration_test.go

@@ -3,9 +3,11 @@ package executor_test
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -14,6 +16,7 @@ import (
 	"github.com/block/pg-sprite/pkg/dbconn"
 	"github.com/block/pg-sprite/pkg/executor"
 	"github.com/block/pg-sprite/pkg/preflight"
+	"github.com/block/pg-sprite/pkg/progress"
 	"github.com/block/pg-sprite/pkg/statement"
 )
 
@@ -120,6 +123,71 @@ func TestExecuteCreateOrdersTableBeforeIndexes(t *testing.T) {
 	assert.Equal(t, "i", relationKind(t, f.pool, f.schema, "t_name_idx"))
 }
 
+func TestExecuteCreateWithProgressReportsQualifiedStepStatementsInOrder(t *testing.T) {
+	f := newCreateFixture(t, "t")
+	ds := desired(t, `
+		CREATE INDEX t_name_idx ON t (name);
+		CREATE TABLE t (id int, name text);
+	`)
+	tracker, err := progress.NewTracker(progress.WallClock{})
+	require.NoError(t, err)
+
+	functionName := pgx.Identifier{f.schema, "delay_create_progress"}.Sanitize()
+	triggerName := pgx.Identifier{f.schema + "_delay_create_progress"}.Sanitize()
+	_, err = f.pool.Exec(t.Context(), fmt.Sprintf(`
+		CREATE FUNCTION %s() RETURNS event_trigger LANGUAGE plpgsql AS $$
+		BEGIN
+			IF current_query() LIKE '%%%s%%' THEN
+				PERFORM pg_sleep(0.25);
+			END IF;
+		END
+		$$;
+		CREATE EVENT TRIGGER %s ON ddl_command_start EXECUTE FUNCTION %s()`,
+		functionName, f.schema, triggerName, functionName))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		ctx := context.WithoutCancel(t.Context())
+		_, cleanupErr := f.pool.Exec(ctx, fmt.Sprintf("DROP EVENT TRIGGER IF EXISTS %s", triggerName))
+		assert.NoError(t, cleanupErr)
+		_, cleanupErr = f.pool.Exec(ctx, fmt.Sprintf("DROP FUNCTION IF EXISTS %s()", functionName))
+		assert.NoError(t, cleanupErr)
+	})
+
+	type result struct {
+		rep executor.SequenceReport
+		err error
+	}
+	results := make(chan result, 1)
+	var workers sync.WaitGroup
+	workers.Go(func() {
+		rep, executeErr := executor.ExecuteCreateWithProgress(t.Context(), f.pool, f.at, f.cr, ds,
+			createBudget, executor.DefaultRetryPolicy(), tracker)
+		results <- result{rep: rep, err: executeErr}
+	})
+	t.Cleanup(workers.Wait)
+
+	var observed []string
+	require.Eventually(t, func() bool {
+		snapshot, progressErr := tracker.Progress(t.Context())
+		if progressErr != nil || snapshot.Detail.Statement == "" {
+			return false
+		}
+		if len(observed) == 0 || observed[len(observed)-1] != snapshot.Detail.Statement {
+			observed = append(observed, snapshot.Detail.Statement)
+		}
+		return len(observed) == 2
+	}, 5*time.Second, 10*time.Millisecond, "both active create steps must publish their statements in order")
+
+	execution := <-results
+	workers.Wait()
+	require.NoError(t, execution.err)
+	require.Len(t, execution.rep.Steps, 2)
+	assert.Equal(t, []string{
+		fmt.Sprintf("CREATE TABLE %s.t (id int, name text)", f.schema),
+		fmt.Sprintf("CREATE INDEX t_name_idx ON %s.t USING btree (name)", f.schema),
+	}, observed)
+}
+
 // The absence proof is time-of-check: a create that takes the name after
 // the check surfaces as the typed collision, and the caller re-diffs
 // rather than assuming what the occupant looks like.

pkg/executor/native.go

@@ -278,7 +278,7 @@ func BuildIndexConcurrentlyWithProgress(ctx context.Context, pool *pgxpool.Pool,
 		return rep, fmt.Errorf("%w: progress tracker is required", ErrInvariantViolation)
 	}
 	tracker.Start(1, progress.OperationConcurrentIndex)
-	tracker.StartStep(1, progress.OperationConcurrentIndex)
+	tracker.StartStep(1, progress.OperationConcurrentIndex, sql)
 	defer func() { tracker.Finish(err) }()
 	return buildIndexConcurrently(ctx, pool, sql, b, tracker)
 }

pkg/executor/native_integration_test.go

@@ -92,13 +92,14 @@ func TestBuildIndexConcurrentlyReportsServerProgressAndFinishes(t *testing.T) {
 	require.NoError(t, err)
 	tracker, err := progress.NewTracker(progress.WallClock{})
 	require.NoError(t, err)
+	statementSQL := fmt.Sprintf("CREATE INDEX CONCURRENTLY progress_idx ON %s.progress_t (payload)", schema)
 
 	type result struct{ err error }
 	results := make(chan result, 1)
 	var workers sync.WaitGroup
 	workers.Go(func() {
 		_, buildErr := executor.BuildIndexConcurrentlyWithProgress(t.Context(), pool,
-			fmt.Sprintf("CREATE INDEX CONCURRENTLY progress_idx ON %s.progress_t (payload)", schema), buildBudget, tracker)
+			statementSQL, buildBudget, tracker)
 		results <- result{err: buildErr}
 	})
 	t.Cleanup(workers.Wait)
@@ -110,6 +111,7 @@ func TestBuildIndexConcurrentlyReportsServerProgressAndFinishes(t *testing.T) {
 		return progressErr == nil && observed.Detail.ServerPhase != ""
 	}, 30*time.Second, 10*time.Millisecond, "the active build must publish server progress")
 	require.NotNil(t, observed.Detail.Work)
+	assert.Equal(t, statementSQL, observed.Detail.Statement)
 	assert.LessOrEqual(t, observed.Detail.Work.BlocksDone, observed.Detail.Work.BlocksTotal)
 	assert.LessOrEqual(t, observed.Detail.Work.TuplesDone, observed.Detail.Work.TuplesTotal)
 

pkg/executor/optimistic.go

@@ -195,7 +195,7 @@ func ExecuteNativeWithProgress(ctx context.Context, pool *pgxpool.Pool, pt prefl
 		return fmt.Errorf("%w: progress tracker is required", ErrInvariantViolation)
 	}
 	tracker.Start(1, progress.OperationOptimistic)
-	tracker.StartStep(1, progress.OperationOptimistic)
+	tracker.StartStep(1, progress.OperationOptimistic, st.SQL())
 	defer func() { tracker.Finish(err) }()
 	return executeNative(ctx, pool, pt, st, b, retry, tracker)
 }

pkg/executor/optimistic_integration_test.go

@@ -3,6 +3,7 @@ package executor_test
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -223,7 +224,23 @@ func TestExecuteNativeWithProgressReportsRetriesAndFailure(t *testing.T) {
 	st := mustParse(t, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int", schema))
 	retry := executor.RetryPolicy{MaxAttempts: 2, InitialBackoff: 10 * time.Millisecond, MaxBackoff: 20 * time.Millisecond}
 	tight := executor.Budget{LockTimeout: 100 * time.Millisecond, StatementTimeout: time.Second}
-	err = executor.ExecuteNativeWithProgress(t.Context(), pool, pt, st, tight, retry, tracker)
+	done := make(chan error, 1)
+	var workers sync.WaitGroup
+	workers.Go(func() {
+		done <- executor.ExecuteNativeWithProgress(t.Context(), pool, pt, st, tight, retry, tracker)
+	})
+	t.Cleanup(workers.Wait)
+
+	var running progress.Snapshot
+	assert.Eventually(t, func() bool {
+		var progressErr error
+		running, progressErr = tracker.Progress(t.Context())
+		return progressErr == nil && running.Detail.Attempt > 0
+	}, time.Second, 5*time.Millisecond, "a blocked optimistic attempt must publish its active statement")
+	assert.Equal(t, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int", schema), running.Detail.Statement)
+
+	err = <-done
+	workers.Wait()
 
 	var budgetErr *executor.BudgetError
 	require.ErrorAs(t, err, &budgetErr)

pkg/executor/sequence.go

@@ -294,7 +294,7 @@ func runSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflight
 	for i, step := range admitted {
 		start := time.Now()
 		if tracker != nil {
-			tracker.StartStep(i+1, progressOperation(step.kind))
+			tracker.StartStep(i+1, progressOperation(step.kind), step.st.SQL())
 			start = tracker.Now()
 		}
 		var indexReport *IndexBuildReport

pkg/executor/sequence_integration_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -102,11 +103,61 @@ func TestRunSequenceWithProgressTracksStepsAndFinishes(t *testing.T) {
 	tracker, err := progress.NewTracker(progress.WallClock{})
 	require.NoError(t, err)
 
-	steps := saferSequence(t, fmt.Sprintf("ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0)", schema))
-	rep, err := executor.RunSequenceWithProgress(t.Context(), pool, pt, steps, runBudget,
-		executor.DefaultRetryPolicy(), tracker)
+	functionName := pgx.Identifier{schema, "delay_sequence_progress"}.Sanitize()
+	triggerName := pgx.Identifier{schema + "_delay_sequence_progress"}.Sanitize()
+	_, err = pool.Exec(t.Context(), fmt.Sprintf(`
+		CREATE FUNCTION %s() RETURNS event_trigger LANGUAGE plpgsql AS $$
+		BEGIN
+			IF current_query() LIKE '%%%s%%' THEN
+				PERFORM pg_sleep(0.25);
+			END IF;
+		END
+		$$;
+		CREATE EVENT TRIGGER %s ON ddl_command_start EXECUTE FUNCTION %s()`,
+		functionName, schema, triggerName, functionName))
 	require.NoError(t, err)
-	require.Len(t, rep.Steps, 2)
+	t.Cleanup(func() {
+		ctx := context.WithoutCancel(t.Context())
+		_, cleanupErr := pool.Exec(ctx, fmt.Sprintf("DROP EVENT TRIGGER IF EXISTS %s", triggerName))
+		assert.NoError(t, cleanupErr)
+		_, cleanupErr = pool.Exec(ctx, fmt.Sprintf("DROP FUNCTION IF EXISTS %s()", functionName))
+		assert.NoError(t, cleanupErr)
+	})
+
+	steps := saferSequence(t, fmt.Sprintf("ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0)", schema))
+	type result struct {
+		rep executor.SequenceReport
+		err error
+	}
+	results := make(chan result, 1)
+	var workers sync.WaitGroup
+	workers.Go(func() {
+		rep, runErr := executor.RunSequenceWithProgress(t.Context(), pool, pt, steps, runBudget,
+			executor.DefaultRetryPolicy(), tracker)
+		results <- result{rep: rep, err: runErr}
+	})
+	t.Cleanup(workers.Wait)
+
+	var observed []string
+	assert.Eventually(t, func() bool {
+		snapshot, progressErr := tracker.Progress(t.Context())
+		if progressErr != nil || snapshot.Detail.Statement == "" {
+			return false
+		}
+		if len(observed) == 0 || observed[len(observed)-1] != snapshot.Detail.Statement {
+			observed = append(observed, snapshot.Detail.Statement)
+		}
+		return len(observed) == 2
+	}, 5*time.Second, 10*time.Millisecond, "both active sequence steps must publish their exact statements")
+	assert.Equal(t, []string{
+		fmt.Sprintf("ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0) NOT VALID", schema),
+		fmt.Sprintf(`ALTER TABLE %s VALIDATE CONSTRAINT "v_positive"`, pgx.Identifier{schema, "t"}.Sanitize()),
+	}, observed)
+
+	runResult := <-results
+	workers.Wait()
+	require.NoError(t, runResult.err)
+	require.Len(t, runResult.rep.Steps, 2)
 
 	snapshot, err := tracker.Progress(t.Context())
 	require.NoError(t, err)

pkg/progress/progress.go

@@ -40,7 +40,8 @@ const (
 // FormatVersion identifies the snapshot contract. A consumer must reject a
 // snapshot whose format_version it does not recognize rather than guess at
 // field semantics. Adding a phase or operation value is a contract change
-// and bumps this version, even when no field is added or renamed.
+// and bumps this version, even when no field is added or renamed. Adding an
+// optional field tagged omitempty does not bump this version.
 const FormatVersion = 1
 
 // Operation is the current operation's execution class.
@@ -78,11 +79,14 @@ type Work struct {
 
 // Detail describes the operation currently executing.
 type Detail struct {
-	Operation   Operation `json:"operation,omitempty"`
-	ServerPhase string    `json:"server_phase,omitempty"`
-	Active      bool      `json:"active"`
-	Attempt     int       `json:"attempt,omitempty"`
-	Work        *Work     `json:"work,omitempty"`
+	Operation Operation `json:"operation,omitempty"`
+	// Statement is the canonical, qualified SQL the executor is running for
+	// the current step, never a rendered or prettified form.
+	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"`
 }
 
 // Snapshot is one immutable progress observation. For a terminal phase the
@@ -145,13 +149,14 @@ func (t *Tracker) Start(total int, operation Operation) {
 	t.session, t.buildPID = nil, 0
 }
 
-// StartStep advances a sequence to a 1-based step and drops any build
-// session from a prior step, so a later step can never poll a stale build.
-func (t *Tracker) StartStep(step int, operation Operation) {
+// StartStep advances a sequence to a 1-based step, records the exact SQL the
+// executor will run, and drops any build session from a prior step, so a later
+// step can never poll a stale build.
+func (t *Tracker) StartStep(step int, operation Operation, statement string) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	t.step, t.stepStart = step, t.clock.Now()
-	t.detail = Detail{Operation: operation, Active: true}
+	t.detail = Detail{Operation: operation, Statement: statement, Active: true}
 	t.session, t.buildPID = nil, 0
 }
 
@@ -194,6 +199,7 @@ func (t *Tracker) Finish(err error) {
 	}
 	t.ended = now
 	t.detail.Active = false
+	t.detail.Statement = ""
 	t.session, t.buildPID = nil, 0
 }