diff --git a/docs/progress-report.md b/docs/progress-report.md index 0314014..4d4efe0 100644 --- a/docs/progress-report.md +++ b/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. +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`. + 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,11 +46,15 @@ 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 | after a step starts | The exact SQL string the executor executes for this step, after front-door qualification and canonicalization — not a display rendering. It remains present on terminal snapshots so an observer can identify the statement that produced the outcome. | | `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. | | `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 @@ -91,7 +98,7 @@ A poll during step 2 of a 3-step sequence, mid concurrent index build: ```json { - "format_version": 1, + "format_version": 2, "phase": "running", "step": 2, "total_steps": 3, @@ -99,6 +106,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, diff --git a/pkg/executor/create.go b/pkg/executor/create.go index 5e416e7..c5240ef 100644 --- a/pkg/executor/create.go +++ b/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 { diff --git a/pkg/executor/create_integration_test.go b/pkg/executor/create_integration_test.go index 7fbdb74..5910769 100644 --- a/pkg/executor/create_integration_test.go +++ b/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. diff --git a/pkg/executor/native.go b/pkg/executor/native.go index 782cef8..7cc4d1c 100644 --- a/pkg/executor/native.go +++ b/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) } diff --git a/pkg/executor/native_integration_test.go b/pkg/executor/native_integration_test.go index b0b6538..f0b645c 100644 --- a/pkg/executor/native_integration_test.go +++ b/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) diff --git a/pkg/executor/optimistic.go b/pkg/executor/optimistic.go index fe5f56f..110cea5 100644 --- a/pkg/executor/optimistic.go +++ b/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) } diff --git a/pkg/executor/optimistic_integration_test.go b/pkg/executor/optimistic_integration_test.go index bb712d7..4b2d4a9 100644 --- a/pkg/executor/optimistic_integration_test.go +++ b/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) diff --git a/pkg/executor/sequence.go b/pkg/executor/sequence.go index f719a23..68055a0 100644 --- a/pkg/executor/sequence.go +++ b/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 diff --git a/pkg/executor/sequence_integration_test.go b/pkg/executor/sequence_integration_test.go index 5ab9cc0..aa78843 100644 --- a/pkg/executor/sequence_integration_test.go +++ b/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) diff --git a/pkg/progress/progress.go b/pkg/progress/progress.go index 2f88189..7c5624d 100644 --- a/pkg/progress/progress.go +++ b/pkg/progress/progress.go @@ -40,8 +40,9 @@ 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. -const FormatVersion = 1 +// 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 // Operation is the current operation's execution class. type Operation string @@ -78,11 +79,15 @@ 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. 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"` } // Snapshot is one immutable progress observation. For a terminal phase the @@ -145,13 +150,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 } diff --git a/pkg/progress/progress_test.go b/pkg/progress/progress_test.go index fc7238e..f8c4973 100644 --- a/pkg/progress/progress_test.go +++ b/pkg/progress/progress_test.go @@ -42,7 +42,7 @@ func runningTrackerWithBuild(t *testing.T, session fakeSession) *progress.Tracke tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)}) require.NoError(t, err) tracker.Start(1, progress.OperationConcurrentIndex) - tracker.StartStep(1, progress.OperationConcurrentIndex) + tracker.StartStep(1, progress.OperationConcurrentIndex, "CREATE INDEX CONCURRENTLY idx ON public.t (id)") tracker.SetConcurrentBuild(session, 4242) return tracker } @@ -54,7 +54,7 @@ func TestTrackerReportsSequencePositionAndInjectedElapsed(t *testing.T) { tracker.Start(3, progress.OperationBrief) clock.now = clock.now.Add(2 * time.Second) - tracker.StartStep(2, progress.OperationValidate) + tracker.StartStep(2, progress.OperationValidate, "ALTER TABLE public.t VALIDATE CONSTRAINT c") tracker.SetAttempt(2) clock.now = clock.now.Add(750 * time.Millisecond) @@ -77,7 +77,7 @@ func TestTrackerSequenceStepsAdvanceMonotonically(t *testing.T) { tracker.Start(3, progress.OperationBrief) for step := 1; step <= 3; step++ { - tracker.StartStep(step, progress.OperationBrief) + tracker.StartStep(step, progress.OperationBrief, "ALTER TABLE public.t ADD COLUMN c int") snapshot, progressErr := tracker.Progress(t.Context()) require.NoError(t, progressErr) assert.Equal(t, step, snapshot.Step) @@ -110,6 +110,7 @@ func TestNewTrackerRequiresClock(t *testing.T) { // A terminal snapshot is terminal: elapsed values freeze at the instant // Finish recorded and do not grow with the clock, for both outcomes. func TestTerminalSnapshotFreezesElapsed(t *testing.T) { + const sql = "ALTER TABLE public.t ADD COLUMN c int" cases := []struct { name string outcome error @@ -124,7 +125,7 @@ func TestTerminalSnapshotFreezesElapsed(t *testing.T) { tracker, err := progress.NewTracker(clock) require.NoError(t, err) tracker.Start(1, progress.OperationOptimistic) - tracker.StartStep(1, progress.OperationOptimistic) + tracker.StartStep(1, progress.OperationOptimistic, sql) clock.now = clock.now.Add(3 * time.Second) tracker.Finish(tc.outcome) @@ -134,6 +135,7 @@ func TestTerminalSnapshotFreezesElapsed(t *testing.T) { assert.Equal(t, tc.phase, snapshot.Phase) assert.Equal(t, 3*time.Second, snapshot.Elapsed, "elapsed must freeze at Finish") assert.Equal(t, 3*time.Second, snapshot.StepElapsed, "step elapsed must freeze at Finish") + assert.Equal(t, sql, snapshot.Detail.Statement, "terminal snapshot must retain its statement") clock.now = clock.now.Add(time.Hour) again, err := tracker.Progress(t.Context()) @@ -150,7 +152,7 @@ func TestStartResetsPriorRunState(t *testing.T) { tracker, err := progress.NewTracker(clock) require.NoError(t, err) tracker.Start(3, progress.OperationBrief) - tracker.StartStep(2, progress.OperationValidate) + tracker.StartStep(2, progress.OperationValidate, "ALTER TABLE public.t VALIDATE CONSTRAINT c") tracker.SetConcurrentBuild(fakeSession{query: func(context.Context, string, ...any) pgx.Row { return fakeRow{scan: func(...any) error { t.Fatal("a new run must not poll the prior run's session") @@ -174,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 1 against this test. +// actually receives. A consumer pins format_version 2 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 { @@ -191,7 +193,7 @@ func TestSnapshotJSONShape(t *testing.T) { require.NoError(t, err) tracker.Start(3, progress.OperationAdmitting) clock.now = clock.now.Add(2 * time.Second) - tracker.StartStep(2, progress.OperationConcurrentIndex) + tracker.StartStep(2, progress.OperationConcurrentIndex, "CREATE INDEX CONCURRENTLY idx ON public.t (id)") tracker.SetAttempt(2) tracker.SetConcurrentBuild(session, 4242) clock.now = clock.now.Add(750 * time.Millisecond) @@ -201,7 +203,7 @@ func TestSnapshotJSONShape(t *testing.T) { raw, err := json.Marshal(snapshot) require.NoError(t, err) assert.JSONEq(t, `{ - "format_version": 1, + "format_version": 2, "phase": "running", "step": 2, "total_steps": 3, @@ -209,6 +211,7 @@ func TestSnapshotJSONShape(t *testing.T) { "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, @@ -226,6 +229,23 @@ func TestSnapshotJSONShape(t *testing.T) { }`, string(raw)) } +func TestTrackerReportsCurrentStatementAndRetainsItOnFinish(t *testing.T) { + tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)}) + require.NoError(t, err) + tracker.Start(3, progress.OperationAdmitting) + + const sql = "CREATE INDEX idx ON public.t (id)" + tracker.StartStep(2, progress.OperationBrief, sql) + snapshot, err := tracker.Progress(t.Context()) + require.NoError(t, err) + assert.Equal(t, sql, snapshot.Detail.Statement) + + tracker.Finish(nil) + finished, err := tracker.Progress(t.Context()) + require.NoError(t, err) + assert.Equal(t, sql, finished.Detail.Statement) +} + // Optional fields are omitted, not emitted as zero values — but the always-on // keys (format_version, phase, both elapsed counters, active) are present // even on an idle tracker, so a consumer never guesses whether zero means @@ -239,7 +259,7 @@ func TestSnapshotJSONOmitsUnsetOptionalFields(t *testing.T) { raw, err := json.Marshal(snapshot) require.NoError(t, err) assert.JSONEq(t, `{ - "format_version": 1, + "format_version": 2, "phase": "pending", "elapsed_ns": 0, "step_elapsed_ns": 0, @@ -412,7 +432,7 @@ func TestStateMutatorsDoNotWaitForInFlightObservation(t *testing.T) { mutated := make(chan struct{}) workers.Go(func() { tracker.SetAttempt(2) - tracker.StartStep(1, progress.OperationConcurrentIndex) + tracker.StartStep(1, progress.OperationConcurrentIndex, "CREATE INDEX CONCURRENTLY idx ON public.t (id)") close(mutated) }) mutatorDeadline := time.After(5 * time.Second)