From 007ef26666e6b68cc004e37ec1cd24acf7cc6fbb Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sat, 15 Aug 2026 13:29:31 +0800 Subject: [PATCH 01/12] chore(deps): bump spirit to pick up the status API resume and throttle fields Spirit's progress API now reports whether a run resumed from a checkpoint and whether (and why) the current phase is throttled. Co-Authored-By: Claude Fable 5 --- go.mod | 4 ++-- go.sum | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/go.mod b/go.mod index a9a5bc3eb..7747f8a54 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/block/schemabot -go 1.26.5 +go 1.26.6 require ( github.com/alecthomas/kong v1.16.0 @@ -10,7 +10,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.44.5 github.com/aws/aws-sdk-go-v2/service/sts v1.43.3 github.com/block/pg-sprite v0.0.0-20260814025010-d6cf677e4feb - github.com/block/spirit v0.16.1-0.20260808013537-864cf2ee6b6b + github.com/block/spirit v0.16.1-0.20260815035524-fc3efbddfa34 github.com/bradleyfalzon/ghinstallation/v2 v2.18.0 github.com/charmbracelet/bubbles v1.0.0 github.com/charmbracelet/bubbletea v1.3.10 diff --git a/go.sum b/go.sum index 4f1ecbd62..7d03fc8c7 100644 --- a/go.sum +++ b/go.sum @@ -117,8 +117,8 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/block/pg-sprite v0.0.0-20260814025010-d6cf677e4feb h1:SppN5gzk5OT9XQtMopxlUNXzyE8d791Q+kz/HldCips= github.com/block/pg-sprite v0.0.0-20260814025010-d6cf677e4feb/go.mod h1:vZxHdTMrCOPAYgswveB7PSjOaOuRgnDLGRw6WoOizRg= -github.com/block/spirit v0.16.1-0.20260808013537-864cf2ee6b6b h1:Z12hyvAmkzjVlZeKHejfRbr+nytBB6DeneF+aAmI/Rs= -github.com/block/spirit v0.16.1-0.20260808013537-864cf2ee6b6b/go.mod h1:aR9KJ8sca3Lo5IpwH6ZGRqK2zdl4Ukhuh7jNdBCRdDk= +github.com/block/spirit v0.16.1-0.20260815035524-fc3efbddfa34 h1:J8YV309/3aJzIdvX/3TV//UnQs1plvNW+a0RMz9Vzlc= +github.com/block/spirit v0.16.1-0.20260815035524-fc3efbddfa34/go.mod h1:mhA5cAELBCJq1jM7uaWnmzLFYBVY1XZ++yhiGJzpwps= github.com/block/tidb/pkg/parser v0.0.0-20260506200501-e528fd979fc8 h1:+OfdTacrEyjlqcRUpBFX9uJ6ROBq6cUjwY4DClhnsdU= github.com/block/tidb/pkg/parser v0.0.0-20260506200501-e528fd979fc8/go.mod h1:zDLDsfNBU5+L6T4J9/OgWAHc/WZvMUjbpgHqQ/t3yKo= github.com/block/vitess v0.0.0-20260703150944-881ec2298245 h1:R7e7uAxl6WIZpeY957JDsrZtuihck6vm7QgRooI295U= From 2695403d2aeb888a736e55aaf7f5e3aebfe0c4fd Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sat, 15 Aug 2026 13:56:55 +0800 Subject: [PATCH 02/12] feat(github): surface engine throttle pauses and checkpoint resumes A table paused by the engine's throttler now renders the pause and its trigger (e.g. replica lag) on the PR comment and CLI progress surfaces, so a stalled progress bar reads as a deliberate slowdown rather than a hang. The flag and display reason ride the stored task row through the durable read-model pipeline; the engine stamps them only on tables in a paced phase (row copy, checksum verify) and the drive's write-through clears them when the pause lifts. Reasons are sanitized at the engine boundary before storage so they are safe to render in markdown. A drive claim that reattaches to the engine's durable checkpoint now records one timeline event, so operators can tell a resumed copy from a fresh start after a pod restart or lease handover. Co-Authored-By: Claude Fable 5 --- TEMPLATES.md | 78 +++++++++++++++++++ pkg/api/progress_handlers.go | 6 ++ pkg/apitypes/apitypes.go | 21 +++-- pkg/cmd/internal/templates/preview_comment.go | 2 + .../internal/templates/preview_sequential.go | 20 +++++ pkg/cmd/internal/templates/progress.go | 17 ++++ pkg/cmd/internal/templates/progress_parse.go | 13 +++- .../templates/progress_states_test.go | 42 ++++++++++ pkg/engine/engine.go | 27 +++++-- pkg/engine/spirit/spirit.go | 48 ++++++++++-- pkg/engine/spirit/spirit_test.go | 70 +++++++++++++++++ pkg/proto/tern.proto | 7 ++ pkg/proto/ternv1/tern.pb.go | 33 ++++++-- pkg/schema/mysql/tasks.sql | 2 + pkg/schema/postgres/tasks.sql | 2 + pkg/storage/internal/sqlstore/tasks.go | 20 ++--- pkg/storage/internal/sqlstore/tasks_test.go | 49 ++++++++++++ pkg/storage/types.go | 9 ++- pkg/tern/grpc_client.go | 2 + pkg/tern/grpc_client_test.go | 4 + pkg/tern/local_apply.go | 17 ++++ pkg/tern/local_apply_grouped.go | 3 + pkg/tern/local_apply_sequential.go | 9 +++ pkg/tern/local_client.go | 2 + pkg/tern/local_client_test.go | 36 +++++++++ pkg/webhook/apply.go | 2 + pkg/webhook/templates/apply.go | 25 +++++- pkg/webhook/templates/apply_test.go | 58 ++++++++++++++ pkg/webhook/templates/preview.go | 17 ++++ 29 files changed, 602 insertions(+), 39 deletions(-) diff --git a/TEMPLATES.md b/TEMPLATES.md index 7ea90ff54..9b893b80b 100644 --- a/TEMPLATES.md +++ b/TEMPLATES.md @@ -2828,6 +2828,54 @@ ALTER TABLE `orders` ADD INDEX `idx_user_id`(`user_id`); ``` +--- + +To stop this schema change: +``` +schemabot stop apply-a1b2c3d4e5f6 -e staging +``` + +_Last updated: 2026-01-01 00:00:00 UTC (2026-01-01 00:00:00 UTC)_ + + + +
+Second Table Throttled + + +## Schema Change Status โ€” Staging + +**Database**: `testapp` | **Apply ID**: `apply-a1b2c3d4e5f6` + +*Applied by @jackjackbits at 2026-01-01 00:00:00 UTC* + +**Status**: In Progress + +๐Ÿ“Š 1/3 complete ยท 1 running (62%) ยท 1 queued + +**Schema `testapp`** + +**`users`**: ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ 62% + +```sql +ALTER TABLE `users` ADD INDEX `idx_email`(`email`); +``` +Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s +โธ Paused by throttler: replica-lag 12s > 10s + +**`products`**: โณ Queued + +```sql +ALTER TABLE `products` ADD INDEX `idx_price`(`price_cents`); +``` + +**`orders`**: ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ โœ… Complete + +```sql +ALTER TABLE `orders` ADD INDEX `idx_user_id`(`user_id`); +``` + + --- To stop this schema change: @@ -4970,6 +5018,36 @@ Sequential mode: First complete, second catching up on accumulated changes ALTER TABLE `users` ADD INDEX `idx_email_created`(`email`, `created_at`); +``` +
+ +
+MySQL: Multi-table Second Table Throttled + +``` + +Sequential mode: First complete, second paused by the engine's throttler + +โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” +โ”‚ Apply ID: apply-a1b2c3d4e5f6 โ”‚ +โ”‚ State: Running โ”‚ +โ”‚ Started: Jan 15 14:10:00 UTC โ”‚ +โ”‚ Duration: 20m โ”‚ +โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ + + + ~ orders: ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ 62% + ALTER TABLE `orders` ADD INDEX `idx_user_status`(`user_id`, `status`); + โ€ข Rows: 3,100,000 / 5,000,000 + โ€ข โธ Paused by throttler: replica-lag 12s > 10s + + ~ products: โณ Queued + ALTER TABLE `products` ADD COLUMN `weight_grams` int DEFAULT 0; + + ~ users: ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ โœ“ Complete + ALTER TABLE `users` ADD INDEX `idx_email_created`(`email`, `created_at`); + + ```
diff --git a/pkg/api/progress_handlers.go b/pkg/api/progress_handlers.go index 1efe59cdb..786aa80a7 100644 --- a/pkg/api/progress_handlers.go +++ b/pkg/api/progress_handlers.go @@ -185,6 +185,8 @@ func progressResponseFromProto(resp *ternv1.ProgressResponse) *apitypes.Progress ETASeconds: t.EtaSeconds, ChecksumRowsChecked: t.ChecksumRowsChecked, ChecksumRowsTotal: t.ChecksumRowsTotal, + Throttled: t.Throttled, + ThrottleReason: t.ThrottleReason, IsInstant: t.IsInstant, ProgressDetail: t.ProgressDetail, TaskID: t.TaskId, @@ -1148,6 +1150,8 @@ func (s *Service) progressFromLocalStorage(ctx context.Context, apply *storage.A PercentComplete: int32(task.ProgressPercent), ChecksumRowsChecked: task.ChecksumRowsChecked, ChecksumRowsTotal: task.ChecksumRowsTotal, + Throttled: task.Throttled, + ThrottleReason: task.ThrottleReason, IsInstant: task.IsInstant, TaskID: task.TaskIdentifier, } @@ -1215,6 +1219,8 @@ func (s *Service) syncTasksFromTern(ctx context.Context, apply *storage.Apply, t task.ProgressPercent = int(tp.PercentComplete) task.ChecksumRowsChecked = tp.ChecksumRowsChecked task.ChecksumRowsTotal = tp.ChecksumRowsTotal + task.Throttled = tp.Throttled + task.ThrottleReason = tp.ThrottleReason task.UpdatedAt = now if err := s.storage.Tasks().Update(ctx, task); err != nil { s.logger.Error("sync task failed", append(task.LogAttrs(), "error", err)...) diff --git a/pkg/apitypes/apitypes.go b/pkg/apitypes/apitypes.go index 5a377085c..3cf5ff5fe 100644 --- a/pkg/apitypes/apitypes.go +++ b/pkg/apitypes/apitypes.go @@ -867,14 +867,19 @@ type TableProgressResponse struct { ETASeconds int64 `json:"eta_seconds,omitempty"` // Checksum phase progress: rows verified so far and total to verify. // Non-zero only while the table is checksumming (verifying copied data). - ChecksumRowsChecked int64 `json:"checksum_rows_checked,omitempty"` - ChecksumRowsTotal int64 `json:"checksum_rows_total,omitempty"` - IsInstant bool `json:"is_instant,omitempty"` - ProgressDetail string `json:"progress_detail,omitempty"` - TaskID string `json:"task_id,omitempty"` - StartedAt string `json:"started_at,omitempty"` - CompletedAt string `json:"completed_at,omitempty"` - Shards []*ShardProgressResponse `json:"shards,omitempty"` + ChecksumRowsChecked int64 `json:"checksum_rows_checked,omitempty"` + ChecksumRowsTotal int64 `json:"checksum_rows_total,omitempty"` + // The engine's throttler is pausing this table's active phase (row copy or + // checksum verify). ThrottleReason names the signal for display and is + // empty when Throttled is false. + Throttled bool `json:"throttled,omitempty"` + ThrottleReason string `json:"throttle_reason,omitempty"` + IsInstant bool `json:"is_instant,omitempty"` + ProgressDetail string `json:"progress_detail,omitempty"` + TaskID string `json:"task_id,omitempty"` + StartedAt string `json:"started_at,omitempty"` + CompletedAt string `json:"completed_at,omitempty"` + Shards []*ShardProgressResponse `json:"shards,omitempty"` } // ShardProgressResponse contains per-shard progress for Vitess schema changes. diff --git a/pkg/cmd/internal/templates/preview_comment.go b/pkg/cmd/internal/templates/preview_comment.go index 66fc30eb9..e3b40f897 100644 --- a/pkg/cmd/internal/templates/preview_comment.go +++ b/pkg/cmd/internal/templates/preview_comment.go @@ -238,6 +238,7 @@ func previewCommentApplyFlowAllOutput() { {"FIRST TABLE RUNNING", func() { fmt.Print(webhooktemplates.PreviewCommentApplyFirstRunning()) }}, {"SECOND TABLE RUNNING", func() { fmt.Print(webhooktemplates.PreviewCommentApplyProgress()) }}, {"SECOND TABLE ESTIMATE EXCEEDED", func() { fmt.Print(webhooktemplates.PreviewCommentApplyEstimateExceeded()) }}, + {"SECOND TABLE THROTTLED", func() { fmt.Print(webhooktemplates.PreviewCommentApplyThrottled()) }}, {"SECOND TABLE CATCHING UP", func() { fmt.Print(webhooktemplates.PreviewCommentApplyCatchingUp()) }}, {"SECOND TABLE CHECKSUMMING", func() { fmt.Print(webhooktemplates.PreviewCommentApplyChecksumming()) }}, {"SECOND TABLE POST-CHECKSUM", func() { fmt.Print(webhooktemplates.PreviewCommentApplyPostChecksum()) }}, @@ -372,6 +373,7 @@ func previewCLIApplyAllOutput() { {"MYSQL: MULTI-TABLE FIRST TABLE RUNNING", previewSeqFirstRunOutput}, {"MYSQL: MULTI-TABLE SECOND TABLE RUNNING", previewSeqSecondRunOutput}, {"MYSQL: MULTI-TABLE SECOND TABLE CATCHING UP", previewSeqCatchingUpOutput}, + {"MYSQL: MULTI-TABLE SECOND TABLE THROTTLED", previewSeqThrottledOutput}, {"MYSQL: MULTI-TABLE SECOND TABLE CHECKSUMMING", previewSeqChecksummingOutput}, {"MYSQL: MULTI-TABLE SECOND TABLE POST-CHECKSUM", previewSeqPostChecksumOutput}, {"MYSQL: MULTI-TABLE THIRD TABLE RUNNING", previewSeqThirdRunOutput}, diff --git a/pkg/cmd/internal/templates/preview_sequential.go b/pkg/cmd/internal/templates/preview_sequential.go index c039d90cd..6a401e35d 100644 --- a/pkg/cmd/internal/templates/preview_sequential.go +++ b/pkg/cmd/internal/templates/preview_sequential.go @@ -107,6 +107,26 @@ func previewSeqThirdRunOutput() { WriteProgress(data) } +func previewSeqThrottledOutput() { + fmt.Println("Sequential mode: First complete, second paused by the engine's throttler") + fmt.Println() + + data := ProgressData{ + State: state.Apply.Running, + Engine: "Spirit", + ApplyID: "apply-a1b2c3d4e5f6", + StartedAt: previewTime.Add(-20 * time.Minute).Format(time.RFC3339), + Tables: []TableProgress{ + {TableName: "users", DDL: seqDDLs[0].ddl, Status: state.Apply.Completed}, + {TableName: "orders", DDL: seqDDLs[1].ddl, Status: state.Task.Running, + RowsCopied: 3100000, RowsTotal: 5000000, PercentComplete: 62, + Throttled: true, ThrottleReason: "replica-lag 12s > 10s"}, + {TableName: "products", DDL: seqDDLs[2].ddl, Status: state.Apply.Pending}, + }, + } + WriteProgress(data) +} + func previewSeqCatchingUpOutput() { fmt.Println("Sequential mode: First complete, second catching up on accumulated changes") fmt.Println() diff --git a/pkg/cmd/internal/templates/progress.go b/pkg/cmd/internal/templates/progress.go index 2a1588208..67220a034 100644 --- a/pkg/cmd/internal/templates/progress.go +++ b/pkg/cmd/internal/templates/progress.go @@ -519,6 +519,7 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel b.WriteString(formatProgressDDL(t.DDL)) } } + writeThrottleLine(&b, t) b.WriteString("\n") b.WriteString(FormatShardProgress(t.Shards)) return b.String() @@ -759,6 +760,7 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel } } + writeThrottleLine(&b, t) if len(t.Shards) == 0 { b.WriteString("\n") } @@ -766,6 +768,21 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel return b.String() } +// writeThrottleLine notes that the engine's throttler is pausing the table's +// active phase, so a stalled bar reads as a deliberate pause rather than a +// hang. The drive clears the stored flag when the pause lifts, so the line +// disappears on the next refresh. +func writeThrottleLine(b *strings.Builder, t TableProgress) { + if !t.Throttled { + return + } + if t.ThrottleReason != "" { + fmt.Fprintf(b, indentDetail+"โธ Paused by throttler: %s\n", t.ThrottleReason) + return + } + fmt.Fprint(b, indentDetail+"โธ Paused by throttler\n") +} + func recoveringIsCopyingRows(t TableProgress) bool { return t.RowsTotal > 0 && t.PercentComplete < 100 } diff --git a/pkg/cmd/internal/templates/progress_parse.go b/pkg/cmd/internal/templates/progress_parse.go index 1099f5ba4..37a350455 100644 --- a/pkg/cmd/internal/templates/progress_parse.go +++ b/pkg/cmd/internal/templates/progress_parse.go @@ -70,9 +70,14 @@ type TableProgress struct { // Non-zero only while the table is checksumming (verifying copied data). ChecksumRowsChecked int64 ChecksumRowsTotal int64 - IsInstant bool - ProgressDetail string // e.g., Spirit: "12.5% copyRows ETA 1h 30m" - Shards []ShardProgress + // The engine's throttler is pausing this table's active phase (row copy + // or checksum verify). ThrottleReason names the signal for display and is + // empty when Throttled is false. + Throttled bool + ThrottleReason string + IsInstant bool + ProgressDetail string // e.g., Spirit: "12.5% copyRows ETA 1h 30m" + Shards []ShardProgress } // ShardProgress contains per-shard progress for template rendering. @@ -185,6 +190,8 @@ func ParseProgressResponse(result *apitypes.ProgressResponse) ProgressData { ETASeconds: tbl.ETASeconds, ChecksumRowsChecked: tbl.ChecksumRowsChecked, ChecksumRowsTotal: tbl.ChecksumRowsTotal, + Throttled: tbl.Throttled, + ThrottleReason: tbl.ThrottleReason, IsInstant: tbl.IsInstant, ProgressDetail: tbl.ProgressDetail, } diff --git a/pkg/cmd/internal/templates/progress_states_test.go b/pkg/cmd/internal/templates/progress_states_test.go index 3b715f08d..a15a8ac1b 100644 --- a/pkg/cmd/internal/templates/progress_states_test.go +++ b/pkg/cmd/internal/templates/progress_states_test.go @@ -387,6 +387,48 @@ func TestFormatTableProgress_Checksumming(t *testing.T) { assert.Contains(t, withProgress, "Rows verified: 321,450 / 1,466,232") } +// A table paused by the engine's throttler names the pause and its trigger, +// so a stalled progress bar reads as a deliberate slowdown (e.g. replica lag) +// rather than a hang. The line renders for the active copy and checksum +// phases only โ€” a throttled flag on a terminal table would be stale. +func TestFormatTableProgress_Throttled(t *testing.T) { + copying := FormatTableProgress(TableProgress{ + TableName: "orders", ChangeType: "alter", Status: state.Apply.Running, + RowsCopied: 45000, RowsTotal: 100000, PercentComplete: 45, + Throttled: true, ThrottleReason: "replica-lag 12s > 10s", + }) + assert.Contains(t, copying, "โธ Paused by throttler: replica-lag 12s > 10s") + + noReason := FormatTableProgress(TableProgress{ + TableName: "orders", ChangeType: "alter", Status: state.Apply.Running, + RowsCopied: 45000, RowsTotal: 100000, PercentComplete: 45, + Throttled: true, + }) + assert.Contains(t, noReason, "โธ Paused by throttler") + assert.NotContains(t, noReason, "Paused by throttler:") + + checksumming := FormatTableProgress(TableProgress{ + TableName: "orders", ChangeType: "alter", Status: state.Task.Checksumming, + ChecksumRowsChecked: 321450, ChecksumRowsTotal: 1466232, + Throttled: true, ThrottleReason: "threads-running 130 > 128", + }) + assert.Contains(t, checksumming, "๐Ÿ” Checksumming to verify data (21%)") + assert.Contains(t, checksumming, "โธ Paused by throttler: threads-running 130 > 128") + + notThrottled := FormatTableProgress(TableProgress{ + TableName: "orders", ChangeType: "alter", Status: state.Apply.Running, + RowsCopied: 45000, RowsTotal: 100000, PercentComplete: 45, + }) + assert.NotContains(t, notThrottled, "Paused by throttler") + + completed := FormatTableProgress(TableProgress{ + TableName: "orders", ChangeType: "alter", Status: state.Apply.Completed, + RowsCopied: 100000, RowsTotal: 100000, PercentComplete: 100, + Throttled: true, ThrottleReason: "replica-lag 12s > 10s", + }) + assert.NotContains(t, completed, "Paused by throttler", "a terminal table never renders a stale throttle flag") +} + func TestFormatTableProgress_InstantDDL(t *testing.T) { tp := TableProgress{ TableName: "users", diff --git a/pkg/engine/engine.go b/pkg/engine/engine.go index 4fd1e28a5..95581a3fe 100644 --- a/pkg/engine/engine.go +++ b/pkg/engine/engine.go @@ -435,6 +435,12 @@ type ProgressResult struct { Tables []TableProgress ResumeState *ResumeState // Updated resume state (engines may update MigrationContext/Metadata during polling) + // ResumedFromCheckpoint reports that this run reattached to a durable + // checkpoint left by an earlier run rather than starting the copy from + // scratch, so preserved progress can be told apart from a fresh restart. + // False for engines without checkpoint resume. + ResumedFromCheckpoint bool + // Metadata carries engine-specific display fields for the progress response // (e.g. PlanetScale branch_name, deploy_request_url, is_instant). It lets the // engine surface structured status to the renderer without core decoding the @@ -480,12 +486,21 @@ type TableProgress struct { // Populated while the table is checksumming (verifying copied data), 0 otherwise. ChecksumRowsChecked int64 ChecksumRowsTotal int64 - Shards []ShardProgress // Per-shard breakdown (for Vitess) - IsInstant bool // True if using instant DDL - ProgressDetail string // Human-readable progress (e.g., Spirit: "12.5% copyRows ETA 1h 30m") - DDL string // The DDL statement being applied - StartedAt *time.Time // When execution actually began (from engine, e.g., SHOW VITESS_MIGRATIONS started_timestamp) - CompletedAt *time.Time // When execution completed (from engine) + // Throttled reports that the engine's throttler is currently pausing the + // phase this table's work is in (the row copy or the checksum verify), so + // stalled row counts read as a deliberate pause rather than a hang. False + // in phases nothing paces, and cleared when the pause lifts. + Throttled bool + // ThrottleReason names the signal pausing the work, for display only + // (e.g. "replica-lag 5s >= 2s"). Empty when Throttled is false or the + // engine cannot explain the pause. + ThrottleReason string + Shards []ShardProgress // Per-shard breakdown (for Vitess) + IsInstant bool // True if using instant DDL + ProgressDetail string // Human-readable progress (e.g., Spirit: "12.5% copyRows ETA 1h 30m") + DDL string // The DDL statement being applied + StartedAt *time.Time // When execution actually began (from engine, e.g., SHOW VITESS_MIGRATIONS started_timestamp) + CompletedAt *time.Time // When execution completed (from engine) } // ShardProgress tracks progress for a single shard. diff --git a/pkg/engine/spirit/spirit.go b/pkg/engine/spirit/spirit.go index 7f50f17e0..b600a57e6 100644 --- a/pkg/engine/spirit/spirit.go +++ b/pkg/engine/spirit/spirit.go @@ -19,6 +19,7 @@ import ( "strings" "sync" "time" + "unicode/utf8" spiritmigration "github.com/block/spirit/pkg/migration" "github.com/block/spirit/pkg/migration/check" @@ -793,12 +794,13 @@ func (e *Engine) Progress(ctx context.Context, req *engine.ProgressRequest) (*en } return &engine.ProgressResult{ - State: state, - Message: message, - ErrorMessage: rm.errorMessage, - Retryable: state == engine.StateFailed, - Tables: tableProgress, - ResumeState: req.ResumeState, + State: state, + Message: message, + ErrorMessage: rm.errorMessage, + Retryable: state == engine.StateFailed, + Tables: tableProgress, + ResumeState: req.ResumeState, + ResumedFromCheckpoint: spiritProgress.Resume, }, nil } @@ -857,11 +859,45 @@ func buildSpiritTableProgress(prog status.Progress, spiritState status.State, dd // runs, so the estimate is stamped on all tables unconditionally. tp.ChecksumRowsChecked = int64(prog.Checksum.RowsChecked) tp.ChecksumRowsTotal = int64(prog.Checksum.RowsTotal) + // Spirit's throttle status is likewise runner-wide and already scoped to + // the paced phases (the row copy and the checksum verify; zero-valued + // everywhere else). Stamp it on the tables participating in that paced + // work โ€” a table still copying, or every table during the verify โ€” so a + // completed table is never rendered as paused by another table's copy. + if tableInPacedPhase(st.IsComplete, spiritState) { + tp.Throttled = prog.Throttle.Throttled + tp.ThrottleReason = sanitizeThrottleReason(prog.Throttle.Reason) + } tableProgress = append(tableProgress, tp) } return tableProgress } +// tableInPacedPhase reports whether a table is doing work the runner's +// throttler paces: its own row copy while incomplete, or the runner-wide +// checksum verify (which runs only after every copy finished). +func tableInPacedPhase(copyComplete bool, spiritState status.State) bool { + return !copyComplete || spiritState == status.Checksum +} + +// sanitizeThrottleReason bounds an engine-produced throttle reason for the +// operator surfaces that render it (PR comment tables, CLI rows): newlines and +// table separators are neutralized and the text is clamped, so a reason can +// never break markdown layout no matter what a throttler reports. +func sanitizeThrottleReason(reason string) string { + reason = strings.Join(strings.Fields(reason), " ") + reason = strings.ReplaceAll(reason, "|", "/") + const maxLen = 200 + if len(reason) > maxLen { + cut := maxLen - len("โ€ฆ") + for cut > 0 && !utf8.RuneStart(reason[cut]) { + cut-- + } + reason = reason[:cut] + "โ€ฆ" + } + return reason +} + // spiritPostCopyPhase reports whether the runner is in one of the active // phases between finishing the row copy and finishing the cutover: applying // the accumulated changeset, restoring deferred indexes, analyzing, verifying diff --git a/pkg/engine/spirit/spirit_test.go b/pkg/engine/spirit/spirit_test.go index dbbe6f7c7..7f083e426 100644 --- a/pkg/engine/spirit/spirit_test.go +++ b/pkg/engine/spirit/spirit_test.go @@ -1,6 +1,7 @@ package spirit import ( + "strings" "testing" "time" @@ -245,6 +246,46 @@ func TestBuildSpiritTableProgress(t *testing.T) { assert.Equal(t, int64(1000), got[0].ChecksumRowsTotal) }) + // Spirit's throttle status is runner-wide but only meaningful to tables + // participating in the paced phase: a table still copying, or every table + // during the checksum verify. A table whose copy finished while others + // still copy must not render as paused by their throttling. + t.Run("throttle reaches copying tables, not completed ones", func(t *testing.T) { + prog := status.Progress{ + Throttle: status.ThrottleStatus{Throttled: true, Reason: "replica-lag 12s > 10s"}, + Tables: []status.TableProgress{ + {TableName: "users", RowsCopied: 45000, RowsTotal: 100000}, + {TableName: "orders", RowsCopied: 1000, RowsTotal: 1000, IsComplete: true}, + }, + } + got := buildSpiritTableProgress(prog, status.CopyRows, ddlByTable, tableNamespace) + require.Len(t, got, 2) + + copying := got[0] + assert.True(t, copying.Throttled) + assert.Equal(t, "replica-lag 12s > 10s", copying.ThrottleReason) + + completed := got[1] + assert.False(t, completed.Throttled, "a finished copy is not paused by another table's throttle") + assert.Empty(t, completed.ThrottleReason) + }) + + // The checksum verify runs only after every copy completes, so a throttled + // verify is stamped on the completed tables โ€” otherwise it would never + // surface at all. + t.Run("throttle reaches completed tables during checksum", func(t *testing.T) { + prog := status.Progress{ + Throttle: status.ThrottleStatus{Throttled: true, Reason: "threads-running 130 > 128"}, + Tables: []status.TableProgress{ + {TableName: "users", RowsCopied: 1000, RowsTotal: 1000, IsComplete: true}, + }, + } + got := buildSpiritTableProgress(prog, status.Checksum, ddlByTable, tableNamespace) + require.Len(t, got, 1) + assert.True(t, got[0].Throttled) + assert.Equal(t, "threads-running 130 > 128", got[0].ThrottleReason) + }) + notReady := []struct { name string eta status.ETA @@ -265,3 +306,32 @@ func TestBuildSpiritTableProgress(t *testing.T) { }) } } + +// TestSanitizeThrottleReason verifies that an engine-produced throttle reason +// is safe for the operator surfaces that render it: whitespace runs collapse, +// markdown table separators are neutralized, and overlong text is clamped so +// a reason can never break a PR comment table or a CLI row. +func TestSanitizeThrottleReason(t *testing.T) { + tests := []struct { + name string + in string + want string + }{ + {"plain reason passes through", "replica-lag 12s > 10s", "replica-lag 12s > 10s"}, + {"empty stays empty", "", ""}, + {"newlines and runs collapse to single spaces", "replica-lag\n12s >\t\t10s", "replica-lag 12s > 10s"}, + {"table separators are neutralized", "signal a > 1 | signal b > 2", "signal a > 1 / signal b > 2"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, sanitizeThrottleReason(tt.in)) + }) + } + + t.Run("overlong reason is clamped with an ellipsis", func(t *testing.T) { + long := strings.Repeat("replica-lag 12s > 10s; ", 20) + got := sanitizeThrottleReason(long) + assert.LessOrEqual(t, len(got), 200) + assert.True(t, strings.HasSuffix(got, "โ€ฆ")) + }) +} diff --git a/pkg/proto/tern.proto b/pkg/proto/tern.proto index 7a7d2f972..79ca2719b 100644 --- a/pkg/proto/tern.proto +++ b/pkg/proto/tern.proto @@ -633,6 +633,13 @@ message TableProgress { // rejection), distinct from the apply-level error_message on // ProgressResponse. Empty when the table has not failed. string error_message = 16; + // The engine's throttler is pausing this table's active phase (row copy or + // checksum verify), so stalled row counts read as a deliberate pause rather + // than a hang. Cleared when the pause lifts. + bool throttled = 17; + // Names the signal pausing the work, for display (e.g. "replica-lag 5s >= + // 2s"). Empty when throttled is false. + string throttle_reason = 18; } // ProgressResponse contains detailed progress information. diff --git a/pkg/proto/ternv1/tern.pb.go b/pkg/proto/ternv1/tern.pb.go index 934ba0988..07ce33171 100644 --- a/pkg/proto/ternv1/tern.pb.go +++ b/pkg/proto/ternv1/tern.pb.go @@ -2288,9 +2288,16 @@ type TableProgress struct { // The table's own failure reason (for example an engine preflight // rejection), distinct from the apply-level error_message on // ProgressResponse. Empty when the table has not failed. - ErrorMessage string `protobuf:"bytes,16,opt,name=error_message,json=errorMessage,proto3" json:"error_message,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + ErrorMessage string `protobuf:"bytes,16,opt,name=error_message,json=errorMessage,proto3" json:"error_message,omitempty"` + // The engine's throttler is pausing this table's active phase (row copy or + // checksum verify), so stalled row counts read as a deliberate pause rather + // than a hang. Cleared when the pause lifts. + Throttled bool `protobuf:"varint,17,opt,name=throttled,proto3" json:"throttled,omitempty"` + // Names the signal pausing the work, for display (e.g. "replica-lag 5s >= + // 2s"). Empty when throttled is false. + ThrottleReason string `protobuf:"bytes,18,opt,name=throttle_reason,json=throttleReason,proto3" json:"throttle_reason,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *TableProgress) Reset() { @@ -2435,6 +2442,20 @@ func (x *TableProgress) GetErrorMessage() string { return "" } +func (x *TableProgress) GetThrottled() bool { + if x != nil { + return x.Throttled + } + return false +} + +func (x *TableProgress) GetThrottleReason() string { + if x != nil { + return x.ThrottleReason + } + return "" +} + // ProgressResponse contains detailed progress information. type ProgressResponse struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -3688,7 +3709,7 @@ const file_tern_proto_rawDesc = "" + "\veta_seconds\x18\x05 \x01(\x03R\n" + "etaSeconds\x12)\n" + "\x10cutover_attempts\x18\x06 \x01(\x05R\x0fcutoverAttempts\x120\n" + - "\x14last_cutover_attempt\x18\a \x01(\tR\x12lastCutoverAttemptJ\x04\b\b\x10\tR\x11ready_to_complete\"\xd2\x04\n" + + "\x14last_cutover_attempt\x18\a \x01(\tR\x12lastCutoverAttemptJ\x04\b\b\x10\tR\x11ready_to_complete\"\x99\x05\n" + "\rTableProgress\x12\x17\n" + "\atask_id\x18\x01 \x01(\tR\x06taskId\x12\x1c\n" + "\tnamespace\x18\x02 \x01(\tR\tnamespace\x12\x1d\n" + @@ -3712,7 +3733,9 @@ const file_tern_proto_rawDesc = "" + "changeType\x122\n" + "\x15checksum_rows_checked\x18\x0e \x01(\x03R\x13checksumRowsChecked\x12.\n" + "\x13checksum_rows_total\x18\x0f \x01(\x03R\x11checksumRowsTotal\x12#\n" + - "\rerror_message\x18\x10 \x01(\tR\ferrorMessage\"\xc7\x03\n" + + "\rerror_message\x18\x10 \x01(\tR\ferrorMessage\x12\x1c\n" + + "\tthrottled\x18\x11 \x01(\bR\tthrottled\x12'\n" + + "\x0fthrottle_reason\x18\x12 \x01(\tR\x0ethrottleReason\"\xc7\x03\n" + "\x10ProgressResponse\x12\x19\n" + "\bapply_id\x18\x01 \x01(\tR\aapplyId\x12$\n" + "\x05state\x18\x02 \x01(\x0e2\x0e.tern.v1.StateR\x05state\x12'\n" + diff --git a/pkg/schema/mysql/tasks.sql b/pkg/schema/mysql/tasks.sql index ceee484ff..49f868ad0 100644 --- a/pkg/schema/mysql/tasks.sql +++ b/pkg/schema/mysql/tasks.sql @@ -25,6 +25,8 @@ CREATE TABLE `tasks` ( `eta_seconds` int DEFAULT NULL, `checksum_rows_checked` bigint DEFAULT '0', `checksum_rows_total` bigint DEFAULT '0', + `throttled` tinyint(1) NOT NULL DEFAULT '0', + `throttle_reason` varchar(255) NOT NULL DEFAULT '', `cutover_attempts` int NOT NULL DEFAULT '0', `is_instant` tinyint(1) DEFAULT '0', `engine_migration_id` varchar(255) DEFAULT NULL, diff --git a/pkg/schema/postgres/tasks.sql b/pkg/schema/postgres/tasks.sql index 3ca0925c7..93f6baa9b 100644 --- a/pkg/schema/postgres/tasks.sql +++ b/pkg/schema/postgres/tasks.sql @@ -25,6 +25,8 @@ CREATE TABLE tasks ( eta_seconds integer DEFAULT NULL, checksum_rows_checked bigint DEFAULT 0, checksum_rows_total bigint DEFAULT 0, + throttled boolean NOT NULL DEFAULT FALSE, + throttle_reason varchar(255) NOT NULL DEFAULT '', cutover_attempts integer NOT NULL DEFAULT 0, is_instant boolean DEFAULT FALSE, engine_migration_id varchar(255) DEFAULT NULL, diff --git a/pkg/storage/internal/sqlstore/tasks.go b/pkg/storage/internal/sqlstore/tasks.go index 35d1eea23..b0cb5194c 100644 --- a/pkg/storage/internal/sqlstore/tasks.go +++ b/pkg/storage/internal/sqlstore/tasks.go @@ -18,7 +18,7 @@ import ( const taskColumns = `id, task_identifier, apply_id, apply_operation_id, plan_id, database_name, database_type, namespace, table_name, shard, ddl, ddl_action, engine, repository, pull_request, environment, state, error_message, options, attempt, - rows_copied, rows_total, progress_percent, eta_seconds, checksum_rows_checked, checksum_rows_total, cutover_attempts, + rows_copied, rows_total, progress_percent, eta_seconds, checksum_rows_checked, checksum_rows_total, throttled, throttle_reason, cutover_attempts, is_instant, engine_migration_id, started_at, completed_at, created_at, updated_at` @@ -62,16 +62,16 @@ func insertTask(ctx context.Context, exec queryExecer, identity identityInserter task_identifier, apply_id, apply_operation_id, plan_id, database_name, database_type, namespace, table_name, shard, ddl, ddl_action, engine, repository, pull_request, environment, state, error_message, options, attempt, - rows_copied, rows_total, progress_percent, eta_seconds, checksum_rows_checked, checksum_rows_total, cutover_attempts, + rows_copied, rows_total, progress_percent, eta_seconds, checksum_rows_checked, checksum_rows_total, throttled, throttle_reason, cutover_attempts, is_instant, engine_migration_id, started_at, completed_at, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `, task.TaskIdentifier, task.ApplyID, nullInt64Ptr(task.ApplyOperationID), task.PlanID, task.Database, task.DatabaseType, task.Namespace, nullString(task.TableName), task.Shard, nullString(task.DDL), nullString(task.DDLAction), task.Engine, task.Repository, task.PullRequest, task.Environment, task.State, nullString(task.ErrorMessage), string(options), task.Attempt, - task.RowsCopied, task.RowsTotal, task.ProgressPercent, task.ETASeconds, task.ChecksumRowsChecked, task.ChecksumRowsTotal, task.CutoverAttempts, + task.RowsCopied, task.RowsTotal, task.ProgressPercent, task.ETASeconds, task.ChecksumRowsChecked, task.ChecksumRowsTotal, task.Throttled, task.ThrottleReason, task.CutoverAttempts, task.IsInstant, nullString(task.EngineMigrationID), task.StartedAt, task.CompletedAt, task.CreatedAt, task.UpdatedAt, ) @@ -103,7 +103,7 @@ func (s *taskStore) Get(ctx context.Context, taskIdentifier string) (*storage.Ta func (s *taskStore) Update(ctx context.Context, task *storage.Task) error { args := []any{ task.State, nullString(task.ErrorMessage), nullJSON(task.Options), task.Attempt, - task.RowsCopied, task.RowsTotal, task.ProgressPercent, task.ETASeconds, task.ChecksumRowsChecked, task.ChecksumRowsTotal, task.CutoverAttempts, + task.RowsCopied, task.RowsTotal, task.ProgressPercent, task.ETASeconds, task.ChecksumRowsChecked, task.ChecksumRowsTotal, task.Throttled, task.ThrottleReason, task.CutoverAttempts, task.IsInstant, nullString(task.EngineMigrationID), task.StartedAt, task.CompletedAt, task.ID, @@ -138,7 +138,7 @@ func (s *taskStore) Update(ctx context.Context, task *storage.Task) error { result, err := s.db.ExecContext(ctx, ` UPDATE tasks SET state = ?, error_message = ?, options = ?, attempt = ?, - rows_copied = ?, rows_total = ?, progress_percent = ?, eta_seconds = ?, checksum_rows_checked = ?, checksum_rows_total = ?, cutover_attempts = ?, + rows_copied = ?, rows_total = ?, progress_percent = ?, eta_seconds = ?, checksum_rows_checked = ?, checksum_rows_total = ?, throttled = ?, throttle_reason = ?, cutover_attempts = ?, is_instant = ?, engine_migration_id = ?, started_at = ?, completed_at = ?, updated_at = NOW() WHERE id = ?`+leasePredicate+` @@ -269,7 +269,7 @@ const shardTaskInsertColumns = ` task_identifier, apply_id, apply_operation_id, plan_id, database_name, database_type, namespace, table_name, shard, ddl, ddl_action, engine, repository, pull_request, environment, state, error_message, options, attempt, - rows_copied, rows_total, progress_percent, eta_seconds, checksum_rows_checked, checksum_rows_total, cutover_attempts, + rows_copied, rows_total, progress_percent, eta_seconds, checksum_rows_checked, checksum_rows_total, throttled, throttle_reason, cutover_attempts, is_instant, engine_migration_id, started_at, completed_at, created_at, updated_at` @@ -282,13 +282,13 @@ func shardTaskInsertValues(task *storage.Task) (string, []any) { if len(options) == 0 { options = []byte("{}") } - return `?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?`, + return `?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?`, []any{ task.TaskIdentifier, task.ApplyID, nullInt64Ptr(task.ApplyOperationID), task.PlanID, task.Database, task.DatabaseType, task.Namespace, nullString(task.TableName), task.Shard, nullString(task.DDL), nullString(task.DDLAction), task.Engine, task.Repository, task.PullRequest, task.Environment, task.State, nullString(task.ErrorMessage), string(options), task.Attempt, - task.RowsCopied, task.RowsTotal, task.ProgressPercent, task.ETASeconds, task.ChecksumRowsChecked, task.ChecksumRowsTotal, task.CutoverAttempts, + task.RowsCopied, task.RowsTotal, task.ProgressPercent, task.ETASeconds, task.ChecksumRowsChecked, task.ChecksumRowsTotal, task.Throttled, task.ThrottleReason, task.CutoverAttempts, task.IsInstant, nullString(task.EngineMigrationID), task.StartedAt, task.CompletedAt, task.CreatedAt, task.UpdatedAt, } @@ -660,6 +660,8 @@ func scanTaskInto(s scanner) (*storage.Task, error) { &etaSeconds, &task.ChecksumRowsChecked, &task.ChecksumRowsTotal, + &task.Throttled, + &task.ThrottleReason, &task.CutoverAttempts, &task.IsInstant, &engineMigrationID, diff --git a/pkg/storage/internal/sqlstore/tasks_test.go b/pkg/storage/internal/sqlstore/tasks_test.go index 23f16114b..585cb1292 100644 --- a/pkg/storage/internal/sqlstore/tasks_test.go +++ b/pkg/storage/internal/sqlstore/tasks_test.go @@ -310,6 +310,55 @@ func TestTaskStore_PerShardTaskRoundTrip(t *testing.T) { assert.Equal(t, "-80", reloaded.Shard, "shard is fixed at creation, not changed by Update") } +// A table paused by the engine's throttler stores the pause flag and its +// display reason on the task row, so readers (PR comment, CLI) render the +// pause from storage without polling the engine. Both fields are written by +// the drive's progress sync every tick: they set while the pause is active and +// clear on the first unpaced tick. +func TestTaskStore_ThrottleRoundTrip(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := NewMySQL(testDB) + + lock := createTestLock(t, store, "testapp", "mysql", "staging") + apply := createTestApply(t, store, lock, "apply_throttle", 1) + + now := time.Now() + _, err := store.Tasks().Create(ctx, &storage.Task{ + TaskIdentifier: "task_throttle", + ApplyID: apply.ID, + PlanID: apply.PlanID, + Database: apply.Database, + DatabaseType: apply.DatabaseType, + Engine: storage.EngineSpirit, + Environment: apply.Environment, + State: state.Task.Running, + Namespace: "testapp", + TableName: "users", + DDL: "ALTER TABLE `users` ADD COLUMN `email` varchar(255)", + DDLAction: "ALTER", + Throttled: true, + ThrottleReason: "replica-lag 12s > 10s", + CreatedAt: now, + UpdatedAt: now, + }) + require.NoError(t, err) + + got, err := store.Tasks().Get(ctx, "task_throttle") + require.NoError(t, err) + require.NotNil(t, got) + assert.True(t, got.Throttled) + assert.Equal(t, "replica-lag 12s > 10s", got.ThrottleReason) + + got.Throttled = false + got.ThrottleReason = "" + require.NoError(t, store.Tasks().Update(ctx, got)) + cleared, err := store.Tasks().Get(ctx, "task_throttle") + require.NoError(t, err) + assert.False(t, cleared.Throttled, "the pause clears when the throttler releases") + assert.Empty(t, cleared.ThrottleReason) +} + // A sharded work operation's operation key identifies which shard task is real // drive input. Other shard rows remain progress detail and must not be replayed // as extra table changes if the operation is resumed. diff --git a/pkg/storage/types.go b/pkg/storage/types.go index 18008788f..02004171d 100644 --- a/pkg/storage/types.go +++ b/pkg/storage/types.go @@ -1199,7 +1199,14 @@ type Task struct { // Non-zero only while the task is checksumming (verifying copied data). ChecksumRowsChecked int64 ChecksumRowsTotal int64 - CutoverAttempts int // Number of cutover attempts for this shard + // Throttled reports that the engine's throttler is pausing this table's + // active phase (row copy or checksum verify), so stalled row counts read + // as a deliberate pause rather than a hang. Cleared when the pause lifts. + Throttled bool + // ThrottleReason names the signal pausing the work, for display (e.g. + // "replica-lag 5s >= 2s"). Empty when Throttled is false. + ThrottleReason string + CutoverAttempts int // Number of cutover attempts for this shard // Execution flags IsInstant bool // True if INSTANT DDL (no copy needed) diff --git a/pkg/tern/grpc_client.go b/pkg/tern/grpc_client.go index 8d2beb700..01d9cb5bc 100644 --- a/pkg/tern/grpc_client.go +++ b/pkg/tern/grpc_client.go @@ -3294,6 +3294,8 @@ func (c *GRPCClient) syncStoredTasksFromRemoteTasks( storedTask.ETASeconds = int(remoteTask.EtaSeconds) storedTask.ChecksumRowsChecked = remoteTask.ChecksumRowsChecked storedTask.ChecksumRowsTotal = remoteTask.ChecksumRowsTotal + storedTask.Throttled = remoteTask.Throttled + storedTask.ThrottleReason = remoteTask.ThrottleReason } // Adopt the remote task's own failure reason (for example an engine // preflight rejection) so the operation row derived from the stored task diff --git a/pkg/tern/grpc_client_test.go b/pkg/tern/grpc_client_test.go index ce089a103..e5d8d465a 100644 --- a/pkg/tern/grpc_client_test.go +++ b/pkg/tern/grpc_client_test.go @@ -4550,6 +4550,8 @@ func TestGRPCClient_SyncRemoteProgressMirrorsRowCopySnapshot(t *testing.T) { EtaSeconds: 340, ChecksumRowsChecked: 200, ChecksumRowsTotal: 1000, + Throttled: true, + ThrottleReason: "replica-lag 12s > 10s", }, }, time.Now()) require.NoError(t, err) @@ -4561,6 +4563,8 @@ func TestGRPCClient_SyncRemoteProgressMirrorsRowCopySnapshot(t *testing.T) { assert.Equal(t, 340, task.ETASeconds) assert.Equal(t, int64(200), task.ChecksumRowsChecked) assert.Equal(t, int64(1000), task.ChecksumRowsTotal) + assert.True(t, task.Throttled, "throttle pause mirrors from the remote table progress") + assert.Equal(t, "replica-lag 12s > 10s", task.ThrottleReason) } func TestGRPCClient_PollSetsTerminalTaskMetadataFromRemoteTaskProgress(t *testing.T) { diff --git a/pkg/tern/local_apply.go b/pkg/tern/local_apply.go index 45518e171..8a1c3cbad 100644 --- a/pkg/tern/local_apply.go +++ b/pkg/tern/local_apply.go @@ -281,6 +281,23 @@ func (c *LocalClient) logApplyEvent(ctx context.Context, applyID int64, taskID * } } +// logEngineResumeOnce records a timeline event the first time a drive claim +// observes the engine reattached to a durable checkpoint instead of starting +// the copy fresh. One event per drive claim is intentional: every claim that +// resumes (pod restart, lease handover, operator start after a stop) is a +// real engine resume the operator should see on the timeline. eventLogged is +// the per-drive latch; the durable checkpoint itself lives in the engine. +func (c *LocalClient) logEngineResumeOnce(ctx context.Context, logger *slog.Logger, apply *storage.Apply, resumed bool, eventLogged *bool) { + if !resumed || *eventLogged { + return + } + *eventLogged = true + const msg = "Engine resumed from checkpoint; row copy continues from durable progress" + c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelInfo, storage.LogEventInfo, storage.LogSourceSchemaBot, + msg, "", "") + logger.Info("engine resumed from checkpoint") +} + // setupSpiritLogging wires up Spirit's log callback to route engine logs to the apply_logs table. // Returns a cleanup function that must be deferred. func (c *LocalClient) setupSpiritLogging(ctx context.Context, apply *storage.Apply, tasks []*storage.Task) func() { diff --git a/pkg/tern/local_apply_grouped.go b/pkg/tern/local_apply_grouped.go index 5688fff7c..20afe4f0b 100644 --- a/pkg/tern/local_apply_grouped.go +++ b/pkg/tern/local_apply_grouped.go @@ -661,6 +661,7 @@ func (c *LocalClient) handleAtomicProgressTick(ctx context.Context, eng engine.E return false } ps.consecutiveErrors = 0 + c.logEngineResumeOnce(ctx, logger, apply, result.ResumedFromCheckpoint, &ps.resumeEventLogged) // Update resumeState if the engine returned a newer one (e.g., with // updated metadata like deploy request URL or migration context). @@ -1265,6 +1266,8 @@ func (c *LocalClient) syncAtomicTaskProgress(ctx context.Context, logger *slog.L task.ETASeconds = int(tp.ETASeconds) task.ChecksumRowsChecked = tp.ChecksumRowsChecked task.ChecksumRowsTotal = tp.ChecksumRowsTotal + task.Throttled = tp.Throttled + task.ThrottleReason = tp.ThrottleReason task.IsInstant = tp.IsInstant if tp.StartedAt != nil && task.StartedAt == nil { task.StartedAt = tp.StartedAt diff --git a/pkg/tern/local_apply_sequential.go b/pkg/tern/local_apply_sequential.go index bd81e52a5..fa180d0ae 100644 --- a/pkg/tern/local_apply_sequential.go +++ b/pkg/tern/local_apply_sequential.go @@ -296,6 +296,11 @@ type atomicPollState struct { // revertSkipped is set after SkipRevert is called to prevent repeated calls. revertSkipped bool + // resumeEventLogged is set after this drive claim records the + // engine-resumed-from-checkpoint timeline event, so the flag the engine + // reports on every subsequent poll produces one event per claim. + resumeEventLogged bool + // cutoverTriggerLogged is set after the drive records the auto-cutover // trigger event, so retries of a not-yet-accepted cutover do not fill the // user-visible timeline with duplicate triggers. @@ -449,6 +454,7 @@ func (c *LocalClient) pollTaskToCompletion(ctx context.Context, apply *storage.A logger := c.logger.With(apply.IdentityLogAttrs()...) var consecutiveErrors int + var resumeEventLogged bool for { select { @@ -528,6 +534,7 @@ func (c *LocalClient) pollTaskToCompletion(ctx context.Context, apply *storage.A continue } consecutiveErrors = 0 + c.logEngineResumeOnce(ctx, logger, apply, result.ResumedFromCheckpoint, &resumeEventLogged) now := time.Now() prevState := task.State @@ -556,6 +563,8 @@ func (c *LocalClient) pollTaskToCompletion(ctx context.Context, apply *storage.A task.ETASeconds = int(tp.ETASeconds) task.ChecksumRowsChecked = tp.ChecksumRowsChecked task.ChecksumRowsTotal = tp.ChecksumRowsTotal + task.Throttled = tp.Throttled + task.ThrottleReason = tp.ThrottleReason task.IsInstant = tp.IsInstant } diff --git a/pkg/tern/local_client.go b/pkg/tern/local_client.go index a83d566c2..1c07269a5 100644 --- a/pkg/tern/local_client.go +++ b/pkg/tern/local_client.go @@ -2451,6 +2451,8 @@ func (c *LocalClient) Progress(ctx context.Context, req *ternv1.ProgressRequest) tp.RowsTotal = t.RowsTotal tp.ChecksumRowsChecked = t.ChecksumRowsChecked tp.ChecksumRowsTotal = t.ChecksumRowsTotal + tp.Throttled = t.Throttled + tp.ThrottleReason = t.ThrottleReason // For Spirit the stored figure is the runner-wide remaining-copy // estimate stamped on every still-copying table (see // buildSpiritTableProgress), so in a multi-table apply each table diff --git a/pkg/tern/local_client_test.go b/pkg/tern/local_client_test.go index 59ee9af51..7d53e4bf3 100644 --- a/pkg/tern/local_client_test.go +++ b/pkg/tern/local_client_test.go @@ -1997,6 +1997,38 @@ func TestLocalClient_ProcessPendingCutoverControlRequestRetriesWhenCutoverNotRea // The drive's auto-cutover records one trigger in the timeline per drive and // retries quietly while the engine backend stages the cutover. A rejection // that outlives the staging window records a one-time timeline error and +// A drive claim that reattaches to the engine's durable checkpoint records one +// timeline event, so an operator can tell a resumed copy from a fresh start. +// The engine reports the resume flag on every subsequent poll; the per-drive +// latch keeps the timeline to one event per claim. +func TestLocalClient_LogEngineResumeOnce(t *testing.T) { + apply := &storage.Apply{ + ID: 7, + ApplyIdentifier: "apply-resume", + Database: "testdb", + Environment: "staging", + } + logs := &mockApplyLogStore{} + client := &LocalClient{ + storage: &exactProgressStorage{logs: logs}, + logger: slog.Default(), + } + + var latch bool + client.logEngineResumeOnce(t.Context(), slog.Default(), apply, false, &latch) + assert.Empty(t, logs.logs, "a fresh start records no resume event") + assert.False(t, latch) + + client.logEngineResumeOnce(t.Context(), slog.Default(), apply, true, &latch) + require.Len(t, logs.logs, 1) + assert.Equal(t, storage.LogEventInfo, logs.logs[0].EventType) + assert.Equal(t, storage.LogLevelInfo, logs.logs[0].Level) + assert.Contains(t, logs.logs[0].Message, "resumed from checkpoint") + + client.logEngineResumeOnce(t.Context(), slog.Default(), apply, true, &latch) + assert.Len(t, logs.logs, 1, "the flag on every later poll produces one event per drive claim") +} + // escalates to Error logging on every further tick, so a backend that never // stages the cutover is visible to operators instead of idling; the retry // loop itself never stops, and an accepted cutover resets the escalation. @@ -3892,6 +3924,8 @@ func TestLocalClient_ProgressRendersNonShardedTableFromStoredTask(t *testing.T) ETASeconds: 90, ChecksumRowsChecked: 10, ChecksumRowsTotal: 1000, + Throttled: true, + ThrottleReason: "replica-lag 12s > 10s", } client := &LocalClient{ config: LocalConfig{Database: "testdb", Type: storage.DatabaseTypeMySQL}, @@ -3914,6 +3948,8 @@ func TestLocalClient_ProgressRendersNonShardedTableFromStoredTask(t *testing.T) assert.Equal(t, int64(90), tp.EtaSeconds, "ETA comes from the stored task row") assert.Equal(t, int64(10), tp.ChecksumRowsChecked) assert.Equal(t, int64(1000), tp.ChecksumRowsTotal) + assert.True(t, tp.Throttled, "throttle pause comes from the stored task row") + assert.Equal(t, "replica-lag 12s > 10s", tp.ThrottleReason) assert.Empty(t, tp.Shards, "a non-sharded table has no per-shard breakdown") } diff --git a/pkg/webhook/apply.go b/pkg/webhook/apply.go index 0eddfaaf7..626dc74fa 100644 --- a/pkg/webhook/apply.go +++ b/pkg/webhook/apply.go @@ -133,6 +133,8 @@ func tableProgressFromTasks(databaseFallback string, tasks []*storage.Task, shar ETASeconds: int64(t.ETASeconds), ChecksumRowsChecked: t.ChecksumRowsChecked, ChecksumRowsTotal: t.ChecksumRowsTotal, + Throttled: t.Throttled, + ThrottleReason: t.ThrottleReason, IsInstant: t.IsInstant, ErrorMessage: t.ErrorMessage, Shards: shardProgressForTable(shardsByTable, t.ApplyOperationID, t.Namespace, t.TableName), diff --git a/pkg/webhook/templates/apply.go b/pkg/webhook/templates/apply.go index 5a60ec258..165183a96 100644 --- a/pkg/webhook/templates/apply.go +++ b/pkg/webhook/templates/apply.go @@ -29,7 +29,12 @@ type TableProgressData struct { // Non-zero only while the table is checksumming (verifying copied data). ChecksumRowsChecked int64 ChecksumRowsTotal int64 - IsInstant bool + // The engine's throttler is pausing this table's active phase (row copy + // or checksum verify). ThrottleReason names the signal for display and is + // empty when Throttled is false. + Throttled bool + ThrottleReason string + IsInstant bool // ErrorMessage is the task's last error. Rendered for states where the // per-table error explains what the user is seeing (e.g. a retrying or @@ -727,6 +732,7 @@ func renderTableProgress(sb *strings.Builder, table TableProgressData, applyAtte fmt.Fprintf(sb, "**`%s`**: %s \U0001f50d Checksumming to verify data...\n", table.TableName, ui.ProgressBarRowCopy(100)) writeDDLLine(sb, table.DDL) } + writeThrottleLine(sb, table) case state.Task.PostChecksum: // The verify passed and the engine is draining the changes that @@ -930,6 +936,7 @@ func isCopyingShardStatus(status string) bool { // renderRunningTable renders a table that is actively copying rows. func renderRunningTable(sb *strings.Builder, table TableProgressData) { + defer writeThrottleLine(sb, table) if table.RowsTotal > 0 { if ui.EstimateExceeded(table.RowsCopied, table.RowsTotal) { fmt.Fprintf(sb, "**`%s`**: %s Finalizing copy\n", table.TableName, ui.ProgressBarActivity()) @@ -960,6 +967,22 @@ func renderRunningTable(sb *strings.Builder, table TableProgressData) { } } +// writeThrottleLine notes that the engine's throttler is pausing the table's +// active phase, so a stalled bar reads as a deliberate pause rather than a +// hang. The drive clears the stored flag when the pause lifts, so the line +// disappears on the next refresh. The reason is sanitized at the engine +// boundary before it is stored, so it is safe to render in markdown. +func writeThrottleLine(sb *strings.Builder, table TableProgressData) { + if !table.Throttled { + return + } + if table.ThrottleReason != "" { + fmt.Fprintf(sb, "โธ Paused by throttler: %s\n", table.ThrottleReason) + return + } + sb.WriteString("โธ Paused by throttler\n") +} + func recoveringIsCopyingRows(table TableProgressData) bool { return table.RowsTotal > 0 && table.PercentComplete < 100 } diff --git a/pkg/webhook/templates/apply_test.go b/pkg/webhook/templates/apply_test.go index de3cc07b8..d851319be 100644 --- a/pkg/webhook/templates/apply_test.go +++ b/pkg/webhook/templates/apply_test.go @@ -260,6 +260,64 @@ func TestRenderApplyStatusComment_Checksumming(t *testing.T) { assert.Contains(t, result, "1 checksumming") } +// A table paused by the engine's throttler names the pause and its trigger in +// the PR comment, so a stalled progress bar reads as a deliberate slowdown +// (e.g. replica lag) rather than a hang. The reason is sanitized at the engine +// boundary, and the line renders only on active tables โ€” a throttled flag on a +// terminal table would be stale. +func TestRenderApplyStatusComment_Throttled(t *testing.T) { + data := ApplyStatusCommentData{ + Database: "testapp", + Environment: "staging", + RequestedBy: "aparajon", + State: "running", + Engine: "Spirit", + Tables: []TableProgressData{ + {TableName: "orders", DDL: "ALTER TABLE `orders` ADD INDEX `idx_user_id` (`user_id`)", Status: "running", + RowsCopied: 45000, RowsTotal: 100000, PercentComplete: 45, + Throttled: true, ThrottleReason: "replica-lag 12s > 10s"}, + {TableName: "users", DDL: "ALTER TABLE `users` ADD INDEX `idx_email` (`email`)", Status: "pending"}, + }, + } + + result := RenderApplyStatusComment(data) + + assert.Contains(t, result, "**`orders`**") + assert.Contains(t, result, "โธ Paused by throttler: replica-lag 12s > 10s") + + data.Tables[0].ThrottleReason = "" + noReason := RenderApplyStatusComment(data) + assert.Contains(t, noReason, "โธ Paused by throttler") + assert.NotContains(t, noReason, "Paused by throttler:") + + data.Tables[0].Throttled = false + notThrottled := RenderApplyStatusComment(data) + assert.NotContains(t, notThrottled, "Paused by throttler") +} + +// A throttled checksum verify surfaces the pause alongside the verify +// progress, since the checksum is the other phase the engine's throttler +// paces. +func TestRenderApplyStatusComment_ThrottledChecksumming(t *testing.T) { + data := ApplyStatusCommentData{ + Database: "testapp", + Environment: "staging", + RequestedBy: "aparajon", + State: "running", + Engine: "Spirit", + Tables: []TableProgressData{ + {TableName: "orders", DDL: "ALTER TABLE `orders` ADD INDEX `idx_user_id` (`user_id`)", Status: "checksumming", + ChecksumRowsChecked: 321450, ChecksumRowsTotal: 1466232, + Throttled: true, ThrottleReason: "threads-running 130 > 128"}, + }, + } + + result := RenderApplyStatusComment(data) + + assert.Contains(t, result, "๐Ÿ” Checksumming to verify data (21%)") + assert.Contains(t, result, "โธ Paused by throttler: threads-running 130 > 128") +} + func TestUnsafeDropIndexUsageTargets(t *testing.T) { tests := []struct { name string diff --git a/pkg/webhook/templates/preview.go b/pkg/webhook/templates/preview.go index 182dd91f7..5ef31f2b4 100644 --- a/pkg/webhook/templates/preview.go +++ b/pkg/webhook/templates/preview.go @@ -1174,6 +1174,23 @@ func PreviewCommentApplyProgress() string { return RenderApplyStatusComment(sampleApplyData(state.Apply.Running, tables)) } +// PreviewCommentApplyThrottled renders an apply comment where the active row +// copy is paused by the engine's throttler, so the stalled bar reads as a +// deliberate slowdown with its trigger named rather than a hang. +func PreviewCommentApplyThrottled() string { + tables := sampleApplyTables() + tables[0].Status = state.Task.Completed + tables[1].Status = state.Task.Running + tables[1].RowsCopied = 914707 + tables[1].RowsTotal = 1466232 + tables[1].PercentComplete = 62 + tables[1].ETASeconds = 195 + tables[1].Throttled = true + tables[1].ThrottleReason = "replica-lag 12s > 10s" + tables[2].Status = state.Task.Pending + return RenderApplyStatusComment(sampleApplyData(state.Apply.Running, tables)) +} + // PreviewCommentApplyCatchingUp renders an apply comment where a table has // finished copying and is draining the changes that accumulated during the copy. func PreviewCommentApplyCatchingUp() string { From 62627fed9e5a67752d3982b6d61c2315bd3e5763 Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sat, 15 Aug 2026 15:09:05 +0800 Subject: [PATCH 03/12] fix(github): stamp the throttle reason only with the throttled flag An unthrottled table must carry no reason, even if the runner reports leftover reason text, so a stale reason can never leak through the read model with the flag down. Co-Authored-By: Claude Fable 5 --- pkg/engine/spirit/spirit.go | 6 ++++-- pkg/engine/spirit/spirit_test.go | 13 +++++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/pkg/engine/spirit/spirit.go b/pkg/engine/spirit/spirit.go index b600a57e6..6984b17ac 100644 --- a/pkg/engine/spirit/spirit.go +++ b/pkg/engine/spirit/spirit.go @@ -864,8 +864,10 @@ func buildSpiritTableProgress(prog status.Progress, spiritState status.State, dd // everywhere else). Stamp it on the tables participating in that paced // work โ€” a table still copying, or every table during the verify โ€” so a // completed table is never rendered as paused by another table's copy. - if tableInPacedPhase(st.IsComplete, spiritState) { - tp.Throttled = prog.Throttle.Throttled + // The reason is stamped only with the flag, keeping the contract that + // an unthrottled table carries no reason. + if tableInPacedPhase(st.IsComplete, spiritState) && prog.Throttle.Throttled { + tp.Throttled = true tp.ThrottleReason = sanitizeThrottleReason(prog.Throttle.Reason) } tableProgress = append(tableProgress, tp) diff --git a/pkg/engine/spirit/spirit_test.go b/pkg/engine/spirit/spirit_test.go index 7f083e426..068dafc3b 100644 --- a/pkg/engine/spirit/spirit_test.go +++ b/pkg/engine/spirit/spirit_test.go @@ -270,6 +270,19 @@ func TestBuildSpiritTableProgress(t *testing.T) { assert.Empty(t, completed.ThrottleReason) }) + // The reason travels only with the flag: an unthrottled table carries no + // reason, even if the runner reports leftover reason text. + t.Run("a reason without the throttled flag is not stamped", func(t *testing.T) { + prog := status.Progress{ + Throttle: status.ThrottleStatus{Throttled: false, Reason: "replica-lag 2s > 10s"}, + Tables: []status.TableProgress{{TableName: "users", RowsCopied: 45000, RowsTotal: 100000}}, + } + got := buildSpiritTableProgress(prog, status.CopyRows, ddlByTable, tableNamespace) + require.Len(t, got, 1) + assert.False(t, got[0].Throttled) + assert.Empty(t, got[0].ThrottleReason) + }) + // The checksum verify runs only after every copy completes, so a throttled // verify is stamped on the completed tables โ€” otherwise it would never // surface at all. From 448d635deeadb2fd6cc77f4d015f079d7696c3a3 Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sat, 15 Aug 2026 15:13:29 +0800 Subject: [PATCH 04/12] fix(github): render the throttle pause as its own paragraph A single newline merges into the preceding rows/ETA line wherever standard markdown line-break rules apply, so the pause gets a blank line before it and stands apart from the progress detail. Co-Authored-By: Claude Fable 5 --- TEMPLATES.md | 1 + pkg/webhook/templates/apply.go | 8 +++++--- pkg/webhook/templates/apply_test.go | 7 ++++--- 3 files changed, 10 insertions(+), 6 deletions(-) diff --git a/TEMPLATES.md b/TEMPLATES.md index 9b893b80b..442e8f19a 100644 --- a/TEMPLATES.md +++ b/TEMPLATES.md @@ -2861,6 +2861,7 @@ _Last updated: 2026-01-01 00:00:0 ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s + โธ Paused by throttler: replica-lag 12s > 10s **`products`**: โณ Queued diff --git a/pkg/webhook/templates/apply.go b/pkg/webhook/templates/apply.go index 165183a96..a929c0931 100644 --- a/pkg/webhook/templates/apply.go +++ b/pkg/webhook/templates/apply.go @@ -971,16 +971,18 @@ func renderRunningTable(sb *strings.Builder, table TableProgressData) { // active phase, so a stalled bar reads as a deliberate pause rather than a // hang. The drive clears the stored flag when the pause lifts, so the line // disappears on the next refresh. The reason is sanitized at the engine -// boundary before it is stored, so it is safe to render in markdown. +// boundary before it is stored, so it is safe to render in markdown. The +// pause is its own paragraph so it never merges into the preceding progress +// detail line, whichever markdown line-break mode renders the comment. func writeThrottleLine(sb *strings.Builder, table TableProgressData) { if !table.Throttled { return } if table.ThrottleReason != "" { - fmt.Fprintf(sb, "โธ Paused by throttler: %s\n", table.ThrottleReason) + fmt.Fprintf(sb, "\nโธ Paused by throttler: %s\n", table.ThrottleReason) return } - sb.WriteString("โธ Paused by throttler\n") + sb.WriteString("\nโธ Paused by throttler\n") } func recoveringIsCopyingRows(table TableProgressData) bool { diff --git a/pkg/webhook/templates/apply_test.go b/pkg/webhook/templates/apply_test.go index d851319be..78df8b83f 100644 --- a/pkg/webhook/templates/apply_test.go +++ b/pkg/webhook/templates/apply_test.go @@ -283,11 +283,12 @@ func TestRenderApplyStatusComment_Throttled(t *testing.T) { result := RenderApplyStatusComment(data) assert.Contains(t, result, "**`orders`**") - assert.Contains(t, result, "โธ Paused by throttler: replica-lag 12s > 10s") + assert.Contains(t, result, "\n\nโธ Paused by throttler: replica-lag 12s > 10s", + "the pause is its own paragraph, not a continuation of the progress detail line") data.Tables[0].ThrottleReason = "" noReason := RenderApplyStatusComment(data) - assert.Contains(t, noReason, "โธ Paused by throttler") + assert.Contains(t, noReason, "\n\nโธ Paused by throttler") assert.NotContains(t, noReason, "Paused by throttler:") data.Tables[0].Throttled = false @@ -315,7 +316,7 @@ func TestRenderApplyStatusComment_ThrottledChecksumming(t *testing.T) { result := RenderApplyStatusComment(data) assert.Contains(t, result, "๐Ÿ” Checksumming to verify data (21%)") - assert.Contains(t, result, "โธ Paused by throttler: threads-running 130 > 128") + assert.Contains(t, result, "\n\nโธ Paused by throttler: threads-running 130 > 128") } func TestUnsafeDropIndexUsageTargets(t *testing.T) { From bdb445f0e5f2f32113882ae8e942644828cecdf1 Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sat, 15 Aug 2026 15:19:03 +0800 Subject: [PATCH 05/12] fix(github): say Throttled, not Paused, on the throttle line MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pause reads as stopped and collides with the operator stop vocabulary; the throttler is backpressure โ€” the copy is slowed and resumes on its own โ€” so the line names it that way on both surfaces. Co-Authored-By: Claude Fable 5 --- TEMPLATES.md | 4 ++-- pkg/cmd/internal/templates/progress.go | 13 +++++++------ .../templates/progress_states_test.go | 12 ++++++------ pkg/webhook/templates/apply.go | 19 ++++++++++--------- pkg/webhook/templates/apply_test.go | 10 +++++----- 5 files changed, 30 insertions(+), 28 deletions(-) diff --git a/TEMPLATES.md b/TEMPLATES.md index 442e8f19a..df384ee3d 100644 --- a/TEMPLATES.md +++ b/TEMPLATES.md @@ -2862,7 +2862,7 @@ ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s -โธ Paused by throttler: replica-lag 12s > 10s +๐Ÿšฆ Throttled: replica-lag 12s > 10s **`products`**: โณ Queued @@ -5040,7 +5040,7 @@ Sequential mode: First complete, second paused by the engine's throttler ~ orders: ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ 62% ALTER TABLE `orders` ADD INDEX `idx_user_status`(`user_id`, `status`); โ€ข Rows: 3,100,000 / 5,000,000 - โ€ข โธ Paused by throttler: replica-lag 12s > 10s + โ€ข ๐Ÿšฆ Throttled: replica-lag 12s > 10s ~ products: โณ Queued ALTER TABLE `products` ADD COLUMN `weight_grams` int DEFAULT 0; diff --git a/pkg/cmd/internal/templates/progress.go b/pkg/cmd/internal/templates/progress.go index 67220a034..a5781a0eb 100644 --- a/pkg/cmd/internal/templates/progress.go +++ b/pkg/cmd/internal/templates/progress.go @@ -768,19 +768,20 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel return b.String() } -// writeThrottleLine notes that the engine's throttler is pausing the table's -// active phase, so a stalled bar reads as a deliberate pause rather than a -// hang. The drive clears the stored flag when the pause lifts, so the line -// disappears on the next refresh. +// writeThrottleLine notes that the engine's throttler is holding back the +// table's active phase, so a stalled bar reads as deliberate backpressure โ€” +// slowed, not stopped โ€” never to be confused with an operator stop. The drive +// clears the stored flag when the throttle lifts, so the line disappears on +// the next refresh. func writeThrottleLine(b *strings.Builder, t TableProgress) { if !t.Throttled { return } if t.ThrottleReason != "" { - fmt.Fprintf(b, indentDetail+"โธ Paused by throttler: %s\n", t.ThrottleReason) + fmt.Fprintf(b, indentDetail+"๐Ÿšฆ Throttled: %s\n", t.ThrottleReason) return } - fmt.Fprint(b, indentDetail+"โธ Paused by throttler\n") + fmt.Fprint(b, indentDetail+"๐Ÿšฆ Throttled\n") } func recoveringIsCopyingRows(t TableProgress) bool { diff --git a/pkg/cmd/internal/templates/progress_states_test.go b/pkg/cmd/internal/templates/progress_states_test.go index a15a8ac1b..8de1fcc54 100644 --- a/pkg/cmd/internal/templates/progress_states_test.go +++ b/pkg/cmd/internal/templates/progress_states_test.go @@ -397,15 +397,15 @@ func TestFormatTableProgress_Throttled(t *testing.T) { RowsCopied: 45000, RowsTotal: 100000, PercentComplete: 45, Throttled: true, ThrottleReason: "replica-lag 12s > 10s", }) - assert.Contains(t, copying, "โธ Paused by throttler: replica-lag 12s > 10s") + assert.Contains(t, copying, "๐Ÿšฆ Throttled: replica-lag 12s > 10s") noReason := FormatTableProgress(TableProgress{ TableName: "orders", ChangeType: "alter", Status: state.Apply.Running, RowsCopied: 45000, RowsTotal: 100000, PercentComplete: 45, Throttled: true, }) - assert.Contains(t, noReason, "โธ Paused by throttler") - assert.NotContains(t, noReason, "Paused by throttler:") + assert.Contains(t, noReason, "๐Ÿšฆ Throttled") + assert.NotContains(t, noReason, "Throttled:") checksumming := FormatTableProgress(TableProgress{ TableName: "orders", ChangeType: "alter", Status: state.Task.Checksumming, @@ -413,20 +413,20 @@ func TestFormatTableProgress_Throttled(t *testing.T) { Throttled: true, ThrottleReason: "threads-running 130 > 128", }) assert.Contains(t, checksumming, "๐Ÿ” Checksumming to verify data (21%)") - assert.Contains(t, checksumming, "โธ Paused by throttler: threads-running 130 > 128") + assert.Contains(t, checksumming, "๐Ÿšฆ Throttled: threads-running 130 > 128") notThrottled := FormatTableProgress(TableProgress{ TableName: "orders", ChangeType: "alter", Status: state.Apply.Running, RowsCopied: 45000, RowsTotal: 100000, PercentComplete: 45, }) - assert.NotContains(t, notThrottled, "Paused by throttler") + assert.NotContains(t, notThrottled, "Throttled") completed := FormatTableProgress(TableProgress{ TableName: "orders", ChangeType: "alter", Status: state.Apply.Completed, RowsCopied: 100000, RowsTotal: 100000, PercentComplete: 100, Throttled: true, ThrottleReason: "replica-lag 12s > 10s", }) - assert.NotContains(t, completed, "Paused by throttler", "a terminal table never renders a stale throttle flag") + assert.NotContains(t, completed, "Throttled", "a terminal table never renders a stale throttle flag") } func TestFormatTableProgress_InstantDDL(t *testing.T) { diff --git a/pkg/webhook/templates/apply.go b/pkg/webhook/templates/apply.go index a929c0931..51ccbc5e2 100644 --- a/pkg/webhook/templates/apply.go +++ b/pkg/webhook/templates/apply.go @@ -967,22 +967,23 @@ func renderRunningTable(sb *strings.Builder, table TableProgressData) { } } -// writeThrottleLine notes that the engine's throttler is pausing the table's -// active phase, so a stalled bar reads as a deliberate pause rather than a -// hang. The drive clears the stored flag when the pause lifts, so the line -// disappears on the next refresh. The reason is sanitized at the engine -// boundary before it is stored, so it is safe to render in markdown. The -// pause is its own paragraph so it never merges into the preceding progress -// detail line, whichever markdown line-break mode renders the comment. +// writeThrottleLine notes that the engine's throttler is holding back the +// table's active phase, so a stalled bar reads as deliberate backpressure โ€” +// slowed, not stopped โ€” never to be confused with an operator stop. The drive +// clears the stored flag when the throttle lifts, so the line disappears on +// the next refresh. The reason is sanitized at the engine boundary before it +// is stored, so it is safe to render in markdown. The line is its own +// paragraph so it never merges into the preceding progress detail line, +// whichever markdown line-break mode renders the comment. func writeThrottleLine(sb *strings.Builder, table TableProgressData) { if !table.Throttled { return } if table.ThrottleReason != "" { - fmt.Fprintf(sb, "\nโธ Paused by throttler: %s\n", table.ThrottleReason) + fmt.Fprintf(sb, "\n๐Ÿšฆ Throttled: %s\n", table.ThrottleReason) return } - sb.WriteString("\nโธ Paused by throttler\n") + sb.WriteString("\n๐Ÿšฆ Throttled\n") } func recoveringIsCopyingRows(table TableProgressData) bool { diff --git a/pkg/webhook/templates/apply_test.go b/pkg/webhook/templates/apply_test.go index 78df8b83f..9f1c53b9b 100644 --- a/pkg/webhook/templates/apply_test.go +++ b/pkg/webhook/templates/apply_test.go @@ -283,17 +283,17 @@ func TestRenderApplyStatusComment_Throttled(t *testing.T) { result := RenderApplyStatusComment(data) assert.Contains(t, result, "**`orders`**") - assert.Contains(t, result, "\n\nโธ Paused by throttler: replica-lag 12s > 10s", + assert.Contains(t, result, "\n\n๐Ÿšฆ Throttled: replica-lag 12s > 10s", "the pause is its own paragraph, not a continuation of the progress detail line") data.Tables[0].ThrottleReason = "" noReason := RenderApplyStatusComment(data) - assert.Contains(t, noReason, "\n\nโธ Paused by throttler") - assert.NotContains(t, noReason, "Paused by throttler:") + assert.Contains(t, noReason, "\n\n๐Ÿšฆ Throttled") + assert.NotContains(t, noReason, "Throttled:") data.Tables[0].Throttled = false notThrottled := RenderApplyStatusComment(data) - assert.NotContains(t, notThrottled, "Paused by throttler") + assert.NotContains(t, notThrottled, "Throttled") } // A throttled checksum verify surfaces the pause alongside the verify @@ -316,7 +316,7 @@ func TestRenderApplyStatusComment_ThrottledChecksumming(t *testing.T) { result := RenderApplyStatusComment(data) assert.Contains(t, result, "๐Ÿ” Checksumming to verify data (21%)") - assert.Contains(t, result, "\n\nโธ Paused by throttler: threads-running 130 > 128") + assert.Contains(t, result, "\n\n๐Ÿšฆ Throttled: threads-running 130 > 128") } func TestUnsafeDropIndexUsageTargets(t *testing.T) { From 1f77e09fdf64fe8f914824315c62cf4d89c0b275 Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sat, 15 Aug 2026 15:41:43 +0800 Subject: [PATCH 06/12] fix(github): render table detail lines as a uniform list The lines under a table header (rows, ETA, verified counts) rendered as bare paragraphs everywhere except the estimate-exceeded branch, which bulleted them. Bullet them all: every line under a header is now a list item, matching the CLI's bulleted detail lines, and list items never merge under any markdown line-break mode. Co-Authored-By: Claude Fable 5 --- TEMPLATES.md | 36 +++++++++++++++++----------------- pkg/webhook/templates/apply.go | 12 ++++++------ 2 files changed, 24 insertions(+), 24 deletions(-) diff --git a/TEMPLATES.md b/TEMPLATES.md index df384ee3d..bff037fe4 100644 --- a/TEMPLATES.md +++ b/TEMPLATES.md @@ -2525,7 +2525,7 @@ Verify the database name, or run the command against the SchemaBot instance that ```sql ALTER TABLE `users` ADD INDEX `idx_email_created`(`email`, `created_at`); ``` -Rows: 3,500,000 / 7,200,000 ยท ETA: 5m 30s +- Rows: 3,500,000 / 7,200,000 ยท ETA: 5m 30s --- @@ -2556,7 +2556,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `users` ADD INDEX `idx_email_created`(`email`, `created_at`); ``` -Rows: 3,500,000 / 7,200,000 ยท ETA: 5m 30s +- Rows: 3,500,000 / 7,200,000 ยท ETA: 5m 30s --- @@ -2639,7 +2639,7 @@ schemabot apply -e staging ```sql ALTER TABLE `users` ADD INDEX `idx_email_created`(`email`, `created_at`); ``` -Rows: 156,342 / 397,453 +- Rows: 156,342 / 397,453 --- @@ -2718,7 +2718,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `orders` ADD INDEX `idx_user_id`(`user_id`); ``` -Rows: 321,450 / 1,466,232 ยท ETA: 5m 40s +- Rows: 321,450 / 1,466,232 ยท ETA: 5m 40s **`users`**: โณ Queued @@ -2765,7 +2765,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` -Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s +- Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s **`products`**: โณ Queued @@ -2860,7 +2860,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` -Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s +- Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s ๐Ÿšฆ Throttled: replica-lag 12s > 10s @@ -2909,7 +2909,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` -Rows copied: 1,466,232 +- Rows copied: 1,466,232 **`products`**: โณ Queued @@ -2956,7 +2956,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` -Rows verified: 321,450 / 1,466,232 +- Rows verified: 321,450 / 1,466,232 **`products`**: โณ Queued @@ -3003,7 +3003,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` -Rows copied: 1,466,232 +- Rows copied: 1,466,232 **`products`**: โณ Queued @@ -3050,7 +3050,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `products` ADD INDEX `idx_price`(`price_cents`); ``` -Rows: 87,231 / 523,140 ยท ETA: 7m 0s +- Rows: 87,231 / 523,140 ยท ETA: 7m 0s **`orders`**: ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ โœ… Complete @@ -3095,7 +3095,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` -Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s +- Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s โ”” shards: โœ“ -40 ยท โ— 40-80 62% ยท โ— 80-c0 31% ยท โณ c0- @@ -3129,7 +3129,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `orders` ADD COLUMN `region` varchar(32); ``` -Rows: 4,200,000,000 / 6,000,000,000 ยท ETA: 1h 30m +- Rows: 4,200,000,000 / 6,000,000,000 ยท ETA: 1h 30m โ”” 256 shards: 200 โœ“ ยท 52 โ— copying ยท 4 โณ ยท slowest f7- 12% @@ -3502,7 +3502,7 @@ schemabot apply -e staging ```sql ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` -Rows: 1,055,687 / 1,466,232 +- Rows: 1,055,687 / 1,466,232 **`orders`**: ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ๐ŸŸฉ โœ… Complete @@ -3977,7 +3977,7 @@ Use `schemabot status -e ` to find the apply ID. ```sql ALTER TABLE `users` ADD INDEX `idx_email_created`(`email`, `created_at`); ``` -Rows: 2,300,000 / 7,200,000 ยท ETA: 13m 0s +- Rows: 2,300,000 / 7,200,000 ยท ETA: 13m 0s --- @@ -4016,7 +4016,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `users` ADD INDEX `idx_email_created`(`email`, `created_at`); ``` -Rows: 2,300,000 / 7,200,000 +- Rows: 2,300,000 / 7,200,000 --- @@ -4176,7 +4176,7 @@ _Last updated: 2026-01-01 00:00:0 ```sql ALTER TABLE `users` ADD INDEX `idx_email_created`(`email`, `created_at`); ``` -Rows: 2,300,000 / 7,200,000 +- Rows: 2,300,000 / 7,200,000 --- @@ -4702,7 +4702,7 @@ ALTER TABLE `events` ADD INDEX `idx_created_at`(`created_at`); ```sql ALTER TABLE `users` DROP INDEX `idx_email`; ``` -Rows: 45,000 / 100,000 +- Rows: 45,000 / 100,000 --- @@ -6439,7 +6439,7 @@ SchemaBot triggers cutover automatically โ€” no action needed. ```sql ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` -Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s +- Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s **`products`**: โณ Queued diff --git a/pkg/webhook/templates/apply.go b/pkg/webhook/templates/apply.go index 51ccbc5e2..71aeb790b 100644 --- a/pkg/webhook/templates/apply.go +++ b/pkg/webhook/templates/apply.go @@ -715,7 +715,7 @@ func renderTableProgress(sb *strings.Builder, table TableProgressData, applyAtte fmt.Fprintf(sb, "**`%s`**: %s โฉ Catching up on accumulated changes...\n", table.TableName, ui.ProgressBarRowCopy(100)) writeDDLLine(sb, table.DDL) if table.RowsCopied > 0 { - fmt.Fprintf(sb, "Rows copied: %s\n", ui.FormatNumber(table.RowsCopied)) + fmt.Fprintf(sb, "- Rows copied: %s\n", ui.FormatNumber(table.RowsCopied)) } case state.Task.Checksumming: @@ -726,7 +726,7 @@ func renderTableProgress(sb *strings.Builder, table TableProgressData, applyAtte pct := ui.ClampPercent(int(table.ChecksumRowsChecked * 100 / table.ChecksumRowsTotal)) fmt.Fprintf(sb, "**`%s`**: %s \U0001f50d Checksumming to verify data (%d%%)\n", table.TableName, ui.ProgressBarRowCopy(pct), pct) writeDDLLine(sb, table.DDL) - fmt.Fprintf(sb, "Rows verified: %s / %s\n", + fmt.Fprintf(sb, "- Rows verified: %s / %s\n", ui.FormatNumber(ui.ClampRows(table.ChecksumRowsChecked, table.ChecksumRowsTotal)), ui.FormatNumber(table.ChecksumRowsTotal)) } else { fmt.Fprintf(sb, "**`%s`**: %s \U0001f50d Checksumming to verify data...\n", table.TableName, ui.ProgressBarRowCopy(100)) @@ -741,7 +741,7 @@ func renderTableProgress(sb *strings.Builder, table TableProgressData, applyAtte fmt.Fprintf(sb, "**`%s`**: %s โฉ Data verified, applying final changes...\n", table.TableName, ui.ProgressBarRowCopy(100)) writeDDLLine(sb, table.DDL) if table.RowsCopied > 0 { - fmt.Fprintf(sb, "Rows copied: %s\n", ui.FormatNumber(table.RowsCopied)) + fmt.Fprintf(sb, "- Rows copied: %s\n", ui.FormatNumber(table.RowsCopied)) } case state.Task.WaitingForCutover: @@ -1021,7 +1021,7 @@ func renderStoppedTable(sb *strings.Builder, table TableProgressData) { // Show rows (no ETA) for stopped tables with progress if table.RowsTotal > 0 && (table.PercentComplete > 0 || table.RowsCopied > 0) { - fmt.Fprintf(sb, "Rows: %s / %s\n", + fmt.Fprintf(sb, "- Rows: %s / %s\n", ui.FormatNumber(ui.ClampRows(table.RowsCopied, table.RowsTotal)), ui.FormatNumber(table.RowsTotal)) } @@ -1041,12 +1041,12 @@ func writeRowsAndETA(sb *strings.Builder, table TableProgressData) { } copied := ui.ClampRows(table.RowsCopied, table.RowsTotal) if table.ETASeconds > 0 { - fmt.Fprintf(sb, "Rows: %s / %s \u00b7 ETA: %s\n", + fmt.Fprintf(sb, "- Rows: %s / %s \u00b7 ETA: %s\n", ui.FormatNumber(copied), ui.FormatNumber(table.RowsTotal), ui.FormatETA(table.ETASeconds)) } else { - fmt.Fprintf(sb, "Rows: %s / %s\n", + fmt.Fprintf(sb, "- Rows: %s / %s\n", ui.FormatNumber(copied), ui.FormatNumber(table.RowsTotal)) } From a56b92a38f530129fdab6efa3f3ce142d64ee99e Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sat, 15 Aug 2026 15:45:53 +0800 Subject: [PATCH 07/12] fix(github): annotate the throttled header instead of a standalone line MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The throttle marker was its own glyph-led paragraph below the detail lines, visually misaligned with them and inventing a new template element. Move the state to where the eye checks progress โ€” a plain "(throttled)" suffix on the table's header line โ€” and explain the trigger with the same tooltip idiom the estimate-exceeded note uses, as an โ„น๏ธ bullet in the detail list. When the engine reports throttled without a reason, the header annotation stands alone. Both the PR comment and CLI render the same shape. Co-Authored-By: Claude Fable 5 --- TEMPLATES.md | 9 ++-- pkg/cmd/internal/templates/progress.go | 50 +++++++++++-------- .../templates/progress_states_test.go | 22 ++++---- pkg/webhook/templates/apply.go | 47 +++++++++-------- pkg/webhook/templates/apply_test.go | 31 ++++++------ 5 files changed, 89 insertions(+), 70 deletions(-) diff --git a/TEMPLATES.md b/TEMPLATES.md index bff037fe4..c56294068 100644 --- a/TEMPLATES.md +++ b/TEMPLATES.md @@ -2855,14 +2855,13 @@ _Last updated: 2026-01-01 00:00:0 **Schema `testapp`** -**`users`**: ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ 62% +**`users`**: ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ 62% (throttled) ```sql ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` - Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s - -๐Ÿšฆ Throttled: replica-lag 12s > 10s +- โ„น๏ธ _Throttled: replica-lag 12s > 10s_ **`products`**: โณ Queued @@ -5037,10 +5036,10 @@ Sequential mode: First complete, second paused by the engine's throttler โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ - ~ orders: ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ 62% + ~ orders: ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ 62% (throttled) ALTER TABLE `orders` ADD INDEX `idx_user_status`(`user_id`, `status`); โ€ข Rows: 3,100,000 / 5,000,000 - โ€ข ๐Ÿšฆ Throttled: replica-lag 12s > 10s + โ€ข โ„น๏ธ Throttled: replica-lag 12s > 10s ~ products: โณ Queued ALTER TABLE `products` ADD COLUMN `weight_grams` int DEFAULT 0; diff --git a/pkg/cmd/internal/templates/progress.go b/pkg/cmd/internal/templates/progress.go index a5781a0eb..3529c6adb 100644 --- a/pkg/cmd/internal/templates/progress.go +++ b/pkg/cmd/internal/templates/progress.go @@ -507,19 +507,19 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel // verify has progressed once Spirit has reported a total. if t.ChecksumRowsTotal > 0 { pct := ui.ClampPercent(int(t.ChecksumRowsChecked * 100 / t.ChecksumRowsTotal)) - fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s ๐Ÿ” Checksumming to verify data (%d%%)\n", t.TableName, ui.ProgressBarRowCopy(pct), pct) + fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s ๐Ÿ” Checksumming to verify data (%d%%)%s\n", t.TableName, ui.ProgressBarRowCopy(pct), pct, throttledSuffix(t)) if t.DDL != "" { b.WriteString(formatProgressDDL(t.DDL)) } fmt.Fprintf(&b, indentDetail+"Rows verified: %s / %s\n", ui.FormatNumber(ui.ClampRows(t.ChecksumRowsChecked, t.ChecksumRowsTotal)), ui.FormatNumber(t.ChecksumRowsTotal)) } else { - fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s ๐Ÿ” Checksumming to verify data...\n", t.TableName, ui.ProgressBarRowCopy(100)) + fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s ๐Ÿ” Checksumming to verify data...%s\n", t.TableName, ui.ProgressBarRowCopy(100), throttledSuffix(t)) if t.DDL != "" { b.WriteString(formatProgressDDL(t.DDL)) } } - writeThrottleLine(&b, t) + writeThrottleTooltip(&b, t) b.WriteString("\n") b.WriteString(FormatShardProgress(t.Shards)) return b.String() @@ -693,7 +693,7 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel // Parsed successfully - show emoji progress bar with structured data displayPercent := ui.RowCopyDisplayPercent(info.Percent, info.RowsCopied) bar := ui.ProgressBarRowCopy(displayPercent) - fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s %d%%\n", t.TableName, bar, displayPercent) + fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s %d%%%s\n", t.TableName, bar, displayPercent, throttledSuffix(t)) if t.DDL != "" { b.WriteString(formatProgressDDL(t.DDL)) } @@ -716,7 +716,7 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel // (Vitess VReplication / Spirit ramp-up โ€” can take a while on a large // table). Show a starting indicator and the row total instead of a 0% // bar that reads as stuck. - fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: โณ Starting copy...\n", t.TableName) + fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: โณ Starting copy...%s\n", t.TableName, throttledSuffix(t)) if t.DDL != "" { b.WriteString(formatProgressDDL(t.DDL)) } @@ -730,7 +730,7 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel // Row copy in progress โ€” show progress bar with structured fields displayPercent := ui.RowCopyDisplayPercent(t.PercentComplete, t.RowsCopied) bar := ui.ProgressBarRowCopy(displayPercent) - fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s %d%%\n", t.TableName, bar, displayPercent) + fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s %d%%%s\n", t.TableName, bar, displayPercent, throttledSuffix(t)) if t.DDL != "" { b.WriteString(formatProgressDDL(t.DDL)) @@ -749,18 +749,18 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel op := ddl.OpToStatementType(t.ChangeType) switch { case t.IsInstant: - fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s Applying instantly...\n", t.TableName, bar) + fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s Applying instantly...%s\n", t.TableName, bar, throttledSuffix(t)) case op == ddl.StatementCreateTable || op == ddl.StatementDropTable: - fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s Applying...\n", t.TableName, bar) + fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s Applying...%s\n", t.TableName, bar, throttledSuffix(t)) default: - fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s Running...\n", t.TableName, bar) + fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s Running...%s\n", t.TableName, bar, throttledSuffix(t)) } if t.DDL != "" { b.WriteString(formatProgressDDL(t.DDL)) } } - writeThrottleLine(&b, t) + writeThrottleTooltip(&b, t) if len(t.Shards) == 0 { b.WriteString("\n") } @@ -768,20 +768,27 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel return b.String() } -// writeThrottleLine notes that the engine's throttler is holding back the -// table's active phase, so a stalled bar reads as deliberate backpressure โ€” -// slowed, not stopped โ€” never to be confused with an operator stop. The drive -// clears the stored flag when the throttle lifts, so the line disappears on -// the next refresh. -func writeThrottleLine(b *strings.Builder, t TableProgress) { +// throttledSuffix annotates a paced-phase header when the engine's throttler +// is holding the phase back, so a slow bar reads as deliberate backpressure โ€” +// slowed, not stopped โ€” right where the eye checks progress, and never to be +// confused with an operator stop. The drive clears the stored flag when the +// throttle lifts, so the annotation disappears on the next refresh. +func throttledSuffix(t TableProgress) string { if !t.Throttled { - return + return "" } - if t.ThrottleReason != "" { - fmt.Fprintf(b, indentDetail+"๐Ÿšฆ Throttled: %s\n", t.ThrottleReason) + return " (throttled)" +} + +// writeThrottleTooltip explains the header's "(throttled)" annotation with the +// engine's reason, using the same dimmed tooltip idiom as the estimate-exceeded +// note. When the engine reports throttled without a reason, the header +// annotation stands alone. +func writeThrottleTooltip(b *strings.Builder, t TableProgress) { + if !t.Throttled || t.ThrottleReason == "" { return } - fmt.Fprint(b, indentDetail+"๐Ÿšฆ Throttled\n") + fmt.Fprintf(b, indentDetail+"%sโ„น๏ธ Throttled: %s%s\n", ANSIDim, t.ThrottleReason, ANSIReset) } func recoveringIsCopyingRows(t TableProgress) bool { @@ -798,12 +805,13 @@ func writeStructuredRowsAndETA(b *strings.Builder, t TableProgress) { func formatEstimateExceededTable(t TableProgress, rowsCopied int64, activityBar, activityLabel string) string { var b strings.Builder - fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s %s\n", t.TableName, activityBar, activityLabel) + fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s %s%s\n", t.TableName, activityBar, activityLabel, throttledSuffix(t)) if t.DDL != "" { b.WriteString(formatProgressDDL(t.DDL)) } fmt.Fprintf(&b, indentDetail+"Rows copied: %s so far\n", ui.FormatNumber(rowsCopied)) fmt.Fprintf(&b, indentDetail+"%sโ„น๏ธ %s%s\n", ANSIDim, ui.EstimateExceededTooltip, ANSIReset) + writeThrottleTooltip(&b, t) return b.String() } diff --git a/pkg/cmd/internal/templates/progress_states_test.go b/pkg/cmd/internal/templates/progress_states_test.go index 8de1fcc54..ef4c7b4e3 100644 --- a/pkg/cmd/internal/templates/progress_states_test.go +++ b/pkg/cmd/internal/templates/progress_states_test.go @@ -387,38 +387,42 @@ func TestFormatTableProgress_Checksumming(t *testing.T) { assert.Contains(t, withProgress, "Rows verified: 321,450 / 1,466,232") } -// A table paused by the engine's throttler names the pause and its trigger, -// so a stalled progress bar reads as a deliberate slowdown (e.g. replica lag) -// rather than a hang. The line renders for the active copy and checksum -// phases only โ€” a throttled flag on a terminal table would be stale. +// A table slowed by the engine's throttler carries a "(throttled)" annotation +// on its header line with the trigger explained in a dimmed tooltip, so a slow +// progress bar reads as deliberate backpressure (e.g. replica lag) rather than +// a hang. The annotation renders for the active copy and checksum phases only โ€” +// a throttled flag on a terminal table would be stale. func TestFormatTableProgress_Throttled(t *testing.T) { copying := FormatTableProgress(TableProgress{ TableName: "orders", ChangeType: "alter", Status: state.Apply.Running, RowsCopied: 45000, RowsTotal: 100000, PercentComplete: 45, Throttled: true, ThrottleReason: "replica-lag 12s > 10s", }) - assert.Contains(t, copying, "๐Ÿšฆ Throttled: replica-lag 12s > 10s") + assert.Contains(t, copying, "45% (throttled)", + "the annotation lands on the header line next to the percent") + assert.Contains(t, copying, "โ„น๏ธ Throttled: replica-lag 12s > 10s") noReason := FormatTableProgress(TableProgress{ TableName: "orders", ChangeType: "alter", Status: state.Apply.Running, RowsCopied: 45000, RowsTotal: 100000, PercentComplete: 45, Throttled: true, }) - assert.Contains(t, noReason, "๐Ÿšฆ Throttled") - assert.NotContains(t, noReason, "Throttled:") + assert.Contains(t, noReason, "45% (throttled)") + assert.NotContains(t, noReason, "โ„น๏ธ Throttled", "no tooltip without a reason") checksumming := FormatTableProgress(TableProgress{ TableName: "orders", ChangeType: "alter", Status: state.Task.Checksumming, ChecksumRowsChecked: 321450, ChecksumRowsTotal: 1466232, Throttled: true, ThrottleReason: "threads-running 130 > 128", }) - assert.Contains(t, checksumming, "๐Ÿ” Checksumming to verify data (21%)") - assert.Contains(t, checksumming, "๐Ÿšฆ Throttled: threads-running 130 > 128") + assert.Contains(t, checksumming, "๐Ÿ” Checksumming to verify data (21%) (throttled)") + assert.Contains(t, checksumming, "โ„น๏ธ Throttled: threads-running 130 > 128") notThrottled := FormatTableProgress(TableProgress{ TableName: "orders", ChangeType: "alter", Status: state.Apply.Running, RowsCopied: 45000, RowsTotal: 100000, PercentComplete: 45, }) + assert.NotContains(t, notThrottled, "(throttled)") assert.NotContains(t, notThrottled, "Throttled") completed := FormatTableProgress(TableProgress{ diff --git a/pkg/webhook/templates/apply.go b/pkg/webhook/templates/apply.go index 71aeb790b..e60414295 100644 --- a/pkg/webhook/templates/apply.go +++ b/pkg/webhook/templates/apply.go @@ -724,15 +724,15 @@ func renderTableProgress(sb *strings.Builder, table TableProgressData, applyAtte // show how far the verify has progressed once Spirit reports a total. if table.ChecksumRowsTotal > 0 { pct := ui.ClampPercent(int(table.ChecksumRowsChecked * 100 / table.ChecksumRowsTotal)) - fmt.Fprintf(sb, "**`%s`**: %s \U0001f50d Checksumming to verify data (%d%%)\n", table.TableName, ui.ProgressBarRowCopy(pct), pct) + fmt.Fprintf(sb, "**`%s`**: %s \U0001f50d Checksumming to verify data (%d%%)%s\n", table.TableName, ui.ProgressBarRowCopy(pct), pct, throttledSuffix(table)) writeDDLLine(sb, table.DDL) fmt.Fprintf(sb, "- Rows verified: %s / %s\n", ui.FormatNumber(ui.ClampRows(table.ChecksumRowsChecked, table.ChecksumRowsTotal)), ui.FormatNumber(table.ChecksumRowsTotal)) } else { - fmt.Fprintf(sb, "**`%s`**: %s \U0001f50d Checksumming to verify data...\n", table.TableName, ui.ProgressBarRowCopy(100)) + fmt.Fprintf(sb, "**`%s`**: %s \U0001f50d Checksumming to verify data...%s\n", table.TableName, ui.ProgressBarRowCopy(100), throttledSuffix(table)) writeDDLLine(sb, table.DDL) } - writeThrottleLine(sb, table) + writeThrottleTooltip(sb, table) case state.Task.PostChecksum: // The verify passed and the engine is draining the changes that @@ -936,10 +936,10 @@ func isCopyingShardStatus(status string) bool { // renderRunningTable renders a table that is actively copying rows. func renderRunningTable(sb *strings.Builder, table TableProgressData) { - defer writeThrottleLine(sb, table) + defer writeThrottleTooltip(sb, table) if table.RowsTotal > 0 { if ui.EstimateExceeded(table.RowsCopied, table.RowsTotal) { - fmt.Fprintf(sb, "**`%s`**: %s Finalizing copy\n", table.TableName, ui.ProgressBarActivity()) + fmt.Fprintf(sb, "**`%s`**: %s Finalizing copy%s\n", table.TableName, ui.ProgressBarActivity(), throttledSuffix(table)) writeDDLLine(sb, table.DDL) fmt.Fprintf(sb, "- Rows copied: %s so far\n", ui.FormatNumber(table.RowsCopied)) fmt.Fprintf(sb, "- โ„น๏ธ _%s_\n", ui.EstimateExceededTooltip) @@ -951,39 +951,44 @@ func renderRunningTable(sb *strings.Builder, table TableProgressData) { // Row total is known but the copy hasn't reported progress yet // (VReplication / Spirit ramp-up). A 0% bar reads as stuck, so show // a starting indicator and the row total instead. - fmt.Fprintf(sb, "**`%s`**: โณ Starting copy...\n", table.TableName) + fmt.Fprintf(sb, "**`%s`**: โณ Starting copy...%s\n", table.TableName, throttledSuffix(table)) writeDDLLine(sb, table.DDL) writeRowsAndETA(sb, table) return } bar := ui.ProgressBarRowCopy(pct) - fmt.Fprintf(sb, "**`%s`**: %s %d%%\n", table.TableName, bar, pct) + fmt.Fprintf(sb, "**`%s`**: %s %d%%%s\n", table.TableName, bar, pct, throttledSuffix(table)) writeDDLLine(sb, table.DDL) writeRowsAndETA(sb, table) } else { // No row data yet (initializing or instant DDL) - fmt.Fprintf(sb, "**`%s`**: Running...\n", table.TableName) + fmt.Fprintf(sb, "**`%s`**: Running...%s\n", table.TableName, throttledSuffix(table)) writeDDLLine(sb, table.DDL) } } -// writeThrottleLine notes that the engine's throttler is holding back the -// table's active phase, so a stalled bar reads as deliberate backpressure โ€” -// slowed, not stopped โ€” never to be confused with an operator stop. The drive -// clears the stored flag when the throttle lifts, so the line disappears on -// the next refresh. The reason is sanitized at the engine boundary before it -// is stored, so it is safe to render in markdown. The line is its own -// paragraph so it never merges into the preceding progress detail line, -// whichever markdown line-break mode renders the comment. -func writeThrottleLine(sb *strings.Builder, table TableProgressData) { +// throttledSuffix annotates a paced-phase header when the engine's throttler +// is holding the phase back, so a slow bar reads as deliberate backpressure โ€” +// slowed, not stopped โ€” right where the eye checks progress, and never to be +// confused with an operator stop. The drive clears the stored flag when the +// throttle lifts, so the annotation disappears on the next refresh. +func throttledSuffix(table TableProgressData) string { if !table.Throttled { - return + return "" } - if table.ThrottleReason != "" { - fmt.Fprintf(sb, "\n๐Ÿšฆ Throttled: %s\n", table.ThrottleReason) + return " (throttled)" +} + +// writeThrottleTooltip explains the header's "(throttled)" annotation with the +// engine's reason, using the same tooltip idiom as the estimate-exceeded note. +// The reason is sanitized at the engine boundary before it is stored, so it is +// safe to render in markdown. When the engine reports throttled without a +// reason, the header annotation stands alone. +func writeThrottleTooltip(sb *strings.Builder, table TableProgressData) { + if !table.Throttled || table.ThrottleReason == "" { return } - sb.WriteString("\n๐Ÿšฆ Throttled\n") + fmt.Fprintf(sb, "- โ„น๏ธ _Throttled: %s_\n", table.ThrottleReason) } func recoveringIsCopyingRows(table TableProgressData) bool { diff --git a/pkg/webhook/templates/apply_test.go b/pkg/webhook/templates/apply_test.go index 9f1c53b9b..e3fbd8c3a 100644 --- a/pkg/webhook/templates/apply_test.go +++ b/pkg/webhook/templates/apply_test.go @@ -260,11 +260,12 @@ func TestRenderApplyStatusComment_Checksumming(t *testing.T) { assert.Contains(t, result, "1 checksumming") } -// A table paused by the engine's throttler names the pause and its trigger in -// the PR comment, so a stalled progress bar reads as a deliberate slowdown +// A table slowed by the engine's throttler carries a "(throttled)" annotation +// on its header line โ€” right where the eye checks progress โ€” with the trigger +// explained in a tooltip bullet, so a slow bar reads as deliberate backpressure // (e.g. replica lag) rather than a hang. The reason is sanitized at the engine -// boundary, and the line renders only on active tables โ€” a throttled flag on a -// terminal table would be stale. +// boundary, and the annotation renders only on active tables โ€” a throttled flag +// on a terminal table would be stale. func TestRenderApplyStatusComment_Throttled(t *testing.T) { data := ApplyStatusCommentData{ Database: "testapp", @@ -282,23 +283,25 @@ func TestRenderApplyStatusComment_Throttled(t *testing.T) { result := RenderApplyStatusComment(data) - assert.Contains(t, result, "**`orders`**") - assert.Contains(t, result, "\n\n๐Ÿšฆ Throttled: replica-lag 12s > 10s", - "the pause is its own paragraph, not a continuation of the progress detail line") + assert.Contains(t, result, "45% (throttled)", + "the annotation lands on the header line next to the percent") + assert.Contains(t, result, "- โ„น๏ธ _Throttled: replica-lag 12s > 10s_", + "the reason renders as a tooltip bullet under the detail list") data.Tables[0].ThrottleReason = "" noReason := RenderApplyStatusComment(data) - assert.Contains(t, noReason, "\n\n๐Ÿšฆ Throttled") - assert.NotContains(t, noReason, "Throttled:") + assert.Contains(t, noReason, "45% (throttled)") + assert.NotContains(t, noReason, "โ„น๏ธ Throttled", "no tooltip without a reason") data.Tables[0].Throttled = false notThrottled := RenderApplyStatusComment(data) + assert.NotContains(t, notThrottled, "(throttled)") assert.NotContains(t, notThrottled, "Throttled") } -// A throttled checksum verify surfaces the pause alongside the verify -// progress, since the checksum is the other phase the engine's throttler -// paces. +// A throttled checksum verify carries the same header annotation and tooltip +// alongside the verify progress, since the checksum is the other phase the +// engine's throttler paces. func TestRenderApplyStatusComment_ThrottledChecksumming(t *testing.T) { data := ApplyStatusCommentData{ Database: "testapp", @@ -315,8 +318,8 @@ func TestRenderApplyStatusComment_ThrottledChecksumming(t *testing.T) { result := RenderApplyStatusComment(data) - assert.Contains(t, result, "๐Ÿ” Checksumming to verify data (21%)") - assert.Contains(t, result, "\n\n๐Ÿšฆ Throttled: threads-running 130 > 128") + assert.Contains(t, result, "๐Ÿ” Checksumming to verify data (21%) (throttled)") + assert.Contains(t, result, "- โ„น๏ธ _Throttled: threads-running 130 > 128_") } func TestUnsafeDropIndexUsageTargets(t *testing.T) { From ecf2e42b827a196df9cfb3d86d27c6924901e1db Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sat, 15 Aug 2026 15:49:08 +0800 Subject: [PATCH 08/12] test(tern): say throttle state, not pause, in assertion messages Co-Authored-By: Claude Fable 5 --- pkg/tern/grpc_client_test.go | 2 +- pkg/tern/local_client_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/tern/grpc_client_test.go b/pkg/tern/grpc_client_test.go index e5d8d465a..45785c375 100644 --- a/pkg/tern/grpc_client_test.go +++ b/pkg/tern/grpc_client_test.go @@ -4563,7 +4563,7 @@ func TestGRPCClient_SyncRemoteProgressMirrorsRowCopySnapshot(t *testing.T) { assert.Equal(t, 340, task.ETASeconds) assert.Equal(t, int64(200), task.ChecksumRowsChecked) assert.Equal(t, int64(1000), task.ChecksumRowsTotal) - assert.True(t, task.Throttled, "throttle pause mirrors from the remote table progress") + assert.True(t, task.Throttled, "throttle state mirrors from the remote table progress") assert.Equal(t, "replica-lag 12s > 10s", task.ThrottleReason) } diff --git a/pkg/tern/local_client_test.go b/pkg/tern/local_client_test.go index 7d53e4bf3..3caa285dd 100644 --- a/pkg/tern/local_client_test.go +++ b/pkg/tern/local_client_test.go @@ -3948,7 +3948,7 @@ func TestLocalClient_ProgressRendersNonShardedTableFromStoredTask(t *testing.T) assert.Equal(t, int64(90), tp.EtaSeconds, "ETA comes from the stored task row") assert.Equal(t, int64(10), tp.ChecksumRowsChecked) assert.Equal(t, int64(1000), tp.ChecksumRowsTotal) - assert.True(t, tp.Throttled, "throttle pause comes from the stored task row") + assert.True(t, tp.Throttled, "throttle state comes from the stored task row") assert.Equal(t, "replica-lag 12s > 10s", tp.ThrottleReason) assert.Empty(t, tp.Shards, "a non-sharded table has no per-shard breakdown") } From ae354db12414e1b11360791172acef1eeff12acc Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sat, 15 Aug 2026 16:06:55 +0800 Subject: [PATCH 09/12] fix(github): use a realistic Aurora throttle reason in previews The sample reason was an invented replica-lag string in a format Spirit never emits, and replica-lag throttling is rare on the Aurora targets operators actually run. Use the commit-latency throttler's real output shape with Spirit's default 100ms threshold instead. Co-Authored-By: Claude Fable 5 --- TEMPLATES.md | 4 ++-- pkg/cmd/internal/templates/preview_sequential.go | 2 +- pkg/webhook/templates/preview.go | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/TEMPLATES.md b/TEMPLATES.md index c56294068..8ae908346 100644 --- a/TEMPLATES.md +++ b/TEMPLATES.md @@ -2861,7 +2861,7 @@ _Last updated: 2026-01-01 00:00:0 ALTER TABLE `users` ADD INDEX `idx_email`(`email`); ``` - Rows: 914,707 / 1,466,232 ยท ETA: 3m 15s -- โ„น๏ธ _Throttled: replica-lag 12s > 10s_ +- โ„น๏ธ _Throttled: commit-latency 112.4ms >= 100ms_ **`products`**: โณ Queued @@ -5039,7 +5039,7 @@ Sequential mode: First complete, second paused by the engine's throttler ~ orders: ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆ๐ŸŸฆโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ 62% (throttled) ALTER TABLE `orders` ADD INDEX `idx_user_status`(`user_id`, `status`); โ€ข Rows: 3,100,000 / 5,000,000 - โ€ข โ„น๏ธ Throttled: replica-lag 12s > 10s + โ€ข โ„น๏ธ Throttled: commit-latency 112.4ms >= 100ms ~ products: โณ Queued ALTER TABLE `products` ADD COLUMN `weight_grams` int DEFAULT 0; diff --git a/pkg/cmd/internal/templates/preview_sequential.go b/pkg/cmd/internal/templates/preview_sequential.go index 6a401e35d..4e31da0b0 100644 --- a/pkg/cmd/internal/templates/preview_sequential.go +++ b/pkg/cmd/internal/templates/preview_sequential.go @@ -120,7 +120,7 @@ func previewSeqThrottledOutput() { {TableName: "users", DDL: seqDDLs[0].ddl, Status: state.Apply.Completed}, {TableName: "orders", DDL: seqDDLs[1].ddl, Status: state.Task.Running, RowsCopied: 3100000, RowsTotal: 5000000, PercentComplete: 62, - Throttled: true, ThrottleReason: "replica-lag 12s > 10s"}, + Throttled: true, ThrottleReason: "commit-latency 112.4ms >= 100ms"}, {TableName: "products", DDL: seqDDLs[2].ddl, Status: state.Apply.Pending}, }, } diff --git a/pkg/webhook/templates/preview.go b/pkg/webhook/templates/preview.go index 5ef31f2b4..52647347c 100644 --- a/pkg/webhook/templates/preview.go +++ b/pkg/webhook/templates/preview.go @@ -1186,7 +1186,7 @@ func PreviewCommentApplyThrottled() string { tables[1].PercentComplete = 62 tables[1].ETASeconds = 195 tables[1].Throttled = true - tables[1].ThrottleReason = "replica-lag 12s > 10s" + tables[1].ThrottleReason = "commit-latency 112.4ms >= 100ms" tables[2].Status = state.Task.Pending return RenderApplyStatusComment(sampleApplyData(state.Apply.Running, tables)) } From b53089da8188b60ef39978509bc8484e411fc3c2 Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sun, 16 Aug 2026 11:23:24 +0800 Subject: [PATCH 10/12] fix(github): enforce the throttle-state contract at every boundary The stored throttle pair now behaves as a point-in-time signal everywhere: the gRPC mirror refreshes it on every tick (including ticks that keep stored row totals), so a lifted throttle clears promptly; a reason never travels without the flag; transitionTaskState clears the pair when a task comes to rest, alongside the ETA. Reason sanitization moves to pkg/engine as SanitizeThrottleReason so the remote-ingest boundary shares the engine stamp site's bounds, and the PR comment tooltip escapes inline markdown so an engine reason cannot cut the italic span short. Co-Authored-By: Claude Fable 5 --- pkg/engine/spirit/spirit.go | 21 +------- pkg/engine/spirit/spirit_test.go | 30 ----------- pkg/engine/throttle.go | 27 ++++++++++ pkg/engine/throttle_test.go | 37 +++++++++++++ pkg/tern/grpc_client.go | 13 ++++- pkg/tern/grpc_client_test.go | 82 +++++++++++++++++++++++++++++ pkg/tern/local_apply.go | 13 +++-- pkg/webhook/templates/apply.go | 2 +- pkg/webhook/templates/apply_test.go | 6 +++ pkg/webhook/templates/common.go | 19 +++++++ 10 files changed, 192 insertions(+), 58 deletions(-) create mode 100644 pkg/engine/throttle.go create mode 100644 pkg/engine/throttle_test.go diff --git a/pkg/engine/spirit/spirit.go b/pkg/engine/spirit/spirit.go index 6984b17ac..14cc8d7ce 100644 --- a/pkg/engine/spirit/spirit.go +++ b/pkg/engine/spirit/spirit.go @@ -19,7 +19,6 @@ import ( "strings" "sync" "time" - "unicode/utf8" spiritmigration "github.com/block/spirit/pkg/migration" "github.com/block/spirit/pkg/migration/check" @@ -868,7 +867,7 @@ func buildSpiritTableProgress(prog status.Progress, spiritState status.State, dd // an unthrottled table carries no reason. if tableInPacedPhase(st.IsComplete, spiritState) && prog.Throttle.Throttled { tp.Throttled = true - tp.ThrottleReason = sanitizeThrottleReason(prog.Throttle.Reason) + tp.ThrottleReason = engine.SanitizeThrottleReason(prog.Throttle.Reason) } tableProgress = append(tableProgress, tp) } @@ -882,24 +881,6 @@ func tableInPacedPhase(copyComplete bool, spiritState status.State) bool { return !copyComplete || spiritState == status.Checksum } -// sanitizeThrottleReason bounds an engine-produced throttle reason for the -// operator surfaces that render it (PR comment tables, CLI rows): newlines and -// table separators are neutralized and the text is clamped, so a reason can -// never break markdown layout no matter what a throttler reports. -func sanitizeThrottleReason(reason string) string { - reason = strings.Join(strings.Fields(reason), " ") - reason = strings.ReplaceAll(reason, "|", "/") - const maxLen = 200 - if len(reason) > maxLen { - cut := maxLen - len("โ€ฆ") - for cut > 0 && !utf8.RuneStart(reason[cut]) { - cut-- - } - reason = reason[:cut] + "โ€ฆ" - } - return reason -} - // spiritPostCopyPhase reports whether the runner is in one of the active // phases between finishing the row copy and finishing the cutover: applying // the accumulated changeset, restoring deferred indexes, analyzing, verifying diff --git a/pkg/engine/spirit/spirit_test.go b/pkg/engine/spirit/spirit_test.go index 068dafc3b..f7715336e 100644 --- a/pkg/engine/spirit/spirit_test.go +++ b/pkg/engine/spirit/spirit_test.go @@ -1,7 +1,6 @@ package spirit import ( - "strings" "testing" "time" @@ -319,32 +318,3 @@ func TestBuildSpiritTableProgress(t *testing.T) { }) } } - -// TestSanitizeThrottleReason verifies that an engine-produced throttle reason -// is safe for the operator surfaces that render it: whitespace runs collapse, -// markdown table separators are neutralized, and overlong text is clamped so -// a reason can never break a PR comment table or a CLI row. -func TestSanitizeThrottleReason(t *testing.T) { - tests := []struct { - name string - in string - want string - }{ - {"plain reason passes through", "replica-lag 12s > 10s", "replica-lag 12s > 10s"}, - {"empty stays empty", "", ""}, - {"newlines and runs collapse to single spaces", "replica-lag\n12s >\t\t10s", "replica-lag 12s > 10s"}, - {"table separators are neutralized", "signal a > 1 | signal b > 2", "signal a > 1 / signal b > 2"}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - assert.Equal(t, tt.want, sanitizeThrottleReason(tt.in)) - }) - } - - t.Run("overlong reason is clamped with an ellipsis", func(t *testing.T) { - long := strings.Repeat("replica-lag 12s > 10s; ", 20) - got := sanitizeThrottleReason(long) - assert.LessOrEqual(t, len(got), 200) - assert.True(t, strings.HasSuffix(got, "โ€ฆ")) - }) -} diff --git a/pkg/engine/throttle.go b/pkg/engine/throttle.go new file mode 100644 index 000000000..31cc82d88 --- /dev/null +++ b/pkg/engine/throttle.go @@ -0,0 +1,27 @@ +package engine + +import ( + "strings" + "unicode/utf8" +) + +// SanitizeThrottleReason bounds an engine-produced throttle reason for the +// operator surfaces that render it (PR comment tables, CLI rows): newlines and +// table separators are neutralized and the text is clamped, so a reason can +// never break markdown layout no matter what a throttler reports. Every +// boundary that ingests a throttle reason โ€” an engine stamping its own +// throttler's reason, or a client mirroring one reported by a remote data +// plane โ€” must pass it through here before persisting it. +func SanitizeThrottleReason(reason string) string { + reason = strings.Join(strings.Fields(reason), " ") + reason = strings.ReplaceAll(reason, "|", "/") + const maxLen = 200 + if len(reason) > maxLen { + cut := maxLen - len("โ€ฆ") + for cut > 0 && !utf8.RuneStart(reason[cut]) { + cut-- + } + reason = reason[:cut] + "โ€ฆ" + } + return reason +} diff --git a/pkg/engine/throttle_test.go b/pkg/engine/throttle_test.go new file mode 100644 index 000000000..19df06a1a --- /dev/null +++ b/pkg/engine/throttle_test.go @@ -0,0 +1,37 @@ +package engine + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/assert" +) + +// TestSanitizeThrottleReason verifies that an engine-produced throttle reason +// is safe for the operator surfaces that render it: whitespace runs collapse, +// markdown table separators are neutralized, and overlong text is clamped so +// a reason can never break a PR comment table or a CLI row. +func TestSanitizeThrottleReason(t *testing.T) { + tests := []struct { + name string + in string + want string + }{ + {"plain reason passes through", "replica-lag 12s > 10s", "replica-lag 12s > 10s"}, + {"empty stays empty", "", ""}, + {"newlines and runs collapse to single spaces", "replica-lag\n12s >\t\t10s", "replica-lag 12s > 10s"}, + {"table separators are neutralized", "signal a > 1 | signal b > 2", "signal a > 1 / signal b > 2"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, SanitizeThrottleReason(tt.in)) + }) + } + + t.Run("overlong reason is clamped with an ellipsis", func(t *testing.T) { + long := strings.Repeat("replica-lag 12s > 10s; ", 20) + got := SanitizeThrottleReason(long) + assert.LessOrEqual(t, len(got), 200) + assert.True(t, strings.HasSuffix(got, "โ€ฆ")) + }) +} diff --git a/pkg/tern/grpc_client.go b/pkg/tern/grpc_client.go index 01d9cb5bc..b3855da52 100644 --- a/pkg/tern/grpc_client.go +++ b/pkg/tern/grpc_client.go @@ -3294,8 +3294,17 @@ func (c *GRPCClient) syncStoredTasksFromRemoteTasks( storedTask.ETASeconds = int(remoteTask.EtaSeconds) storedTask.ChecksumRowsChecked = remoteTask.ChecksumRowsChecked storedTask.ChecksumRowsTotal = remoteTask.ChecksumRowsTotal - storedTask.Throttled = remoteTask.Throttled - storedTask.ThrottleReason = remoteTask.ThrottleReason + } + // Throttle state is a point-in-time signal, not cumulative progress: + // it is mirrored on every tick โ€” including ticks whose row totals are + // kept โ€” so a lifted throttle clears promptly instead of lingering on + // the PR comment. The reason travels only with an active throttle and + // is bounded here because the remote data plane's reason is untrusted + // input for the operator surfaces that render it. + storedTask.Throttled = remoteTask.Throttled + storedTask.ThrottleReason = "" + if remoteTask.Throttled { + storedTask.ThrottleReason = engine.SanitizeThrottleReason(remoteTask.ThrottleReason) } // Adopt the remote task's own failure reason (for example an engine // preflight rejection) so the operation row derived from the stored task diff --git a/pkg/tern/grpc_client_test.go b/pkg/tern/grpc_client_test.go index 45785c375..8663bcc26 100644 --- a/pkg/tern/grpc_client_test.go +++ b/pkg/tern/grpc_client_test.go @@ -4567,6 +4567,88 @@ func TestGRPCClient_SyncRemoteProgressMirrorsRowCopySnapshot(t *testing.T) { assert.Equal(t, "replica-lag 12s > 10s", task.ThrottleReason) } +func TestGRPCClient_SyncRemoteProgressThrottleContract(t *testing.T) { + // Throttle state is a point-in-time signal the PR comment and CLI render + // verbatim, so the mirror must enforce its contract on every tick: a + // lifted throttle clears even on a snapshot whose row totals are kept, a + // reason never survives without the flag, and a remote reason is bounded + // before it can reach an operator surface. + apply := &storage.Apply{ + ID: 22, + ApplyIdentifier: "apply-throttle-contract", + Database: "testdb", + Environment: "staging", + ExternalID: "remote-throttle-contract", + State: state.Apply.Running, + } + newThrottledTask := func() *storage.Task { + return &storage.Task{ + ID: 35, + TaskIdentifier: "task-throttle-contract", + ApplyID: apply.ID, + Namespace: "default", + TableName: "orders", + State: state.Task.Running, + RowsCopied: 450, + RowsTotal: 1000, + Throttled: true, + ThrottleReason: "replica-lag 12s > 10s", + } + } + sync := func(t *testing.T, task *storage.Task, remote *ternv1.TableProgress) { + t.Helper() + client := &GRPCClient{ + storage: &mockStorage{ + tasks: &mockTaskStore{tasks: []*storage.Task{task}}, + logs: &mockApplyLogStore{}, + }, + } + remote.Namespace = "default" + remote.TableName = "orders" + remote.Status = state.Task.Running + err := client.syncStoredTasksFromRemoteTasks(t.Context(), apply, []*storage.Task{task}, []*ternv1.TableProgress{remote}, time.Now()) + require.NoError(t, err) + } + + t.Run("a lifted throttle clears on a snapshot with omitted row totals", func(t *testing.T) { + task := newThrottledTask() + sync(t, task, &ternv1.TableProgress{}) + + assert.Equal(t, int64(450), task.RowsCopied, "the stored row-copy progress survives the omitted totals") + assert.Equal(t, int64(1000), task.RowsTotal) + assert.False(t, task.Throttled, "the throttle flag mirrors the snapshot even when row totals are kept") + assert.Empty(t, task.ThrottleReason) + }) + + t.Run("a reason without the flag is dropped", func(t *testing.T) { + task := newThrottledTask() + sync(t, task, &ternv1.TableProgress{ + RowsCopied: 500, + RowsTotal: 1000, + ThrottleReason: "replica-lag 12s > 10s", + }) + + assert.False(t, task.Throttled) + assert.Empty(t, task.ThrottleReason, "an unthrottled task never carries a reason") + }) + + t.Run("a remote reason is sanitized before it is stored", func(t *testing.T) { + task := newThrottledTask() + sync(t, task, &ternv1.TableProgress{ + RowsCopied: 500, + RowsTotal: 1000, + Throttled: true, + ThrottleReason: "replica-lag\n12s | " + strings.Repeat("threads-running 130 > 128; ", 20), + }) + + assert.True(t, task.Throttled) + assert.NotContains(t, task.ThrottleReason, "\n", "newlines collapse before storage") + assert.NotContains(t, task.ThrottleReason, "|", "table separators are neutralized before storage") + assert.LessOrEqual(t, len(task.ThrottleReason), 200, "an overlong remote reason is clamped") + assert.True(t, strings.HasSuffix(task.ThrottleReason, "โ€ฆ")) + }) +} + func TestGRPCClient_PollSetsTerminalTaskMetadataFromRemoteTaskProgress(t *testing.T) { // Terminal remote task progress marks the stored task terminal and fills // local metadata before the stored apply row is marked completed. diff --git a/pkg/tern/local_apply.go b/pkg/tern/local_apply.go index 8a1c3cbad..04e667a8a 100644 --- a/pkg/tern/local_apply.go +++ b/pkg/tern/local_apply.go @@ -351,13 +351,16 @@ func (c *LocalClient) transitionTaskState(ctx context.Context, task *storage.Tas oldState := task.State task.State = newState task.UpdatedAt = time.Now() - // An ETA is only meaningful while an engine is actively driving the task - // and refreshing the estimate. Clear it when the task comes to rest - // (terminal, stopped, retryable, pending) so the stored row never carries - // a frozen estimate; a resumed or retried task gets a fresh figure from - // the first engine poll. + // An ETA and throttle state are only meaningful while an engine is + // actively driving the task and refreshing them. Clear them when the task + // comes to rest (terminal, stopped, retryable, pending) so the stored row + // never carries a frozen estimate or renders as paused with no copy in + // flight; a resumed or retried task gets fresh figures from the first + // engine poll. if !state.IsInFlightTaskState(newState) { task.ETASeconds = 0 + task.Throttled = false + task.ThrottleReason = "" } if err := c.storage.Tasks().Update(ctx, task); err != nil { c.logger.Error("failed to update task state", append(task.LogAttrs(), "error", err)...) diff --git a/pkg/webhook/templates/apply.go b/pkg/webhook/templates/apply.go index e60414295..8b0109c0e 100644 --- a/pkg/webhook/templates/apply.go +++ b/pkg/webhook/templates/apply.go @@ -988,7 +988,7 @@ func writeThrottleTooltip(sb *strings.Builder, table TableProgressData) { if !table.Throttled || table.ThrottleReason == "" { return } - fmt.Fprintf(sb, "- โ„น๏ธ _Throttled: %s_\n", table.ThrottleReason) + fmt.Fprintf(sb, "- โ„น๏ธ _Throttled: %s_\n", escapeInlineMarkdown(table.ThrottleReason)) } func recoveringIsCopyingRows(table TableProgressData) bool { diff --git a/pkg/webhook/templates/apply_test.go b/pkg/webhook/templates/apply_test.go index e3fbd8c3a..06de086ca 100644 --- a/pkg/webhook/templates/apply_test.go +++ b/pkg/webhook/templates/apply_test.go @@ -297,6 +297,12 @@ func TestRenderApplyStatusComment_Throttled(t *testing.T) { notThrottled := RenderApplyStatusComment(data) assert.NotContains(t, notThrottled, "(throttled)") assert.NotContains(t, notThrottled, "Throttled") + + data.Tables[0].Throttled = true + data.Tables[0].ThrottleReason = "signal_a 1_000ms >= `500ms` [gradual]" + escaped := RenderApplyStatusComment(data) + assert.Contains(t, escaped, "- โ„น๏ธ _Throttled: signal\\_a 1\\_000ms >= \\`500ms\\` \\[gradual\\]_", + "markdown delimiters in an engine reason are escaped so they cannot cut the italic span short") } // A throttled checksum verify carries the same header annotation and tooltip diff --git a/pkg/webhook/templates/common.go b/pkg/webhook/templates/common.go index f83e73128..80990cb16 100644 --- a/pkg/webhook/templates/common.go +++ b/pkg/webhook/templates/common.go @@ -147,6 +147,25 @@ func escapeMarkdownLinkText(text string) string { return strings.ReplaceAll(text, "]", "\\]") } +// escapeInlineMarkdown backslash-escapes the characters that can change how +// engine-influenced text renders inside an inline markdown span: emphasis and +// code delimiters, link brackets, and HTML tag openers. Use it when untrusted +// text is interpolated into styled inline markdown (for example inside an +// italic span), where a stray delimiter would otherwise cut the styling short +// or start a new construct. +func escapeInlineMarkdown(text string) string { + var b strings.Builder + b.Grow(len(text)) + for _, r := range text { + switch r { + case '\\', '`', '*', '_', '[', ']', '<': + b.WriteByte('\\') + } + b.WriteRune(r) + } + return b.String() +} + // maxCommentErrorLen bounds an error message rendered into a PR comment so a // pathological engine error cannot flood the comment. Genuine engine errors, // such as a Spirit preflight check reason, are a few hundred characters and From 330f9d8ec0fb0b6844587e2b49ca30c9ce4f2a67 Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sun, 16 Aug 2026 11:24:19 +0800 Subject: [PATCH 11/12] test(tern): cover the one-per-claim resume event at the poll pipeline level Drive the sequential poll against an engine that reports a resumed copy on every tick and assert a single timeline event, proving the latch holds through the real poll loop rather than only in the helper's unit test. Also reunite an auto-cutover scenario comment with its test. Co-Authored-By: Claude Fable 5 --- .../local_apply_sequential_progress_test.go | 54 +++++++++++++++++++ pkg/tern/local_client_test.go | 6 +-- 2 files changed, 57 insertions(+), 3 deletions(-) diff --git a/pkg/tern/local_apply_sequential_progress_test.go b/pkg/tern/local_apply_sequential_progress_test.go index 050dadbe3..d89f4d4c4 100644 --- a/pkg/tern/local_apply_sequential_progress_test.go +++ b/pkg/tern/local_apply_sequential_progress_test.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "strings" "testing" "github.com/stretchr/testify/assert" @@ -164,6 +165,59 @@ func TestPollTaskToCompletion_RefinesPostCopyPhases(t *testing.T) { }, taskStore.states) } +// A drive claim that reattaches to an engine's durable checkpoint must surface +// that in the apply timeline exactly once, even though the engine reports the +// resume flag on every subsequent poll โ€” so an operator reading the timeline +// can tell a resumed copy from a fresh start without the event repeating on +// every tick. Drive the sequential poll against an engine that reports a +// resumed copy across several polls and assert a single timeline event. +func TestPollTaskToCompletion_LogsResumeOnce(t *testing.T) { + task := &storage.Task{ + ID: 1, ApplyID: 1, TaskIdentifier: "task-1", + Database: "appdb", DatabaseType: storage.DatabaseTypeMySQL, + TableName: "mutes", State: state.Task.Running, + } + apply := &storage.Apply{ + ID: 1, ApplyIdentifier: "apply-1", Database: "appdb", + DatabaseType: storage.DatabaseTypeMySQL, Environment: "staging", + } + resumedResult := func(engineState engine.State) *engine.ProgressResult { + return &engine.ProgressResult{ + State: engineState, + ResumedFromCheckpoint: true, + Tables: []engine.TableProgress{{Table: "mutes", State: spiritstatus.CopyRows.String()}}, + } + } + logs := &mockApplyLogStore{} + eng := &phaseSequenceEngine{results: []*engine.ProgressResult{ + resumedResult(engine.StateRunning), + resumedResult(engine.StateRunning), + resumedResult(engine.StateCompleted), + }} + client := &LocalClient{ + config: LocalConfig{Database: "appdb", Type: storage.DatabaseTypeMySQL}, + spiritEngine: eng, + storage: &exactProgressStorage{ + tasks: &exactProgressTaskStore{tasks: []*storage.Task{task}}, + controlRequests: &testControlRequestStore{}, + logs: logs, + }, + logger: slog.Default(), + } + + action := client.pollTaskToCompletion(t.Context(), apply, task, nil, nil) + + assert.Equal(t, taskContinue, action) + assert.Equal(t, state.Task.Completed, task.State) + resumeEvents := 0 + for _, entry := range logs.logs { + if strings.Contains(entry.Message, "resumed from checkpoint") { + resumeEvents++ + } + } + assert.Equal(t, 1, resumeEvents, "the resume flag on every poll records one timeline event per drive claim") +} + // permanentProgressErrorEngine always fails Progress with a permanent error. type permanentProgressErrorEngine struct{ engine.Engine } diff --git a/pkg/tern/local_client_test.go b/pkg/tern/local_client_test.go index 3caa285dd..652ee39ae 100644 --- a/pkg/tern/local_client_test.go +++ b/pkg/tern/local_client_test.go @@ -1994,9 +1994,6 @@ func TestLocalClient_ProcessPendingCutoverControlRequestRetriesWhenCutoverNotRea assert.True(t, hasLogMessageContaining(logs.logs, "Cutover triggered (caller: cli:alice)")) } -// The drive's auto-cutover records one trigger in the timeline per drive and -// retries quietly while the engine backend stages the cutover. A rejection -// that outlives the staging window records a one-time timeline error and // A drive claim that reattaches to the engine's durable checkpoint records one // timeline event, so an operator can tell a resumed copy from a fresh start. // The engine reports the resume flag on every subsequent poll; the per-drive @@ -2029,6 +2026,9 @@ func TestLocalClient_LogEngineResumeOnce(t *testing.T) { assert.Len(t, logs.logs, 1, "the flag on every later poll produces one event per drive claim") } +// The drive's auto-cutover records one trigger in the timeline per drive and +// retries quietly while the engine backend stages the cutover. A rejection +// that outlives the staging window records a one-time timeline error and // escalates to Error logging on every further tick, so a backend that never // stages the cutover is visible to operators instead of idling; the retry // loop itself never stops, and an accepted cutover resets the escalation. From 694e5688cd18213a25866ee29f304940ef28c74d Mon Sep 17 00:00:00 2001 From: Armand Parajon Date: Sun, 16 Aug 2026 11:24:27 +0800 Subject: [PATCH 12/12] docs: cover resume and throttle in the spirit progress pipeline doc Co-Authored-By: Claude Fable 5 --- docs/spirit_progress.md | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/docs/spirit_progress.md b/docs/spirit_progress.md index 66a120805..437e51a26 100644 --- a/docs/spirit_progress.md +++ b/docs/spirit_progress.md @@ -50,6 +50,8 @@ key to debugging stale-progress issues. | `CurrentState` | `status.State` | Atomic int32 enum: `Initial`, `CopyRows`, `WaitingOnSentinelTable`, `Checksum`, `CutOver`, `Close`, ... | | `Summary` | `string` | `"71436/221193 32.30% copyRows ETA 5m 30s"` | | `Tables[]` | `[]TableProgress` | Per-table: `TableName`, `RowsCopied` (uint64), `RowsTotal` (uint64), `IsComplete` (bool) | +| `Resume` | `bool` | True only after the runner successfully resumed from its durable checkpoint; a fresh start (or an abandoned resume attempt) reports false. | +| `Throttle` | `status.ThrottleStatus` | `Throttled` (bool), `Reason` (display-only string, `" "`), `Utilization` (float64, 0 means unknown โ€” never render it as idle). | Key details: - **ETA is embedded in `Summary`**, not a separate field. Downstream layers parse it out with a regex. @@ -59,6 +61,12 @@ Key details: `CurrentState == Checksum`** โ€” it reads zero before the verify starts and again after it finishes (including during `PostChecksum`). A renderer that shows checksum counters for any other phase shows zeros. +- **`Throttle` is populated only for the paced phases** โ€” the row copy and the checksum + verify report the throttler's live status; every other phase reports the zero + `ThrottleStatus{}`, so a finished schema change can never look paused on an aged-out + signal. A composite throttler joins the reasons of every currently-throttling child with + `"; "`, and `Throttled` can be true with an empty `Reason` (a throttler that implements + no reason extension). ## Spirit runner lifecycle @@ -114,6 +122,14 @@ Two properties matter for display: - Stamps the runner-wide checksum estimate (`ChecksumRowsChecked`/`Total`) on every table unconditionally โ€” Spirit populates it only during the verify phase, and every copy is complete by then. + - Stamps the runner-wide throttle status (`Throttled`/`ThrottleReason`) on the + tables participating in paced work โ€” a table still copying, or every table + during the verify (`tableInPacedPhase`) โ€” so a completed table is never + rendered as paused by another table's copy. The reason passes through + `engine.SanitizeThrottleReason` so it can never break an operator surface. +4. Sets `ProgressResult.ResumedFromCheckpoint` from `Progress.Resume`, which the + drive turns into a one-per-claim "resumed from checkpoint" timeline event + (`logEngineResumeOnce`). Key types: `engine.ProgressResult`, `engine.TableProgress` (`pkg/engine/engine.go`). @@ -188,7 +204,7 @@ no live result for a reader on another pod, and the drive keeps stored current. 2. Picks the most relevant task (priority: active > stopped > pending > terminal). 3. Builds the per-table response from the stored task rows: `Status`, `RowsCopied`, `RowsTotal`, `ProgressPercent`, `ETASeconds`, - `ChecksumRowsChecked/Total`. When per-shard rows are persisted, the table + `ChecksumRowsChecked/Total`, `Throttled`, `ThrottleReason`. When per-shard rows are persisted, the table headline is the aggregate of those rows, computed at read time. Reading stored state matters for: @@ -212,6 +228,8 @@ Task fields updated during polling (`storage.Task` in `pkg/storage/types.go`): | `ETASeconds` | int | From `engine.TableProgress.ETASeconds` | | `ChecksumRowsChecked` | int64 | From `engine.TableProgress.ChecksumRowsChecked` | | `ChecksumRowsTotal` | int64 | From `engine.TableProgress.ChecksumRowsTotal` | +| `Throttled` | bool | From `engine.TableProgress.Throttled`; cleared when the task comes to rest (`transitionTaskState`) | +| `ThrottleReason` | string | From `engine.TableProgress.ThrottleReason`, sanitized at every ingest boundary; empty whenever `Throttled` is false | | `IsInstant` | bool | From `engine.TableProgress.IsInstant` | | `State` | string | Mapped from `engine.State` | | `StartedAt` | time | Set when task transitions to RUNNING | @@ -222,7 +240,7 @@ Task fields updated during polling (`storage.Task` in `pkg/storage/types.go`): | Trigger | What writes | Frequency | Fields updated | |---------|-------------|-----------|----------------| -| `pollForCompletionAtomic` (atomic mode) | Poller goroutine | Every 500ms | `State`, `RowsCopied`, `RowsTotal`, `ProgressPercent`, `ETASeconds`, `UpdatedAt`, `CompletedAt` (on terminal) | +| `pollForCompletionAtomic` (atomic mode) | Poller goroutine | Every 500ms | `State`, `RowsCopied`, `RowsTotal`, `ProgressPercent`, `ETASeconds`, `Throttled`, `ThrottleReason`, `UpdatedAt`, `CompletedAt` (on terminal) | | `pollTaskToCompletion` (sequential mode) | Poller goroutine | Every 500ms | Same as above, plus `IsInstant` | | `LocalClient.Stop()` | Stop handler | Once, after `eng.Stop()` blocks | `State` โ†’ STOPPED (or COMPLETED if table finished), `RowsCopied`, `RowsTotal`, `ProgressPercent`, `ETASeconds`, `CompletedAt` | | `LocalClient.Progress()` | Progress handler | On each API call (if engine state changed and task is non-terminal) | `State`, `UpdatedAt`, `CompletedAt` |