diff --git a/TEMPLATES.md b/TEMPLATES.md index 7ea90ff54..8ae908346 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 @@ -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% (throttled) + +```sql +ALTER TABLE `users` ADD INDEX `idx_email`(`email`); +``` +- Rows: 914,707 / 1,466,232 · ETA: 3m 15s +- ℹ️ _Throttled: commit-latency 112.4ms >= 100ms_ + +**`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: @@ -2860,7 +2908,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 @@ -2907,7 +2955,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 @@ -2954,7 +3002,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 @@ -3001,7 +3049,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 @@ -3046,7 +3094,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- @@ -3080,7 +3128,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% @@ -3453,7 +3501,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 @@ -3928,7 +3976,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 --- @@ -3967,7 +4015,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 --- @@ -4127,7 +4175,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 --- @@ -4653,7 +4701,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 --- @@ -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% (throttled) + ALTER TABLE `orders` ADD INDEX `idx_user_status`(`user_id`, `status`); + • Rows: 3,100,000 / 5,000,000 + • ℹ️ Throttled: commit-latency 112.4ms >= 100ms + + ~ 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`); + + ```
@@ -6360,7 +6438,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/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` | diff --git a/e2e/consumermodule/go.mod b/e2e/consumermodule/go.mod index cc58481e8..0b1dfbe55 100644 --- a/e2e/consumermodule/go.mod +++ b/e2e/consumermodule/go.mod @@ -8,7 +8,7 @@ // version imported by pkg/api/telemetry.go; the test guards this precondition. module github.com/block/schemabot/e2e/consumermodule -go 1.26.5 +go 1.26.6 require ( github.com/block/schemabot v0.0.0 @@ -37,7 +37,7 @@ require ( github.com/aws/smithy-go v1.27.7 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/block/pg-sprite v0.0.0-20260814025010-d6cf677e4feb // indirect - github.com/block/spirit v0.16.1-0.20260808013537-864cf2ee6b6b // indirect + github.com/block/spirit v0.16.1-0.20260815035524-fc3efbddfa34 // indirect github.com/bradleyfalzon/ghinstallation/v2 v2.18.0 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect diff --git a/e2e/consumermodule/go.sum b/e2e/consumermodule/go.sum index aab236f3d..748e8ef4b 100644 --- a/e2e/consumermodule/go.sum +++ b/e2e/consumermodule/go.sum @@ -50,8 +50,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= 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= 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..4e31da0b0 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: "commit-latency 112.4ms >= 100ms"}, + {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..3529c6adb 100644 --- a/pkg/cmd/internal/templates/progress.go +++ b/pkg/cmd/internal/templates/progress.go @@ -507,18 +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)) } } + writeThrottleTooltip(&b, t) b.WriteString("\n") b.WriteString(FormatShardProgress(t.Shards)) return b.String() @@ -692,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)) } @@ -715,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)) } @@ -729,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)) @@ -748,17 +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)) } } + writeThrottleTooltip(&b, t) if len(t.Shards) == 0 { b.WriteString("\n") } @@ -766,6 +768,29 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel return b.String() } +// 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 " (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.Fprintf(b, indentDetail+"%sℹ️ Throttled: %s%s\n", ANSIDim, t.ThrottleReason, ANSIReset) +} + func recoveringIsCopyingRows(t TableProgress) bool { return t.RowsTotal > 0 && t.PercentComplete < 100 } @@ -780,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_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..ef4c7b4e3 100644 --- a/pkg/cmd/internal/templates/progress_states_test.go +++ b/pkg/cmd/internal/templates/progress_states_test.go @@ -387,6 +387,52 @@ func TestFormatTableProgress_Checksumming(t *testing.T) { assert.Contains(t, withProgress, "Rows verified: 321,450 / 1,466,232") } +// 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, "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, "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%) (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{ + 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, "Throttled", "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..14cc8d7ce 100644 --- a/pkg/engine/spirit/spirit.go +++ b/pkg/engine/spirit/spirit.go @@ -793,12 +793,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 +858,29 @@ 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. + // 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 = engine.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 +} + // 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..f7715336e 100644 --- a/pkg/engine/spirit/spirit_test.go +++ b/pkg/engine/spirit/spirit_test.go @@ -245,6 +245,59 @@ 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 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. + 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 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/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 2652f17e5..e21d086c1 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 f780125b9..74223d560 100644 --- a/pkg/tern/grpc_client.go +++ b/pkg/tern/grpc_client.go @@ -3298,6 +3298,17 @@ func (c *GRPCClient) syncStoredTasksFromRemoteTasks( storedTask.ChecksumRowsChecked = remoteTask.ChecksumRowsChecked storedTask.ChecksumRowsTotal = remoteTask.ChecksumRowsTotal } + // 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 // carries a per-table error. An empty remote error never clears a stored diff --git a/pkg/tern/grpc_client_test.go b/pkg/tern/grpc_client_test.go index ce089a103..8663bcc26 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,90 @@ 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 state mirrors from the remote table progress") + 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) { diff --git a/pkg/tern/local_apply.go b/pkg/tern/local_apply.go index 45518e171..04e667a8a 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() { @@ -334,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/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_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.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 5e118ed36..c0c2b24a7 100644 --- a/pkg/tern/local_client_test.go +++ b/pkg/tern/local_client_test.go @@ -1994,6 +1994,38 @@ func TestLocalClient_ProcessPendingCutoverControlRequestRetriesWhenCutoverNotRea assert.True(t, hasLogMessageContaining(logs.logs, "Cutover triggered (caller: cli:alice)")) } +// 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") +} + // 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 @@ -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 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") } 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..8b0109c0e 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 @@ -710,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: @@ -719,14 +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", + 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) } + writeThrottleTooltip(sb, table) case state.Task.PostChecksum: // The verify passed and the engine is draining the changes that @@ -735,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: @@ -930,9 +936,10 @@ func isCopyingShardStatus(status string) bool { // renderRunningTable renders a table that is actively copying rows. func renderRunningTable(sb *strings.Builder, table TableProgressData) { + 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) @@ -944,22 +951,46 @@ 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) } } +// 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 " (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 + } + fmt.Fprintf(sb, "- ℹ️ _Throttled: %s_\n", escapeInlineMarkdown(table.ThrottleReason)) +} + func recoveringIsCopyingRows(table TableProgressData) bool { return table.RowsTotal > 0 && table.PercentComplete < 100 } @@ -995,7 +1026,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)) } @@ -1015,12 +1046,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)) } diff --git a/pkg/webhook/templates/apply_test.go b/pkg/webhook/templates/apply_test.go index de3cc07b8..06de086ca 100644 --- a/pkg/webhook/templates/apply_test.go +++ b/pkg/webhook/templates/apply_test.go @@ -260,6 +260,74 @@ func TestRenderApplyStatusComment_Checksumming(t *testing.T) { assert.Contains(t, result, "1 checksumming") } +// 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 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", + 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, "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, "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") + + 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 +// 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%) (throttled)") + assert.Contains(t, result, "- ℹ️ _Throttled: threads-running 130 > 128_") +} + func TestUnsafeDropIndexUsageTargets(t *testing.T) { tests := []struct { name string 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 diff --git a/pkg/webhook/templates/preview.go b/pkg/webhook/templates/preview.go index 182dd91f7..52647347c 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 = "commit-latency 112.4ms >= 100ms" + 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 {