diff --git a/doc/rfc/submitqueue/speculation.md b/doc/rfc/submitqueue/speculation.md index ceeabc118..19a118232 100644 --- a/doc/rfc/submitqueue/speculation.md +++ b/doc/rfc/submitqueue/speculation.md @@ -55,7 +55,7 @@ Every write is a compare-and-swap: a writer that loses re-reads on a later run. Verdicts are controller-owned facts: the Speculator can neither compute nor veto them. - **Land.** Each path carries an assumption about every dependency — *succeeds* (built on top of) or *fails* (built without). Normally, once a path's build has passed and every dependency has finished the way the path assumed — one assumed *succeeds* has landed, one assumed *fails* has failed or been cancelled — the speculate controller moves the head to Landing and hands it to Runway. A dependency that is merely *landing* has not finished, because a land can fail, so a single matching path still waits for the answer. Complete passed coverage is the exception described in Bypass large diff: it lets a head land before those answers arrive. If the hand-off is lost, the next run re-sends it. The same run sets the head's remaining in-flight paths *cancelling*: once the head can land they cannot help, and they hold CI slots until they stop. The landsignal controller records Runway's terminal result: success marks the head Succeeded, while failure marks it Failed. The result publishes a single dirty signal — no per-dependent fan-out — and the next run refutes paths whose assumption disagrees with the result: *fails* assumptions after success, *succeeds* assumptions after failure. The hand-off is idempotent, so Runway reports success without another land when the change is already present. A chain ordinarily lands one at a time, but a fully covered head can bypass its unsettled predecessors. -- **Failure (no viable path).** A batch fails when every possible future has a failed build — no path can pass, so it can never land. +- **Failure (no viable path).** A Speculating batch fails only after every dependency is terminal, at least one stored path remains unbroken, and every unbroken path is Failed or Cancelled. Cancellation consumes the execution attempt without rewriting the observed Cancelled fact as a test failure. Unfunded futures and unresolved dependencies still wait; a passed alternative still wins. Terminal paths cannot be retried automatically. - **Cancel.** A cancelled batch is driven terminal: its in-flight paths are set *cancelling*, then the batch is marked Cancelled once they stop (see Cancellation). ### Conflict relaxation @@ -84,7 +84,7 @@ Cancellation is best-effort: a batch marked *cancelling* may still land if a lan Two kinds of cancel, split by owner: -- **Preemption (extension).** The Speculator may propose `Cancel` on an in-flight path whose head is Speculating to free budget for a better candidate — its only cancel power. The controller validates it (never a passed path) and enacts it. +- **Preemption (extension).** The Speculator may propose `Cancel` on an in-flight path whose head is Speculating to free budget for a better candidate — its only cancel power. The controller validates it (never a passed path) and enacts it. A viable path cancelled by a custom preempting Speculator consumes that attempt and can exhaust the head once dependencies settle; the shipped sticky allocator does not preempt. - **Correctness cancels (controller).** Refutation — a resolved dependency breaks a path's assumption, so the controller cancels that path — and batch cancellation (below). The Speculator is not consulted. Batch cancellation (user-initiated) marks the batch *cancelling*: a non-terminal intent that halts new work. The speculate run drives it terminal — it sets its in-flight paths *cancelling*, marks the batch *cancelled* once they stop, recomputes dependent paths against that outcome (those that assume it fails keep building; those that assume it succeeds are refuted), and concludes the contained requests. @@ -121,3 +121,9 @@ Signatures live in code and are not copied here, so they cannot drift. This sect **Scorer** — [`submitqueue/extension/speculation/scorer`](../../../submitqueue/extension/speculation/scorer/README.md). `Score(ctx, batch, paths)` is how likely the batch is to reach Succeeded. The default impl is evidence wrapping a base (heuristic or composite): the base prices the change and ignores `paths`; evidence revises that price from the path set and batch state. See [outcome-scorer.md](outcome-scorer.md). **Generator and Allocator** — [`generator`](../../../submitqueue/extension/speculation/generator/README.md) and [`allocator`](../../../submitqueue/extension/speculation/allocator/README.md), the two composition points inside the default Speculator. The Generator opens a pull-based stream of candidate paths over the batches and path sets; the Allocator spends the build budget over that stream, reconciling it against the path sets. Both abort on a cancelled context. + +## Recovering quiet queues with cancelled paths + +After deploying the exhaustion policy and draining older speculation consumers, run `./tool/bazel run //tool/submitqueue/recovercancelledpaths -- --queue --recovery-id --dry-run` with the existing `MYSQL_DSN` configuration. Repeat `--queue` to explicitly bound the selection. The command reports the numerically earliest affected Speculating head per queue; a missing path set is unfunded, not affected. Dry-run is the default and never opens the queue database. + +Review the reported IDs, then repeat with `--apply` instead of `--dry-run` and configure `QUEUE_MYSQL_DSN`. Applying publishes one ordinary speculate nudge per affected queue; it never edits outcomes or invokes the planner directly. Retry an interrupted invocation with the same recovery ID so successful nudges deduplicate; use a new ID for a deliberately new recovery. Any inspection or publication error exits nonzero, while other selected queues are still attempted. Verify the ordinary conclude/request pipeline reaches terminal status. Do not apply against older consumers: they can acknowledge the wake-up without resolving the exhausted head. Rolling back the policy can reintroduce the stall but must not reverse terminal outcomes or replay submissions. diff --git a/submitqueue/core/messagequeue/BUILD.bazel b/submitqueue/core/messagequeue/BUILD.bazel index 7dc3faf37..9d7066c99 100644 --- a/submitqueue/core/messagequeue/BUILD.bazel +++ b/submitqueue/core/messagequeue/BUILD.bazel @@ -13,6 +13,7 @@ go_library( "//service/submitqueue:__subpackages__", "//submitqueue:__subpackages__", "//test:__subpackages__", + "//tool/submitqueue/recovercancelledpaths:__pkg__", ], deps = [ "//api/base/change/protopb:go_default_library", diff --git a/submitqueue/entity/speculation.go b/submitqueue/entity/speculation.go index 734dfe1f6..df592f51f 100644 --- a/submitqueue/entity/speculation.go +++ b/submitqueue/entity/speculation.go @@ -151,10 +151,8 @@ type SpeculationPathEntry struct { Path SpeculationPath // Status is the lifecycle status of the current build attempt. Status SpeculationPathStatus - // Attempt is the build attempt number for this path, starting at 1. A path - // can be built more than once (e.g. after a prior build is cancelled to free - // budget, or fails and is retried), so ID alone does not identify an - // execution — (ID, Attempt) does. It increments with each new build. + // Attempt is the build attempt number for this path, starting at 1. + // The pair (ID, Attempt) uniquely identifies an execution. Attempt int // Version is the version of the object. It is used for optimistic locking. // Versioning starts at 1 and is incremented for each change to the object. @@ -192,7 +190,7 @@ type PathAction string const ( // PathActionUnknown is the zero-value sentinel; it is never a valid action. PathActionUnknown PathAction = "" - // PathActionBuild proposes starting (or resurrecting) a build for the path. + // PathActionBuild proposes starting a build for the path. PathActionBuild PathAction = "build" // PathActionCancel proposes preempting an in-flight path to free build budget. PathActionCancel PathAction = "cancel" diff --git a/submitqueue/orchestrator/controller/speculate/check_test.go b/submitqueue/orchestrator/controller/speculate/check_test.go index 3700ff56a..cc3e947c7 100644 --- a/submitqueue/orchestrator/controller/speculate/check_test.go +++ b/submitqueue/orchestrator/controller/speculate/check_test.go @@ -134,6 +134,13 @@ func TestFilterProposals_Rejects(t *testing.T) { entryFor(valid, entity.SpeculationPathStatusPassed)), want: rejectCancelPassed, }, + { + name: "build would resurrect a cancelled path", + proposal: entity.Speculation{Path: valid, Action: entity.PathActionBuild}, + snap: checkSnapshot(entity.BatchStateSpeculating, + entryFor(valid, entity.SpeculationPathStatusCancelled)), + want: rejectPathTerminal, + }, { name: "build would resurrect a finished path", proposal: entity.Speculation{Path: valid, Action: entity.PathActionBuild}, diff --git a/submitqueue/orchestrator/controller/speculate/doc.go b/submitqueue/orchestrator/controller/speculate/doc.go index 79ec04ee7..3a70d9fe0 100644 --- a/submitqueue/orchestrator/controller/speculate/doc.go +++ b/submitqueue/orchestrator/controller/speculate/doc.go @@ -54,7 +54,7 @@ // cancelled to free the slot. // - A fails and P2 passed: B lands without A, again with no new build. // P1's guess is broken. -// - A resolved either way, and every unbroken path failed: no future +// - A resolved either way, and every unbroken path failed or was cancelled: no future // remains in which B passes, so B fails. // // # The life of a path @@ -85,9 +85,9 @@ // stops it, and only an observation of that stop (or proof nothing was ever // dispatched) moves the path to cancelled. The intent needs no dispatch of its // own — the poll loop reads it off the set and asks the runner to stop the -// build. A terminal path can be resurrected by a new build proposal — status -// returns to pending and Attempt increments, the one backwards step in the -// diagram. +// build. Terminal paths cannot be rebuilt. Failed and cancelled attempts are +// exhausted once dependencies settle; preempting a viable path consumes its +// attempt too. The sticky allocator does not preempt. // // # The life of a batch, as seen from here // diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go index 331bde6d6..777f9d5d5 100644 --- a/submitqueue/orchestrator/controller/speculate/finalize.go +++ b/submitqueue/orchestrator/controller/speculate/finalize.go @@ -365,7 +365,7 @@ func (c *Controller) applyOutcome(ctx context.Context, store orchstorage.Storage // Cancellation carries none — the cancel path owns that reason. var concludeMeta map[string]string if decision == outcomeFail { - concludeMeta = map[string]string{topickey.MetadataKeyFailureReason: "no speculation path could pass; every candidate build failed"} + concludeMeta = map[string]string{topickey.MetadataKeyFailureReason: "no speculation path could pass; every viable build failed or was cancelled"} } // Named for the run that decided it, so a redelivery re-deriving the // same outcome does not conclude the batch twice, and so it stays diff --git a/submitqueue/orchestrator/controller/speculate/outcome.go b/submitqueue/orchestrator/controller/speculate/outcome.go index c7a64a525..ee31a1dcc 100644 --- a/submitqueue/orchestrator/controller/speculate/outcome.go +++ b/submitqueue/orchestrator/controller/speculate/outcome.go @@ -154,8 +154,8 @@ func allAssumptionsSettled(path entity.SpeculationPath, snap snapshot) bool { } // hasNoViableFuture reports whether the head can never pass: every dependency -// has resolved, and every path consistent with how they resolved has a failed -// build. +// has resolved, and every path consistent with how they resolved has exhausted +// its attempt with a failed or cancelled build. // // Deliberately conservative — failing a batch that could still have landed is // far worse than failing it a tick later, so this errs toward waiting: @@ -178,7 +178,7 @@ func hasNoViableFuture(head entity.Batch, set entity.SpeculationPathSet, snap sn continue } live++ - if entry.Status != entity.SpeculationPathStatusFailed { + if entry.Status != entity.SpeculationPathStatusFailed && entry.Status != entity.SpeculationPathStatusCancelled { return false } } diff --git a/submitqueue/orchestrator/controller/speculate/outcome_test.go b/submitqueue/orchestrator/controller/speculate/outcome_test.go index e14dced6e..72df64518 100644 --- a/submitqueue/orchestrator/controller/speculate/outcome_test.go +++ b/submitqueue/orchestrator/controller/speculate/outcome_test.go @@ -352,41 +352,38 @@ func TestHasNoViableFuture(t *testing.T) { pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionSucceeds), entity.SpeculationPathStatusFailed) - t.Run("waits while a dependency is unresolved", func(t *testing.T) { - // An unresolved dependency means futures the queue has not tried yet. - snap := snapWith(entity.BatchStateSucceeded, entity.BatchStateSpeculating) - assert.False(t, hasNoViableFuture(headBatch, setOf(failed), snap)) - }) - - t.Run("fails once everything is resolved and every live path failed", func(t *testing.T) { - snap := snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded) - assert.True(t, hasNoViableFuture(headBatch, setOf(failed), snap)) - }) - - t.Run("does not fail while a path is still running", func(t *testing.T) { - running := entryFor( - pathOver(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionSucceeds), - entity.SpeculationPathStatusBuilding) - running.ID = "still-running" - snap := snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded) - assert.False(t, hasNoViableFuture(headBatch, setOf(failed, running), snap)) - }) - - t.Run("does not fail a head with nothing funded", func(t *testing.T) { - snap := snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded) - assert.False(t, hasNoViableFuture(headBatch, setOf(), snap), - "an unfunded head is waiting for the Speculator, not out of options") - }) - - t.Run("ignores broken paths when judging", func(t *testing.T) { - // This path assumed dep1 would fail; it succeeded, so the failed build - // tells us nothing about a future that can still happen. - brokenFail := entryFor( - pathOver(entity.DependencyAssumptionFails, entity.DependencyAssumptionSucceeds), - entity.SpeculationPathStatusFailed) - snap := snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded) - assert.False(t, hasNoViableFuture(headBatch, setOf(brokenFail), snap)) - }) + cancelled := failed + cancelled.Status = entity.SpeculationPathStatusCancelled + settled := snapWith(entity.BatchStateSucceeded, entity.BatchStateSucceeded) + for _, status := range []entity.SpeculationPathStatus{ + entity.SpeculationPathStatusFailed, entity.SpeculationPathStatusCancelled, + } { + t.Run(string(status), func(t *testing.T) { + entry := failed + entry.Status = status + assert.True(t, hasNoViableFuture(headBatch, setOf(entry), settled)) + assert.False(t, hasNoViableFuture(headBatch, setOf(entry), + snapWith(entity.BatchStateSucceeded, entity.BatchStateSpeculating))) + assert.True(t, hasNoViableFuture(entity.Batch{ID: head}, + setOf(entryFor(entity.SpeculationPath{Head: head}, status)), snapshot{})) + broken := entry + broken.Path = pathOver(entity.DependencyAssumptionFails, entity.DependencyAssumptionSucceeds) + assert.False(t, hasNoViableFuture(headBatch, setOf(broken), settled)) + }) + } + assert.True(t, hasNoViableFuture(headBatch, setOf(failed, cancelled), settled)) + assert.False(t, hasNoViableFuture(headBatch, setOf(), settled)) + for _, status := range []entity.SpeculationPathStatus{ + entity.SpeculationPathStatusUnknown, entity.SpeculationPathStatusPending, + entity.SpeculationPathStatusBuilding, entity.SpeculationPathStatusCancelling, + entity.SpeculationPathStatusPassed, + } { + t.Run("not exhausted "+string(status), func(t *testing.T) { + entry := failed + entry.Status = status + assert.False(t, hasNoViableFuture(headBatch, setOf(cancelled, entry), settled)) + }) + } } func TestDecide(t *testing.T) { @@ -406,6 +403,10 @@ func TestDecide(t *testing.T) { assert.Equal(t, outcomeLand, decide(headBatch, setOf(passed), allResolved)) assert.Equal(t, outcomeFail, decide(headBatch, setOf(failed), allResolved)) + cancelled := failed + cancelled.Status = entity.SpeculationPathStatusCancelled + assert.Equal(t, outcomeFail, decide(headBatch, setOf(cancelled), allResolved)) + assert.Equal(t, outcomeLand, decide(headBatch, setOf(cancelled, passed), allResolved)) assert.Equal(t, outcomeWait, decide(headBatch, setOf(building), allResolved)) // A passed path wins over a failed sibling: one way through is enough. diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go index 009181751..9288a545d 100644 --- a/submitqueue/orchestrator/controller/speculate/run_test.go +++ b/submitqueue/orchestrator/controller/speculate/run_test.go @@ -792,10 +792,41 @@ func TestRun_FailedHeadConcludesAfterTheStateWrite(t *testing.T) { // The failure reason rides the conclude message so conclude can stamp it on // the failed requests' terminal log, without persisting it as batch state. require.Len(t, h.messages, 1) - assert.Equal(t, "no speculation path could pass; every candidate build failed", + assert.Equal(t, "no speculation path could pass; every viable build failed or was cancelled", h.messages[0].Metadata[topickey.MetadataKeyFailureReason]) } +func TestRun_ProviderCancellationExhaustsHead(t *testing.T) { + ctrl := gomock.NewController(t) + batch := entity.Batch{ID: head, Queue: "q", State: entity.BatchStateSpeculating, Version: 1} + h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{batch}) + path := entity.SpeculationPath{Head: head} + h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{ + Head: head, Paths: []entity.SpeculationPathEntry{entryFor(path, entity.SpeculationPathStatusBuilding)}, Version: 1, + }, nil) + h.pathBuilds.EXPECT().Get(gomock.Any(), path.ID(), 1).Return(entity.PathBuild{BuildID: "b1"}, nil) + h.builds.EXPECT().Get(gomock.Any(), "b1").Return(entity.Build{ID: "b1", Status: entity.BuildStatusCancelled}, nil) + gomock.InOrder( + h.pathSets.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).DoAndReturn( + func(_ context.Context, set entity.SpeculationPathSet, _, _ int32) error { + assert.Equal(t, entity.SpeculationPathStatusCancelled, set.Paths[0].Status) + assert.Equal(t, 1, set.Paths[0].Attempt) + assert.Empty(t, h.published) + return nil + }), + h.batches.EXPECT().Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateFailed}, int32(1), int32(2)).DoAndReturn( + func(context.Context, entity.Batch, int32, int32) error { + assert.Empty(t, h.published) + return nil + }), + ) + + require.NoError(t, h.run(head)) + assert.Equal(t, []string{"conclude"}, h.published) + assert.Equal(t, "no speculation path could pass; every viable build failed or was cancelled", h.messages[0].Metadata[topickey.MetadataKeyFailureReason]) + assert.Zero(t, h.spec.calls) +} + // A failure resolves a dependency, which can fail everything stacked on it. The // whole cascade finalizes inside one run: a single pass would only catch the // dependents that happen to come later in queue order, and the rest would wait diff --git a/tool/submitqueue/recovercancelledpaths/BUILD.bazel b/tool/submitqueue/recovercancelledpaths/BUILD.bazel new file mode 100644 index 000000000..ace262f5e --- /dev/null +++ b/tool/submitqueue/recovercancelledpaths/BUILD.bazel @@ -0,0 +1,48 @@ +load("@rules_go//go:def.bzl", "go_binary", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["main.go"], + importpath = "github.com/uber/submitqueue/tool/submitqueue/recovercancelledpaths", + visibility = ["//visibility:private"], + deps = [ + "//platform/base/id:go_default_library", + "//platform/consumer:go_default_library", + "//platform/extension/messagequeue/mysql:go_default_library", + "//platform/publish:go_default_library", + "//submitqueue/core/messagequeue:go_default_library", + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/storage:go_default_library", + "//submitqueue/orchestrator/core/batch:go_default_library", + "//submitqueue/orchestrator/extension/storage:go_default_library", + "//submitqueue/orchestrator/extension/storage/mysql:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_binary( + name = "recovercancelledpaths", + embed = [":go_default_library"], + visibility = ["//visibility:public"], +) + +go_test( + name = "go_default_test", + srcs = ["main_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", + "//submitqueue/core/messagequeue:go_default_library", + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/storage:go_default_library", + "//submitqueue/extension/storage/mock:go_default_library", + "//submitqueue/orchestrator/extension/storage:go_default_library", + "//submitqueue/orchestrator/extension/storage/mock:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/tool/submitqueue/recovercancelledpaths/main.go b/tool/submitqueue/recovercancelledpaths/main.go new file mode 100644 index 000000000..288891d49 --- /dev/null +++ b/tool/submitqueue/recovercancelledpaths/main.go @@ -0,0 +1,229 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Command recovercancelledpaths nudges explicitly selected queues through the +// ordinary speculation consumer without changing any stored outcome. +package main + +import ( + "context" + "database/sql" + "errors" + "flag" + "fmt" + "io" + "os" + "strings" + + "github.com/uber-go/tally" + resourceid "github.com/uber/submitqueue/platform/base/id" + "github.com/uber/submitqueue/platform/consumer" + queuemysql "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + "github.com/uber/submitqueue/platform/publish" + sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + corebatch "github.com/uber/submitqueue/submitqueue/orchestrator/core/batch" + orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage" + mysqlstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mysql" + "go.uber.org/zap" +) + +const recoveryCause = "cancelled-path-recovery" + +type options struct { + queues []string + recoveryID string + apply bool +} + +func parseOptions(args []string, output io.Writer) (options, error) { + var opts options + var dryRun bool + flags := flag.NewFlagSet("recovercancelledpaths", flag.ContinueOnError) + flags.SetOutput(output) + flags.Func("queue", "queue to inspect (required, repeatable)", func(queue string) error { + if queue == "" || strings.TrimSpace(queue) != queue { + return fmt.Errorf("queue must be non-empty without surrounding whitespace") + } + for _, existing := range opts.queues { + if existing == queue { + return fmt.Errorf("duplicate queue %q", queue) + } + } + opts.queues = append(opts.queues, queue) + return nil + }) + flags.StringVar(&opts.recoveryID, "recovery-id", "", "stable operator invocation identity (required with --apply)") + flags.BoolVar(&opts.apply, "apply", false, "publish recovery nudges") + flags.BoolVar(&dryRun, "dry-run", false, "inspect only (the default)") + if err := flags.Parse(args); err != nil { + return options{}, err + } + if len(opts.queues) == 0 || flags.NArg() != 0 { + return options{}, fmt.Errorf("explicit --queue selections and no positional arguments are required") + } + if opts.apply && dryRun { + return options{}, fmt.Errorf("--apply conflicts with --dry-run") + } + if opts.apply && strings.TrimSpace(opts.recoveryID) == "" { + return options{}, fmt.Errorf("--recovery-id is required with --apply") + } + return opts, nil +} + +func main() { + if err := run(context.Background(), os.Args[1:], os.Stdout); err != nil { + fmt.Fprintln(os.Stderr, "Error:", err) + os.Exit(1) + } +} + +func run(ctx context.Context, args []string, output io.Writer) error { + opts, err := parseOptions(args, output) + if err != nil { + return err + } + appDB, err := openDatabase("MYSQL_DSN") + if err != nil { + return err + } + defer appDB.Close() + backend, err := mysqlstorage.NewStorage(appDB, tally.NoopScope) + if err != nil { + return err + } + + var registry consumer.TopicRegistry + if opts.apply { + queueDB, err := openDatabase("QUEUE_MYSQL_DSN") + if err != nil { + return err + } + defer queueDB.Close() + queue, err := queuemysql.NewQueue(queuemysql.Params{DB: queueDB, Logger: zap.NewNop(), MetricsScope: tally.NoopScope}) + if err != nil { + return err + } + defer queue.Close() + registry, err = consumer.NewTopicRegistry([]consumer.TopicConfig{{Key: sqmq.TopicKeySpeculate, Name: "speculate", Queue: queue}}) + if err != nil { + return err + } + } + return publishCancelledPathRecovery(ctx, opts, backend.For, registry, output) +} + +func openDatabase(env string) (*sql.DB, error) { + dsn := os.Getenv(env) + if dsn == "" { + return nil, fmt.Errorf("%s environment variable is required", env) + } + db, err := sql.Open("mysql", dsn) + if err != nil { + return nil, fmt.Errorf("invalid %s configuration", env) + } + return db, nil +} + +func publishCancelledPathRecovery(ctx context.Context, opts options, storeFor func(string) (orchstorage.Storage, error), registry consumer.TopicRegistry, output io.Writer) error { + var failures []error + for _, queue := range opts.queues { + if err := recoverQueue(ctx, opts, queue, storeFor, registry, output); err != nil { + failures = append(failures, fmt.Errorf("queue %q: %w", queue, err)) + } + } + return errors.Join(failures...) +} + +func recoverQueue(ctx context.Context, opts options, queue string, storeFor func(string) (orchstorage.Storage, error), registry consumer.TopicRegistry, output io.Writer) error { + store, err := storeFor(queue) + if err != nil { + return err + } + if store == nil { + return fmt.Errorf("storage is required") + } + headID, err := affectedHead(ctx, store) + if err != nil { + return err + } + if headID == "" { + _, err := fmt.Fprintf(output, "queue=%q unaffected\n", queue) + return err + } + if _, err := fmt.Fprintf(output, "queue=%q affected-head=%s\n", queue, headID); err != nil { + return err + } + if !opts.apply { + return nil + } + payload, err := sqmq.MarshalID(sqmq.TopicKeySpeculate, headID, queue) + if err != nil { + return err + } + // This nudge depends only on stored facts; the consumer owns all decisions. + if err := publish.Message(ctx, registry, sqmq.TopicKeySpeculate, publish.MessageParams{ + Tenant: queue, PartitionKey: queue, Payload: payload, + ID: publish.IntentID(headID, recoveryCause, opts.recoveryID), + }); err != nil { + return err + } + _, err = fmt.Fprintf(output, "queue=%q queued-head=%s recovery-id=%s\n", queue, headID, opts.recoveryID) + return err +} + +func affectedHead(ctx context.Context, store orchstorage.Storage) (string, error) { + batches, err := corebatch.ListByStates(ctx, store, []entity.BatchState{entity.BatchStateSpeculating}) + if err != nil { + return "", err + } + var selected string + for _, batch := range batches { + set, err := store.GetSpeculationPathSetStore().Get(ctx, batch.ID) + if errors.Is(err, storage.ErrNotFound) { + continue + } + if err != nil { + return "", err + } + if !hasCancelledPath(set) { + continue + } + if err := resourceid.Validate(batch.ID); err != nil { + return "", err + } + if selected == "" { + selected = batch.ID + continue + } + order, err := resourceid.Compare(batch.ID, selected) + if err != nil { + return "", err + } + if order < 0 { + selected = batch.ID + } + } + return selected, nil +} + +func hasCancelledPath(set entity.SpeculationPathSet) bool { + for _, path := range set.Paths { + if path.Status == entity.SpeculationPathStatusCancelled { + return true + } + } + return false +} diff --git a/tool/submitqueue/recovercancelledpaths/main_test.go b/tool/submitqueue/recovercancelledpaths/main_test.go new file mode 100644 index 000000000..8d7317c3b --- /dev/null +++ b/tool/submitqueue/recovercancelledpaths/main_test.go @@ -0,0 +1,199 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "bytes" + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + sqmq "github.com/uber/submitqueue/submitqueue/core/messagequeue" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/storage" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + orchstorage "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage" + orchmock "github.com/uber/submitqueue/submitqueue/orchestrator/extension/storage/mock" + "go.uber.org/mock/gomock" +) + +func TestParseOptions(t *testing.T) { + for _, tt := range []struct { + name string + args []string + valid bool + apply bool + }{ + {name: "dry run default", args: []string{"--queue", "q"}, valid: true}, + {name: "explicit dry run", args: []string{"--queue", "q", "--dry-run"}, valid: true}, + {name: "apply", args: []string{"--queue", "q", "--apply", "--recovery-id", "operator-run"}, valid: true, apply: true}, + {name: "no queues"}, + {name: "empty queue", args: []string{"--queue", ""}}, + {name: "whitespace queue", args: []string{"--queue", " q"}}, + {name: "duplicate queue", args: []string{"--queue", "q", "--queue", "q"}}, + {name: "apply requires identity", args: []string{"--queue", "q", "--apply"}}, + {name: "conflicting modes", args: []string{"--queue", "q", "--apply", "--dry-run", "--recovery-id", "run"}}, + {name: "positional argument", args: []string{"--queue", "q", "extra"}}, + } { + t.Run(tt.name, func(t *testing.T) { + opts, err := parseOptions(tt.args, &bytes.Buffer{}) + if !tt.valid { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.apply, opts.apply) + }) + } +} + +func recoveryStore(ctrl *gomock.Controller, batches []entity.Batch, sets map[string]entity.SpeculationPathSet) *orchmock.MockStorage { + store := orchmock.NewMockStorage(ctrl) + states := storagemock.NewMockQueueBatchStateStore(ctrl) + batchStore := storagemock.NewMockBatchStore(ctrl) + paths := storagemock.NewMockSpeculationPathSetStore(ctrl) + store.EXPECT().GetQueueBatchStateStore().Return(states).AnyTimes() + store.EXPECT().GetBatchStore().Return(batchStore).AnyTimes() + store.EXPECT().GetSpeculationPathSetStore().Return(paths).AnyTimes() + var records []entity.QueueBatchState + for _, batch := range batches { + records = append(records, entity.QueueBatchState{BatchID: batch.ID}) + batchStore.EXPECT().Get(gomock.Any(), batch.ID).Return(batch, nil).AnyTimes() + } + states.EXPECT().List(gomock.Any(), entity.BatchStateSpeculating).Return(records, nil).AnyTimes() + paths.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, id string) (entity.SpeculationPathSet, error) { + set, ok := sets[id] + if !ok { + return entity.SpeculationPathSet{}, storage.ErrNotFound + } + return set, nil + }).AnyTimes() + return store +} + +func TestCancelledPathRecovery(t *testing.T) { + for _, apply := range []bool{false, true} { + t.Run(map[bool]string{false: "dry run", true: "apply partial failure and repeat"}[apply], func(t *testing.T) { + ctrl := gomock.NewController(t) + cancelled := entity.SpeculationPathSet{Paths: []entity.SpeculationPathEntry{{Status: entity.SpeculationPathStatusCancelled}}} + store := recoveryStore(ctrl, []entity.Batch{ + {ID: "10", State: entity.BatchStateSpeculating}, + {ID: "2", State: entity.BatchStateSpeculating}, + {ID: "3", State: entity.BatchStateSpeculating}, + {ID: "1", State: entity.BatchStateLanding}, + }, map[string]entity.SpeculationPathSet{"10": cancelled, "2": cancelled, "1": cancelled}) + publisher := queuemock.NewMockPublisher(ctrl) + queue := queuemock.NewMockQueue(ctrl) + queue.EXPECT().Publisher().Return(publisher).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{{Key: sqmq.TopicKeySpeculate, Name: "speculate", Queue: queue}}) + require.NoError(t, err) + var messages []entityqueue.Message + if apply { + publisher.EXPECT().Publish(gomock.Any(), "speculate", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + messages = append(messages, msg) + if len(messages) == 1 { + return assert.AnError + } + return nil + }).Times(3) + } + storeFor := func(string) (orchstorage.Storage, error) { return store, nil } + opts := options{queues: []string{"q", "other"}, recoveryID: "operator-run", apply: apply} + var output bytes.Buffer + err = publishCancelledPathRecovery(context.Background(), opts, storeFor, registry, &output) + if apply { + require.ErrorIs(t, err, assert.AnError) + opts.queues = []string{"q"} + require.NoError(t, publishCancelledPathRecovery(context.Background(), opts, storeFor, registry, &output)) + require.Len(t, messages, 3) + assert.Equal(t, messages[0].ID, messages[2].ID) + assert.Equal(t, "2/cancelled-path-recovery/operator-run", messages[0].ID) + for _, msg := range messages { + id, queue, err := sqmq.UnmarshalID(sqmq.TopicKeySpeculate, msg.Payload) + require.NoError(t, err) + assert.Equal(t, "2", id) + assert.Equal(t, queue, msg.Tenant) + assert.Equal(t, queue, msg.PartitionKey) + } + assert.Contains(t, output.String(), "queued-head=2") + } else { + require.NoError(t, err) + assert.Empty(t, messages) + assert.NotContains(t, output.String(), "queued-head") + } + assert.Contains(t, output.String(), "affected-head=2") + }) + } +} + +func TestRecoveryInspection(t *testing.T) { + for _, tt := range []struct { + name string + state entity.BatchState + set entity.SpeculationPathSet + missing bool + want string + }{ + {name: "no cancelled paths", state: entity.BatchStateSpeculating}, + {name: "missing path set", state: entity.BatchStateSpeculating, missing: true}, + {name: "cancelled live head", state: entity.BatchStateSpeculating, set: entity.SpeculationPathSet{Paths: []entity.SpeculationPathEntry{{Status: entity.SpeculationPathStatusCancelled}}}, want: "2"}, + {name: "terminal head is stale membership", state: entity.BatchStateCancelled, set: entity.SpeculationPathSet{Paths: []entity.SpeculationPathEntry{{Status: entity.SpeculationPathStatusCancelled}}}}, + } { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + sets := map[string]entity.SpeculationPathSet{} + if !tt.missing { + sets["2"] = tt.set + } + store := recoveryStore(ctrl, []entity.Batch{{ID: "2", State: tt.state}}, sets) + id, err := affectedHead(context.Background(), store) + require.NoError(t, err) + assert.Equal(t, tt.want, id) + }) + } + t.Run("unavailable storage", func(t *testing.T) { + for _, failure := range []error{nil, assert.AnError} { + err := publishCancelledPathRecovery(context.Background(), options{queues: []string{"q"}}, func(string) (orchstorage.Storage, error) { return nil, failure }, consumer.TopicRegistry{}, &bytes.Buffer{}) + require.Error(t, err) + } + }) + t.Run("path read fails", func(t *testing.T) { + ctrl := gomock.NewController(t) + store := orchmock.NewMockStorage(ctrl) + states := storagemock.NewMockQueueBatchStateStore(ctrl) + batches := storagemock.NewMockBatchStore(ctrl) + paths := storagemock.NewMockSpeculationPathSetStore(ctrl) + store.EXPECT().GetQueueBatchStateStore().Return(states) + store.EXPECT().GetBatchStore().Return(batches) + store.EXPECT().GetSpeculationPathSetStore().Return(paths) + states.EXPECT().List(gomock.Any(), entity.BatchStateSpeculating).Return([]entity.QueueBatchState{{BatchID: "2"}}, nil) + batches.EXPECT().Get(gomock.Any(), "2").Return(entity.Batch{ID: "2", State: entity.BatchStateSpeculating}, nil) + paths.EXPECT().Get(gomock.Any(), "2").Return(entity.SpeculationPathSet{}, assert.AnError) + _, err := affectedHead(context.Background(), store) + require.ErrorIs(t, err, assert.AnError) + }) + t.Run("missing publisher", func(t *testing.T) { + ctrl := gomock.NewController(t) + store := recoveryStore(ctrl, []entity.Batch{{ID: "2", State: entity.BatchStateSpeculating}}, map[string]entity.SpeculationPathSet{ + "2": {Paths: []entity.SpeculationPathEntry{{Status: entity.SpeculationPathStatusCancelled}}}, + }) + err := publishCancelledPathRecovery(context.Background(), options{queues: []string{"q"}, apply: true, recoveryID: "run"}, func(string) (orchstorage.Storage, error) { return store, nil }, consumer.TopicRegistry{}, &bytes.Buffer{}) + require.Error(t, err) + }) +}