diff --git a/TEMPLATES.md b/TEMPLATES.md index 7ea90ff54..face2881f 100644 --- a/TEMPLATES.md +++ b/TEMPLATES.md @@ -3259,7 +3259,7 @@ _Last updated: 2026-01-01 00:00:0 **Schema `testapp`** -**`users`**: ๐ŸŸง๐ŸŸง๐ŸŸง๐ŸŸง๐ŸŸง๐ŸŸงโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ ๐Ÿ”„ Interrupted โ€” retrying automatically (attempt 2/10) +**`users`**: ๐ŸŸง๐ŸŸง๐ŸŸง๐ŸŸง๐ŸŸง๐ŸŸงโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ ๐Ÿ”„ Retrying ยท attempt 4/10 ยท next 14:31 UTC ```sql ALTER TABLE `users` ADD INDEX `idx_email`(`email`); @@ -4741,6 +4741,31 @@ Single table progress (default): โœ“ Apply complete! +``` + + +
+MySQL: Single Table Retrying + +``` + +โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” +โ”‚ Apply ID: apply-a1b2c3d4e5f6 โ”‚ +โ”‚ State: Retrying โ”‚ +โ”‚ Retry: attempt 4/10 ยท next 14:31:00 UTC โ”‚ +โ”‚ Started: Jan 15 14:26:00 UTC โ”‚ +โ”‚ Duration: 4m โ”‚ +โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ + + connection reset by peer + + + โ”€โ”€ testapp โ”€โ”€ + + ~ users: ๐ŸŸง๐ŸŸง๐ŸŸง๐ŸŸง๐ŸŸง๐ŸŸง๐ŸŸงโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœโฌœ Retrying + ALTER TABLE `users` ADD INDEX `idx_email_created`(`email`, `created_at`); + + ```
diff --git a/docs/apply-lifecycle.md b/docs/apply-lifecycle.md index 7789a6527..245c55666 100644 --- a/docs/apply-lifecycle.md +++ b/docs/apply-lifecycle.md @@ -159,9 +159,18 @@ data plane, a network blip, an unclean kill mid-drive. Nothing about the work needs to change; it just needs to be picked up again. Those failures move the apply to `failed_retryable` instead of `failed`, and recovery is automatic: a recovery driver reclaims the apply and continues it from where it left off, -using engine checkpoints so completed work is not redone. Failures where -retrying cannot help โ€” the engine rejected a statement, the target refused the -change โ€” skip this state and go straight to permanent `failed`. +using engine checkpoints so completed work is not redone. + +Failures where retrying cannot help skip this state and go straight to permanent +`failed`. Those are the ones the target rejects for a reason that belongs to the +statement or to the data already in the table: a duplicate value under a new +unique index, existing rows that do not fit a narrowed column, a column the +table does not have, DDL the target will not perform at all. Nothing about the +target changes between attempts, so the apply is failed on the spot with the +target's own reason rather than spending its whole budget reproducing it โ€” and +the database's active-apply slot is released for the corrected apply that +follows. Anything that might read differently once a lock, a lagging replica, or +a busy target has moved on keeps its retries. A clean shutdown is not one of these failures. A process that stops on purpose hands its claims back and leaves the apply active, so a peer driver resumes it @@ -172,10 +181,24 @@ currently **10 attempts**, on top of the original run. The attempt counter is visible in the PR progress comment, so you can watch how much budget an apply has burned. +Attempts are also **paced**. A failure that reproduces instantly โ€” a refused +lock, a connection error โ€” would otherwise spend the whole budget in under a +minute, on exactly the kind of failure a little time was most likely to clear. +So each attempt arms a wait for the next one: the first couple of retries are +immediate, and the wait then steps up and holds flat, spreading a fully spent +budget over roughly ten minutes. The wait is measured from the *start* of an +attempt, so an attempt that ran longer than its own wait has already spaced +itself out and retries as soon as it fails. The PR comment and the CLI's +progress view name the clock time the next attempt is due. + +The pacing applies to automatic retries only. An operator `start` runs now, and +a driver that dies mid-attempt is picked up by a peer as soon as its lease goes +stale โ€” neither waits out a retry nobody asked for. + ``` running --(recoverable failure)--> failed_retryable ^ | - | recovery driver reclaims, | + | recovery driver reclaims, | wait out the backoff +---- attempt counter +1, resume <----+ from checkpoints | | budget spent (10 attempts), diff --git a/pkg/api/ensure_schema_postgres_test.go b/pkg/api/ensure_schema_postgres_test.go index b081b16b2..1be8726b8 100644 --- a/pkg/api/ensure_schema_postgres_test.go +++ b/pkg/api/ensure_schema_postgres_test.go @@ -65,7 +65,7 @@ func TestPostgresCreateTableColumns_EmbeddedFiles(t *testing.T) { "id", "apply_identifier", "lock_id", "plan_id", "database_name", "database_type", "repository", "pull_request", "environment", "deployment", "caller", "installation_id", "external_id", "idempotency_key", "engine", "state", "error_message", "options", "attempt", - "lease_owner", "lease_token", "lease_acquired_at", "started_at", "completed_at", + "retry_after", "lease_owner", "lease_token", "lease_acquired_at", "started_at", "completed_at", "revert_skipped_at", "created_at", "updated_at", }, applies) diff --git a/pkg/api/handlers_test.go b/pkg/api/handlers_test.go index fa07546ed..ba25def2b 100644 --- a/pkg/api/handlers_test.go +++ b/pkg/api/handlers_test.go @@ -6385,3 +6385,27 @@ func TestSetRevertSkippedMetadata(t *testing.T) { setRevertSkippedMetadata(resp, &storage.Apply{RevertSkippedAt: &now}) assert.Equal(t, "true", resp.Metadata["revert_skipped"], "flag set once revert_skipped_at is present") } + +// overlayRetryBudget reports the spent retry budget in every state, so a +// permanently failed apply still shows how many recoveries were tried, but names +// the next attempt only while the apply is waiting for one. A claim leaves the +// armed deadline behind on the row it resumes, so a resumed apply must not +// advertise a retry that is not coming. +func TestOverlayRetryBudget(t *testing.T) { + due := time.Now().Add(90 * time.Second) + + retrying := &apitypes.ProgressResponse{} + overlayRetryBudget(retrying, &storage.Apply{State: state.Apply.FailedRetryable, Attempt: 3, RetryAfter: &due}) + assert.Equal(t, int32(3), retrying.Attempt) + assert.Equal(t, due.Format(time.RFC3339), retrying.RetryAfter) + + resumed := &apitypes.ProgressResponse{} + overlayRetryBudget(resumed, &storage.Apply{State: state.Apply.Running, Attempt: 3, RetryAfter: &due}) + assert.Equal(t, int32(3), resumed.Attempt, "the spent budget stays visible after redispatch") + assert.Empty(t, resumed.RetryAfter, "a running apply is not waiting on a retry") + + failed := &apitypes.ProgressResponse{} + overlayRetryBudget(failed, &storage.Apply{State: state.Apply.Failed, Attempt: storage.MaxRecoveryAttempts, RetryAfter: &due}) + assert.Equal(t, int32(storage.MaxRecoveryAttempts), failed.Attempt, "an exhausted budget is the record of what was tried") + assert.Empty(t, failed.RetryAfter) +} diff --git a/pkg/api/progress_handlers.go b/pkg/api/progress_handlers.go index 1efe59cdb..ac187a184 100644 --- a/pkg/api/progress_handlers.go +++ b/pkg/api/progress_handlers.go @@ -448,6 +448,8 @@ func (s *Service) handleProgressByApplyID(w http.ResponseWriter, r *http.Request overlayApplyOptions(httpResp, apply) + overlayRetryBudget(httpResp, apply) + setRevertSkippedMetadata(httpResp, apply) // Overlay per-table timestamps from task records. The proto response @@ -474,6 +476,28 @@ func (s *Service) handleProgressByApplyID(w http.ResponseWriter, r *http.Request s.writeJSON(w, http.StatusOK, httpResp) } +// overlayRetryBudget surfaces the apply's automatic-retry state: how much of the +// budget it has spent and, while it is waiting to be retried, when its next +// attempt becomes eligible. Both are control-plane bookkeeping โ€” the claim path +// owns them โ€” so they are read from the stored apply on every progress path, +// including the one whose per-table detail comes from a remote engine. +// +// The spent budget is reported in every state: on a permanently failed apply it +// is the record of how many recoveries were tried. The wait is reported only +// while the apply is retrying, because the stored deadline outlives the state +// that gave it meaning โ€” a claim leaves the armed deadline behind on the row it +// resumes, and reporting that on a running or completed apply would name a +// retry that is not coming. +func overlayRetryBudget(resp *apitypes.ProgressResponse, apply *storage.Apply) { + if apply == nil { + return + } + resp.Attempt = int32(apply.Attempt) + if apply.RetryAfter != nil && state.IsState(apply.State, state.Apply.FailedRetryable) { + resp.RetryAfter = apply.RetryAfter.Format(time.RFC3339) + } +} + // setRevertSkippedMetadata surfaces the skip-revert flag from the apply's stored // revert_skipped_at, so progress consumers can show that revert was skipped and // finalization is in progress. It reads apply state โ€” no engine-specific side @@ -1130,6 +1154,7 @@ func (s *Service) progressFromLocalStorage(ctx context.Context, apply *storage.A httpResp.ErrorMessage = apply.ErrorMessage } overlayApplyOptions(httpResp, apply) + overlayRetryBudget(httpResp, apply) setRevertSkippedMetadata(httpResp, apply) operations, deploymentByOperationID, released := s.bestEffortProgressOperations(ctx, apply) httpResp.Operations = operations diff --git a/pkg/apitypes/apitypes.go b/pkg/apitypes/apitypes.go index 5a377085c..00408874c 100644 --- a/pkg/apitypes/apitypes.go +++ b/pkg/apitypes/apitypes.go @@ -814,6 +814,16 @@ type ProgressResponse struct { PullRequest string `json:"pull_request,omitempty"` // PR URL (blank for CLI context) StartedAt string `json:"started_at,omitempty"` CompletedAt string `json:"completed_at,omitempty"` + // Attempt is how many automatic redispatches an interrupted apply has + // already consumed of its retry budget. Zero on an apply that has not been + // redispatched. + Attempt int32 `json:"attempt,omitempty"` + // RetryAfter is the RFC3339 time the next automatic retry becomes eligible. + // An interrupted apply backs off between attempts, so this distinguishes an + // apply that is waiting from one that is due and about to be picked up. Set + // only while the apply is retrying and a wait was armed; empty otherwise, + // including on an apply that has already been redispatched. + RetryAfter string `json:"retry_after,omitempty"` // Operations carries per-deployment operation rows for multi-deployment applies. // Empty for single-deployment applies. Operations []*ProgressOperationResponse `json:"operations,omitempty"` diff --git a/pkg/cmd/commands/preview.go b/pkg/cmd/commands/preview.go index 0fac8c3d3..13a55bcbd 100644 --- a/pkg/cmd/commands/preview.go +++ b/pkg/cmd/commands/preview.go @@ -35,8 +35,8 @@ func (cmd *PreviewCmd) Run(g *Globals) error { switch previewType { // Basic types case templates.PreviewPlan, templates.PreviewProgress, templates.PreviewWaitingForDeploy, templates.PreviewWaitingForCutover, - templates.PreviewCuttingOver, templates.PreviewCompleted, templates.PreviewFailed, - templates.PreviewStopped, templates.PreviewStates: + templates.PreviewCuttingOver, templates.PreviewCompleted, templates.PreviewRetrying, + templates.PreviewFailed, templates.PreviewStopped, templates.PreviewStates: templates.PreviewCLIOutput(previewType) // Lock types case templates.PreviewLockAcquired, templates.PreviewLockConflict, @@ -173,6 +173,7 @@ Basic Types: waiting_for_cutover Show sample waiting for cutover output cutting_over Show sample cutting over output completed Show sample completed output + retrying Show sample interrupted output (waiting on an automatic retry) failed Show sample failed output stopped Show sample stopped output (mid-apply stop) states Show state display formatting @@ -276,7 +277,7 @@ Comment Templates (GitHub PR comments): comment_apply_completed Multi-table: completed (all tables done) comment_apply_failed Multi-table: failed (with error and cancelled tables) comment_apply_failed_before_row_copy Multi-table: failed before row copy (preflight rejection, per-table error) - comment_apply_retrying Multi-table: interrupted, retrying automatically (attempt counter) + comment_apply_retrying Multi-table: interrupted, retrying automatically (attempt counter + next attempt time) comment_apply_stopped Multi-table: stopped (partial progress) comment_apply_waiting_cutover Waiting for cutover (deferred, operator triggers) comment_apply_waiting_cutover_automatic Waiting for cutover (non-deferred, drive triggers) diff --git a/pkg/cmd/internal/templates/preview.go b/pkg/cmd/internal/templates/preview.go index 95b9e8fa7..fca2ff4de 100644 --- a/pkg/cmd/internal/templates/preview.go +++ b/pkg/cmd/internal/templates/preview.go @@ -13,6 +13,7 @@ var previewTime = time.Date(2026, 1, 15, 14, 30, 0, 0, time.UTC) // SetPreviewMode configures the package to use fixed timestamps for deterministic output. func SetPreviewMode() { nowFunc = func() time.Time { return previewTime } + localZone = time.UTC ui.NowFunc = func() time.Time { return previewTime } } @@ -26,6 +27,7 @@ const ( PreviewWaitingForCutover PreviewType = "waiting_for_cutover" PreviewCuttingOver PreviewType = "cutting_over" PreviewCompleted PreviewType = "completed" + PreviewRetrying PreviewType = "retrying" PreviewFailed PreviewType = "failed" PreviewStopped PreviewType = "stopped" PreviewStates PreviewType = "states" @@ -143,7 +145,7 @@ const ( PreviewCommentApplyCompleted PreviewType = "comment_apply_completed" // Apply completed (all tables done) PreviewCommentApplyFailed PreviewType = "comment_apply_failed" // Apply failed (1 done, 1 failed, 1 cancelled) PreviewCommentApplyFailedBeforeRowCopy PreviewType = "comment_apply_failed_before_row_copy" // Apply failed before row copy (preflight rejection, per-table error) - PreviewCommentApplyRetrying PreviewType = "comment_apply_retrying" // Apply interrupted, retrying automatically (attempt counter) + PreviewCommentApplyRetrying PreviewType = "comment_apply_retrying" // Apply interrupted, retrying automatically (attempt counter + next attempt time) PreviewCommentApplyStopped PreviewType = "comment_apply_stopped" // Apply stopped (1 done, 1 stopped) PreviewCommentApplyWaitingCutover PreviewType = "comment_apply_waiting_cutover" // Waiting for cutover (deferred, operator triggers) PreviewCommentApplyWaitingCutoverAutomatic PreviewType = "comment_apply_waiting_cutover_automatic" // Waiting for cutover (non-deferred, drive triggers) diff --git a/pkg/cmd/internal/templates/preview_comment.go b/pkg/cmd/internal/templates/preview_comment.go index 66fc30eb9..cd9539b47 100644 --- a/pkg/cmd/internal/templates/preview_comment.go +++ b/pkg/cmd/internal/templates/preview_comment.go @@ -363,6 +363,7 @@ func previewCLIApplyAllOutput() { // MySQL: single table {"MYSQL: SINGLE TABLE RUNNING", previewProgressOutput}, {"MYSQL: SINGLE TABLE COMPLETED", previewCompletedOutput}, + {"MYSQL: SINGLE TABLE RETRYING", previewRetryingOutput}, {"MYSQL: SINGLE TABLE FAILED", previewFailedOutput}, {"MYSQL: SINGLE TABLE STOPPED", previewStoppedOutput}, {"MYSQL: SINGLE TABLE WAITING FOR CUTOVER", previewWaitingForCutoverOutput}, diff --git a/pkg/cmd/internal/templates/preview_dispatch.go b/pkg/cmd/internal/templates/preview_dispatch.go index 19c92fdc4..c63c454e3 100644 --- a/pkg/cmd/internal/templates/preview_dispatch.go +++ b/pkg/cmd/internal/templates/preview_dispatch.go @@ -29,6 +29,8 @@ func PreviewCLIOutput(previewType PreviewType) { previewCuttingOverOutput() case PreviewCompleted: previewCompletedOutput() + case PreviewRetrying: + previewRetryingOutput() case PreviewFailed: previewFailedOutput() case PreviewStopped: diff --git a/pkg/cmd/internal/templates/preview_progress.go b/pkg/cmd/internal/templates/preview_progress.go index 7b3a3e01c..6250edc8e 100644 --- a/pkg/cmd/internal/templates/preview_progress.go +++ b/pkg/cmd/internal/templates/preview_progress.go @@ -9,6 +9,7 @@ import ( "github.com/block/schemabot/pkg/apitypes" "github.com/block/schemabot/pkg/state" + "github.com/block/schemabot/pkg/storage" "vitess.io/vitess/go/vt/key" ) @@ -872,6 +873,34 @@ func previewFailedOutput() { WriteProgress(data) } +// previewRetryingOutput shows an apply between automatic recovery attempts: the +// drive was interrupted mid-copy, part of the retry budget is spent, and the +// next attempt is armed for a fixed time. The Retry row is what tells an +// operator this apply is waiting rather than wedged. The wait is the one the +// real policy arms for this attempt, so the preview cannot drift from it. +func previewRetryingOutput() { + data := ProgressData{ + State: state.Apply.FailedRetryable, + Engine: "Spirit", + ApplyID: "apply-a1b2c3d4e5f6", + StartedAt: previewTime.Add(-4 * time.Minute).Format(time.RFC3339), + ErrorMessage: "connection reset by peer", + Attempt: 3, + RetryAfter: previewTime.Add(storage.RetryBackoff(3)).Format(time.RFC3339), + Tables: []TableProgress{ + { + TableName: "users", Namespace: "testapp", + DDL: "ALTER TABLE `users` ADD INDEX `idx_email_created` (`email`, `created_at`)", + Status: state.Task.FailedRetryable, + RowsCopied: 156342, + RowsTotal: 397453, + PercentComplete: 39, + }, + }, + } + WriteProgress(data) +} + func previewStoppedOutput() { // Sample progress with stopped state (mid-apply stop) startedAt := previewTime.Add(-3 * time.Minute).Format(time.RFC3339) diff --git a/pkg/cmd/internal/templates/progress.go b/pkg/cmd/internal/templates/progress.go index 2a1588208..dd574cc23 100644 --- a/pkg/cmd/internal/templates/progress.go +++ b/pkg/cmd/internal/templates/progress.go @@ -56,6 +56,11 @@ func FormatKeyspaceHeader(ns string) string { // nowFunc returns the current time. Overridden in previews for deterministic output. var nowFunc = time.Now +// localZone is the zone absolute clock times render in โ€” an operator reads them +// against the wall clock in front of them. Overridden in previews so a rendered +// snapshot does not depend on the zone of the machine that produced it. +var localZone = time.Local + // volumeBoxRow returns the detail-box row for an operator-set volume level. // The level only matters while the engine is actively working (copying, // draining, or verifying โ€” volume stays adjustable through the post-copy @@ -69,6 +74,36 @@ func volumeBoxRow(volume int, applyState string) (BoxRow, bool) { return BoxRow{"Volume", fmt.Sprintf("%d/%d", volume, storage.MaxVolume)}, true } +// showsFailureCause reports whether the progress view prints the apply's error +// beneath the detail box. A permanently failed apply and an interrupted one ask +// the operator the same question โ€” what went wrong, and is it worth waiting on โ€” +// and only the raw message answers it. Other states have no cause to explain. +func showsFailureCause(applyState string) bool { + return state.IsState(applyState, state.Apply.Failed) || + state.IsState(applyState, state.Apply.FailedRetryable) +} + +// retryBoxRow renders the automatic-retry state of an interrupted apply: how +// much of the retry budget the next attempt consumes, and the clock time that +// attempt becomes eligible. The wait is shown as a time rather than a countdown +// because a watch redraws far more often than the backoff advances. It is +// omitted once the wait has elapsed โ€” the retry is then due and naming a time in +// the past would read as a missed deadline โ€” and outside the retrying state, +// where a spent attempt counter carries no signal. +func retryBoxRow(data ProgressData) (BoxRow, bool) { + if !state.IsState(data.State, state.Apply.FailedRetryable) { + return BoxRow{}, false + } + announced, retryComing := storage.AnnouncedRetryAttempt(data.Attempt) + retry := fmt.Sprintf("attempt %d/%d", announced, storage.MaxRecoveryAttempts) + if retryComing { + if due, err := time.Parse(time.RFC3339, data.RetryAfter); err == nil && due.After(nowFunc()) { + retry += " ยท next " + due.In(localZone).Format("15:04:05 MST") + } + } + return BoxRow{"Retry", retry}, true +} + // WriteProgress writes the schema change progress to stdout. func WriteProgress(data ProgressData) { // No active schema change @@ -106,6 +141,9 @@ func WriteProgress(data ProgressData) { rows = append(rows, BoxRow{"Environment", data.Environment}) } rows = append(rows, BoxRow{"State", displayState}) + if row, ok := retryBoxRow(data); ok { + rows = append(rows, row) + } if row, ok := volumeBoxRow(data.Volume, data.State); ok { rows = append(rows, row) } @@ -160,7 +198,7 @@ func WriteProgress(data ProgressData) { WriteBox(rows, "State", colorFn) // Error below the box - if data.State == state.Apply.Failed && data.ErrorMessage != "" { + if showsFailureCause(data.State) && data.ErrorMessage != "" { fmt.Printf("\n %s%s%s\n", ANSIRed, data.ErrorMessage, ANSIReset) } @@ -410,7 +448,7 @@ func FormatProgressState(s string) string { case state.Apply.Completed: return ANSIGreen + "โœ“ Completed" + ANSIReset case state.Apply.FailedRetryable: - return ANSIYellow + "โ†ป Retrying" + ANSIReset + return ANSIOrange + "โ†ป Retrying" + ANSIReset case state.Apply.Failed: return ANSIRed + "โœ— Failed" + ANSIReset case state.Apply.Stopped: @@ -592,7 +630,7 @@ func FormatTableProgressWithActivity(t TableProgress, activityBar, activityLabel case state.Apply.FailedRetryable: if t.PercentComplete > 0 || t.RowsCopied > 0 { retryPercent := ui.RowCopyDisplayPercent(t.PercentComplete, t.RowsCopied) - bar := ui.ProgressBar(retryPercent, ui.ColorYellow) + bar := ui.ProgressBar(retryPercent, ui.ColorOrange) fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: %s Retrying\n", t.TableName, bar) } else { fmt.Fprintf(&b, indentTable+progressSymbol(t.ChangeType)+"%s: Retrying\n", t.TableName) @@ -1361,7 +1399,7 @@ func stateColorFunc(s string) func(string) string { case state.Apply.Failed: return colorWrap(ANSIRed) case state.Apply.FailedRetryable: - return colorWrap(ANSIYellow) + return colorWrap(ANSIOrange) case state.Apply.Running, state.Apply.RunningDegraded, state.Apply.CatchingUp, state.Apply.Checksumming, state.Apply.PostChecksum: return colorWrap(ANSICyan) diff --git a/pkg/cmd/internal/templates/progress_parse.go b/pkg/cmd/internal/templates/progress_parse.go index 1099f5ba4..32844a20b 100644 --- a/pkg/cmd/internal/templates/progress_parse.go +++ b/pkg/cmd/internal/templates/progress_parse.go @@ -26,10 +26,16 @@ type ProgressData struct { ErrorMessage string StartedAt string // RFC3339 format CompletedAt string // RFC3339 format - Operations []ProgressOperation - Tables []TableProgress - Options map[string]string // Apply options (defer_cutover, skip_revert, etc.) - Metadata map[string]string // Engine metadata (e.g., deploy_request_url, branch_name) + // Attempt is how many automatic redispatches an interrupted apply has + // already spent of its retry budget. + Attempt int + // RetryAfter is the RFC3339 time the next automatic retry becomes eligible. + // Empty when no wait is in force and the retry is due. + RetryAfter string + Operations []ProgressOperation + Tables []TableProgress + Options map[string]string // Apply options (defer_cutover, skip_revert, etc.) + Metadata map[string]string // Engine metadata (e.g., deploy_request_url, branch_name) // Volume is the apply's current volume level (1=slowest, 11=fastest). // Zero means the operator never set one, so the display stays quiet. Volume int @@ -149,6 +155,8 @@ func ParseProgressResponse(result *apitypes.ProgressResponse) ProgressData { ErrorMessage: result.ErrorMessage, StartedAt: result.StartedAt, CompletedAt: result.CompletedAt, + Attempt: int(result.Attempt), + RetryAfter: result.RetryAfter, Options: result.Options, Metadata: result.Metadata, Volume: int(result.Volume), diff --git a/pkg/cmd/internal/templates/progress_parse_test.go b/pkg/cmd/internal/templates/progress_parse_test.go index c32c409b0..5327b361b 100644 --- a/pkg/cmd/internal/templates/progress_parse_test.go +++ b/pkg/cmd/internal/templates/progress_parse_test.go @@ -2,12 +2,14 @@ package templates import ( "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/block/schemabot/pkg/apitypes" "github.com/block/schemabot/pkg/state" + "github.com/block/schemabot/pkg/storage" ) func TestParseProgressResponseIncludesOperationsAndTableDeployments(t *testing.T) { @@ -96,6 +98,47 @@ func TestVolumeBoxRow(t *testing.T) { } } +// The retry row reports how much of the budget the next attempt spends and when +// it becomes eligible, so an operator can tell an apply that is waiting from one +// that is wedged. It appears only while the apply is retrying, and names a time +// only while one is still ahead. +func TestRetryBoxRow(t *testing.T) { + now := time.Date(2026, 1, 15, 14, 30, 0, 0, time.UTC) + originalNow, originalZone := nowFunc, localZone + t.Cleanup(func() { nowFunc, localZone = originalNow, originalZone }) + nowFunc = func() time.Time { return now } + localZone = time.UTC + + row, ok := retryBoxRow(ProgressData{ + State: state.Apply.FailedRetryable, + Attempt: 3, + RetryAfter: now.Add(90 * time.Second).Format(time.RFC3339), + }) + require.True(t, ok) + assert.Equal(t, "Retry", row.Label) + assert.Equal(t, "attempt 4/10 ยท next 14:31:30 UTC", row.Value) + + due, ok := retryBoxRow(ProgressData{ + State: state.Apply.FailedRetryable, + Attempt: 3, + RetryAfter: now.Add(-time.Second).Format(time.RFC3339), + }) + require.True(t, ok) + assert.Equal(t, "attempt 4/10", due.Value, "an elapsed wait names no time โ€” the retry is due") + + _, ok = retryBoxRow(ProgressData{State: state.Apply.Running, Attempt: 3}) + assert.False(t, ok, "a spent attempt counter carries no signal outside the retrying state") + + spent, ok := retryBoxRow(ProgressData{ + State: state.Apply.FailedRetryable, + Attempt: storage.MaxRecoveryAttempts, + RetryAfter: now.Add(90 * time.Second).Format(time.RFC3339), + }) + require.True(t, ok) + assert.Equal(t, "attempt 10/10", spent.Value, + "a spent budget names the last attempt made, not one the claim path will never admit") +} + func TestParseProgressResponseWithoutOperationsKeepsDeploymentEmpty(t *testing.T) { result := &apitypes.ProgressResponse{ State: state.Apply.Completed, diff --git a/pkg/cmd/internal/templates/progress_states_test.go b/pkg/cmd/internal/templates/progress_states_test.go index 3b715f08d..ec7ebe657 100644 --- a/pkg/cmd/internal/templates/progress_states_test.go +++ b/pkg/cmd/internal/templates/progress_states_test.go @@ -485,7 +485,7 @@ func TestFormatTableProgress_FailedRetryableKeepsProgress(t *testing.T) { } output := FormatTableProgress(tp) - assert.Contains(t, output, ui.ProgressBar(45, ui.ColorYellow)+" Retrying") + assert.Contains(t, output, ui.ProgressBar(45, ui.ColorOrange)+" Retrying") }) t.Run("without progress", func(t *testing.T) { @@ -497,7 +497,7 @@ func TestFormatTableProgress_FailedRetryableKeepsProgress(t *testing.T) { output := FormatTableProgress(tp) assert.Contains(t, output, "users: Retrying") - assert.NotContains(t, output, ui.ColorYellow) + assert.NotContains(t, output, ui.ColorOrange) }) } @@ -618,10 +618,10 @@ func TestStateColorsReserveRedForFailure(t *testing.T) { } } - for _, s := range []string{state.Apply.Stopped, state.Apply.Cancelled, state.Apply.Reverted} { + for _, s := range []string{state.Apply.Stopped, state.Apply.Cancelled, state.Apply.Reverted, state.Apply.FailedRetryable} { fn := stateColorFunc(s) require.NotNil(t, fn, "expected color function for state %q", s) - assert.Contains(t, fn(state.Label(s)), ANSIOrange, "operator-halted state %q must render orange", s) + assert.Contains(t, fn(state.Label(s)), ANSIOrange, "halted state %q must render orange", s) } } diff --git a/pkg/engine/spirit/execution.go b/pkg/engine/spirit/execution.go index 2a585f949..aafd9decd 100644 --- a/pkg/engine/spirit/execution.go +++ b/pkg/engine/spirit/execution.go @@ -471,7 +471,11 @@ func (e *Engine) executeSpiritMigration(ctx context.Context, host, username, pas logger.Error("schema change failed", "error", err, ) - e.setSchemaChangeFailed(fmt.Errorf("schema change failed: %w", err)) + failure := fmt.Errorf("schema change failed: %w", err) + if checksumRejectedUniqueIndex(runner, parsed) { + failure = &engine.PermanentError{Err: failure} + } + e.setSchemaChangeFailed(failure) utils.CloseAndLog(runner) return err } @@ -489,14 +493,28 @@ func (e *Engine) setSchemaChangeCompleted() { } } -// setSchemaChangeFailed sets the state to failed with an error message. +// setSchemaChangeFailed sets the state to failed with an error message, and +// records whether the failure is one a later attempt could clear. A rejection +// the target reproduces every time is reported as a permanent failure so the +// apply goes terminal on the spot instead of spending its recovery budget โ€” and +// the target's active-apply slot โ€” reproducing the same rejection. func (e *Engine) setSchemaChangeFailed(err error) { + err = classifyExecutionFailure(err) + if err != nil && !engine.IsRetryable(err) { + attrs := []any{"error", err} + if number, rejected := permanentDDLRejection(err); rejected { + attrs = append(attrs, "mysql_error_number", number) + } + e.changeLogger().Error("the target reproduces this failure on every attempt; the apply will not be retried", attrs...) + } + e.mu.Lock() defer e.mu.Unlock() if e.runningSchemaChange != nil { e.runningSchemaChange.state = engine.StateFailed if err != nil { e.runningSchemaChange.errorMessage = err.Error() + e.runningSchemaChange.permanentFailure = !engine.IsRetryable(err) } } } diff --git a/pkg/engine/spirit/failure_classification.go b/pkg/engine/spirit/failure_classification.go new file mode 100644 index 000000000..f6c4a6efc --- /dev/null +++ b/pkg/engine/spirit/failure_classification.go @@ -0,0 +1,139 @@ +package spirit + +import ( + "errors" + + spiritmigration "github.com/block/spirit/pkg/migration" + "github.com/block/spirit/pkg/statement" + "github.com/block/spirit/pkg/status" + "github.com/go-sql-driver/mysql" + + "github.com/block/schemabot/pkg/engine" +) + +// MySQL server error numbers the target returns when it rejects a statement for +// a reason that belongs to the statement or to the data already in the table. +// Re-running the identical statement against the same target reproduces each of +// them exactly, so an apply that meets one has nothing to wait for. +const ( + // The rows already in the table do not satisfy the schema being applied. + // Only a change to the data can make the statement succeed. + errNullNotAllowed = 1048 // ER_BAD_NULL_ERROR + errDuplicateEntry = 1062 // ER_DUP_ENTRY + errDataOutOfRange = 1264 // ER_WARN_DATA_OUT_OF_RANGE + errTruncatedWrongValue = 1292 // ER_TRUNCATED_WRONG_VALUE + errTruncatedWrongFieldValue = 1366 // ER_TRUNCATED_WRONG_VALUE_FOR_FIELD + errDataTooLong = 1406 // ER_DATA_TOO_LONG + errCheckConstraintViolated = 3819 // ER_CHECK_CONSTRAINT_VIOLATED + + // The statement contradicts the schema it is being applied to, or is not + // valid DDL. Only a change to the schema files can make it succeed. + errUnknownColumn = 1054 // ER_BAD_FIELD_ERROR + errDuplicateColumnName = 1060 // ER_DUP_FIELDNAME + errDuplicateKeyName = 1061 // ER_DUP_KEYNAME + errParse = 1064 // ER_PARSE_ERROR + errInvalidDefault = 1067 // ER_INVALID_DEFAULT + errKeyColumnNotFound = 1072 // ER_KEY_COLUMN_DOES_NOT_EXITS + errCantDropFieldOrKey = 1091 // ER_CANT_DROP_FIELD_OR_KEY + errBlobKeyWithoutLength = 1170 // ER_BLOB_KEY_WITHOUT_LENGTH + errPrimaryCantHaveNull = 1171 // ER_PRIMARY_CANT_HAVE_NULL + + // The target refuses to perform the operation at all. A later attempt asks + // for the same unsupported operation and is refused the same way. + errNotSupportedYet = 1235 // ER_NOT_SUPPORTED_YET + errAlterOperationNotSupported = 1845 // ER_ALTER_OPERATION_NOT_SUPPORTED + errAlterOperationNotSupportedReason = 1846 // ER_ALTER_OPERATION_NOT_SUPPORTED_REASON +) + +// permanentDDLErrors is the set of target rejections an apply cannot retry its +// way out of. Membership is deliberately narrow: a rejection left out of the set +// keeps the automatic retries and costs the operator only the wait they were +// already spending, while a rejection wrongly added to it ends an apply that +// would have recovered on its own. Anything the target might answer differently +// once a lock, a lagging replica, disk, or a peer transaction has moved on โ€” +// lock wait timeouts, deadlocks, read-only, connection loss โ€” stays out. +var permanentDDLErrors = map[uint16]struct{}{ + errNullNotAllowed: {}, + errDuplicateEntry: {}, + errDataOutOfRange: {}, + errTruncatedWrongValue: {}, + errTruncatedWrongFieldValue: {}, + errDataTooLong: {}, + errCheckConstraintViolated: {}, + errUnknownColumn: {}, + errDuplicateColumnName: {}, + errDuplicateKeyName: {}, + errParse: {}, + errInvalidDefault: {}, + errKeyColumnNotFound: {}, + errCantDropFieldOrKey: {}, + errBlobKeyWithoutLength: {}, + errPrimaryCantHaveNull: {}, + errNotSupportedYet: {}, + errAlterOperationNotSupported: {}, + errAlterOperationNotSupportedReason: {}, +} + +// permanentDDLRejection returns the target's error number when err carries a +// rejection an apply cannot retry its way out of. The number is what names the +// rejection in the log that records why an apply skipped operator recovery. +func permanentDDLRejection(err error) (uint16, bool) { + var mysqlErr *mysql.MySQLError + if !errors.As(err, &mysqlErr) { + return 0, false + } + if _, permanent := permanentDDLErrors[mysqlErr.Number]; !permanent { + return 0, false + } + return mysqlErr.Number, true +} + +// checksumRejectedUniqueIndex reports whether a failed Spirit run failed the way +// a unique index over data that is not unique fails. That rejection never +// reaches the engine as a target error: the row copy drops the duplicate rows +// instead of refusing them, so the copy succeeds and the checksum that follows +// can never be made to agree with the source. The condition is the one Spirit +// itself uses to explain the failure โ€” the run ended inside the checksum phase +// and the batch adds a unique constraint โ€” read from the runner's structured +// status and the parsed statements rather than from the message Spirit renders. +// +// The caller reaches this only after ruling out a cancelled context, so a stop +// during the checksum is never mistaken for the target's verdict. +// +// The condition is coarser than the failure it names, because the error that +// would separate the two is not available here: a checksum that kept finding +// row differences is the unique-index rejection, while one that errored on +// every attempt is the transient failure the retries exist for, and both +// arrive with the differences discarded. The coarse read costs a transiently +// errored checksum its retries and the operator a re-apply; the alternative +// costs a rejected unique index its whole budget, and each attempt re-verifies +// the table in full, because a checksum that found differences is not allowed +// to persist a watermark to resume from. Narrow this to the differences case +// once the run failure carries which one it was. +func checksumRejectedUniqueIndex(runner *spiritmigration.Runner, parsed []*statement.AbstractStatement) bool { + if runner.Progress().CurrentState != status.Checksum { + return false + } + for _, stmt := range parsed { + if errors.Is(stmt.AlterContainsAddUnique(), statement.ErrAlterContainsUnique) { + return true + } + } + return false +} + +// classifyExecutionFailure marks err permanent when the target rejected the +// statement for a reason a later attempt cannot change, and returns it unchanged +// otherwise. Recording the classification lets a failed schema change either +// enter operator recovery or go terminal immediately: spending a recovery budget +// on a rejection that reproduces every time buys nothing and holds the target's +// active-apply slot for the whole budget. +func classifyExecutionFailure(err error) error { + if err == nil { + return nil + } + if _, permanent := permanentDDLRejection(err); !permanent { + return err + } + return &engine.PermanentError{Err: err} +} diff --git a/pkg/engine/spirit/failure_classification_integration_test.go b/pkg/engine/spirit/failure_classification_integration_test.go new file mode 100644 index 000000000..dffc2f8a5 --- /dev/null +++ b/pkg/engine/spirit/failure_classification_integration_test.go @@ -0,0 +1,88 @@ +//go:build integration + +package spirit + +import ( + "database/sql" + "log/slog" + "os" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/schemabot/pkg/engine" +) + +// runToTerminalProgress runs ddl through the engine's own execution path and +// returns the progress an operator's poller would read once it settles. +func runToTerminalProgress(t *testing.T, dsn, table, ddl string) *engine.ProgressResult { + t.Helper() + + host, username, password, database, err := parseDSN(dsn) + require.NoError(t, err, "parseDSN") + + eng := New(Config{Logger: slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError}))}) + defer eng.Drain() + + eng.mu.Lock() + eng.runningSchemaChange = &runningSchemaChange{ + database: database, + tables: []string{table}, + state: engine.StateRunning, + started: time.Now(), + } + eng.mu.Unlock() + + eng.executeSchemaChange(t.Context(), host, username, password, database, []string{ddl}, false, directPolicy{}) + + result, err := eng.Progress(t.Context(), &engine.ProgressRequest{}) + require.NoError(t, err, "Progress()") + return result +} + +// A unique index over data that already contains duplicates is rejected by the +// target on every attempt. The apply fails permanently rather than entering +// operator recovery: the automatic retries can only reproduce the rejection, +// and while they run the apply holds the database's active-apply slot and the +// operator reads an apply that still looks like it might recover. +func TestDuplicateDataFailsTheApplyWithoutRetrying(t *testing.T) { + dsn, db := setupTestMySQL(t) + cleanupTables(t, db) + + _, err := db.ExecContext(t.Context(), "CREATE TABLE `duplicate_unique_index` ("+ + "id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, "+ + "email VARCHAR(100) NOT NULL"+ + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci") + require.NoError(t, err, "create table") + + _, err = db.ExecContext(t.Context(), + "INSERT INTO `duplicate_unique_index` (email) VALUES ('a@example.com'), ('a@example.com')") + require.NoError(t, err, "seed duplicate rows") + + result := runToTerminalProgress(t, dsn, "duplicate_unique_index", + "ALTER TABLE `duplicate_unique_index` ADD UNIQUE INDEX `idx_email` (`email`)") + + assert.Equal(t, engine.StateFailed, result.State) + assert.False(t, result.Retryable, + "the duplicate rows are still there on the next attempt, so retrying can only reproduce the rejection") + assert.NotEmpty(t, result.ErrorMessage, + "a permanent failure must still say what the target objected to") + assert.Zero(t, indexCount(t, db, "duplicate_unique_index", "idx_email"), + "the index the operator asked for was not added") +} + +// indexCount reports how many parts of the named index exist on the table, and +// zero when the index was not created. +func indexCount(t *testing.T, db *sql.DB, tableName, indexName string) int { + t.Helper() + + var count int + require.NoError(t, db.QueryRowContext(t.Context(), + "SELECT COUNT(*) FROM information_schema.STATISTICS "+ + "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?", + tableName, indexName, + ).Scan(&count), "read index %s on %s", indexName, tableName) + return count +} diff --git a/pkg/engine/spirit/failure_classification_test.go b/pkg/engine/spirit/failure_classification_test.go new file mode 100644 index 000000000..89bb5c65c --- /dev/null +++ b/pkg/engine/spirit/failure_classification_test.go @@ -0,0 +1,103 @@ +package spirit + +import ( + "errors" + "fmt" + "testing" + + "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + + "github.com/block/schemabot/pkg/engine" +) + +// The rejections an apply cannot retry its way out of are reported as permanent +// failures, and everything else keeps the automatic retries. The transient cases +// are the ones the retry budget exists for: they are the boundary the set must +// not cross. +func TestExecutionFailureClassification(t *testing.T) { + tests := []struct { + name string + err error + retryable bool + }{ + { + name: "duplicate value for a new unique index", + err: &mysql.MySQLError{Number: errDuplicateEntry, Message: "Duplicate entry 'a@example.com' for key 'idx_email'"}, + retryable: false, + }, + { + name: "existing rows violate a new NOT NULL column", + err: &mysql.MySQLError{Number: errNullNotAllowed, Message: "Column 'name' cannot be null"}, + retryable: false, + }, + { + name: "existing data does not fit the narrowed column", + err: &mysql.MySQLError{Number: errDataTooLong, Message: "Data too long for column 'name'"}, + retryable: false, + }, + { + name: "dropping a column the table does not have", + err: &mysql.MySQLError{Number: errCantDropFieldOrKey, Message: "Can't DROP 'missing'"}, + retryable: false, + }, + { + name: "the target refuses the operation outright", + err: &mysql.MySQLError{Number: errAlterOperationNotSupportedReason, Message: "ALGORITHM=INPLACE is not supported"}, + retryable: false, + }, + { + name: "wrapped rejections are classified through the chain", + err: fmt.Errorf("schema change failed: %w", &mysql.MySQLError{Number: errDuplicateEntry}), + retryable: false, + }, + { + name: "lock wait timeout clears once the holder commits", + err: &mysql.MySQLError{Number: 1205, Message: "Lock wait timeout exceeded"}, + retryable: true, + }, + { + name: "deadlock victims succeed on a later attempt", + err: &mysql.MySQLError{Number: 1213, Message: "Deadlock found when trying to get lock"}, + retryable: true, + }, + { + name: "a read-only target accepts writes again after failover", + err: &mysql.MySQLError{Number: 1290, Message: "The MySQL server is running with the --read-only option"}, + retryable: true, + }, + { + name: "a lost connection is reconnected on the next attempt", + err: &mysql.MySQLError{Number: 2013, Message: "Lost connection to MySQL server during query"}, + retryable: true, + }, + { + name: "failures that never reached the target stay retryable", + err: errors.New("dial tcp: connect: connection refused"), + retryable: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + classified := classifyExecutionFailure(tt.err) + assert.Equal(t, tt.retryable, engine.IsRetryable(classified)) + assert.ErrorIs(t, classified, tt.err, "classification must preserve the original error for the caller to report") + }) + } +} + +// A classified failure keeps the target's own message, which is the only place +// an operator reads what the target objected to. +func TestPermanentClassificationPreservesTheTargetMessage(t *testing.T) { + rejection := &mysql.MySQLError{Number: errDuplicateEntry, Message: "Duplicate entry 'a@example.com' for key 'idx_email'"} + + classified := classifyExecutionFailure(fmt.Errorf("schema change failed: %w", rejection)) + + assert.False(t, engine.IsRetryable(classified)) + assert.Contains(t, classified.Error(), "Duplicate entry 'a@example.com' for key 'idx_email'") +} + +func TestClassifyExecutionFailureIgnoresNil(t *testing.T) { + assert.NoError(t, classifyExecutionFailure(nil)) +} diff --git a/pkg/engine/spirit/spirit.go b/pkg/engine/spirit/spirit.go index 7f50f17e0..90f8e2694 100644 --- a/pkg/engine/spirit/spirit.go +++ b/pkg/engine/spirit/spirit.go @@ -95,6 +95,7 @@ type runningSchemaChange struct { progressCallback func() string // returns Summary from Spirit's Progress API state engine.State errorMessage string // Error details when state is StateFailed + permanentFailure bool // Set when the recorded failure reproduces on every later attempt started time.Time deferCutover bool // Whether to defer cutover until manual trigger volumeRestartInProgress bool // Set while stored stopped state should still be exposed as running progress. @@ -796,7 +797,7 @@ func (e *Engine) Progress(ctx context.Context, req *engine.ProgressRequest) (*en State: state, Message: message, ErrorMessage: rm.errorMessage, - Retryable: state == engine.StateFailed, + Retryable: state == engine.StateFailed && !rm.permanentFailure, Tables: tableProgress, ResumeState: req.ResumeState, }, nil diff --git a/pkg/engine/spirit/spirit_integration_test.go b/pkg/engine/spirit/spirit_integration_test.go index ed1bcaede..c1afc138a 100644 --- a/pkg/engine/spirit/spirit_integration_test.go +++ b/pkg/engine/spirit/spirit_integration_test.go @@ -1604,7 +1604,8 @@ func TestEngine_Progress_FailingApplyNeverReportsCompleted(t *testing.T) { assert.Equal(t, engine.StateFailed, result.State) assert.Contains(t, result.ErrorMessage, "schema change failed") assert.Contains(t, result.ErrorMessage, "nonexistent_column") - assert.True(t, result.Retryable) + assert.False(t, result.Retryable, + "a column the table does not have is missing on every attempt, so the failure is permanent") } // TestEngine_ExecuteMigration_MultipleStatements tests running multiple diff --git a/pkg/schema/mysql/applies.sql b/pkg/schema/mysql/applies.sql index 0ecd10311..66a8faee2 100644 --- a/pkg/schema/mysql/applies.sql +++ b/pkg/schema/mysql/applies.sql @@ -18,6 +18,7 @@ CREATE TABLE `applies` ( `error_message` text, `options` json NOT NULL, `attempt` int NOT NULL DEFAULT '0', + `retry_after` datetime(6) DEFAULT NULL, `lease_owner` varchar(255) NOT NULL DEFAULT '', `lease_token` varchar(64) NOT NULL DEFAULT '', `lease_acquired_at` datetime DEFAULT NULL, diff --git a/pkg/schema/postgres/applies.sql b/pkg/schema/postgres/applies.sql index 3d76b00f5..9cac7844d 100644 --- a/pkg/schema/postgres/applies.sql +++ b/pkg/schema/postgres/applies.sql @@ -18,6 +18,7 @@ CREATE TABLE applies ( error_message text, options jsonb NOT NULL, attempt integer NOT NULL DEFAULT 0, + retry_after timestamp DEFAULT NULL, lease_owner varchar(255) NOT NULL DEFAULT '', lease_token varchar(64) NOT NULL DEFAULT '', lease_acquired_at timestamp DEFAULT NULL, diff --git a/pkg/storage/internal/sqlstore/applies.go b/pkg/storage/internal/sqlstore/applies.go index 4cbd8b839..8711dcb5e 100644 --- a/pkg/storage/internal/sqlstore/applies.go +++ b/pkg/storage/internal/sqlstore/applies.go @@ -24,13 +24,13 @@ import ( // applyColumns lists all columns for SELECT queries. const applyColumns = `id, apply_identifier, lock_id, plan_id, database_name, database_type, repository, pull_request, environment, deployment, caller, installation_id, external_id, idempotency_key, engine, - state, error_message, options, attempt, + state, error_message, options, attempt, retry_after, lease_owner, lease_token, lease_acquired_at, created_at, started_at, completed_at, updated_at, revert_skipped_at` const applyColumnsForApplyAlias = `a.id, a.apply_identifier, a.lock_id, a.plan_id, a.database_name, a.database_type, a.repository, a.pull_request, a.environment, a.deployment, a.caller, a.installation_id, a.external_id, a.idempotency_key, a.engine, - a.state, a.error_message, a.options, a.attempt, + a.state, a.error_message, a.options, a.attempt, a.retry_after, a.lease_owner, a.lease_token, a.lease_acquired_at, a.created_at, a.started_at, a.completed_at, a.updated_at, a.revert_skipped_at` @@ -51,6 +51,31 @@ const ( applyTargetLockReleaseTimeout = 5 * time.Second ) +// retryBackoffDeadline renders the time an apply admitted on this attempt may be +// claimed again. The deadline is computed by the database rather than bound as a +// Go timestamp so the wait never depends on the app and database clocks agreeing. +// +// Both the deadline and the predicate that reads it back are expressed at the +// precision the retry policy is written in, and retry_after stores that same +// precision on every dialect. A backoff rounded to a coarser column would shift +// the wait by up to that rounding step โ€” in the direction that holds back a +// retry carrying no wait at all โ€” and the shift would depend on which dialect +// the deployment runs. +func retryBackoffDeadline(dialect Dialect, attempt int) string { + backoff := uint64(storage.RetryBackoff(attempt).Microseconds()) + return dialect.RelativeTime(TimestampPrecisionMicrosecond, AfterCurrentTime, LiteralIntervalAmount(backoff), IntervalMicrosecond) +} + +// retryBackoffElapsed renders the predicate that holds a failed_retryable apply +// unclaimable until the backoff armed by its last attempt has run out. It gates +// the retryable claim clause only: a stale-active claim is crash recovery and a +// control-request claim is an operator command, and neither should be made to +// wait on a retry the operator is not asking for. A NULL retry_after means no +// wait was armed, so the row is claimable immediately. +func retryBackoffElapsed(dialect Dialect, alias string) string { + return "(" + alias + ".retry_after IS NULL OR " + alias + ".retry_after <= " + dialect.CurrentTimestamp(TimestampPrecisionMicrosecond) + ")" +} + // applyStore implements storage.ApplyStore using MySQL. type applyStore struct { db *rebindDB @@ -1449,7 +1474,7 @@ func (s *applyStore) ClaimApplyByID(ctx context.Context, applyID int64, owner st OR EXISTS (SELECT 1 FROM apply_operations ao WHERE ao.apply_id = a.id) )) OR (a.state IN (%s) AND a.updated_at < %s) - OR (a.state = ? AND a.attempt < ? AND a.updated_at >= %s) + OR (a.state = ? AND a.attempt < ? AND a.updated_at >= %s AND %s) OR ( a.state = ? AND EXISTS ( @@ -1482,7 +1507,7 @@ func (s *applyStore) ClaimApplyByID(ctx context.Context, applyID int64, owner st ) LIMIT 1 FOR UPDATE SKIP LOCKED - `, applyColumns, activeStatePlaceholders, staleClaimCutoff, retryFreshnessCutoff, staleClaimCutoff), queryArgs...) + `, applyColumns, activeStatePlaceholders, staleClaimCutoff, retryFreshnessCutoff, retryBackoffElapsed(s.dialect, "a"), staleClaimCutoff), queryArgs...) apply, err := scanApplyInto(row) if errors.Is(err, sql.ErrNoRows) { @@ -1765,7 +1790,7 @@ func persistApplyClaim(ctx context.Context, db *rebindDB, locker namedlock.Locke } if isStartingClaim(apply.State) { - claimed, err := transitionClaimToState(ctx, tx, apply, state.Apply.Running, owner, leaseToken) + claimed, err := transitionClaimToState(ctx, dialect, tx, apply, state.Apply.Running, owner, leaseToken) if err != nil { return claimLostRace, err } @@ -1818,7 +1843,7 @@ func claimStoppedApplyUnderTargetLock(ctx context.Context, db *rebindDB, locker return refuseStoppedClaimForActiveTarget(ctx, tx, apply, owner, database, dbType, environment) } - claimed, err := transitionClaimToState(ctx, tx, apply, state.Apply.Resuming, owner, leaseToken) + claimed, err := transitionClaimToState(ctx, dialect, tx, apply, state.Apply.Resuming, owner, leaseToken) if err != nil { return claimLostRace, err } @@ -1846,16 +1871,25 @@ func isStartingClaim(applyState string) bool { // landing between the SELECT and this UPDATE; a zero rows-affected result means // another driver already moved the row, so the caller backs off cleanly. Reports // false on that lost race. -func transitionClaimToState(ctx context.Context, tx *rebindTx, apply *storage.Apply, targetState, owner, leaseToken string) (bool, error) { - result, err := tx.ExecContext(ctx, ` +// +// A retryable claim also arms retry_after for the attempt after this one, so a +// failure that reproduces instantly waits before burning the next unit of the +// budget. Arming it here โ€” when the attempt is admitted, not when it fails โ€” +// measures the wait from the start of the attempt, so an attempt that ran +// longer than its own backoff has already spaced itself out and retries as soon +// as it fails. +func transitionClaimToState(ctx context.Context, dialect Dialect, tx *rebindTx, apply *storage.Apply, targetState, owner, leaseToken string) (bool, error) { + result, err := tx.ExecContext(ctx, fmt.Sprintf(` UPDATE applies SET state = ?, updated_at = NOW(), lease_owner = ?, lease_token = ?, lease_acquired_at = NOW(), attempt = CASE WHEN ? = ? THEN attempt + 1 ELSE attempt END, + retry_after = CASE WHEN ? = ? THEN %s ELSE retry_after END, completed_at = NULL, error_message = CASE WHEN ? = ? THEN '' ELSE error_message END WHERE id = ? AND state = ? - `, targetState, owner, leaseToken, apply.State, state.Apply.FailedRetryable, apply.State, state.Apply.FailedRetryable, apply.ID, apply.State) + `, retryBackoffDeadline(dialect, apply.Attempt+1)), + targetState, owner, leaseToken, apply.State, state.Apply.FailedRetryable, apply.State, state.Apply.FailedRetryable, apply.State, state.Apply.FailedRetryable, apply.ID, apply.State) if err != nil { return false, fmt.Errorf("claim apply %d (%s) in state %s: %w", apply.ID, apply.ApplyIdentifier, apply.State, err) } @@ -2340,7 +2374,7 @@ func scanApplies(rows *sql.Rows) ([]*storage.Apply, error) { // scanApplyInto scans apply data from any scanner (Row or Rows). func scanApplyInto(s scanner) (*storage.Apply, error) { var apply storage.Apply - var leaseAcquiredAt, startedAt, completedAt, revertSkippedAt sql.NullTime + var retryAfter, leaseAcquiredAt, startedAt, completedAt, revertSkippedAt sql.NullTime var idempotencyKey sql.NullString var options []byte @@ -2349,7 +2383,7 @@ func scanApplyInto(s scanner) (*storage.Apply, error) { &apply.Database, &apply.DatabaseType, &apply.Repository, &apply.PullRequest, &apply.Environment, &apply.Deployment, &apply.Caller, &apply.InstallationID, &apply.ExternalID, &idempotencyKey, &apply.Engine, - &apply.State, &apply.ErrorMessage, &options, &apply.Attempt, + &apply.State, &apply.ErrorMessage, &options, &apply.Attempt, &retryAfter, &apply.LeaseOwner, &apply.LeaseToken, &leaseAcquiredAt, &apply.CreatedAt, &startedAt, &completedAt, &apply.UpdatedAt, &revertSkippedAt, ) @@ -2360,6 +2394,10 @@ func scanApplyInto(s scanner) (*storage.Apply, error) { apply.IdempotencyKey = idempotencyKey.String apply.Options = options + if retryAfter.Valid { + apply.RetryAfter = &retryAfter.Time + } + if leaseAcquiredAt.Valid { apply.LeaseAcquiredAt = &leaseAcquiredAt.Time } diff --git a/pkg/storage/internal/sqlstore/applies_test.go b/pkg/storage/internal/sqlstore/applies_test.go index d41b4a69c..919a234b4 100644 --- a/pkg/storage/internal/sqlstore/applies_test.go +++ b/pkg/storage/internal/sqlstore/applies_test.go @@ -2416,6 +2416,114 @@ func TestApplyStore_ClaimApplyByIDRefusesUnrecoverableRetryable(t *testing.T) { }) } +// A failure that reproduces instantly would otherwise spend the whole recovery +// budget in seconds, on exactly the failures a little time was most likely to +// clear. Each admitted attempt arms a wait for the next one, and the retryable +// claim arm honours it: the apply is unclaimable until the wait elapses, then +// claimable again with a longer wait armed for the attempt after that. +func TestApplyStore_ClaimApplyByIDBacksOffBetweenRetries(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := NewMySQL(testDB) + + lock := createTestLock(t, store, "testdb", storage.DatabaseTypeMySQL, "staging") + apply := createTestApplyWithStateAndEnv(t, store, lock, "apply_retry_backoff", 620, state.Apply.FailedRetryable, "staging") + + // The first two attempts carry no wait, so a transient blip clears without + // an operator noticing a pause. + for attempt := 1; attempt <= 2; attempt++ { + claimed, err := store.Applies().ClaimApplyByID(ctx, apply.ID, "operator-a") + require.NoError(t, err, "attempt %d", attempt) + require.NotNil(t, claimed, "attempt %d must be admitted without a wait", attempt) + require.NoError(t, failRetryable(t, apply.ID)) + } + + blocked, err := store.Applies().ClaimApplyByID(ctx, apply.ID, "operator-a") + require.NoError(t, err) + assert.Nil(t, blocked, "a retry armed with a wait must not be admitted before the wait elapses") + + persisted, err := store.Applies().Get(ctx, apply.ID) + require.NoError(t, err) + require.NotNil(t, persisted.RetryAfter, "an admitted attempt arms the wait for the next one") + assert.True(t, persisted.RetryAfter.After(time.Now()), "the armed wait is in the future") + + _, err = testDB.ExecContext(ctx, `UPDATE applies SET retry_after = NOW() - INTERVAL 1 SECOND WHERE id = ?`, apply.ID) + require.NoError(t, err) + + claimed, err := store.Applies().ClaimApplyByID(ctx, apply.ID, "operator-a") + require.NoError(t, err) + require.NotNil(t, claimed, "the retry is admitted once its wait has elapsed") + assert.Equal(t, 3, claimed.Attempt) +} + +// The backoff paces automatic retries, not operators. A start request is an +// explicit instruction to run now, and a stale lease means a driver died +// mid-attempt rather than an attempt failing โ€” neither should be made to wait +// out a retry the operator did not ask for. +func TestApplyStore_ClaimApplyByIDIgnoresBackoffOutsideAutomaticRetry(t *testing.T) { + t.Run("operator start request", func(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := NewMySQL(testDB) + + lock := createTestLock(t, store, "testdb", storage.DatabaseTypeMySQL, "staging") + apply := createTestApplyWithStateAndEnv(t, store, lock, "apply_backoff_start", 621, state.Apply.Stopped, "staging") + require.NoError(t, armRetryBackoff(t, apply.ID)) + _, alreadyPending, err := store.ControlRequests().RequestPending(ctx, &storage.ApplyControlRequest{ + ApplyID: apply.ID, + Operation: storage.ControlOperationStart, + Status: storage.ControlRequestPending, + Metadata: []byte(`{}`), + }) + require.NoError(t, err) + require.False(t, alreadyPending) + + claimed, err := store.Applies().ClaimApplyByID(ctx, apply.ID, "operator-a") + require.NoError(t, err) + require.NotNil(t, claimed, "an operator start runs now regardless of an armed retry wait") + }) + + t.Run("stale lease recovery", func(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := NewMySQL(testDB) + + lock := createTestLock(t, store, "testdb", storage.DatabaseTypeMySQL, "staging") + apply := createTestApplyWithStateAndEnv(t, store, lock, "apply_backoff_stale", 622, state.Apply.Running, "staging") + require.NoError(t, armRetryBackoff(t, apply.ID)) + _, err := testDB.ExecContext(ctx, ` + UPDATE applies + SET lease_owner = 'dead-driver', lease_token = 'token', updated_at = NOW() - INTERVAL 1 HOUR + WHERE id = ? + `, apply.ID) + require.NoError(t, err) + + claimed, err := store.Applies().ClaimApplyByID(ctx, apply.ID, "operator-a") + require.NoError(t, err) + require.NotNil(t, claimed, "crash recovery must not wait out an armed retry backoff") + }) +} + +// failRetryable returns a claimed apply to failed_retryable, the state a drive +// leaves behind when its attempt fails with a recoverable error. +func failRetryable(t *testing.T, applyID int64) error { + t.Helper() + _, err := testDB.ExecContext(t.Context(), ` + UPDATE applies SET state = ?, lease_owner = '', lease_token = '', updated_at = NOW() WHERE id = ? + `, state.Apply.FailedRetryable, applyID) + return err +} + +// armRetryBackoff puts a retry wait well into the future, so a claim that is +// admitted anyway proves it does not consult the backoff. +func armRetryBackoff(t *testing.T, applyID int64) error { + t.Helper() + _, err := testDB.ExecContext(t.Context(), ` + UPDATE applies SET retry_after = NOW() + INTERVAL 1 HOUR WHERE id = ? + `, applyID) + return err +} + // ClaimApplyByID is how the operation-level claim loop acquires the parent apply // lease after leasing a stale operation row. When a PlanetScale driver crashes // mid-setup the operation row stays running (stale) while the parent apply sits diff --git a/pkg/storage/internal/sqlstore/apply_operations.go b/pkg/storage/internal/sqlstore/apply_operations.go index ebef5d4a5..166f6c8c2 100644 --- a/pkg/storage/internal/sqlstore/apply_operations.go +++ b/pkg/storage/internal/sqlstore/apply_operations.go @@ -985,6 +985,7 @@ func (s *applyOperationStore) FindNextApplyOperation(ctx context.Context, owner a.state = ? AND a.attempt < ? AND a.updated_at >= %s + AND %s ) OR ( a.state IN (%s) @@ -997,7 +998,7 @@ func (s *applyOperationStore) FindNextApplyOperation(ctx context.Context, owner ORDER BY created_at, id LIMIT 1 FOR UPDATE SKIP LOCKED - `, applyOperationColumns, terminalStatePlaceholders, activeStatePlaceholders, staleClaimCutoff, activeStatePlaceholders, staleClaimCutoff, retryFreshnessCutoff, activeStatePlaceholders, staleClaimCutoff), queryArgs...) + `, applyOperationColumns, terminalStatePlaceholders, activeStatePlaceholders, staleClaimCutoff, activeStatePlaceholders, staleClaimCutoff, retryFreshnessCutoff, retryBackoffElapsed(s.dialect, "a"), activeStatePlaceholders, staleClaimCutoff), queryArgs...) ad, err := scanApplyOperationInto(row) if errors.Is(err, sql.ErrNoRows) { @@ -1101,6 +1102,12 @@ func (s *applyOperationStore) FindNextApplyOperation(ctx context.Context, owner // different failed_retryable operations of the same apply concurrently, and // the row lock this UPDATE takes serializes them, so the second sees the // already-incremented attempt and does not overshoot maxRecoveryAttempts. + // That concurrency holds only for claims racing before the first of them + // commits. Afterwards the parent's freshly armed retry_after gates the + // claim clause, so a sibling operation that failed later waits out the + // same backoff โ€” the pacing is per apply, not per operation, and a + // multi-deployment fan-out spends one shared budget rather than one per + // deployment. if ad.State == state.ApplyOperation.FailedRetryable { // Advance the operation's own attempt only on a genuine deliberate // redispatch โ€” the parent apply is still failed_retryable. A @@ -1133,9 +1140,22 @@ func (s *applyOperationStore) FindNextApplyOperation(ctx context.Context, owner ad.Attempt++ } - if _, err := tx.ExecContext(ctx, ` + // Arming the parent's retry backoff needs the attempt this redispatch + // consumes, so the current value is read inside the claim + // transaction. A sibling operation of the same apply can consume its + // own unit of the budget between this read and the UPDATE, leaving + // the computed wait one step short โ€” bounded by the cap, and re-armed + // by the next claim either way. + var parentAttempt int + if err := tx.QueryRowContext(ctx, ` + SELECT attempt FROM applies WHERE id = ? + `, ad.ApplyID).Scan(&parentAttempt); err != nil { + return nil, fmt.Errorf("read retry budget for apply_operation %d redispatch: %w", ad.ID, err) + } + + if _, err := tx.ExecContext(ctx, fmt.Sprintf(` UPDATE applies - SET attempt = attempt + 1, updated_at = NOW() + SET attempt = attempt + 1, retry_after = %s, updated_at = NOW() WHERE id = ? AND state = ? AND attempt < ? @@ -1143,7 +1163,8 @@ func (s *applyOperationStore) FindNextApplyOperation(ctx context.Context, owner SELECT 1 FROM apply_operations o WHERE o.apply_id = applies.id AND o.id <> ? ) - `, ad.ApplyID, state.Apply.FailedRetryable, maxRecoveryAttempts, ad.ID); err != nil { + `, retryBackoffDeadline(s.dialect, parentAttempt+1)), + ad.ApplyID, state.Apply.FailedRetryable, maxRecoveryAttempts, ad.ID); err != nil { return nil, fmt.Errorf("consume retry budget for apply_operation %d redispatch: %w", ad.ID, err) } } diff --git a/pkg/storage/internal/sqlstore/apply_operations_test.go b/pkg/storage/internal/sqlstore/apply_operations_test.go index a7895b375..df6e9e322 100644 --- a/pkg/storage/internal/sqlstore/apply_operations_test.go +++ b/pkg/storage/internal/sqlstore/apply_operations_test.go @@ -1675,6 +1675,80 @@ func TestApplyOperationStore_FindNextApplyOperation_ClaimsFailedRetryableWithinB assert.WithinDuration(t, time.Now(), persisted.UpdatedAt, 5*time.Second, "heartbeat must be refreshed on re-claim") } +// A multi-deployment apply paces its automatic retries off the same armed wait +// as a single-deployment one: the budget is the parent apply's, so a +// failed_retryable operation stays unclaimable until the parent's wait elapses. +// Without this a fan-out whose failure reproduces instantly burns the whole +// shared budget as fast as the claim loop can spin. +func TestApplyOperationStore_FindNextApplyOperation_HonorsParentRetryBackoff(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := NewMySQL(testDB) + + lock := createTestLock(t, store, "testdb", "mysql", "staging") + apply := createTestApplyWithStateAndEnv(t, store, lock, "apply_op_backoff_gate", 1, state.Apply.FailedRetryable, "staging") + _, err := testDB.ExecContext(ctx, ` + UPDATE applies SET attempt = ?, updated_at = NOW() WHERE id = ? + `, maxRecoveryAttempts-1, apply.ID) + require.NoError(t, err) + require.NoError(t, armRetryBackoff(t, apply.ID)) + + id, err := store.ApplyOperations().Insert(ctx, &storage.ApplyOperation{ + ApplyID: apply.ID, Deployment: "region-a", State: state.ApplyOperation.FailedRetryable, + }) + require.NoError(t, err) + + blocked, err := store.ApplyOperations().FindNextApplyOperation(ctx, "test-operator") + require.NoError(t, err) + assert.Nil(t, blocked, "a redispatch armed with a wait must not be admitted before the wait elapses") + + _, err = testDB.ExecContext(ctx, `UPDATE applies SET retry_after = NOW() - INTERVAL 1 SECOND WHERE id = ?`, apply.ID) + require.NoError(t, err) + + claimed, err := store.ApplyOperations().FindNextApplyOperation(ctx, "test-operator") + require.NoError(t, err) + require.NotNil(t, claimed, "the redispatch is admitted once its wait has elapsed") + assert.Equal(t, id, claimed.ID) +} + +// The operation redispatch arms the next wait on the parent apply, the row the +// claim clause reads it back from. Arming on admission rather than on failure +// measures the wait from the start of the attempt, so an operation that ran +// longer than its own backoff retries as soon as it fails. +func TestApplyOperationStore_FindNextApplyOperation_RedispatchArmsParentRetryBackoff(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := NewMySQL(testDB) + + lock := createTestLock(t, store, "testdb", "mysql", "staging") + apply := createTestApplyWithStateAndEnv(t, store, lock, "apply_op_backoff_arm", 1, state.Apply.FailedRetryable, "staging") + _, err := testDB.ExecContext(ctx, ` + UPDATE applies SET attempt = ?, updated_at = NOW() WHERE id = ? + `, maxRecoveryAttempts-1, apply.ID) + require.NoError(t, err) + + _, err = store.ApplyOperations().Insert(ctx, &storage.ApplyOperation{ + ApplyID: apply.ID, Deployment: "region-a", State: state.ApplyOperation.FailedRetryable, + }) + require.NoError(t, err) + // A sibling operation makes this a genuine multi-deployment redispatch, + // which is what consumes the parent budget and arms the shared wait. + _, err = store.ApplyOperations().Insert(ctx, &storage.ApplyOperation{ + ApplyID: apply.ID, Deployment: "region-b", State: state.ApplyOperation.Pending, + }) + require.NoError(t, err) + + claimed, err := store.ApplyOperations().FindNextApplyOperation(ctx, "test-operator") + require.NoError(t, err) + require.NotNil(t, claimed) + + parent, err := store.Applies().Get(ctx, apply.ID) + require.NoError(t, err) + require.NotNil(t, parent.RetryAfter, "an admitted redispatch arms the parent's wait for the next one") + assert.True(t, parent.RetryAfter.After(time.Now()), "the armed wait is in the future") + assert.Equal(t, maxRecoveryAttempts, parent.Attempt, "the redispatch consumes one unit of the parent budget") +} + // TestApplyOperationStore_FindNextApplyOperation_MultiOpRedispatchConsumesParentBudget // verifies OC-5 Part B2: an operation-only redispatch of a failed_retryable // operation in a multi-deployment apply consumes one unit of the parent apply's diff --git a/pkg/storage/types.go b/pkg/storage/types.go index 18008788f..f17391da6 100644 --- a/pkg/storage/types.go +++ b/pkg/storage/types.go @@ -18,6 +18,52 @@ import ( // the storage claim/expiry paths. const MaxRecoveryAttempts = 10 +// retryBackoffStep and retryBackoffCap shape the wait between the automatic +// redispatches of a failed_retryable apply. Without a wait the budget is spent +// on whatever failure is fastest to reproduce โ€” a refused lock or a connection +// error can burn all of MaxRecoveryAttempts inside a minute, which is the case +// where a retry was most likely to have worked given a little time. Stepping +// the wait up and then holding it flat keeps the first attempts immediate for a +// transient blip while spreading the rest of the budget over a window long +// enough to outlast a rolling restart or a brief target outage. +const ( + retryBackoffStep = 30 * time.Second + retryBackoffCap = 90 * time.Second +) + +// RetryBackoff returns how long an apply admitted on this attempt waits before +// it may be claimed again, where attempt is the value that claim consumed. The +// first admitted attempt arms no wait, so together with a first interruption +// that has no attempt to pace from, a transient blip gets two immediate retries. +// Spending the rest of the budget on waits takes roughly ten minutes of wall +// clock before MaxRecoveryAttempts terminalizes the apply as failed. +func RetryBackoff(attempt int) time.Duration { + if attempt < 1 { + return 0 + } + delay := time.Duration(attempt-1) * retryBackoffStep + if delay > retryBackoffCap { + return retryBackoffCap + } + return delay +} + +// AnnouncedRetryAttempt reports the attempt number an interrupted apply's +// operator-facing surfaces should name, and whether the budget still allows +// that attempt to happen, where attempt is the apply's stored count of +// attempts already admitted. The claim path stops admitting at +// MaxRecoveryAttempts, so past that point the budget is spent and the surfaces +// name the last attempt made rather than counting past the ceiling: an apply +// sitting in failed_retryable with a spent budget is waiting to be terminalized, +// not waiting to retry, and a comment promising one more attempt outlives the +// apply that could have made it. +func AnnouncedRetryAttempt(attempt int) (announced int, retryComing bool) { + if attempt >= MaxRecoveryAttempts { + return MaxRecoveryAttempts, false + } + return attempt + 1, true +} + // MaxWebhookEventAttempts is the claim budget for webhook inbox rows: how many // times FindNext will hand out a given delivery (each claim increments // attempts) before the row stops being claimable. It bounds the blast radius @@ -664,6 +710,14 @@ type Apply struct { // Once the retry budget is exhausted, the apply becomes failed. Attempt int + // RetryAfter is the earliest time a failed_retryable apply may be claimed + // again. It is armed when an attempt is admitted, so the wait is measured + // from the start of that attempt: a failure that took longer than its own + // backoff has already spaced itself out and retries immediately, while a + // failure that returns instantly waits. Nil means no wait โ€” the first + // interruption of an apply retries as soon as a driver notices it. + RetryAfter *time.Time + // LeaseOwner identifies the driver that last claimed this apply. It is // operator-facing context; LeaseToken is the ownership capability used for // correctness. diff --git a/pkg/storage/types_test.go b/pkg/storage/types_test.go index ad911b631..f279de9a5 100644 --- a/pkg/storage/types_test.go +++ b/pkg/storage/types_test.go @@ -3,6 +3,7 @@ package storage import ( "fmt" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -350,3 +351,29 @@ func TestVSchemaPredicates(t *testing.T) { } }) } + +// The retry backoff spends the recovery budget over a window long enough to +// outlast a rolling restart or a brief target outage, without making the first +// interruptions of a transient blip feel stalled: the first attempts are +// immediate, the wait then steps up and holds flat at the cap. +func TestRetryBackoff(t *testing.T) { + // A first interruption has no attempt to compute a wait from and retries at + // once; the attempt it consumes arms no wait either, so a transient blip + // gets two immediate retries before the pacing starts. + assert.Zero(t, RetryBackoff(1)) + assert.Equal(t, 30*time.Second, RetryBackoff(2)) + assert.Equal(t, 60*time.Second, RetryBackoff(3)) + assert.Equal(t, 90*time.Second, RetryBackoff(4), "the wait holds at the cap") + assert.Equal(t, 90*time.Second, RetryBackoff(MaxRecoveryAttempts)) + + // An instantly-reproducing failure spends the whole budget on waits, which + // must add up to minutes of wall clock rather than the seconds the same + // failure would otherwise burn it in. The wait armed by the last attempt is + // never consumed โ€” the budget is already spent โ€” so the window is the sum of + // the waits before it. + var window time.Duration + for attempt := 1; attempt < MaxRecoveryAttempts; attempt++ { + window += RetryBackoff(attempt) + } + assert.Equal(t, 10*time.Minute+30*time.Second, window) +} diff --git a/pkg/ui/progressbar.go b/pkg/ui/progressbar.go index fa02771ce..654c0c2f3 100644 --- a/pkg/ui/progressbar.go +++ b/pkg/ui/progressbar.go @@ -6,9 +6,9 @@ import "strings" // Progress bar colors (emoji). const ( ColorBlue = "๐ŸŸฆ" // In progress (copying rows) - ColorYellow = "๐ŸŸจ" // Waiting for cutover + ColorYellow = "๐ŸŸจ" // Healthy wait (waiting for cutover, revert window open) ColorGreen = "๐ŸŸฉ" // Complete - ColorOrange = "๐ŸŸง" // Operator-halted (stopped, cancelled, reverted) + ColorOrange = "๐ŸŸง" // Halted (stopped, cancelled, reverted, retrying after an interruption) ColorRed = "๐ŸŸฅ" // Failed ColorEmpty = "โฌœ" // Unfilled ) diff --git a/pkg/webhook/apply.go b/pkg/webhook/apply.go index 0eddfaaf7..fd198f40c 100644 --- a/pkg/webhook/apply.go +++ b/pkg/webhook/apply.go @@ -38,6 +38,9 @@ func buildApplyCommentData(apply *storage.Apply, tasks []*storage.Task, display Rollback: apply.IsRollback(), DeferCutover: apply.GetOptions().DeferCutover, } + if apply.RetryAfter != nil { + data.RetryAfter = apply.RetryAfter.Format(time.RFC3339) + } if apply.StartedAt != nil { data.StartedAt = apply.StartedAt.Format(time.RFC3339) } diff --git a/pkg/webhook/multi_apply.go b/pkg/webhook/multi_apply.go index 76e15f406..e37c43da3 100644 --- a/pkg/webhook/multi_apply.go +++ b/pkg/webhook/multi_apply.go @@ -5,6 +5,7 @@ import ( "time" "github.com/block/schemabot/pkg/presentation" + "github.com/block/schemabot/pkg/state" "github.com/block/schemabot/pkg/storage" "github.com/block/schemabot/pkg/webhook/templates" ) @@ -168,6 +169,15 @@ func buildDeploymentDetail(apply *storage.Apply, op *storage.ApplyOperation, tas Rollback: apply.IsRollback(), DeferCutover: apply.GetOptions().DeferCutover, } + // The armed wait belongs to the apply, not to one deployment of it: an + // operation is re-leased for redispatch only while the parent apply is + // itself failed_retryable. Carrying the deadline onto a deployment row of a + // still-active parent would name a time nothing is going to act on, so a + // failed deployment of an otherwise-running rollout reports its state + // without promising a retry at a particular moment. + if apply.RetryAfter != nil && state.IsState(apply.State, state.Apply.FailedRetryable) { + data.RetryAfter = apply.RetryAfter.Format(time.RFC3339) + } if apply.StartedAt != nil { data.StartedAt = apply.StartedAt.Format(time.RFC3339) } diff --git a/pkg/webhook/templates/apply.go b/pkg/webhook/templates/apply.go index 5a60ec258..ab7139893 100644 --- a/pkg/webhook/templates/apply.go +++ b/pkg/webhook/templates/apply.go @@ -82,9 +82,19 @@ type ApplyStatusCommentData struct { // from the derived model to agree with its line. DerivedStatus string - // Attempt is the apply's operator redispatch count so far; the retry the - // comment announces is Attempt+1 of storage.MaxRecoveryAttempts. - Attempt int + // Attempt is the apply's operator redispatch count so far; the attempt the + // comment announces follows from it via storage.AnnouncedRetryAttempt, + // which stops counting once the budget is spent. + Attempt int + + // RetryAfter is the RFC3339 time the next automatic retry becomes eligible. + // An interrupted apply backs off between attempts, and the comment is not + // re-rendered while it waits, so the wait is shown as a clock time rather + // than a countdown that would freeze at whatever it read when posted. Empty + // when no wait is in force and the retry is due as soon as a driver picks + // the apply up. + RetryAfter string + StartedAt string // RFC3339 format CompletedAt string // RFC3339 format Tables []TableProgressData @@ -411,6 +421,26 @@ func revertWindowCountdown(revertExpiresAt string) string { return fmt.Sprintf("Closes in %s", formatDuration(remaining)) } +// nextRetrySegment renders the clock time an interrupted apply's next attempt +// becomes eligible, as a trailing segment of the table's retry line. It renders +// nothing when no wait is in force or the wait has already elapsed: the retry is +// then due now, and naming a time in the past would read as a missed deadline +// rather than as work about to be picked up. +func nextRetrySegment(retryAfter string) string { + if retryAfter == "" { + return "" + } + due, err := time.Parse(time.RFC3339, retryAfter) + if err != nil { + return "" + } + // NowFunc (not time.Now) so previews and tests render deterministically. + if !due.After(NowFunc()) { + return "" + } + return " ยท next " + due.UTC().Format("15:04 UTC") +} + // writeCutoverSummary writes a readiness summary for cutover states, // showing how many tables are ready for cutover vs not yet ready. func writeCutoverSummary(sb *strings.Builder, tables []TableProgressData) { @@ -623,7 +653,7 @@ func writeTableProgressSection(sb *strings.Builder, data ApplyStatusCommentData) renderResumingTable(sb, table) continue } - renderTableProgress(sb, table, data.Attempt, data.ErrorMessage) + renderTableProgress(sb, table, applyRetryFor(data), data.ErrorMessage) } } } @@ -683,12 +713,31 @@ func tableStatePriority(tableStatus string) int { return ui.TableStatePriority(state.NormalizeTaskStatus(tableStatus)) } +// applyRetry is the apply-level retry state an interrupted table row reports: +// how much of the retry budget is spent, and when the next attempt is due. +type applyRetry struct { + attempt int + retryAfter string // RFC3339 format +} + +// applyRetryFor reports the retry state an interrupted table row may show. The +// wait is carried only while the apply is retrying: a claim leaves the armed +// deadline behind on the row it resumes, so once the apply is moving again the +// stored time names a retry that is not coming. +func applyRetryFor(data ApplyStatusCommentData) applyRetry { + retry := applyRetry{attempt: data.Attempt} + if state.IsState(data.State, state.Apply.FailedRetryable) { + retry.retryAfter = data.RetryAfter + } + return retry +} + // renderTableProgress renders a single table's progress as markdown. // Mirrors the CLI's writeTableProgressWithState logic but outputs markdown // instead of ANSI. applyError is the apply-level error message the comment // renders as its own block, so a failed table's identical error is not // repeated below the row. -func renderTableProgress(sb *strings.Builder, table TableProgressData, applyAttempt int, applyError string) { +func renderTableProgress(sb *strings.Builder, table TableProgressData, retry applyRetry, applyError string) { // Normalize to canonical Task state for consistent matching. status := state.NormalizeTaskStatus(table.Status) @@ -780,9 +829,14 @@ func renderTableProgress(sb *strings.Builder, table TableProgressData, applyAtte } case state.Task.FailedRetryable: - bar := ui.ProgressBarStopped(ui.RowCopyDisplayPercent(table.PercentComplete, table.RowsCopied)) - fmt.Fprintf(sb, "**`%s`**: %s \U0001f504 Interrupted โ€” retrying automatically (attempt %d/%d)\n", - table.TableName, bar, applyAttempt+1, storage.MaxRecoveryAttempts) + bar := ui.ProgressBar(ui.RowCopyDisplayPercent(table.PercentComplete, table.RowsCopied), ui.ColorOrange) + announced, retryComing := storage.AnnouncedRetryAttempt(retry.attempt) + next := "" + if retryComing { + next = nextRetrySegment(retry.retryAfter) + } + fmt.Fprintf(sb, "**`%s`**: %s \U0001f504 Retrying ยท attempt %d/%d%s\n", + table.TableName, bar, announced, storage.MaxRecoveryAttempts, next) writeDDLLine(sb, table.DDL) if table.ErrorMessage != "" { writeTableErrorLine(sb, table.ErrorMessage) diff --git a/pkg/webhook/templates/apply_test.go b/pkg/webhook/templates/apply_test.go index de3cc07b8..59d8b0aa7 100644 --- a/pkg/webhook/templates/apply_test.go +++ b/pkg/webhook/templates/apply_test.go @@ -10,6 +10,7 @@ import ( "github.com/block/schemabot/pkg/apitypes" "github.com/block/schemabot/pkg/state" + "github.com/block/schemabot/pkg/storage" "github.com/block/schemabot/pkg/ui" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -1094,7 +1095,7 @@ func TestRenderApplyStatusComment_FailedRetryable(t *testing.T) { assert.Contains(t, result, "**Status**: Retrying") // The retry detail lives on the affected table, not in the headline, and // counts the upcoming retry against the operator redispatch budget. - assert.Contains(t, result, "๐Ÿ”„ Interrupted โ€” retrying automatically (attempt 1/10)") + assert.Contains(t, result, "๐Ÿ”„ Retrying ยท attempt 1/10") assert.Contains(t, result, "> โš ๏ธ Last error: remote deployment unavailable") assert.Contains(t, result, "๐ŸŸง") // orange bar for the interrupted table // Progress summary counts the retrying table. @@ -1131,7 +1132,58 @@ func TestRenderApplyStatusComment_FailedRetryableCountsAttempts(t *testing.T) { result := RenderApplyStatusComment(data) - assert.Contains(t, result, "๐Ÿ”„ Interrupted โ€” retrying automatically (attempt 5/10)") + assert.Contains(t, result, "๐Ÿ”„ Retrying ยท attempt 5/10") +} + +// An interrupted apply backs off between attempts, and the comment is not +// re-rendered while it waits. The retry line names the clock time the next +// attempt becomes eligible so a watcher can tell a waiting apply from a stalled +// one, and drops that segment once the wait has elapsed and the retry is due. +func TestRenderApplyStatusComment_FailedRetryableNextRetry(t *testing.T) { + now := time.Date(2026, 8, 15, 14, 30, 0, 0, time.UTC) + original := NowFunc + t.Cleanup(func() { NowFunc = original }) + NowFunc = func() time.Time { return now } + + data := ApplyStatusCommentData{ + Database: "testapp", + Environment: "staging", + State: state.Apply.FailedRetryable, + ApplyID: "apply-abc123", + Attempt: 2, + Tables: []TableProgressData{ + {TableName: "users", DDL: "ALTER TABLE `users` ADD COLUMN `email` varchar(255)", Status: state.Task.FailedRetryable}, + }, + } + + data.RetryAfter = now.Add(2 * time.Minute).Format(time.RFC3339) + assert.Contains(t, RenderApplyStatusComment(data), "๐Ÿ”„ Retrying ยท attempt 3/10 ยท next 14:32 UTC") + + data.RetryAfter = now.Add(-time.Minute).Format(time.RFC3339) + due := RenderApplyStatusComment(data) + assert.Contains(t, due, "๐Ÿ”„ Retrying ยท attempt 3/10\n") + assert.NotContains(t, due, "next ") + + data.RetryAfter = "" + assert.Contains(t, RenderApplyStatusComment(data), "๐Ÿ”„ Retrying ยท attempt 3/10\n") + + // A claim leaves the armed deadline behind on the row it resumes, so once + // the apply is moving again the stored time must not be named. + data.RetryAfter = now.Add(2 * time.Minute).Format(time.RFC3339) + data.State = state.Apply.Running + resumed := RenderApplyStatusComment(data) + assert.Contains(t, resumed, "๐Ÿ”„ Retrying ยท attempt 3/10\n") + assert.NotContains(t, resumed, "next ") + + // The claim path stops admitting attempts at the budget ceiling, so the + // last attempt names itself rather than a further one, and the deadline it + // leaves behind promises nothing. + data.State = state.Apply.FailedRetryable + data.Attempt = storage.MaxRecoveryAttempts + spent := RenderApplyStatusComment(data) + assert.Contains(t, spent, "๐Ÿ”„ Retrying ยท attempt 10/10\n") + assert.NotContains(t, spent, "attempt 11/10") + assert.NotContains(t, spent, "next ") } // Every apply state must render a human-readable headline. Raw snake_case @@ -1165,7 +1217,7 @@ func TestRenderApplyStatusComment_FailedRetryableUppercaseStatus(t *testing.T) { result := RenderApplyStatusComment(data) - assert.Contains(t, result, "๐Ÿ”„ Interrupted โ€” retrying automatically") + assert.Contains(t, result, "๐Ÿ”„ Retrying ยท attempt") assert.NotContains(t, result, "Running...") } diff --git a/pkg/webhook/templates/preview.go b/pkg/webhook/templates/preview.go index 182dd91f7..3bb700fa2 100644 --- a/pkg/webhook/templates/preview.go +++ b/pkg/webhook/templates/preview.go @@ -7,6 +7,7 @@ import ( "github.com/block/schemabot/pkg/apitypes" "github.com/block/schemabot/pkg/presentation" "github.com/block/schemabot/pkg/state" + "github.com/block/schemabot/pkg/storage" "github.com/block/schemabot/pkg/webhook/action" ) @@ -1414,7 +1415,8 @@ func PreviewCommentApplyFailedBeforeRowCopy() string { // PreviewCommentApplyRetrying renders an apply comment where the middle table // was interrupted by a retryable failure and the driver is redispatching it, -// with the attempt counter showing how much of the retry budget is used. +// with the attempt counter showing how much of the retry budget is used and the +// clock time the next attempt becomes eligible. func PreviewCommentApplyRetrying() string { tables := sampleApplyTables() tables[0].Status = state.Task.Completed @@ -1425,7 +1427,8 @@ func PreviewCommentApplyRetrying() string { tables[1].ErrorMessage = PreviewErrorMiddleFailed tables[2].Status = state.Task.Pending data := sampleApplyData(state.Apply.FailedRetryable, tables) - data.Attempt = 1 + data.Attempt = 3 + data.RetryAfter = NowFunc().Add(storage.RetryBackoff(data.Attempt)).UTC().Format(time.RFC3339) return RenderApplyStatusComment(data) }