diff --git a/service/submitqueue/demo/BUILD.bazel b/service/submitqueue/demo/BUILD.bazel new file mode 100644 index 000000000..f3fe36235 --- /dev/null +++ b/service/submitqueue/demo/BUILD.bazel @@ -0,0 +1,36 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "demo.go", + "run.go", + "workload.go", + ], + importpath = "github.com/uber/submitqueue/service/submitqueue/demo", + visibility = ["//visibility:public"], + deps = [ + "//api/base/mergestrategy/protopb:go_default_library", + "//api/submitqueue/gateway/protopb:go_default_library", + "//submitqueue/client:go_default_library", + "@org_golang_google_protobuf//proto:go_default_library", + "@org_golang_x_sync//errgroup:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = [ + "run_test.go", + "workload_test.go", + ], + embed = [":go_default_library"], + deps = [ + "//api/base/mergestrategy/protopb:go_default_library", + "//api/submitqueue/gateway/protopb:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@org_golang_google_grpc//codes:go_default_library", + "@org_golang_google_grpc//status:go_default_library", + ], +) diff --git a/service/submitqueue/demo/README.md b/service/submitqueue/demo/README.md new file mode 100644 index 000000000..6cb847132 --- /dev/null +++ b/service/submitqueue/demo/README.md @@ -0,0 +1,17 @@ +# Demo workload runner + +This package is the reusable part of the SubmitQueue demo. The OSS command in the sibling `requests` package and internal commands import `github.com/uber/submitqueue/service/submitqueue/demo`; deployment configuration, authentication, source construction, and connection ownership remain with those commands. + +`Run` generates changes through a supplied source. `RunExisting` takes revisions already resolved by the caller and never invokes the source. Both return known changes, accepted request identifiers, and the last observed request histories, including on partial failure. A request identifier is opaque and must always be used with its queue. + +Independent changes are created and submitted with bounded concurrency. Each can enter the queue as soon as it is ready. Burst mode prepares every change before submitting any request. Stacked mode creates a chain sequentially, prepares its members, and submits their ordered URIs as one request. + +Sources own their base branch or snapshot. A stack link receives the preceding change's branch and head; a source must honor that parent or fail rather than silently retargeting it. The OSS providers retain their frozen-base behavior. A remote source that resolves a branch at creation time can use a moving base without promising identical commit ancestry across independent changes. + +Readiness is optional and only runs when submission is enabled. A command can wait for real approvals or checks without baking that policy into the workload generator. Existing-change stack validation is the caller's responsibility. + +The runner owns polling and interactive terminal cleanup. It waits for recorded histories rather than sampling only current status, preserving fast transitions and build metadata. Monitoring failures are not request failures: permanent errors stop the local watch without cancelling creation or submission and without inventing a terminal server state, and temporary failures retain the last known status and history. + +Mutating source and submission calls are never retried automatically. A failing sibling stops new work from starting and cancels readiness waits, but never cancels a source or submission call already in flight. A timeout may hide a successful remote operation; inspect the reported artifacts before retrying. Stopping the command does not cancel submitted work, close PRs, or delete branches. + +Workload layout is deterministic for a supplied `RunID`; an empty identifier receives a timestamp and random suffix so simultaneous runs do not collide. File/folder controls and the existing OSS command flags and Make targets are unchanged. diff --git a/service/submitqueue/demo/demo.go b/service/submitqueue/demo/demo.go new file mode 100644 index 000000000..fdf1a2bb7 --- /dev/null +++ b/service/submitqueue/demo/demo.go @@ -0,0 +1,139 @@ +// Copyright (c) 2026 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 demo generates and submits workloads with injected change sources +// and gateway transports. Deployment configuration and credentials belong to +// the calling command. +package demo + +import ( + "context" + + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" +) + +// Options controls a workload, independently of its provider and transport. +type Options struct { + // Count is the number of changes to generate. + Count int + // Folders is the shard count; zero selects a reproducible count for the run. + Folders int + // Files is the minimum number of files per generated change. + Files int + // Concurrency bounds independent creation, readiness and submission. + Concurrency int + // Stacked submits an ordered chain as one request. + Stacked bool + // Burst prepares all independent changes before submitting any. + Burst bool + // Prefix is the generated branch namespace. + Prefix string + // RunID identifies a workload; empty generates a unique identifier. + RunID string + // Land enables submission; false only creates changes. + Land bool + // Watch waits for terminal request histories when Land is true. + Watch bool + // Queue is the exact queue accepting the requests. + Queue string + // Strategy is the wire strategy used for every submitted URI. + Strategy mergestrategypb.Strategy +} + +// File is one generated file and its source-specific commit description. +type File struct { + // Path is relative to the repository root. + Path string + // Body is the complete file content. + Body string + // Message is the description when a provider commits files individually. + Message string +} + +// Change is a head-pinned revision, or an attempted branch returned on failure. +type Change struct { + // Branch is the source branch. + Branch string + // HeadSHA is the full revision identifier, empty for an unconfirmed artifact. + HeadSHA string + // URI is the head-pinned identifier, empty for an unconfirmed artifact. + URI string + // Label identifies the change in terminal output. + Label string + // URL is an optional browser link. + URL string +} + +// ChangeSpec describes a new change. Sources own their configured base. +type ChangeSpec struct { + // Branch is the branch to create. + Branch string + // Title describes the change. + Title string + // Files contains generated contents. + Files []File + // Parent identifies the preceding change when HasParent is true. + Parent Change + // HasParent distinguishes a stack link from a change on the source's base. + HasParent bool + // Note reports progress and may be called concurrently. + Note func(string, ...any) +} + +// Source creates independent changes concurrently and stack links sequentially. +type Source interface { + Open(context.Context, ChangeSpec) (Change, error) +} + +// Gateway is the transport-independent submission and history contract. +type Gateway interface { + Land(ctx context.Context, queue string, uris []string, strategy mergestrategypb.Strategy) (requestID string, err error) + History(ctx context.Context, queue, requestID string) ([]*pb.HistoryEvent, error) +} + +// Readiness waits for an exact revision to be eligible for submission. +type Readiness interface { + Wait(context.Context, Change) error +} + +// Dependencies supplies provider, transport and readiness integration. +type Dependencies struct { + // Source is required for generation, unused by RunExisting. + Source Source + // Gateway is required when Land is enabled. + Gateway Gateway + // Readiness is optional; nil means no preparation is required. + Readiness Readiness +} + +// RequestResult records one accepted request and its observed history. +type RequestResult struct { + // Changes is the ordered set submitted in this request. + Changes []Change + // ID is opaque and only meaningful together with the run's queue. + ID string + // Status is the last observed server status, not a local monitoring verdict. + Status string + // History is the last successful history read. + History []*pb.HistoryEvent +} + +// RunResult preserves known artifacts, including after a partially failed run. +type RunResult struct { + // Changes contains known artifacts in input order, including unconfirmed branches. + Changes []Change + // Requests contains accepted requests in workload order. + Requests []RequestResult +} diff --git a/service/submitqueue/demo/requests/BUILD.bazel b/service/submitqueue/demo/requests/BUILD.bazel index 5c4cf9eec..af577edd3 100644 --- a/service/submitqueue/demo/requests/BUILD.bazel +++ b/service/submitqueue/demo/requests/BUILD.bazel @@ -12,12 +12,11 @@ go_library( importpath = "github.com/uber/submitqueue/service/submitqueue/demo/requests", visibility = ["//visibility:private"], deps = [ - "//api/base/mergestrategy/protopb:go_default_library", "//platform/base/change/git:go_default_library", "//platform/base/change/github:go_default_library", "//platform/git/exec:go_default_library", + "//service/submitqueue/demo:go_default_library", "//submitqueue/client:go_default_library", - "@org_golang_x_sync//errgroup:go_default_library", ], ) @@ -45,11 +44,9 @@ go_test( "SUBMITQUEUE_TEST_GIT": "$(location @git//:git)", }, deps = [ - "//api/base/mergestrategy/protopb:go_default_library", "//platform/base/change/git:go_default_library", "//platform/git/exec:go_default_library", "//platform/git/exectest:go_default_library", - "//submitqueue/client:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", ], diff --git a/service/submitqueue/demo/requests/main.go b/service/submitqueue/demo/requests/main.go index a43807b27..14b12405a 100644 --- a/service/submitqueue/demo/requests/main.go +++ b/service/submitqueue/demo/requests/main.go @@ -12,69 +12,29 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Command requests creates changes, enqueues each as a land request, and -// watches them move through the pipeline — so the demo stack can be exercised -// repeatedly without authoring changes by hand. -// -// Nothing is awaited until the end, which is the point. Each change is enqueued -// the moment it is created, so the queue is already working on the first while -// the last is still being made. A queue that only ever holds one request in -// flight never batches, never analyzes a conflict against another batch, and -// never speculates; those behaviors only appear when requests overlap. The -// table watches all of them at once. -// -// Independent changes are created several at a time (-concurrency), since -// nothing about them depends on the others — which matters most against a -// provider, where each is several round trips. A stack cannot be: every change -// in it is based on the branch before it, so the next cannot be cut until the -// previous head exists. -// -// Two shapes of change, because the pipeline treats them differently: -// -// - independent (default): each change targets the base branch and is -// enqueued as its own request, immediately after it is created. This is -// what puts requests in flight against each other. -// - stacked (-stacked): each change is based on the one before it, and all of -// them go in as a single request once the chain exists — the atomic-stack -// path, where the whole set reaches the target in one push. -// -// The table is drawn before the first change exists and refreshed for the whole -// run, so there is never a stretch with nothing to look at. Each row is one land -// request and shows the states it has passed through, read from the gateway's -// history API rather than sampled — polling only the current status would miss -// any transition that happens between two ticks, which for a fast queue is most -// of them. -// -// How a change is made depends on -provider, matching the stack the run is -// pointed at: -// -// - fake (default): a change is a URI and nothing else. No repository, no -// credential, no I/O — the fastest way to put traffic through the queue. -// - git: a branch pushed to the sandbox repository the stack lands into. -// Real commits, still no credential. -// - github: a real pull request over the REST API, which needs no clone and -// no git binary, only GITHUB_TOKEN — the same credential the stack uses. +// Command requests supplies OSS providers and transport to the shared demo runner. package main import ( "context" - "crypto/sha256" "flag" "fmt" "os" + "os/signal" "path/filepath" "strings" - "time" + "syscall" - mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" gitexec "github.com/uber/submitqueue/platform/git/exec" + "github.com/uber/submitqueue/service/submitqueue/demo" "github.com/uber/submitqueue/submitqueue/client" - "golang.org/x/sync/errgroup" ) func main() { cfg := parseFlags() - if err := run(context.Background(), cfg); err != nil { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + if err := run(ctx, cfg); err != nil { fmt.Fprintf(os.Stderr, "Error: %v\n", err) os.Exit(1) } @@ -151,98 +111,59 @@ func run(ctx context.Context, cfg config) error { if err != nil { return err } - src, cleanup, err := newChangeSource(ctx, cfg) if err != nil { return err } defer cleanup() - baseSHA, err := src.baseSHA(ctx, cfg.base) if err != nil { return fmt.Errorf("read %s: %w", cfg.base, err) } - - var sq *client.Client + var gateway demo.Gateway if cfg.land { - sq, err = client.New(client.Options{Addr: cfg.addr, TLS: cfg.tls, TokenEnv: cfg.tokenEnv}) + sq, err := client.New(client.Options{Addr: cfg.addr, TLS: cfg.tls, TokenEnv: cfg.tokenEnv}) if err != nil { return err } defer sq.Close() + gateway = sq + } + fmt.Printf("Creating %d change(s) via %s — %s\n\n", cfg.count, target(cfg), shape(cfg)) + result, err := demo.Run(ctx, demo.Options{ + Count: cfg.count, Folders: cfg.folders, Files: cfg.files, Concurrency: cfg.concurrency, + Stacked: cfg.stacked, Burst: cfg.burst, Prefix: cfg.prefix, Land: cfg.land, Watch: cfg.watch, + Queue: cfg.queue, Strategy: strategy, + }, demo.Dependencies{Source: pinnedSource{source: src, branch: cfg.base, sha: baseSHA}, Gateway: gateway}) + hinted := false + if !cfg.land && (!cfg.stacked || err == nil) { + var pinned []demo.Change + for _, change := range result.Changes { + if change.URI != "" && change.HeadSHA != "" { + pinned = append(pinned, change) + } + } + if len(pinned) > 0 { + fmt.Printf("\nEnqueue them with:\n %s\n", enqueueHint(pinned)) + hinted = true + } } - - // A run tag keeps repeated invocations from colliding on branch names, and - // makes it obvious in the repository which changes came from one run. - tag := time.Now().Format("0102-150405") - // Resolved once, so every change in the run is dealt into the same tree and - // the number can be reported rather than inferred from the paths. - cfg.folders = resolveFolders(tag, cfg.folders) - if cfg.provider == providerFake { - fmt.Printf("Creating %d synthetic change(s) via %s — %s\n\n", cfg.count, target(cfg), shape(cfg)) - } else { - fmt.Printf("Creating %d change(s) across %d folder(s) via %s — %s\n\n", - cfg.count, cfg.folders, target(cfg), shape(cfg)) - } - - // Every row is known before anything is created: one per change, or a - // single one for a stack, since the whole chain lands as one request. The - // table is therefore complete from the first draw and only ever fills in. - t := client.NewTracker(client.NewRows(rowCount(cfg))) - t.Note("starting") - - // Statuses are read on their own clock, concurrently with creation. A run - // that only started polling once every change existed would show an - // empty trail for the whole creation phase — which for a large -count is - // most of the run, and is exactly the stretch worth watching, since the - // early requests are already moving through the queue by then. - if cfg.land { - polling, stop := context.WithCancel(ctx) - defer stop() - go t.Poll(polling, sq.Gateway(), cfg.queue) - } - - created, err := createAndEnqueue(ctx, src, sq, cfg, strategy, tag, baseSHA, t) if err != nil { - return err - } - t.Seal() - - if !cfg.land { - t.Note("created %d change(s), not enqueued", len(created)) - fmt.Printf("\nEnqueue them with:\n %s\n", enqueueHint(created)) - return nil - } - if !cfg.watch { - t.Note("enqueued, not watching") - return nil - } - - // A large run has more changes than a window has lines, so the wait happens - // in a full-screen view the reader can scroll. Restored before Conclude, so - // the final table lands in the scrollback and not on a screen that is about - // to be handed back. - stop, quit := t.Interact(ctx) - defer stop() - - select { - case <-ctx.Done(): - stop() - return ctx.Err() - case <-quit: - case <-t.Settled(): + printArtifacts(result, hinted) } - stop() - return t.Conclude() + return err } -// rowCount is how many rows the table needs: one per change, or a single -// one for a stack, which lands as one request however many changes it carries. -func rowCount(cfg config) int { - if cfg.stacked { - return 1 +func printArtifacts(result demo.RunResult, pinnedHinted bool) { + for _, change := range result.Changes { + if pinnedHinted && change.URI != "" && change.HeadSHA != "" { + continue + } + fmt.Printf("change: %s %s\n", change.Label, change.URI) + } + for _, request := range result.Requests { + fmt.Printf("request: %s (last observed: %s)\n", request.ID, request.Status) } - return cfg.count } func shape(cfg config) string { @@ -330,371 +251,18 @@ func target(cfg config) string { } } -// change is one change this run created. -type change struct { - // label identifies the change in the table: a pull request number where - // there is one, and the branch otherwise. - label string - // url is where the change can be opened, empty when it lives nowhere a - // browser can reach. - url string - branch string - // headSHA is the commit the change now points at, which the next change in - // a stack branches from. - headSHA string - // uri is the SubmitQueue change URI pinning the change to its head. - uri string -} - -// enqueueHint is the command that lands what a -land=false run created. -// -// Pull request URLs where there are any, since that is what a reader recognizes -// and can open; change URIs otherwise, which is the only handle a branch in a -// local repository has. -func enqueueHint(cs []change) string { - if len(cs) > 0 && cs[0].url != "" { - return fmt.Sprintf("make land PRS=%q", strings.Join(urlsOf(cs), " ")) - } - return fmt.Sprintf("make land URIS=%q", strings.Join(urisOf(cs), " ")) -} - -func urlsOf(cs []change) []string { - out := make([]string, 0, len(cs)) - for _, c := range cs { - out = append(out, c.url) - } - return out -} - -// How many directories a run spreads its changes across, as a range rather than -// a number: each run picks one, so repeated runs against the same queue do not -// all collide in the same shape. -// -// This range is the whole dial on what a run demonstrates. The default analyzer -// keys on the directory, so a large number means nothing ever collides and a -// run shows only the parallel case, while one means everything serializes and -// it shows only the other. Five to ten is wide enough to look like a real tree -// and small enough that a run of a handful of changes still puts two of them in -// the same place. -// -// It used to be two nested levels of 256, keyed on each file's own name, which -// put a twelve-file run's odds of any collision at roughly one in a thousand -// and scattered a single change across as many directories as it had files. -const ( - minShardDirs = 5 - maxShardDirs = 10 -) - -// resolveFolders is how many directories a run spreads its changes across: -// what -folders asked for, or a number picked from the range above when it -// asked for nothing. -// -// Worth setting deliberately when the run is meant to show one thing. -folders 1 -// puts every change in the same place, so the queue serializes the lot and each -// change speculates on the one before it; a number well above -count keeps them -// apart, so they all go out together. -func resolveFolders(tag string, configured int) int { - if configured > 0 { - return configured - } - sum := sha256.Sum256([]byte("folders#" + tag)) - return minShardDirs + int(sum[0])%(maxShardDirs-minShardDirs+1) -} - -// changeShard is the directory one change writes into, as a fixed-width number -// so a run's collisions are visible at a glance in a table or a diff. -// -// Keyed on the change rather than on each file, so a change occupies exactly -// one directory: a change spread over several would overlap with everything and -// report a conflict that says nothing about it. Derived from the run tag, so -// replaying a tag reproduces the same collisions. -func changeShard(tag string, folders, change int) string { - sum := sha256.Sum256([]byte(fmt.Sprintf("shard#%s#%d", tag, change))) - return fmt.Sprintf("%02d", int(sum[0])%folders) -} - -// changeFilePath returns the repository path for one file of one change. -// -// The leaf name carries the run tag, the change index and the file index, which -// is what makes it unique: no two files in a run, and no two runs against the -// same repository, can ever name the same path. That uniqueness is load-bearing -// — see createAndEnqueue — and it is the leaf that carries it, not the -// directory. Two changes deliberately share a directory (see shardDirs) so the -// conflict analyzer has something to find, and they still never write the same -// file, so a shared directory serializes them rather than making them collide. -func changeFilePath(tag string, folders, change, file int) string { - leaf := fmt.Sprintf("%s-%d-%d.txt", tag, change, file) - return strings.Join([]string{"demo", changeShard(tag, folders, change), leaf}, "/") -} - -// changeFileCount returns how many files a change touches: at least min, varied -// a little so a run does not produce a row of identically shaped changes. -// -// The variation is derived from the run tag and the change index rather than -// from a clock or a global source of randomness, so replaying a tag reproduces -// the same run. A demo that cannot be reproduced is hard to talk about when -// something in it goes wrong. -func changeFileCount(tag string, change, min int) int { - if min < 1 { - min = 1 - } - sum := sha256.Sum256([]byte(fmt.Sprintf("%s#%d", tag, change))) - return min + int(sum[0]%4) -} - -// lander enqueues a request. *client.Client is the real one; a test supplies -// its own to observe when each change is enqueued relative to when it is created. -type lander interface { - Land(ctx context.Context, queue string, uris []string, strategy mergestrategypb.Strategy) (string, error) -} - -// createAndEnqueue creates the changes and puts them on the queue, filling -// in the tracker's rows as it goes and reporting each step beneath the table. -// -// For independent changes the two steps interleave: each change is -// enqueued the moment it exists, so the queue is already working on it while -// the next is being opened. Stacked changes cannot interleave — one request -// carries the whole chain, so it can only be submitted once the chain is -// complete. -// -// Every file a change writes is its own, at a path no other change uses. -// Independent changes would otherwise collide on content and the run would -// measure conflict handling rather than the throughput it is trying to show; a -// caller wanting a conflict can make one deliberately. Each change spreads -// several files across the sharded tree, so it arrives as a multi-file, multi- -// commit change rather than a single-line edit — which is both closer to -// a real change and enough to exercise replaying a range of commits. -func createAndEnqueue( - ctx context.Context, - src changeSource, - sq lander, - cfg config, - strategy mergestrategypb.Strategy, - tag, baseSHA string, - t *client.Tracker, -) ([]change, error) { - if cfg.stacked { - return createStack(ctx, src, sq, cfg, strategy, tag, baseSHA, t) - } - return createIndependent(ctx, src, sq, cfg, strategy, tag, baseSHA, t) -} - -// createIndependent creates the changes concurrently, up to the configured -// limit, enqueuing each the moment it exists. -// -// Independent changes have nothing to say to each other: each branches from the -// same base and writes files no other change touches, so the only reason to -// create them one at a time was that the loop did. Creating a change is -// several round trips to the provider — a branch, a commit per file, the pull -// request itself — and doing that serially is most of what a large run spends -// its time on. It also delays the overlap the demo exists to show, since the -// queue cannot work on requests that have not been submitted yet. -// -// The limit is there because the provider is a shared service with its own -// opinion about burst rates, and because the point is to feed the queue, not to -// find out how fast a repository can be hammered. -func createIndependent( - ctx context.Context, - src changeSource, - sq lander, - cfg config, - strategy mergestrategypb.Strategy, - tag, baseSHA string, - t *client.Tracker, -) ([]change, error) { - if cfg.burst && cfg.land { - return createBurst(ctx, src, sq, cfg, strategy, tag, baseSHA, t) - } - - rows := t.Rows() - // Indexed rather than appended: the workers finish in whatever order the - // provider answers them, and the caller still wants the run's own order. - created := make([]change, cfg.count) - - group, groupCtx := errgroup.WithContext(ctx) - group.SetLimit(cfg.concurrency) - - for i := 1; i <= cfg.count; i++ { - group.Go(func() error { - c, err := createOne(groupCtx, src, cfg, tag, baseSHA, cfg.base, i, t, rows[i-1]) - if err != nil { - return err - } - created[i-1] = c - - if !cfg.land { - return nil - } - t.Note("enqueuing %s", c.label) - sqid, err := sq.Land(groupCtx, cfg.queue, urisOf([]change{c}), strategy) - if err != nil { - return err - } - t.Update(func() { rows[i-1].SQID, rows[i-1].Submitted = sqid, time.Now() }) - return nil - }) - } - - if err := group.Wait(); err != nil { - return nil, err - } - return created, nil -} - -// createBurst creates every change first and only then enqueues them, firing -// all the Land calls together so the requests arrive at the queue in one burst. -// -// The default path lands each change the moment it exists, so the queue starts -// working while later changes are still being created. Burst trades that early -// overlap for a simultaneous arrival: useful for watching the queue admit a -// large batch at once. It does not make creation faster — with -provider git the -// creation phase is still serialized on a single work tree — it only separates -// creation from enqueuing so the enqueues are not spread across it. -func createBurst( - ctx context.Context, - src changeSource, - sq lander, - cfg config, - strategy mergestrategypb.Strategy, - tag, baseSHA string, - t *client.Tracker, -) ([]change, error) { - rows := t.Rows() - created := make([]change, cfg.count) - - create, createCtx := errgroup.WithContext(ctx) - create.SetLimit(cfg.concurrency) - for i := 1; i <= cfg.count; i++ { - create.Go(func() error { - c, err := createOne(createCtx, src, cfg, tag, baseSHA, cfg.base, i, t, rows[i-1]) - if err != nil { - return err - } - created[i-1] = c - return nil - }) - } - if err := create.Wait(); err != nil { - return nil, err - } - - t.Note("enqueuing %d changes at once", cfg.count) - land, landCtx := errgroup.WithContext(ctx) - land.SetLimit(cfg.concurrency) - for i := 1; i <= cfg.count; i++ { - land.Go(func() error { - sqid, err := sq.Land(landCtx, cfg.queue, urisOf([]change{created[i-1]}), strategy) - if err != nil { - return err - } - t.Update(func() { rows[i-1].SQID, rows[i-1].Submitted = sqid, time.Now() }) - return nil - }) - } - if err := land.Wait(); err != nil { - return nil, err - } - return created, nil -} - -// createStack creates the changes one after another, each based on the one -// before it, and submits the whole chain as a single request. -// -// This one cannot be parallelized, and not for want of trying: a change is -// based on the branch of the change before it and must see its content, so the -// next branch cannot be cut until the previous head exists. -func createStack( - ctx context.Context, - src changeSource, - sq lander, - cfg config, - strategy mergestrategypb.Strategy, - tag, baseSHA string, - t *client.Tracker, -) ([]change, error) { - rows := t.Rows() - created := make([]change, 0, cfg.count) - - parentBranch, parentSHA := cfg.base, baseSHA - for i := 1; i <= cfg.count; i++ { - // A stack is one request, so every change lands on the single row. - c, err := createOne(ctx, src, cfg, tag, parentSHA, parentBranch, i, t, rows[0]) - if err != nil { - return nil, err - } - created = append(created, c) - parentBranch, parentSHA = c.branch, c.headSHA - } - - // The stack goes in as one request, which is only possible now that every - // change in it exists. - if cfg.land { - t.Note("enqueuing the stack") - sqid, err := sq.Land(ctx, cfg.queue, urisOf(created), strategy) - if err != nil { - return nil, err +func enqueueHint(changes []demo.Change) string { + values := make([]string, len(changes)) + for i, c := range changes { + if c.URL != "" { + values[i] = c.URL + } else { + values[i] = c.URI } - t.Update(func() { rows[0].SQID, rows[0].Submitted = sqid, time.Now() }) - } - return created, nil -} - -// createOne describes one change — its branch, and the files it writes — hands -// it to the source to be made real, and records it on the given row. -// -// What a change is made of is decided here rather than by the source, so the -// three providers put the same shape of change through the queue and differ -// only in how it comes to exist. -func createOne( - ctx context.Context, - src changeSource, - cfg config, - tag, parentSHA, parentBranch string, - i int, - t *client.Tracker, - target *client.Row, -) (change, error) { - branch := fmt.Sprintf("%s/%s/%d", cfg.prefix, tag, i) - - // Each file is its own commit, so a change arrives as a range of commits - // rather than a single edit. - fileCount := changeFileCount(tag, i, cfg.files) - files := make([]changeFile, 0, fileCount) - for k := 1; k <= fileCount; k++ { - files = append(files, changeFile{ - path: changeFilePath(tag, cfg.folders, i, k), - body: fmt.Sprintf("change %d of run %s\nfile %d of %d\n", i, tag, k, fileCount), - message: fmt.Sprintf("demo change %d (run %s): file %d of %d", i, tag, k, fileCount), - }) - } - - opened, err := src.open(ctx, changeSpec{ - branch: branch, - parentBranch: parentBranch, - parentSHA: parentSHA, - title: fmt.Sprintf("demo change %d (run %s)", i, tag), - files: files, - note: t.Note, - }) - if err != nil { - return change{}, err } - - t.Update(func() { target.Cells = append(target.Cells, opened.cell) }) - return change{ - label: opened.cell.Text, - url: opened.cell.URL, - branch: branch, - headSHA: opened.headSHA, - uri: opened.uri, - }, nil -} - -// urisOf is the change URIs the run pinned, in caller order. -func urisOf(cs []change) []string { - out := make([]string, 0, len(cs)) - for _, c := range cs { - out = append(out, c.uri) + flag := "URIS" + if len(changes) > 0 && changes[0].URL != "" { + flag = "PRS" } - return out + return fmt.Sprintf("make land %s=%q", flag, strings.Join(values, " ")) } diff --git a/service/submitqueue/demo/requests/main_test.go b/service/submitqueue/demo/requests/main_test.go index 7d0a36c89..f971673cc 100644 --- a/service/submitqueue/demo/requests/main_test.go +++ b/service/submitqueue/demo/requests/main_test.go @@ -15,145 +15,12 @@ package main import ( - "context" - "fmt" - "strings" - "sync" - "sync/atomic" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - - mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" - "github.com/uber/submitqueue/submitqueue/client" ) -func TestChangeFilePath_IsUniquePerFileAcrossChangesAndRuns(t *testing.T) { - // Uniqueness is the property the whole layout rests on: two changes writing - // the same path would collide on content, and the run would measure conflict - // handling instead of throughput. - seen := make(map[string]string) - for _, tag := range []string{"0810-1203", "0810-1204"} { - for change := 1; change <= 20; change++ { - for file := 1; file <= 8; file++ { - path := changeFilePath(tag, resolveFolders(tag, 0), change, file) - owner := fmt.Sprintf("%s/%d/%d", tag, change, file) - if prev, ok := seen[path]; ok { - t.Fatalf("path %s produced for both %s and %s", path, prev, owner) - } - seen[path] = owner - } - } - } -} - -func TestChangeFilePath_PutsEveryFileOfAChangeInOneDirectory(t *testing.T) { - // The conflict analyzer keys on the directory, so a change spread across - // several would overlap with everything and report a conflict that says - // nothing about the change. - dirs := make(map[string]struct{}) - for file := 1; file <= 8; file++ { - parts := strings.Split(changeFilePath("0810-1203", 5, 1, file), "/") - require.Len(t, parts, 3, "demo root, one folder, and the leaf") - assert.Equal(t, "demo", parts[0]) - assert.Regexp(t, `^\d{2}$`, parts[1]) - dirs[strings.Join(parts[:2], "/")] = struct{}{} - } - assert.Len(t, dirs, 1, "one change writes into one directory") -} - -// -folders is the dial on what a run demonstrates, so both ends of it have to -// do what they say. -func TestChangeFilePath_FollowsTheFolderCount(t *testing.T) { - folderOf := func(folders, change int) string { - parts := strings.Split(changeFilePath("0810-1203", folders, change, 1), "/") - return strings.Join(parts[:2], "/") - } - - t.Run("one folder puts every change together", func(t *testing.T) { - dirs := make(map[string]struct{}) - for change := 1; change <= 10; change++ { - dirs[folderOf(1, change)] = struct{}{} - } - assert.Len(t, dirs, 1, "every change must conflict with every other") - }) - - t.Run("many folders keep changes apart", func(t *testing.T) { - dirs := make(map[string]struct{}) - for change := 1; change <= 3; change++ { - dirs[folderOf(64, change)] = struct{}{} - } - assert.Len(t, dirs, 3, "three changes in 64 folders should not collide") - }) -} - -func TestResolveFolders(t *testing.T) { - t.Run("honors an explicit count", func(t *testing.T) { - assert.Equal(t, 1, resolveFolders("0810-1203", 1)) - assert.Equal(t, 42, resolveFolders("0810-1203", 42)) - }) - - t.Run("picks within the range when unset", func(t *testing.T) { - // A demo that cannot be replayed is hard to talk about when something in - // it goes wrong, and how the changes were spread is part of what - // happened — so the pick follows the run tag. - assert.Equal(t, resolveFolders("0810-1203", 0), resolveFolders("0810-1203", 0)) - - seen := make(map[int]struct{}) - for minute := range 60 { - folders := resolveFolders(fmt.Sprintf("0810-12%02d", minute), 0) - assert.GreaterOrEqual(t, folders, minShardDirs) - assert.LessOrEqual(t, folders, maxShardDirs) - seen[folders] = struct{}{} - } - assert.Greater(t, len(seen), 1, "runs must not all pick the same number") - }) -} - -func TestChangeFileCount(t *testing.T) { - tests := []struct { - name string - min int - }{ - {name: "default minimum", min: 3}, - {name: "single file floor", min: 1}, - {name: "non-positive is clamped", min: 0}, - {name: "negative is clamped", min: -5}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - floor := tt.min - if floor < 1 { - floor = 1 - } - for change := 1; change <= 50; change++ { - got := changeFileCount("0810-1203", change, tt.min) - assert.GreaterOrEqual(t, got, floor) - assert.LessOrEqual(t, got, floor+3) - } - }) - } -} - -func TestChangeFileCount_VariesButIsReproducible(t *testing.T) { - counts := make(map[int]struct{}) - for change := 1; change <= 30; change++ { - got := changeFileCount("0810-1203", change, 3) - counts[got] = struct{}{} - assert.Equal(t, got, changeFileCount("0810-1203", change, 3), - "replaying a tag must reproduce the run") - } - assert.Greater(t, len(counts), 1, "the count should vary across changes, not be constant") -} - -func TestRowCount(t *testing.T) { - // A stack lands as one request however many pull requests it chains, so it - // gets one row; independent changes get one each. - assert.Equal(t, 3, rowCount(config{count: 3}), "independent changes are one request each") - assert.Equal(t, 1, rowCount(config{count: 3, stacked: true}), "a stack is a single request") -} - func TestConfigValidate(t *testing.T) { valid := config{ provider: providerGitHub, token: "t", repo: "owner/name", @@ -216,58 +83,3 @@ func TestShapeReportsConcurrency(t *testing.T) { assert.NotContains(t, shape(config{count: 10, concurrency: 5, burst: true, land: false}), "at once", "burst is a landing decision; with nothing to land there is no burst") } - -// recordingSource counts how many changes have been created, so a test can -// observe creation relative to enqueuing. It creates through fakeSource, so it -// stands in for any provider — the two-phase orchestration under test is the -// same whichever one is wired. -type recordingSource struct { - created *int64 -} - -func (recordingSource) baseSHA(context.Context, string) (string, error) { return "base", nil } - -func (r recordingSource) open(ctx context.Context, spec changeSpec) (openedChange, error) { - atomic.AddInt64(r.created, 1) - return fakeSource{}.open(ctx, spec) -} - -// landerFunc adapts a function to the lander interface. -type landerFunc func(context.Context, string, []string, mergestrategypb.Strategy) (string, error) - -func (f landerFunc) Land(ctx context.Context, queue string, uris []string, s mergestrategypb.Strategy) (string, error) { - return f(ctx, queue, uris, s) -} - -// TestCreateBurst_EnqueuesOnlyAfterEveryChangeIsCreated is the property -burst -// exists for: no request reaches the queue until the last change has been -// created, so they all arrive together. It is provider-independent — burst runs -// every source through the same two phases. -func TestCreateBurst_EnqueuesOnlyAfterEveryChangeIsCreated(t *testing.T) { - const count = 8 - cfg := config{count: count, concurrency: 4, files: 3, folders: 3, land: true, burst: true, queue: "q", prefix: "demo"} - - var created int64 - src := recordingSource{created: &created} - - var mu sync.Mutex - createdWhenLanded := make([]int64, 0, count) - lander := landerFunc(func(context.Context, string, []string, mergestrategypb.Strategy) (string, error) { - mu.Lock() - createdWhenLanded = append(createdWhenLanded, atomic.LoadInt64(&created)) - mu.Unlock() - return "sqid", nil - }) - - tracker := client.NewTracker(client.NewRows(count)) - got, err := createBurst(context.Background(), src, lander, cfg, - mergestrategypb.Strategy_SQUASH_REBASE, "run", "base", tracker) - require.NoError(t, err) - require.Len(t, got, count) - - require.Len(t, createdWhenLanded, count, "every change is enqueued exactly once") - for _, seen := range createdWhenLanded { - assert.Equal(t, int64(count), seen, - "burst enqueues nothing until every change has been created") - } -} diff --git a/service/submitqueue/demo/requests/source.go b/service/submitqueue/demo/requests/source.go index 44973cd13..54327ba93 100644 --- a/service/submitqueue/demo/requests/source.go +++ b/service/submitqueue/demo/requests/source.go @@ -17,6 +17,7 @@ package main import ( "context" + "github.com/uber/submitqueue/service/submitqueue/demo" "github.com/uber/submitqueue/submitqueue/client" ) @@ -81,3 +82,31 @@ type openedChange struct { // cell is what the table shows for this change. cell client.Cell } + +type pinnedSource struct { + source changeSource + branch string + sha string +} + +func (s pinnedSource) Open(ctx context.Context, spec demo.ChangeSpec) (demo.Change, error) { + parentBranch, parentSHA := s.branch, s.sha + if spec.HasParent { + parentBranch, parentSHA = spec.Parent.Branch, spec.Parent.HeadSHA + } + files := make([]changeFile, len(spec.Files)) + for i, file := range spec.Files { + files[i] = changeFile{path: file.Path, body: file.Body, message: file.Message} + } + opened, err := s.source.open(ctx, changeSpec{ + branch: spec.Branch, parentBranch: parentBranch, parentSHA: parentSHA, + title: spec.Title, files: files, note: spec.Note, + }) + if err != nil { + return demo.Change{Branch: spec.Branch, Label: spec.Branch}, err + } + return demo.Change{ + Branch: spec.Branch, HeadSHA: opened.headSHA, URI: opened.uri, + Label: opened.cell.Text, URL: opened.cell.URL, + }, nil +} diff --git a/service/submitqueue/demo/requests/source_test.go b/service/submitqueue/demo/requests/source_test.go index 49bec3d59..f2a38380d 100644 --- a/service/submitqueue/demo/requests/source_test.go +++ b/service/submitqueue/demo/requests/source_test.go @@ -46,7 +46,7 @@ func quietSpec(branch, parentBranch, parentSHA string, files ...changeFile) chan func TestSources_ProduceSubmittableURIs(t *testing.T) { files := make([]changeFile, 0, 8) for k := 1; k <= 8; k++ { - files = append(files, changeFile{path: changeFilePath("0814-154238", 5, 1, k)}) + files = append(files, changeFile{path: fmt.Sprintf("demo/01/0814-154238-1-%d.txt", k)}) } spec := quietSpec("demo/0814-154238/1", "main", strings.Repeat("b", 40), files...) diff --git a/service/submitqueue/demo/run.go b/service/submitqueue/demo/run.go new file mode 100644 index 000000000..4b58a6ed0 --- /dev/null +++ b/service/submitqueue/demo/run.go @@ -0,0 +1,393 @@ +// Copyright (c) 2026 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 demo + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "fmt" + "strings" + "sync" + "time" + + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/submitqueue/client" + "golang.org/x/sync/errgroup" + "google.golang.org/protobuf/proto" +) + +// Run generates a workload. Returned artifacts remain available on failure. +func Run(ctx context.Context, opts Options, deps Dependencies) (RunResult, error) { + if opts.Prefix == "" { + opts.Prefix = "demo" + } + if err := opts.validate(); err != nil { + return RunResult{}, err + } + if deps.Source == nil { + return RunResult{}, fmt.Errorf("a change source is required") + } + if opts.RunID == "" { + var suffix [4]byte + if _, err := rand.Read(suffix[:]); err != nil { + return RunResult{}, fmt.Errorf("create run identifier: %w", err) + } + opts.RunID = time.Now().Format("0102-150405") + "-" + hex.EncodeToString(suffix[:]) + } + opts.Folders = resolveFolders(opts.RunID, opts.Folders) + return runWorkload(ctx, opts, deps, nil) +} + +// RunExisting submits existing changes in caller order without invoking Source. +// Each input must pin its revision. Stack ancestry is the caller's responsibility. +func RunExisting(ctx context.Context, opts Options, changes []Change, deps Dependencies) (RunResult, error) { + opts.Count = len(changes) + if opts.Files == 0 { + opts.Files = 1 + } + if err := opts.validate(); err != nil { + return RunResult{}, err + } + for _, change := range changes { + if change.URI == "" || change.HeadSHA == "" { + return RunResult{}, fmt.Errorf("existing changes must include a pinned URI and head revision") + } + } + return runWorkload(ctx, opts, deps, append([]Change(nil), changes...)) +} + +func (opts Options) validate() error { + switch { + case opts.Count < 1: + return fmt.Errorf("count must be at least 1") + case opts.Concurrency < 1: + return fmt.Errorf("concurrency must be at least 1") + case opts.Files < 1: + return fmt.Errorf("files must be at least 1") + case opts.Folders < 0: + return fmt.Errorf("folders must not be negative") + case opts.Land && opts.Queue == "": + return fmt.Errorf("queue is required for submission") + } + if strings.ContainsAny(opts.RunID, "/\\") || opts.RunID == "." || opts.RunID == ".." { + return fmt.Errorf("run identifier must be a single path component") + } + for _, part := range strings.Split(opts.Prefix, "/") { + if part == "." || part == ".." { + return fmt.Errorf("prefix must not contain relative path components") + } + } + return nil +} + +type workload struct { + opts Options + deps Dependencies + tracker *client.Tracker + changes []Change + requests [][]Change +} + +func runWorkload(ctx context.Context, opts Options, deps Dependencies, existing []Change) (RunResult, error) { + if opts.Land && deps.Gateway == nil { + return RunResult{}, fmt.Errorf("a gateway is required for submission") + } + rows := opts.Count + if opts.Stacked { + rows = 1 + } + tracker := client.NewTracker(client.NewRows(rows)) + w := workload{ + opts: opts, deps: deps, tracker: tracker, + changes: make([]Change, opts.Count), requests: make([][]Change, rows), + } + tracker.Note("starting") + // Polling has its own context: a failed watch must never cancel an + // in-flight Open or Land, whose outcome would then be unknown. + pollCtx, stopPolling := context.WithCancel(ctx) + defer stopPolling() + recorder := &historyRecorder{gateway: deps.Gateway, histories: make(map[string][]*pb.HistoryEvent)} + pollDone := make(chan error, 1) + polling := opts.Land && opts.Watch + if polling { + go func() { + err := tracker.PollHistory(pollCtx, recorder, opts.Queue) + if err != nil && pollCtx.Err() == nil { + tracker.Note("watch stopped: %v; submission continues", err) + } + pollDone <- err + }() + } + + err := w.createAndSubmit(ctx, existing) + tracker.Seal() + var pollErr error + pollJoined := false + interacted := false + if err == nil && polling { + interacted = true + stop, quit := tracker.Interact(ctx) + select { + case <-tracker.Settled(): + case pollErr = <-pollDone: + pollJoined = true + case <-ctx.Done(): + case <-quit: + err = fmt.Errorf("watch interrupted; submitted requests continue remotely") + } + stop() + } + stopPolling() + if polling && !pollJoined { + pollErr = <-pollDone + } + if errors.Is(pollErr, context.Canceled) { + pollErr = nil + } + err = errors.Join(err, pollErr) + if ctxErr := ctx.Err(); ctxErr != nil && !errors.Is(err, ctxErr) { + err = errors.Join(err, ctxErr) + } + // The interactive view took the table with it; draw it into the scrollback + // even when the watch ended early, but only let its verdict stand alone. + if interacted { + if verdict := tracker.Conclude(); err == nil { + err = verdict + } + } + if err != nil { + tracker.Note("run stopped: %v", err) + } else if !opts.Land { + tracker.Note("created %d change(s), not enqueued", opts.Count) + } else if !opts.Watch { + tracker.Note("enqueued, not watching") + } + return w.snapshotResult(recorder), err +} + +func (w *workload) createAndSubmit(ctx context.Context, existing []Change) error { + if w.opts.Stacked { + return w.createAndSubmitStack(ctx, existing) + } + if w.opts.Burst && w.opts.Land { + if err := w.forEachChange(ctx, func(gate context.Context, i int) error { + if err := w.createChange(ctx, gate, i, Change{}, false, existing); err != nil { + return err + } + return w.prepareChange(gate, i, i) + }); err != nil { + return err + } + w.tracker.Note("enqueuing %d prepared changes", w.opts.Count) + return w.forEachChange(ctx, func(gate context.Context, i int) error { + return w.submitRequest(ctx, gate, i, []Change{w.changes[i]}) + }) + } + return w.forEachChange(ctx, func(gate context.Context, i int) error { + if err := w.createChange(ctx, gate, i, Change{}, false, existing); err != nil { + return err + } + if !w.opts.Land { + return nil + } + if err := w.prepareChange(gate, i, i); err != nil { + return err + } + return w.submitRequest(ctx, gate, i, []Change{w.changes[i]}) + }) +} + +// forEachChange passes fn a gate that closes when any sibling fails. Mutating +// calls check the gate before starting but run on the caller's context, so a +// sibling failure never cancels an Open or Land whose outcome would be unknown. +func (w *workload) forEachChange(ctx context.Context, fn func(gate context.Context, i int) error) error { + group, groupCtx := errgroup.WithContext(ctx) + group.SetLimit(w.opts.Concurrency) + for i := range w.changes { + if groupCtx.Err() != nil { + break + } + group.Go(func() error { + if err := groupCtx.Err(); err != nil { + return err + } + return fn(groupCtx, i) + }) + } + return errors.Join(group.Wait(), ctx.Err()) +} + +func (w *workload) createAndSubmitStack(ctx context.Context, existing []Change) error { + for i := range w.changes { + parent := Change{} + if i > 0 { + parent = w.changes[i-1] + } + if err := w.createChange(ctx, ctx, i, parent, i > 0, existing); err != nil { + return err + } + } + if !w.opts.Land { + return nil + } + // Finish the chain before checks can block its creation. + w.setReadinessNote(0, "waiting for readiness") + if err := w.forEachChange(ctx, func(gate context.Context, i int) error { + return w.waitForReadiness(gate, i) + }); err != nil { + return err + } + w.setReadinessNote(0, "") + return w.submitRequest(ctx, ctx, 0, w.changes) +} + +func (w *workload) createChange(ctx, gate context.Context, i int, parent Change, hasParent bool, existing []Change) error { + if err := gate.Err(); err != nil { + return err + } + var opened Change + var err error + if existing != nil { + opened = existing[i] + } else { + branch := fmt.Sprintf("%s/%s/%d", w.opts.Prefix, w.opts.RunID, i+1) + files := make([]File, changeFileCount(w.opts.RunID, i+1, w.opts.Files)) + for k := range files { + files[k] = File{ + Path: changeFilePath(w.opts.RunID, w.opts.Folders, i+1, k+1), + Body: fmt.Sprintf("change %d of run %s\nfile %d of %d\n", i+1, w.opts.RunID, k+1, len(files)), + Message: fmt.Sprintf("demo change %d (run %s): file %d of %d", i+1, w.opts.RunID, k+1, len(files)), + } + } + opened, err = w.deps.Source.Open(ctx, ChangeSpec{ + Branch: branch, Title: fmt.Sprintf("demo change %d (run %s)", i+1, w.opts.RunID), + Files: files, Parent: parent, HasParent: hasParent, Note: w.tracker.Note, + }) + } + w.changes[i] = opened + row := i + if w.opts.Stacked { + row = 0 + } + if opened.URI != "" || opened.Branch != "" { + w.tracker.Update(func() { + w.tracker.Rows()[row].Cells = append(w.tracker.Rows()[row].Cells, + client.Cell{Text: opened.Label, URL: opened.URL}) + }) + } + if err != nil { + return err + } + if opened.URI == "" || opened.HeadSHA == "" { + return fmt.Errorf("source returned an unpinned change for %s", opened.Branch) + } + return nil +} + +func (w *workload) prepareChange(ctx context.Context, i, row int) error { + w.setReadinessNote(row, "waiting for readiness") + if err := w.waitForReadiness(ctx, i); err != nil { + return err + } + w.setReadinessNote(row, "") + return nil +} + +func (w *workload) waitForReadiness(ctx context.Context, i int) error { + if w.deps.Readiness == nil { + return nil + } + if err := w.deps.Readiness.Wait(ctx, w.changes[i]); err != nil { + return fmt.Errorf("prepare %s: %w", w.changes[i].Label, err) + } + return nil +} + +func (w *workload) setReadinessNote(row int, note string) { + if w.deps.Readiness == nil { + return + } + w.tracker.Update(func() { w.tracker.Rows()[row].Note = note }) +} + +func (w *workload) submitRequest(ctx, gate context.Context, row int, changes []Change) error { + if err := gate.Err(); err != nil { + return err + } + uris := make([]string, len(changes)) + for i, change := range changes { + uris[i] = change.URI + } + w.tracker.Note("enqueuing %s", strings.Join(uris, ", ")) + id, err := w.deps.Gateway.Land(ctx, w.opts.Queue, uris, w.opts.Strategy) + if err != nil { + return err + } + if id == "" { + return fmt.Errorf("gateway accepted a request without returning its identifier") + } + w.requests[row] = append([]Change(nil), changes...) + w.tracker.Update(func() { + r := w.tracker.Rows()[row] + r.SQID, r.Submitted, r.Status = id, time.Now(), "accepted" + }) + return nil +} + +func (w *workload) snapshotResult(recorder *historyRecorder) RunResult { + result := RunResult{} + for _, change := range w.changes { + if change.URI != "" || change.Branch != "" { + result.Changes = append(result.Changes, change) + } + } + for i, row := range w.tracker.SnapshotRows() { + if row.SQID != "" { + result.Requests = append(result.Requests, RequestResult{ + Changes: append([]Change(nil), w.requests[i]...), ID: row.SQID, + Status: row.Status, History: recorder.histories[row.SQID], + }) + } + } + return result +} + +type historyRecorder struct { + gateway Gateway + mu sync.Mutex + histories map[string][]*pb.HistoryEvent +} + +func (r *historyRecorder) History(ctx context.Context, queue, id string) ([]*pb.HistoryEvent, error) { + events, err := r.gateway.History(ctx, queue, id) + if err != nil { + return nil, err + } + if len(events) == 0 { + return events, nil + } + cloned := make([]*pb.HistoryEvent, len(events)) + for i, event := range events { + if event != nil { + cloned[i] = proto.Clone(event).(*pb.HistoryEvent) + } + } + r.mu.Lock() + r.histories[id] = cloned + r.mu.Unlock() + return events, nil +} diff --git a/service/submitqueue/demo/run_test.go b/service/submitqueue/demo/run_test.go new file mode 100644 index 000000000..95b88c702 --- /dev/null +++ b/service/submitqueue/demo/run_test.go @@ -0,0 +1,416 @@ +// Copyright (c) 2026 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 demo + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type sourceFunc func(context.Context, ChangeSpec) (Change, error) + +func (f sourceFunc) Open(ctx context.Context, spec ChangeSpec) (Change, error) { + return f(ctx, spec) +} + +type readinessFunc func(context.Context, Change) error + +func (f readinessFunc) Wait(ctx context.Context, change Change) error { + return f(ctx, change) +} + +type gatewayFixture struct { + land func(context.Context, string, []string, mergestrategypb.Strategy) (string, error) + history func(context.Context, string, string) ([]*pb.HistoryEvent, error) +} + +func (g gatewayFixture) Land(ctx context.Context, queue string, uris []string, strategy mergestrategypb.Strategy) (string, error) { + return g.land(ctx, queue, uris, strategy) +} + +func (g gatewayFixture) History(ctx context.Context, queue, id string) ([]*pb.HistoryEvent, error) { + if g.history == nil { + return nil, nil + } + return g.history(ctx, queue, id) +} + +func fixtureOptions() Options { + return Options{ + Count: 3, Files: 1, Folders: 2, Concurrency: 1, Prefix: "demo", + RunID: "run", Queue: "q", Land: true, Strategy: mergestrategypb.Strategy_SQUASH_REBASE, + } +} + +func fixtureChange(spec ChangeSpec) Change { + return Change{ + Branch: spec.Branch, HeadSHA: strings.Repeat("a", 40), + URI: "change:" + spec.Branch, Label: spec.Branch, + } +} + +func fixtureGateway() gatewayFixture { + return gatewayFixture{land: func(_ context.Context, _ string, uris []string, _ mergestrategypb.Strategy) (string, error) { + return strings.Join(uris, ","), nil + }} +} + +func TestWorkloadSubmissionModes(t *testing.T) { + tests := []struct { + name string + burst bool + stacked bool + wantLandSizes []int + wantCreated []int + }{ + {name: "immediate", wantLandSizes: []int{1, 1, 1}, wantCreated: []int{1, 2, 3}}, + {name: "burst", burst: true, wantLandSizes: []int{1, 1, 1}, wantCreated: []int{3, 3, 3}}, + {name: "stack", stacked: true, wantLandSizes: []int{3}, wantCreated: []int{3}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + opts := fixtureOptions() + opts.Burst, opts.Stacked = tt.burst, tt.stacked + var specs []ChangeSpec + var prepared []string + var landSizes, createdAtLand []int + src := sourceFunc(func(_ context.Context, spec ChangeSpec) (Change, error) { + specs = append(specs, spec) + return fixtureChange(spec), nil + }) + gateway := fixtureGateway() + gateway.land = func(_ context.Context, _ string, uris []string, _ mergestrategypb.Strategy) (string, error) { + landSizes = append(landSizes, len(uris)) + createdAtLand = append(createdAtLand, len(specs)) + return fmt.Sprintf("id-%d", len(landSizes)), nil + } + result, err := Run(context.Background(), opts, Dependencies{ + Source: src, Gateway: gateway, Readiness: readinessFunc(func(_ context.Context, change Change) error { + prepared = append(prepared, change.URI) + return nil + }), + }) + require.NoError(t, err) + assert.Equal(t, tt.wantLandSizes, landSizes) + assert.Equal(t, tt.wantCreated, createdAtLand) + require.Len(t, result.Changes, 3) + assert.Len(t, prepared, 3) + assert.Len(t, result.Requests, len(tt.wantLandSizes)) + for i, spec := range specs { + assert.Equal(t, tt.stacked && i > 0, spec.HasParent) + if spec.HasParent { + assert.Equal(t, result.Changes[i-1], spec.Parent) + } + } + }) + } +} + +func TestBurstWaitsForAllReadiness(t *testing.T) { + opts := fixtureOptions() + opts.Burst = true + var prepared, calls int + expected := errors.New("readiness failed") + gateway := fixtureGateway() + gateway.land = func(context.Context, string, []string, mergestrategypb.Strategy) (string, error) { + calls++ + return "id", nil + } + result, err := Run(context.Background(), opts, Dependencies{ + Source: sourceFunc(func(_ context.Context, spec ChangeSpec) (Change, error) { return fixtureChange(spec), nil }), + Gateway: gateway, Readiness: readinessFunc(func(context.Context, Change) error { + prepared++ + if prepared == 3 { + return expected + } + return nil + }), + }) + require.ErrorIs(t, err, expected) + assert.Len(t, result.Changes, 3) + assert.Empty(t, result.Requests) + assert.Zero(t, calls) +} + +func TestIndependentCreationIsBoundedAndOverlapsSubmission(t *testing.T) { + opts := fixtureOptions() + opts.Concurrency = 2 + opts.Count = 4 + entered := make(chan struct{}, 4) + release := make(chan struct{}) + landed := make(chan struct{}, 4) + done := make(chan error, 1) + src := sourceFunc(func(ctx context.Context, spec ChangeSpec) (Change, error) { + entered <- struct{}{} + if strings.HasSuffix(spec.Branch, "/1") { + return fixtureChange(spec), nil + } + select { + case <-release: + return fixtureChange(spec), nil + case <-ctx.Done(): + return Change{}, ctx.Err() + } + }) + gateway := fixtureGateway() + gateway.land = func(_ context.Context, _ string, uris []string, _ mergestrategypb.Strategy) (string, error) { + landed <- struct{}{} + return uris[0], nil + } + go func() { + _, err := Run(context.Background(), opts, Dependencies{Source: src, Gateway: gateway}) + done <- err + }() + <-landed + <-entered + <-entered + <-entered + select { + case <-entered: + assert.Fail(t, "creation exceeded the two active workers") + default: + } + close(release) + require.NoError(t, <-done) +} + +func TestPartialFailurePreservesAcceptedArtifacts(t *testing.T) { + opts := fixtureOptions() + expected := errors.New("creation failed") + calls := 0 + result, err := Run(context.Background(), opts, Dependencies{ + Gateway: fixtureGateway(), + Source: sourceFunc(func(_ context.Context, spec ChangeSpec) (Change, error) { + calls++ + if calls == 2 { + return Change{Branch: spec.Branch}, expected + } + return fixtureChange(spec), nil + }), + }) + require.ErrorIs(t, err, expected) + assert.Equal(t, 2, calls) + assert.Len(t, result.Changes, 2) + require.Len(t, result.Requests, 1) + assert.Equal(t, "accepted", result.Requests[0].Status) + assert.Equal(t, "change:demo/run/1", result.Requests[0].ID) +} + +func TestExistingChangesAreNotCreatedOrMutated(t *testing.T) { + opts := fixtureOptions() + opts.Stacked = true + changes := []Change{fixtureChange(ChangeSpec{Branch: "first"}), fixtureChange(ChangeSpec{Branch: "second"})} + want := append([]Change(nil), changes...) + var uris []string + gateway := fixtureGateway() + gateway.land = func(_ context.Context, _ string, in []string, _ mergestrategypb.Strategy) (string, error) { + uris = append([]string(nil), in...) + return "opaque/id", nil + } + result, err := RunExisting(context.Background(), opts, changes, Dependencies{ + Gateway: gateway, Source: sourceFunc(func(context.Context, ChangeSpec) (Change, error) { + return Change{}, errors.New("existing changes must not invoke Source") + }), + }) + require.NoError(t, err) + assert.Equal(t, want, changes) + assert.Equal(t, []string{changes[0].URI, changes[1].URI}, uris) + require.Len(t, result.Requests, 1) + result.Changes[0].URI = "mutated" + assert.Equal(t, want, changes) + assert.Equal(t, want, result.Requests[0].Changes) +} + +func TestCreateOnlySkipsGatewayAndReadiness(t *testing.T) { + opts := fixtureOptions() + opts.Land, opts.Watch = false, true + result, err := Run(context.Background(), opts, Dependencies{ + Source: sourceFunc(func(_ context.Context, spec ChangeSpec) (Change, error) { return fixtureChange(spec), nil }), + Readiness: readinessFunc(func(context.Context, Change) error { + return errors.New("create-only must not wait") + }), + }) + require.NoError(t, err) + assert.Len(t, result.Changes, 3) + assert.Empty(t, result.Requests) +} + +func TestWatchReturnsHistoryAndTerminalVerdict(t *testing.T) { + for _, terminal := range []string{"landed", "error", "cancelled"} { + t.Run(terminal, func(t *testing.T) { + opts := fixtureOptions() + opts.Count, opts.Watch = 1, true + gateway := fixtureGateway() + events := []*pb.HistoryEvent{{Status: "accepted"}, {Status: terminal, Metadata: map[string]string{"build_url": "build"}}} + gateway.history = func(context.Context, string, string) ([]*pb.HistoryEvent, error) { return events, nil } + result, err := Run(context.Background(), opts, Dependencies{ + Source: sourceFunc(func(_ context.Context, spec ChangeSpec) (Change, error) { return fixtureChange(spec), nil }), + Gateway: gateway, + }) + if terminal == "landed" { + require.NoError(t, err) + } else { + require.Error(t, err) + } + require.Len(t, result.Requests, 1) + assert.Equal(t, terminal, result.Requests[0].Status) + assert.Equal(t, events, result.Requests[0].History) + result.Requests[0].History[1].Metadata["build_url"] = "changed" + assert.Equal(t, "build", events[1].Metadata["build_url"]) + }) + } +} + +func TestWatchFailureDoesNotCancelCreationOrSubmission(t *testing.T) { + opts := fixtureOptions() + opts.Count, opts.Concurrency, opts.Watch = 2, 2, true + historyFailed := make(chan struct{}) + var once sync.Once + gateway := fixtureGateway() + gateway.history = func(context.Context, string, string) ([]*pb.HistoryEvent, error) { + once.Do(func() { close(historyFailed) }) + return nil, status.Error(codes.PermissionDenied, "denied") + } + result, err := Run(context.Background(), opts, Dependencies{ + Source: sourceFunc(func(ctx context.Context, spec ChangeSpec) (Change, error) { + if !strings.HasSuffix(spec.Branch, "/1") { + <-historyFailed + } + if err := ctx.Err(); err != nil { + return Change{}, err + } + return fixtureChange(spec), nil + }), + Gateway: gateway, + }) + require.Error(t, err) + var grpcErr interface{ GRPCStatus() *status.Status } + require.ErrorAs(t, err, &grpcErr) + assert.Equal(t, codes.PermissionDenied, grpcErr.GRPCStatus().Code()) + require.Len(t, result.Changes, 2) + require.Len(t, result.Requests, 2) + for _, request := range result.Requests { + assert.Equal(t, "accepted", request.Status) + } +} + +func TestSiblingFailureDoesNotCancelInFlightLand(t *testing.T) { + opts := fixtureOptions() + opts.Count, opts.Concurrency = 3, 3 + landStarted := make(chan struct{}) + siblingsCancelled := make(chan struct{}) + var landErr error + gateway := fixtureGateway() + gateway.land = func(ctx context.Context, _ string, uris []string, _ mergestrategypb.Strategy) (string, error) { + close(landStarted) + <-siblingsCancelled + landErr = ctx.Err() + return strings.Join(uris, ","), nil + } + result, err := Run(context.Background(), opts, Dependencies{ + Source: sourceFunc(func(_ context.Context, spec ChangeSpec) (Change, error) { return fixtureChange(spec), nil }), + Gateway: gateway, + Readiness: readinessFunc(func(gate context.Context, change Change) error { + switch { + case strings.HasSuffix(change.Branch, "/1"): + <-landStarted + return fmt.Errorf("not ready") + case strings.HasSuffix(change.Branch, "/3"): + <-gate.Done() + close(siblingsCancelled) + return gate.Err() + } + return nil + }), + }) + require.Error(t, err) + assert.NoError(t, landErr) + require.Len(t, result.Requests, 1) + assert.Equal(t, "change:demo/run/2", result.Requests[0].ID) +} + +func TestInvalidWorkloadsFailBeforeCreation(t *testing.T) { + for _, mutate := range []func(*Options){ + func(o *Options) { o.Count = 0 }, + func(o *Options) { o.Concurrency = 0 }, + func(o *Options) { o.Files = 0 }, + func(o *Options) { o.Folders = -1 }, + func(o *Options) { o.Queue = "" }, + func(o *Options) { o.RunID = "../escape" }, + func(o *Options) { o.Prefix = "../escape" }, + } { + opts := fixtureOptions() + mutate(&opts) + _, err := Run(context.Background(), opts, Dependencies{}) + require.Error(t, err) + } +} + +func TestCallerCancellationDuringWatch(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + opts := fixtureOptions() + opts.Count, opts.Watch = 1, true + gateway := fixtureGateway() + gateway.history = func(context.Context, string, string) ([]*pb.HistoryEvent, error) { + cancel() + return nil, context.Canceled + } + result, err := Run(ctx, opts, Dependencies{ + Source: sourceFunc(func(_ context.Context, spec ChangeSpec) (Change, error) { return fixtureChange(spec), nil }), + Gateway: gateway, + }) + require.ErrorIs(t, err, context.Canceled) + require.Len(t, result.Requests, 1) + assert.Equal(t, "accepted", result.Requests[0].Status) +} + +func TestWatchRetainsKnownHistoryAfterEmptyRead(t *testing.T) { + opts := fixtureOptions() + opts.Count, opts.Watch = 1, true + gateway := fixtureGateway() + calls := 0 + known := []*pb.HistoryEvent{{Status: "started", Metadata: map[string]string{"build_url": "build"}}} + gateway.history = func(context.Context, string, string) ([]*pb.HistoryEvent, error) { + calls++ + switch calls { + case 1: + return known, nil + case 2: + return nil, nil + default: + return nil, status.Error(codes.PermissionDenied, "denied") + } + } + result, err := Run(context.Background(), opts, Dependencies{ + Source: sourceFunc(func(_ context.Context, spec ChangeSpec) (Change, error) { return fixtureChange(spec), nil }), + Gateway: gateway, + }) + require.Error(t, err) + require.Len(t, result.Requests, 1) + assert.Equal(t, "started", result.Requests[0].Status) + assert.Equal(t, known, result.Requests[0].History) +} diff --git a/service/submitqueue/demo/workload.go b/service/submitqueue/demo/workload.go new file mode 100644 index 000000000..e49b00fdb --- /dev/null +++ b/service/submitqueue/demo/workload.go @@ -0,0 +1,53 @@ +// Copyright (c) 2026 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 demo + +import ( + "crypto/sha256" + "fmt" + "strings" +) + +// Five to ten shards produce both shared-directory and independent changes. +const ( + minShardDirs = 5 + maxShardDirs = 10 +) + +func resolveFolders(tag string, configured int) int { + if configured > 0 { + return configured + } + sum := sha256.Sum256([]byte("folders#" + tag)) + return minShardDirs + int(sum[0])%(maxShardDirs-minShardDirs+1) +} + +func changeShard(tag string, folders, change int) string { + sum := sha256.Sum256([]byte(fmt.Sprintf("shard#%s#%d", tag, change))) + return fmt.Sprintf("%02d", int(sum[0])%folders) +} + +func changeFilePath(tag string, folders, change, file int) string { + leaf := fmt.Sprintf("%s-%d-%d.txt", tag, change, file) + return strings.Join([]string{"demo", changeShard(tag, folders, change), leaf}, "/") +} + +func changeFileCount(tag string, change, min int) int { + if min < 1 { + min = 1 + } + sum := sha256.Sum256([]byte(fmt.Sprintf("%s#%d", tag, change))) + return min + int(sum[0]%4) +} diff --git a/service/submitqueue/demo/workload_test.go b/service/submitqueue/demo/workload_test.go new file mode 100644 index 000000000..d0f5f1b21 --- /dev/null +++ b/service/submitqueue/demo/workload_test.go @@ -0,0 +1,130 @@ +// 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 demo + +import ( + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestChangeFilePath_IsUniquePerFileAcrossChangesAndRuns(t *testing.T) { + seen := make(map[string]string) + for _, tag := range []string{"0810-1203", "0810-1204"} { + for change := 1; change <= 20; change++ { + for file := 1; file <= 8; file++ { + path := changeFilePath(tag, resolveFolders(tag, 0), change, file) + owner := fmt.Sprintf("%s/%d/%d", tag, change, file) + _, exists := seen[path] + require.False(t, exists, "path %s duplicated", path) + seen[path] = owner + } + } + } +} + +func TestChangeFilePath_PutsEveryFileOfAChangeInOneDirectory(t *testing.T) { + dirs := make(map[string]struct{}) + for file := 1; file <= 8; file++ { + parts := strings.Split(changeFilePath("0810-1203", 5, 1, file), "/") + require.Len(t, parts, 3, "demo root, one folder, and the leaf") + assert.Equal(t, "demo", parts[0]) + assert.Regexp(t, `^\d{2}$`, parts[1]) + dirs[strings.Join(parts[:2], "/")] = struct{}{} + } + assert.Len(t, dirs, 1, "one change writes into one directory") +} + +func TestChangeFilePath_FollowsTheFolderCount(t *testing.T) { + folderOf := func(folders, change int) string { + parts := strings.Split(changeFilePath("0810-1203", folders, change, 1), "/") + return strings.Join(parts[:2], "/") + } + + t.Run("one folder puts every change together", func(t *testing.T) { + dirs := make(map[string]struct{}) + for change := 1; change <= 10; change++ { + dirs[folderOf(1, change)] = struct{}{} + } + assert.Len(t, dirs, 1, "every change must conflict with every other") + }) + + t.Run("many folders keep changes apart", func(t *testing.T) { + dirs := make(map[string]struct{}) + for change := 1; change <= 3; change++ { + dirs[folderOf(64, change)] = struct{}{} + } + assert.Len(t, dirs, 3, "three changes in 64 folders should not collide") + }) +} + +func TestResolveFolders(t *testing.T) { + t.Run("honors an explicit count", func(t *testing.T) { + assert.Equal(t, 1, resolveFolders("0810-1203", 1)) + assert.Equal(t, 42, resolveFolders("0810-1203", 42)) + }) + + t.Run("picks within the range when unset", func(t *testing.T) { + assert.Equal(t, resolveFolders("0810-1203", 0), resolveFolders("0810-1203", 0)) + + seen := make(map[int]struct{}) + for minute := range 60 { + folders := resolveFolders(fmt.Sprintf("0810-12%02d", minute), 0) + assert.GreaterOrEqual(t, folders, minShardDirs) + assert.LessOrEqual(t, folders, maxShardDirs) + seen[folders] = struct{}{} + } + assert.Greater(t, len(seen), 1, "runs must not all pick the same number") + }) +} + +func TestChangeFileCount(t *testing.T) { + tests := []struct { + name string + min int + }{ + {name: "default minimum", min: 3}, + {name: "single file floor", min: 1}, + {name: "non-positive is clamped", min: 0}, + {name: "negative is clamped", min: -5}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + floor := tt.min + if floor < 1 { + floor = 1 + } + for change := 1; change <= 50; change++ { + got := changeFileCount("0810-1203", change, tt.min) + assert.GreaterOrEqual(t, got, floor) + assert.LessOrEqual(t, got, floor+3) + } + }) + } +} + +func TestChangeFileCount_VariesButIsReproducible(t *testing.T) { + counts := make(map[int]struct{}) + for change := 1; change <= 30; change++ { + got := changeFileCount("0810-1203", change, 3) + counts[got] = struct{}{} + assert.Equal(t, got, changeFileCount("0810-1203", change, 3), + "replaying a tag must reproduce the run") + } + assert.Greater(t, len(counts), 1, "the count should vary across changes, not be constant") +} diff --git a/submitqueue/client/BUILD.bazel b/submitqueue/client/BUILD.bazel index 687597f1e..8885a75e7 100644 --- a/submitqueue/client/BUILD.bazel +++ b/submitqueue/client/BUILD.bazel @@ -16,10 +16,13 @@ go_library( "//api/base/change/protopb:go_default_library", "//api/base/mergestrategy/protopb:go_default_library", "//api/submitqueue/gateway/protopb:go_default_library", + "//platform/errs:go_default_library", "//submitqueue/entity:go_default_library", "@org_golang_google_grpc//:go_default_library", + "@org_golang_google_grpc//codes:go_default_library", "@org_golang_google_grpc//credentials:go_default_library", "@org_golang_google_grpc//credentials/insecure:go_default_library", + "@org_golang_google_grpc//status:go_default_library", "@org_golang_x_term//:go_default_library", ], ) @@ -28,6 +31,7 @@ go_test( name = "go_default_test", srcs = [ "conn_test.go", + "history_read_test.go", "query_test.go", "tui_test.go", "view_test.go", @@ -35,6 +39,7 @@ go_test( embed = [":go_default_library"], deps = [ "//api/submitqueue/gateway/protopb:go_default_library", + "//platform/errs:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@org_golang_google_grpc//:go_default_library", diff --git a/submitqueue/client/history_read_test.go b/submitqueue/client/history_read_test.go new file mode 100644 index 000000000..5c5a16589 --- /dev/null +++ b/submitqueue/client/history_read_test.go @@ -0,0 +1,221 @@ +// Copyright (c) 2026 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 client + +import ( + "context" + "errors" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/platform/errs" +) + +type historyReaderFunc func(context.Context, string, string) ([]*pb.HistoryEvent, error) + +func (f historyReaderFunc) History(ctx context.Context, queue, sqid string) ([]*pb.HistoryEvent, error) { + return f(ctx, queue, sqid) +} + +type historySourceFunc func(context.Context, *pb.GetRequestHistoryByIDRequest) (*pb.GetRequestHistoryByIDResponse, error) + +func (f historySourceFunc) GetRequestHistoryByID( + ctx context.Context, request *pb.GetRequestHistoryByIDRequest, _ ...grpc.CallOption, +) (*pb.GetRequestHistoryByIDResponse, error) { + return f(ctx, request) +} + +func TestPollHistoryPermanentFailures(t *testing.T) { + for _, code := range []codes.Code{codes.PermissionDenied, codes.Unauthenticated, codes.InvalidArgument} { + t.Run(code.String(), func(t *testing.T) { + row := &Row{SQID: "opaque/id", Status: "speculating", Trail: []string{"accepted", "speculating"}} + tracker := NewTracker([]*Row{row}) + failure := status.Error(code, "read rejected") + reader := historyReaderFunc(func(_ context.Context, queue, sqid string) ([]*pb.HistoryEvent, error) { + assert.Equal(t, "dummy-queue", queue) + assert.Equal(t, "opaque/id", sqid) + return nil, failure + }) + var err error + captureStdout(t, func() { + tracker.Seal() + err = tracker.PollHistory(context.Background(), reader, "dummy-queue") + }) + require.ErrorIs(t, err, failure) + assert.Equal(t, code, status.Code(err)) + assert.Equal(t, "speculating", row.Status) + assert.Equal(t, []string{"accepted", "speculating"}, row.Trail) + assert.False(t, row.Done) + assert.True(t, row.Settled.IsZero()) + assert.False(t, isClosed(tracker.Settled())) + assert.NotEmpty(t, row.Note) + }) + } +} + +func TestHistoryTransientDiagnosticsPreserveRecordedError(t *testing.T) { + for _, failure := range []error{ + status.Error(codes.Unavailable, "temporarily unavailable"), + status.Error(codes.DeadlineExceeded, "request deadline exceeded"), + status.Error(codes.ResourceExhausted, "temporarily throttled"), + status.Error(codes.Aborted, "concurrent update"), + status.Error(codes.Internal, "proxy reset"), + errs.NewRetryableError(errors.New("temporary storage failure")), + } { + t.Run(fmt.Sprintf("%T-%v", failure, status.Code(failure)), func(t *testing.T) { + row := &Row{SQID: "1"} + tracker := NewTracker([]*Row{row}) + reader := historyReaderFunc(func(context.Context, string, string) ([]*pb.HistoryEvent, error) { + return []*pb.HistoryEvent{{Status: "speculating", LastError: "build failed"}}, nil + }) + captureStdout(t, func() { + require.NoError(t, tracker.refreshHistory(context.Background(), reader, "queue", true)) + }) + reader = func(context.Context, string, string) ([]*pb.HistoryEvent, error) { return nil, failure } + captureStdout(t, func() { + require.NoError(t, tracker.refreshHistory(context.Background(), reader, "queue", true)) + require.NoError(t, tracker.refreshHistory(context.Background(), reader, "queue", true)) + }) + assert.Equal(t, "build failed; history read: "+failure.Error(), row.Note) + assert.Equal(t, "speculating", row.Status) + assert.False(t, row.Done) + + reader = func(context.Context, string, string) ([]*pb.HistoryEvent, error) { + return []*pb.HistoryEvent{{Status: "landed"}}, nil + } + captureStdout(t, func() { + tracker.Seal() + require.NoError(t, tracker.refreshHistory(context.Background(), reader, "queue", true)) + }) + assert.Empty(t, row.Note) + assert.Equal(t, "landed", row.Status) + assert.True(t, row.Done) + assert.True(t, isClosed(tracker.Settled())) + }) + } +} + +func TestPollHistoryRetriesUntilSettled(t *testing.T) { + tracker := NewTracker([]*Row{{SQID: "1"}}) + calls := 0 + reader := historyReaderFunc(func(context.Context, string, string) ([]*pb.HistoryEvent, error) { + calls++ + if calls == 1 { + return nil, status.Error(codes.Unavailable, "retry") + } + return []*pb.HistoryEvent{{Status: "landed"}}, nil + }) + captureStdout(t, func() { + tracker.Seal() + require.NoError(t, tracker.PollHistory(context.Background(), reader, "queue")) + }) + assert.Equal(t, 2, calls) + assert.Equal(t, "landed", tracker.SnapshotRows()[0].Status) +} + +func TestHistoryPendingReadPreservesStatus(t *testing.T) { + for _, failure := range []error{nil, status.Error(codes.NotFound, "not materialized")} { + t.Run(status.Code(failure).String(), func(t *testing.T) { + row := &Row{SQID: "1", Status: "speculating", Note: "build failed"} + tracker := NewTracker([]*Row{row}) + unavailable := historyReaderFunc(func(context.Context, string, string) ([]*pb.HistoryEvent, error) { + return nil, status.Error(codes.Unavailable, "temporary") + }) + pending := historyReaderFunc(func(context.Context, string, string) ([]*pb.HistoryEvent, error) { + return nil, failure + }) + captureStdout(t, func() { + require.NoError(t, tracker.refreshHistory(context.Background(), unavailable, "queue", true)) + require.NoError(t, tracker.refreshHistory(context.Background(), pending, "queue", true)) + }) + assert.Equal(t, "speculating", row.Status) + assert.Equal(t, "build failed", row.Note) + assert.False(t, row.Done) + }) + } +} + +func TestPollHistoryContextCancellation(t *testing.T) { + for _, cancelDuringRead := range []bool{false, true} { + t.Run(fmt.Sprintf("cancel-during-read-%t", cancelDuringRead), func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + tracker := NewTracker([]*Row{{SQID: "1"}}) + calls := 0 + reader := historyReaderFunc(func(ctx context.Context, _, _ string) ([]*pb.HistoryEvent, error) { + calls++ + cancel() + return nil, ctx.Err() + }) + if !cancelDuringRead { + cancel() + } + captureStdout(t, func() { + require.ErrorIs(t, tracker.PollHistory(ctx, reader, "queue"), context.Canceled) + }) + expectedCalls := 0 + if cancelDuringRead { + expectedCalls = 1 + } + assert.Equal(t, expectedCalls, calls) + assert.False(t, tracker.SnapshotRows()[0].Done) + }) + } +} + +func TestLegacyRefreshStillToleratesPermanentErrors(t *testing.T) { + row := &Row{SQID: "1", Status: "accepted", Note: "existing note"} + tracker := NewTracker([]*Row{row}) + source := historySourceFunc(func(context.Context, *pb.GetRequestHistoryByIDRequest) (*pb.GetRequestHistoryByIDResponse, error) { + return nil, status.Error(codes.PermissionDenied, "legacy ignored") + }) + captureStdout(t, func() { tracker.refresh(context.Background(), source, "queue") }) + assert.Equal(t, "accepted", row.Status) + assert.Equal(t, "existing note", row.Note) + assert.False(t, row.Done) +} + +func TestTrackerSnapshotRowsDoNotAliasTrackedValues(t *testing.T) { + tracker := NewTracker([]*Row{{ + SQID: "1", + Cells: []Cell{{Text: "#12", URL: "https://example.test/12"}}, + Trail: []string{"accepted"}, + }}) + snapshot := tracker.SnapshotRows() + snapshot[0].SQID = "mutated" + snapshot[0].Cells[0].Text = "mutated" + snapshot[0].Trail[0] = "mutated" + + next := tracker.SnapshotRows() + assert.Equal(t, "1", next[0].SQID) + assert.Equal(t, "#12", next[0].Cells[0].Text) + assert.Equal(t, []string{"accepted"}, next[0].Trail) + + captureStdout(t, func() { + tracker.Update(func() { + tracker.Rows()[0].Cells[0].Text = "#13" + tracker.Rows()[0].Trail[0] = "started" + }) + }) + assert.Equal(t, "#12", next[0].Cells[0].Text) + assert.Equal(t, []string{"accepted"}, next[0].Trail) +} diff --git a/submitqueue/client/watch.go b/submitqueue/client/watch.go index a04370e6e..b90a18b42 100644 --- a/submitqueue/client/watch.go +++ b/submitqueue/client/watch.go @@ -22,8 +22,11 @@ import ( "time" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/platform/errs" ) // HistorySource is the part of the gateway a watch reads from. @@ -39,6 +42,23 @@ type HistorySource interface { ) (*pb.GetRequestHistoryByIDResponse, error) } +// HistoryReader decouples request tracking from the gateway's transport. +type HistoryReader interface { + History(ctx context.Context, queue, sqid string) ([]*pb.HistoryEvent, error) +} + +type gatewayHistoryReader struct { + source HistorySource +} + +func (r gatewayHistoryReader) History(ctx context.Context, queue, sqid string) ([]*pb.HistoryEvent, error) { + resp, err := r.source.GetRequestHistoryByID(ctx, &pb.GetRequestHistoryByIDRequest{Sqid: sqid, Queue: queue}) + if err != nil { + return nil, err + } + return resp.GetEvents(), nil +} + // Tracker owns the rows and the table drawn from them. // // Two things move a run forward at once — whatever is producing requests, and @@ -49,6 +69,9 @@ type Tracker struct { rows []*Row r *renderer status string + // historyNotes excludes monitoring diagnostics, which must not replace a + // request's recorded LastError. + historyNotes map[*Row]string // sealed records that every request that will be watched is known. Without // it, polling would find nothing outstanding before the first request // existed and call the run finished. @@ -70,6 +93,19 @@ func (t *Tracker) Rows() []*Row { return t.rows } +// SnapshotRows returns independent row values and slices under the tracker lock. +func (t *Tracker) SnapshotRows() []Row { + t.mu.Lock() + defer t.mu.Unlock() + rows := make([]Row, len(t.rows)) + for i, row := range t.rows { + rows[i] = *row + rows[i].Cells = append([]Cell(nil), row.Cells...) + rows[i].Trail = append([]string(nil), row.Trail...) + } + return rows +} + // Settled closes once every request has reached a terminal status. func (t *Tracker) Settled() <-chan struct{} { return t.settled @@ -219,17 +255,35 @@ func (t *Tracker) signalLocked() { // Poll re-reads statuses until the run finishes or the context ends. func (t *Tracker) Poll(ctx context.Context, src HistorySource, queue string) { + _ = t.pollHistory(ctx, gatewayHistoryReader{source: src}, queue, false) +} + +// PollHistory returns nil after all sealed rows settle, or a context/permanent +// read error. Temporary errors and missing histories are retried; read failures +// never change a request's recorded status. +func (t *Tracker) PollHistory(ctx context.Context, reader HistoryReader, queue string) error { + return t.pollHistory(ctx, reader, queue, true) +} + +func (t *Tracker) pollHistory(ctx context.Context, reader HistoryReader, queue string, reportErrors bool) error { ticker := time.NewTicker(pollInterval) defer ticker.Stop() + if reportErrors { + if err := t.refreshHistory(ctx, reader, queue, true); err != nil { + return err + } + } for { select { case <-ctx.Done(): - return + return ctx.Err() case <-t.settled: - return + return nil case <-ticker.C: } - t.refresh(ctx, src, queue) + if err := t.refreshHistory(ctx, reader, queue, reportErrors); err != nil { + return err + } } } @@ -239,11 +293,22 @@ func (t *Tracker) Poll(ctx context.Context, src HistorySource, queue string) { // stall whatever is producing requests behind the network, and letting that run // ahead is the whole point of watching each request the moment it exists. func (t *Tracker) refresh(ctx context.Context, src HistorySource, queue string) { + _ = t.refreshHistory(ctx, gatewayHistoryReader{source: src}, queue, false) +} + +func (t *Tracker) refreshHistory(ctx context.Context, reader HistoryReader, queue string, reportErrors bool) error { + if err := ctx.Err(); err != nil { + return err + } t.mu.Lock() - outstanding := make([]*Row, 0, len(t.rows)) + type requestRow struct { + row *Row + sqid string + } + outstanding := make([]requestRow, 0, len(t.rows)) for _, rw := range t.rows { if rw.SQID != "" && !rw.Done { - outstanding = append(outstanding, rw) + outstanding = append(outstanding, requestRow{row: rw, sqid: rw.SQID}) } } total := len(t.rows) @@ -254,26 +319,66 @@ func (t *Tracker) refresh(ctx context.Context, src HistorySource, queue string) trail []string status string note string + err error + empty bool } readings := make([]reading, 0, len(outstanding)) - for _, rw := range outstanding { - // SQID is written once, before the row becomes outstanding, so reading - // it here without the lock is safe. - resp, err := src.GetRequestHistoryByID(ctx, &pb.GetRequestHistoryByIDRequest{Sqid: rw.SQID, Queue: queue}) - if err != nil || resp == nil || len(resp.Events) == 0 { - // A history that is not readable yet is normal right after Land; - // the next tick picks it up. + var readErr error + for _, request := range outstanding { + events, err := reader.History(ctx, queue, request.sqid) + if err != nil { + if !reportErrors { + continue + } + if status.Code(err) == codes.NotFound { + readings = append(readings, reading{rw: request.row, empty: true}) + continue + } + readings = append(readings, reading{rw: request.row, err: err}) + if ctx.Err() != nil { + readErr = ctx.Err() + break + } + if !retryableHistoryError(err) { + readErr = fmt.Errorf("history for %s: %w", request.sqid, err) + break + } continue } - trail, status, note := digest(resp.Events) - readings = append(readings, reading{rw: rw, trail: trail, status: status, note: note}) + if len(events) == 0 { + if reportErrors { + readings = append(readings, reading{rw: request.row, empty: true}) + } + continue + } + trail, current, note := digest(events) + readings = append(readings, reading{rw: request.row, trail: trail, status: current, note: note}) } t.mu.Lock() defer t.mu.Unlock() settled := 0 + if t.historyNotes == nil { + t.historyNotes = make(map[*Row]string) + } for _, got := range readings { + if _, known := t.historyNotes[got.rw]; !known { + t.historyNotes[got.rw] = got.rw.Note + } + if got.err != nil { + note := fmt.Sprintf("history read: %v", got.err) + if recorded := t.historyNotes[got.rw]; recorded != "" { + note = recorded + "; " + note + } + got.rw.Note = note + continue + } + if got.empty { + got.rw.Note = t.historyNotes[got.rw] + continue + } + t.historyNotes[got.rw] = got.note got.rw.Trail, got.rw.Status, got.rw.Note = got.trail, got.status, got.note if terminalStatuses[got.status] && !got.rw.Done { // Stamped from the local clock rather than the event timestamp so @@ -290,4 +395,17 @@ func (t *Tracker) refresh(ctx context.Context, src HistorySource, queue string) t.status = fmt.Sprintf("%d of %d settled", settled, total) t.r.draw(t.rows, t.status) t.signalLocked() + return readErr +} + +func retryableHistoryError(err error) bool { + if errs.IsRetryable(err) { + return true + } + switch status.Code(err) { + case codes.Unavailable, codes.DeadlineExceeded, codes.ResourceExhausted, codes.Aborted, codes.Internal: + return true + default: + return false + } }