From 0651e0ac64a6a1e9cc5a7146451fc9a2c70aef7a Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Fri, 7 Aug 2026 15:54:47 -0700 Subject: [PATCH 1/5] feat(example): create, enqueue, and watch demo pull requests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Exercising the demo meant opening pull requests by hand and pasting their URLs into a land command — slow enough to discourage running it, and it made the interesting case impossible to reach in practice. A queue that only ever holds one request in flight never batches, never analyzes a conflict against another batch, and never speculates. ### What? `make demo-pr` creates pull requests in the scratch repository, enqueues them, and watches them settle. Each pull request is enqueued the moment it is created, so the queue is already working on the first while the last is still being opened. Nothing is awaited until everything is in; the wait happens once, against all of them. That overlap is what puts requests in flight against each other, which is the whole point of running more than one. `STACKED=true` is the exception, and necessarily so: one request carries the whole chain, so it can only be submitted once every change in it exists. That is the atomic-stack path, where the set reaches the target in a single push. Progress is a table keyed on the gateway's status API, redrawn as requests move, so it is visible which stage each one is in rather than only whether it finished. On a terminal it redraws in place; piped to a file it prints a fresh table whenever something changes, so logs stay readable. It lives beside the configuration it drives, under `service/submitqueue/example/`, and reaches GitHub through the REST API rather than a clone — so it needs no checkout and no git binary, only `GITHUB_TOKEN`, the credential the stack already uses. ## Test Plan ✅ `bazel build //service/submitqueue/example/pr`; `make help` lists `demo-pr`. Running it end to end needs a scratch repository and a token, so it is exercised the same way the rest of the provider path is — manually. See `doc/howto/PROVIDER-E2E.md`. # Conflicts: # service/submitqueue/demo/provider/github/profiles.yaml # Please enter the commit message for your changes. Lines starting # with '#' will be kept; you may remove them yourself if you want to. # An empty message aborts the commit. # # interactive rebase in progress; onto a9eb7496 # Last command done (1 command done): # pick 528556f7 # feat(example): create, enqueue, and watch demo pull requests # Next commands to do (4 remaining commands): # pick 74a87687 # feat(demo): live table with the full state trail per request # pick e5e50dd1 # feat(messagequeue): level the queue's own logs separately # You are currently rebasing. # # Changes to be committed: # modified: Makefile # modified: doc/howto/PROVIDER-E2E.md # new file: service/submitqueue/demo/pr/BUILD.bazel # new file: service/submitqueue/demo/pr/main.go # modified: service/submitqueue/demo/provider/github/profiles.yaml # --- Makefile | 17 +- doc/howto/PROVIDER-E2E.md | 31 +- service/submitqueue/demo/pr/BUILD.bazel | 22 + service/submitqueue/demo/pr/main.go | 555 ++++++++++++++++++ .../demo/provider/github/profiles.yaml | 7 +- 5 files changed, 629 insertions(+), 3 deletions(-) create mode 100644 service/submitqueue/demo/pr/BUILD.bazel create mode 100644 service/submitqueue/demo/pr/main.go diff --git a/Makefile b/Makefile index 7d29deeb..adf42144 100644 --- a/Makefile +++ b/Makefile @@ -52,7 +52,12 @@ export REPO_ROOT := $(shell pwd) PROVIDER ?= github export SQ_PROVIDER_CONFIG_DIR ?= $(REPO_ROOT)/service/submitqueue/demo/provider/$(PROVIDER) -# Defaults for `make land` against the provider demo stack. +# Defaults for `make land` / `make demo-pr` against the provider demo stack. +DEMO_REPO ?= behinddwalls/sq-demo +COUNT ?= 3 +STACKED ?= false +LAND ?= true +WATCH ?= true QUEUE ?= demo-queue STRATEGY ?= SQUASH_REBASE GATEWAY_ADDR ?= localhost:8081 @@ -147,6 +152,16 @@ clean-proto: ## Clean generated proto files @rm -f $(foreach p,$(PROTO_PACKAGES),$(p)/protopb/*.pb.go $(p)/protopb/*.pb.yarpc.go) @echo "Proto clean complete!" +demo-pr: ## Create N PRs in the demo repo, enqueue each as it is created, and watch (COUNT=3; needs GITHUB_TOKEN) + @$(BAZEL) run //service/submitqueue/demo/pr -- \ + -repo $(DEMO_REPO) \ + -count $(COUNT) \ + -stacked=$(STACKED) \ + -gateway $(GATEWAY_ADDR) \ + -queue $(QUEUE) \ + -strategy $(STRATEGY) \ + -land=$(LAND) -watch=$(WATCH) + deps: tidy-go ## Download and tidy Go dependencies @echo "Dependencies installed!" diff --git a/doc/howto/PROVIDER-E2E.md b/doc/howto/PROVIDER-E2E.md index 4437349b..f044b869 100644 --- a/doc/howto/PROVIDER-E2E.md +++ b/doc/howto/PROVIDER-E2E.md @@ -25,7 +25,7 @@ For a **fine-grained** token, grant these repository permissions. Each is here b | Metadata | Read | mandatory on every fine-grained token; GitHub adds it for you | | Contents | Read and write | the git merger — clone, fetch, push to the target branch, and force-move each landed change's head branch | | Pull requests | Read | the change provider reads pull request metadata, and `land -pr` reads the head commit | -| Pull requests | Read **and write** | only for `make demo-prs`, which opens pull requests | +| Pull requests | Read **and write** | only for `make demo-pr`, which opens pull requests | | Actions | Read and write | only if you switch the build runner to GitHub Actions — dispatch a run, poll it, cancel it | A **classic** PAT needs `repo`, plus `workflow` if you use the GitHub Actions build runner. @@ -89,6 +89,35 @@ make land PRS="https://github.com///pull/1 \ The order of `PRS` is the stack order. All three land as **one push** to `main` — there is no window where a reader sees the stack half-applied — and all three show as merged. Tier 2 asserts the single-push property mechanically, by counting ref updates in the target's reflog. +## Simulating traffic + +Opening pull requests by hand gets old fast. `demo-pr` creates them, enqueues them, and shows you where each one is: + +```bash +make demo-pr # 3 independent PRs, each enqueued as it is created +make demo-pr COUNT=8 # more traffic +make demo-pr STACKED=true # one stack, enqueued as a single request +make demo-pr LAND=false # create only, print the land command +``` + +Each pull request is enqueued the moment it exists, so the queue is already working on the first while the last is still being opened. That overlap is the point: a queue holding one request at a time never batches, never analyzes a conflict against another batch, and never speculates. Nothing is awaited until every request is in. + +Then it watches all of them at once, redrawing a table as they move: + +``` + REQUEST CHANGES STATUS ELAPSED + ------------------------ ---------- ------------ ---------- + demo-queue/12 #31 landed 34s + demo-queue/13 #32 processing 31s + demo-queue/14 #33 batched 28s +``` + +`STACKED=true` is the exception to the overlap: one request carries the whole chain, so it can only go in once every pull request in it exists. That is the atomic-stack path — the whole set reaches `main` in a single push. + +It talks to GitHub over the REST API with the same `GITHUB_TOKEN`, so it needs no clone and no git binary. Each run tags its branches with a timestamp so repeated runs do not collide, and each change edits its own file so independent changes do not conflict by accident. + +The command exits non-zero if any request settles anywhere other than `landed`, so it works in a script. Piped to a file it prints a fresh table whenever something moves instead of redrawing in place. + ## Watching it work ```bash diff --git a/service/submitqueue/demo/pr/BUILD.bazel b/service/submitqueue/demo/pr/BUILD.bazel new file mode 100644 index 00000000..a889d597 --- /dev/null +++ b/service/submitqueue/demo/pr/BUILD.bazel @@ -0,0 +1,22 @@ +load("@rules_go//go:def.bzl", "go_binary", "go_library") + +go_library( + name = "go_default_library", + srcs = ["main.go"], + importpath = "github.com/uber/submitqueue/service/submitqueue/demo/pr", + visibility = ["//visibility:private"], + deps = [ + "//api/base/change/protopb:go_default_library", + "//api/base/mergestrategy/protopb:go_default_library", + "//api/submitqueue/gateway/protopb:go_default_library", + "//platform/base/change/github:go_default_library", + "@org_golang_google_grpc//:go_default_library", + "@org_golang_google_grpc//credentials/insecure:go_default_library", + ], +) + +go_binary( + name = "pr", + embed = [":go_default_library"], + visibility = ["//visibility:public"], +) diff --git a/service/submitqueue/demo/pr/main.go b/service/submitqueue/demo/pr/main.go new file mode 100644 index 00000000..f4244df7 --- /dev/null +++ b/service/submitqueue/demo/pr/main.go @@ -0,0 +1,555 @@ +// 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 pr populates a scratch repository with pull requests, enqueues them, +// and watches them move through the pipeline — so the demo stack can be +// exercised repeatedly without opening pull requests by hand. +// +// Nothing is awaited until the end, which is the point. Each pull request is +// enqueued the moment it is created, so the queue is already working on the +// first while the last is still being opened. 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 status table at the end watches all of them at once. +// +// Two shapes of change, because the pipeline treats them differently: +// +// - independent (default): each pull request 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 pull request 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. +// +// Everything goes through GitHub's REST API rather than a local clone, so the +// tool needs no checkout and no git binary — only GITHUB_TOKEN, the same +// credential the stack itself uses. +package main + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "flag" + "fmt" + "net/http" + "os" + "sort" + "strings" + "time" + + changepb "github.com/uber/submitqueue/api/base/change/protopb" + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + githubchange "github.com/uber/submitqueue/platform/base/change/github" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// pollInterval bounds how often the watcher re-reads every request's status. +const pollInterval = 2 * time.Second + +// terminalStatuses are the states a land request settles on, in the gateway's +// customer-facing vocabulary. +var terminalStatuses = map[string]bool{"landed": true, "error": true, "cancelled": true} + +func main() { + cfg := parseFlags() + if err := run(context.Background(), cfg); err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } +} + +// config is everything the run needs, resolved from flags and the environment. +type config struct { + repo string + base string + count int + stacked bool + prefix string + land bool + watch bool + gateway string + queue string + strategy string + token string + apiRoot string + host string +} + +func parseFlags() config { + var c config + flag.StringVar(&c.repo, "repo", "behinddwalls/sq-demo", "scratch repository as owner/name") + flag.StringVar(&c.base, "base", "main", "branch the changes target") + flag.IntVar(&c.count, "count", 3, "how many pull requests to create") + flag.BoolVar(&c.stacked, "stacked", false, "chain the pull requests and enqueue them as one stack") + flag.StringVar(&c.prefix, "prefix", "demo", "branch name prefix") + flag.BoolVar(&c.land, "land", true, "enqueue each pull request as it is created") + flag.BoolVar(&c.watch, "watch", true, "watch the requests until they all settle") + flag.StringVar(&c.gateway, "gateway", "localhost:8081", "gateway address") + flag.StringVar(&c.queue, "queue", "demo-queue", "queue to land on") + flag.StringVar(&c.strategy, "strategy", "SQUASH_REBASE", "merge strategy") + flag.Parse() + + c.token = os.Getenv("GITHUB_TOKEN") + c.apiRoot = "https://api.github.com" + c.host = "github.com" + return c +} + +func run(ctx context.Context, cfg config) error { + if cfg.token == "" { + return fmt.Errorf("GITHUB_TOKEN is not set; it is the same credential the stack uses") + } + if cfg.count < 1 { + return fmt.Errorf("-count must be at least 1") + } + owner, repo, ok := strings.Cut(cfg.repo, "/") + if !ok || owner == "" || repo == "" { + return fmt.Errorf("-repo %q must be owner/name", cfg.repo) + } + strategy, err := parseStrategy(cfg.strategy) + if err != nil { + return err + } + + gh := &githubClient{root: cfg.apiRoot, token: cfg.token, owner: owner, repo: repo} + baseSHA, err := gh.branchSHA(ctx, cfg.base) + if err != nil { + return fmt.Errorf("read %s: %w", cfg.base, err) + } + + var client pb.SubmitQueueGatewayClient + if cfg.land { + conn, err := grpc.NewClient(cfg.gateway, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return fmt.Errorf("connect to gateway %s: %w", cfg.gateway, err) + } + defer conn.Close() + client = pb.NewSubmitQueueGatewayClient(conn) + } + + // 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") + fmt.Printf("Creating %d pull request(s) in %s — %s\n\n", cfg.count, cfg.repo, shape(cfg)) + + created, requests, err := createAndEnqueue(ctx, gh, client, cfg, strategy, tag, baseSHA) + if err != nil { + return err + } + + if !cfg.land { + fmt.Printf("\nEnqueue them with:\n make land PRS=\"%s\"\n", strings.Join(urlsOf(created), " ")) + return nil + } + if !cfg.watch { + return nil + } + return watch(ctx, client, cfg.queue, requests) +} + +func shape(cfg config) string { + if cfg.stacked { + return "stacked, enqueued as one request once the chain exists" + } + return "independent, each enqueued as soon as it is created" +} + +// change is one pull request this run created. +type change struct { + number int + url string + branch string + // uri is the SubmitQueue change URI pinning the pull request to its head. + uri string +} + +func urlsOf(cs []change) []string { + out := make([]string, 0, len(cs)) + for _, c := range cs { + out = append(out, c.url) + } + return out +} + +// createAndEnqueue opens the pull requests and puts them on the queue. +// +// For independent changes the two steps interleave: each pull request 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. +// +// Each change edits its own file. 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. +func createAndEnqueue( + ctx context.Context, + gh *githubClient, + client pb.SubmitQueueGatewayClient, + cfg config, + strategy mergestrategypb.Strategy, + tag, baseSHA string, +) ([]change, []*request, error) { + created := make([]change, 0, cfg.count) + requests := make([]*request, 0, cfg.count) + + parentBranch, parentSHA := cfg.base, baseSHA + for i := 1; i <= cfg.count; i++ { + branch := fmt.Sprintf("%s/%s/%d", cfg.prefix, tag, i) + if err := gh.createBranch(ctx, branch, parentSHA); err != nil { + return nil, nil, fmt.Errorf("create branch %s: %w", branch, err) + } + + path := fmt.Sprintf("demo/%s-%d.txt", tag, i) + body := fmt.Sprintf("change %d of run %s\n", i, tag) + headSHA, err := gh.commitFile(ctx, branch, path, body, fmt.Sprintf("demo change %d (run %s)", i, tag)) + if err != nil { + return nil, nil, fmt.Errorf("commit to %s: %w", branch, err) + } + + number, url, err := gh.openPR(ctx, fmt.Sprintf("demo change %d (run %s)", i, tag), branch, parentBranch) + if err != nil { + return nil, nil, fmt.Errorf("open pull request for %s: %w", branch, err) + } + + c := change{ + number: number, url: url, branch: branch, + uri: githubchange.ChangeID{ + Scheme: "github", Host: cfg.host, Org: gh.owner, Repo: gh.repo, + PRNumber: number, HeadCommitSHA: headSHA, + }.String(), + } + created = append(created, c) + fmt.Printf(" created #%-5d %s\n", number, url) + + if cfg.stacked { + // The next change builds on this one, so it sees this change's + // content and its pull request is based on this branch. + parentBranch, parentSHA = branch, headSHA + continue + } + if !cfg.land { + continue + } + req, err := enqueue(ctx, client, cfg, strategy, []change{c}) + if err != nil { + return nil, nil, err + } + requests = append(requests, req) + fmt.Printf(" enqueued %-6s %s\n", req.changes, req.sqid) + } + + // The stack goes in as one request, which is only possible now that every + // change in it exists. + if cfg.stacked && cfg.land { + req, err := enqueue(ctx, client, cfg, strategy, created) + if err != nil { + return nil, nil, err + } + requests = append(requests, req) + fmt.Printf(" enqueued %-6s %s\n", req.changes, req.sqid) + } + return created, requests, nil +} + +// request is one land request being watched. +type request struct { + sqid string + // changes names the pull requests the request carries, for the table. + changes string + // submitted is when the gateway accepted it, so elapsed time is per request + // rather than per run. + submitted time.Time + + status string + note string + done bool +} + +// enqueue submits one land request carrying the given changes, in order. +func enqueue( + ctx context.Context, + client pb.SubmitQueueGatewayClient, + cfg config, + strategy mergestrategypb.Strategy, + changes []change, +) (*request, error) { + uris := make([]string, 0, len(changes)) + labels := make([]string, 0, len(changes)) + for _, c := range changes { + uris = append(uris, c.uri) + labels = append(labels, fmt.Sprintf("#%d", c.number)) + } + + resp, err := client.Land(ctx, &pb.LandRequest{ + Queue: cfg.queue, + Change: &changepb.Change{Uris: uris}, + Strategy: strategy, + }) + if err != nil { + return nil, fmt.Errorf("land %s failed: %w", strings.Join(labels, ","), err) + } + return &request{ + sqid: resp.Sqid, + changes: strings.Join(labels, ","), + submitted: time.Now(), + status: "submitted", + }, nil +} + +// watch polls every request until they have all settled, redrawing a table as +// statuses change so the pipeline is visible while it runs. +func watch(ctx context.Context, client pb.SubmitQueueGatewayClient, queue string, requests []*request) error { + if len(requests) == 0 { + return nil + } + fmt.Printf("\n") + r := newRenderer(len(requests) + 2) + r.draw(requests) + + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + + changed, pending := refresh(ctx, client, queue, requests) + if changed || r.inPlace { + // On a terminal the elapsed column moves every tick, so redraw + // regardless; piped output only reprints when something happened. + r.draw(requests) + } + if pending == 0 { + return summarize(requests) + } + } +} + +// refresh re-reads every unsettled request, reporting whether anything moved +// and how many are still in flight. +func refresh(ctx context.Context, client pb.SubmitQueueGatewayClient, queue string, requests []*request) (changed bool, pending int) { + for _, req := range requests { + if req.done { + continue + } + resp, err := client.GetRequestSummaryByID(ctx, &pb.GetRequestSummaryByIDRequest{Sqid: req.sqid, Queue: queue}) + if err != nil || resp.Request == nil { + // A summary that is not readable yet is normal right after Land; + // the next tick picks it up. + pending++ + continue + } + if resp.Request.Status != req.status || resp.Request.LastError != req.note { + changed = true + } + req.status = resp.Request.Status + req.note = resp.Request.LastError + req.done = terminalStatuses[req.status] + if !req.done { + pending++ + } + } + return changed, pending +} + +// summarize fails the run if anything did not land, so a scripted demo notices. +func summarize(requests []*request) error { + var failed []string + for _, req := range requests { + if req.status != "landed" { + failed = append(failed, fmt.Sprintf("%s=%s", req.sqid, req.status)) + } + } + if len(failed) > 0 { + sort.Strings(failed) + return fmt.Errorf("%d of %d request(s) did not land: %s", len(failed), len(requests), strings.Join(failed, ", ")) + } + fmt.Printf("\nAll %d request(s) landed.\n", len(requests)) + return nil +} + +// renderer draws the status table, redrawing in place on a terminal and +// appending a fresh block otherwise, so piping the output to a file stays +// readable instead of filling with escape codes. +type renderer struct { + lines int + inPlace bool + drawn bool +} + +func newRenderer(lines int) *renderer { + info, err := os.Stdout.Stat() + tty := err == nil && info.Mode()&os.ModeCharDevice != 0 + return &renderer{lines: lines, inPlace: tty} +} + +func (r *renderer) draw(requests []*request) { + if r.inPlace && r.drawn { + // Move back over the previous table and overwrite it in place. + fmt.Printf("\033[%dA", r.lines) + } + r.line(fmt.Sprintf(" %-24s %-10s %-12s %8s", "REQUEST", "CHANGES", "STATUS", "ELAPSED")) + r.line(" " + strings.Repeat("-", 24) + " " + strings.Repeat("-", 10) + " " + strings.Repeat("-", 12) + " " + strings.Repeat("-", 8)) + for _, req := range requests { + line := fmt.Sprintf(" %-24s %-10s %-12s %7ds", req.sqid, req.changes, req.status, int(time.Since(req.submitted).Seconds())) + if req.note != "" { + line += " " + truncate(req.note, 60) + } + r.line(line) + } + r.drawn = true +} + +// line writes one table row, clearing whatever the previous draw left on it +// when redrawing in place. +func (r *renderer) line(s string) { + if r.inPlace { + fmt.Printf("\033[K%s\n", s) + return + } + fmt.Println(s) +} + +func truncate(s string, n int) string { + s = strings.ReplaceAll(s, "\n", " ") + if len(s) <= n { + return s + } + return s[:n-1] + "…" +} + +func parseStrategy(name string) (mergestrategypb.Strategy, error) { + switch strings.ToUpper(strings.TrimSpace(name)) { + case "", "DEFAULT": + return mergestrategypb.Strategy_DEFAULT, nil + case "REBASE": + return mergestrategypb.Strategy_REBASE, nil + case "SQUASH_REBASE": + return mergestrategypb.Strategy_SQUASH_REBASE, nil + case "MERGE": + return mergestrategypb.Strategy_MERGE, nil + case "PROMOTE": + return mergestrategypb.Strategy_PROMOTE, nil + default: + return mergestrategypb.Strategy_DEFAULT, fmt.Errorf("unknown strategy %q", name) + } +} + +// githubClient is the slice of GitHub's REST API this tool needs: read a +// branch, create a branch, commit a file, open a pull request. +type githubClient struct { + root string + token string + owner string + repo string +} + +func (g *githubClient) branchSHA(ctx context.Context, branch string) (string, error) { + var out struct { + Object struct { + SHA string `json:"sha"` + } `json:"object"` + } + if err := g.do(ctx, http.MethodGet, "/git/ref/heads/"+branch, nil, &out); err != nil { + return "", err + } + return out.Object.SHA, nil +} + +func (g *githubClient) createBranch(ctx context.Context, branch, fromSHA string) error { + return g.do(ctx, http.MethodPost, "/git/refs", + map[string]string{"ref": "refs/heads/" + branch, "sha": fromSHA}, nil) +} + +// commitFile writes a file on a branch and returns the resulting commit SHA — +// the commit a change URI pins the pull request to. +func (g *githubClient) commitFile(ctx context.Context, branch, path, content, message string) (string, error) { + body := map[string]string{ + "message": message, + "content": base64.StdEncoding.EncodeToString([]byte(content)), + "branch": branch, + } + var out struct { + Commit struct { + SHA string `json:"sha"` + } `json:"commit"` + } + if err := g.do(ctx, http.MethodPut, "/contents/"+path, body, &out); err != nil { + return "", err + } + return out.Commit.SHA, nil +} + +func (g *githubClient) openPR(ctx context.Context, title, head, base string) (int, string, error) { + body := map[string]string{"title": title, "head": head, "base": base, "body": "Opened by service/submitqueue/demo/pr."} + var out struct { + Number int `json:"number"` + HTMLURL string `json:"html_url"` + } + if err := g.do(ctx, http.MethodPost, "/pulls", body, &out); err != nil { + return 0, "", err + } + return out.Number, out.HTMLURL, nil +} + +// do issues one authenticated request against the repository, decoding into out +// when it is non-nil. +func (g *githubClient) do(ctx context.Context, method, path string, body any, out any) error { + endpoint := fmt.Sprintf("%s/repos/%s/%s%s", g.root, g.owner, g.repo, path) + + var payload []byte + if body != nil { + var err error + if payload, err = json.Marshal(body); err != nil { + return fmt.Errorf("encode request for %s: %w", endpoint, err) + } + } + + req, err := http.NewRequestWithContext(ctx, method, endpoint, bytes.NewReader(payload)) + if err != nil { + return fmt.Errorf("build request for %s: %w", endpoint, err) + } + req.Header.Set("Accept", "application/vnd.github+json") + req.Header.Set("Authorization", "Bearer "+g.token) + if payload != nil { + req.Header.Set("Content-Type", "application/json") + } + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return fmt.Errorf("%s %s: %w", method, endpoint, err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + var detail bytes.Buffer + _, _ = detail.ReadFrom(resp.Body) + return fmt.Errorf("%s %s returned %s: %s", method, endpoint, resp.Status, strings.TrimSpace(detail.String())) + } + if out == nil { + return nil + } + if err := json.NewDecoder(resp.Body).Decode(out); err != nil { + return fmt.Errorf("decode response from %s: %w", endpoint, err) + } + return nil +} diff --git a/service/submitqueue/demo/provider/github/profiles.yaml b/service/submitqueue/demo/provider/github/profiles.yaml index e08887de..1cf4cf56 100644 --- a/service/submitqueue/demo/provider/github/profiles.yaml +++ b/service/submitqueue/demo/provider/github/profiles.yaml @@ -27,7 +27,12 @@ queues: # Every build succeeds instantly, so a land completes in seconds and the # demo exercises the merge rather than waiting on CI. - buildRunner: {type: fake} + buildRunner: + type: githubactions + owner: behinddwalls + repo: sq-demo + workflow: ci.yml # file name or numeric workflow id + ref: main # the branch the workflow definition is read from # # To run real CI instead, replace the line above with the block below. It # needs a workflow in the target repository that is triggerable by From ce1d51a3f7dda01cb3686b7f7eba3f5945eddbe2 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Sat, 8 Aug 2026 08:20:49 -0700 Subject: [PATCH 2/5] feat(demo): live table with the full state trail per request MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The table only appeared once every pull request existed, and then showed a single word per row. Both are the wrong way round for the thing the demo is meant to make visible. Creation scrolled past as loose `created` / `enqueued` lines while the table waited, so the most interesting stretch — early requests moving through the queue while later ones are still being opened — had nothing to look at. At `COUNT=20` that is most of the run. The single word came from polling `GetRequestSummaryByID` every two seconds, which does not just fail to show where a request has been: it misses transitions outright. A request that batches and speculates between two ticks shows neither. That a request went `accepted → started → batched → speculating → landed` is the story worth telling, and sampling cannot tell it. ### What? The rows are allocated before anything is created — one per land request, or a single row for a stack — so the table is complete from the first draw and only ever fills in. Whatever is happening right now is one line underneath it rather than output that pushes the table away. Statuses come from `GetRequestHistoryByID`, which returns every recorded transition, so the trail is complete rather than sampled. It replaces the summary call rather than adding to it, and current status and last error fall out of the same response. A `tracker` owns the rows behind a mutex and polls on its own clock, concurrently with creation. Polling only after the last pull request existed is what left the trail empty for the whole creation phase. `CHANGES` links to the pull request: an OSC 8 hyperlink on a terminal, the full URL when piped. `ELAPSED` starts when the gateway accepts the request and freezes when it settles, so a finished row keeps the time it took. An error is wrapped onto continuation lines indented under its row rather than truncated into a cell. The tail of a pipeline error is usually the part that says what went wrong. Two latent bugs in the renderer are fixed along the way. The cursor-up distance was fixed at construction and would have drifted once the table gained a status line; it now counts the lines it actually emitted. And OSC 8 escapes carry many bytes at zero display width, so padding and the truncation budget both had to stop measuring them — measuring the rendered prefix cut a 51-character trail down to 24 with 85 columns free. ## Test Plan ✅ `bazel test //service/submitqueue/demo/pr:go_default_test`, covering trail derivation, the frozen clock, visible-width padding, wrapping, and the line accounting the redraw depends on. ✅ Same suite under `--@rules_go//go/config:race`, which is what the concurrent poll-and-update test is for. ✅ `make test` — 98 targets. The alignment test was checked against the bug it guards: reinstating the rune-count budget fails `TestRowLineAlignment`. Running it end to end needs a scratch repository and a token, so it is exercised the same way the rest of the provider path is — manually. See `doc/howto/PROVIDER-E2E.md`. --- doc/howto/PROVIDER-E2E.md | 20 +- service/submitqueue/demo/pr/BUILD.bazel | 15 +- service/submitqueue/demo/pr/main.go | 681 +++++++++++++++++++---- service/submitqueue/demo/pr/main_test.go | 658 ++++++++++++++++++++++ 4 files changed, 1251 insertions(+), 123 deletions(-) create mode 100644 service/submitqueue/demo/pr/main_test.go diff --git a/doc/howto/PROVIDER-E2E.md b/doc/howto/PROVIDER-E2E.md index f044b869..26b5a8ee 100644 --- a/doc/howto/PROVIDER-E2E.md +++ b/doc/howto/PROVIDER-E2E.md @@ -102,21 +102,25 @@ make demo-pr LAND=false # create only, print the land command Each pull request is enqueued the moment it exists, so the queue is already working on the first while the last is still being opened. That overlap is the point: a queue holding one request at a time never batches, never analyzes a conflict against another batch, and never speculates. Nothing is awaited until every request is in. -Then it watches all of them at once, redrawing a table as they move: +The table is there from the start — one row per land request, drawn before the first pull request exists and filled in as the run proceeds. Whatever is happening right now is a single line underneath it, so creating and enqueuing does not scroll the table away: ``` - REQUEST CHANGES STATUS ELAPSED - ------------------------ ---------- ------------ ---------- - demo-queue/12 #31 landed 34s - demo-queue/13 #32 processing 31s - demo-queue/14 #33 batched 28s + REQUEST CHANGES ELAPSED STAGE + ───────────── ─────── ─────── ─────────────────────────────────────────────────── + demo-queue/12 #31 34s accepted → started → batched → speculating → landed + demo-queue/13 #32 31s accepted → started → batched → speculating + demo-queue/14 #33 28s accepted → started + + ▸ 1 of 3 settled ``` -`STACKED=true` is the exception to the overlap: one request carries the whole chain, so it can only go in once every pull request in it exists. That is the atomic-stack path — the whole set reaches `main` in a single push. +Each row shows the states its request passed through, not just the one it is in. That comes from the gateway's history API rather than from sampling the current status, so a request that batches and speculates between two ticks still shows both. `CHANGES` links to the pull request: on a terminal `#31` is clickable, and in a redirected run it is written out as a full URL instead. `ELAPSED` runs from the moment the gateway accepted the request and stops when it settles, so a finished row keeps the time it took rather than counting on. + +`STACKED=true` is the exception to the overlap: one request carries the whole chain, so it can only go in once every pull request in it exists. That is the atomic-stack path — the whole set reaches `main` in a single push, and the table shows it as the single row it is. It talks to GitHub over the REST API with the same `GITHUB_TOKEN`, so it needs no clone and no git binary. Each run tags its branches with a timestamp so repeated runs do not collide, and each change edits its own file so independent changes do not conflict by accident. -The command exits non-zero if any request settles anywhere other than `landed`, so it works in a script. Piped to a file it prints a fresh table whenever something moves instead of redrawing in place. +The command exits non-zero if any request settles anywhere other than `landed`, so it works in a script. Piped to a file it prints a fresh table whenever a request moves — and not when only the clock did — instead of redrawing in place. ## Watching it work diff --git a/service/submitqueue/demo/pr/BUILD.bazel b/service/submitqueue/demo/pr/BUILD.bazel index a889d597..28729f35 100644 --- a/service/submitqueue/demo/pr/BUILD.bazel +++ b/service/submitqueue/demo/pr/BUILD.bazel @@ -1,4 +1,4 @@ -load("@rules_go//go:def.bzl", "go_binary", "go_library") +load("@rules_go//go:def.bzl", "go_binary", "go_library", "go_test") go_library( name = "go_default_library", @@ -10,6 +10,7 @@ go_library( "//api/base/mergestrategy/protopb:go_default_library", "//api/submitqueue/gateway/protopb:go_default_library", "//platform/base/change/github:go_default_library", + "//submitqueue/entity:go_default_library", "@org_golang_google_grpc//:go_default_library", "@org_golang_google_grpc//credentials/insecure:go_default_library", ], @@ -20,3 +21,15 @@ go_binary( embed = [":go_default_library"], visibility = ["//visibility:public"], ) + +go_test( + name = "go_default_test", + srcs = ["main_test.go"], + embed = [":go_default_library"], + deps = [ + "//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//:go_default_library", + ], +) diff --git a/service/submitqueue/demo/pr/main.go b/service/submitqueue/demo/pr/main.go index f4244df7..228c26ec 100644 --- a/service/submitqueue/demo/pr/main.go +++ b/service/submitqueue/demo/pr/main.go @@ -21,7 +21,7 @@ // first while the last is still being opened. 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 status table at the end watches all of them at once. +// overlap. The table watches all of them at once. // // Two shapes of change, because the pipeline treats them differently: // @@ -32,6 +32,13 @@ // 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 pull request 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. +// // Everything goes through GitHub's REST API rather than a local clone, so the // tool needs no checkout and no git binary — only GITHUB_TOKEN, the same // credential the stack itself uses. @@ -48,22 +55,43 @@ import ( "os" "sort" "strings" + "sync" "time" + "unicode/utf8" changepb "github.com/uber/submitqueue/api/base/change/protopb" mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" githubchange "github.com/uber/submitqueue/platform/base/change/github" + "github.com/uber/submitqueue/submitqueue/entity" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" ) -// pollInterval bounds how often the watcher re-reads every request's status. -const pollInterval = 2 * time.Second +const ( + // pollInterval bounds how often the watcher re-reads every request's history. + pollInterval = 2 * time.Second + + // maxLineWidth caps a redrawn line. A line that wraps occupies two physical + // rows, which permanently desyncs the cursor arithmetic the in-place redraw + // depends on; capping is cheaper than asking the terminal how wide it is. + maxLineWidth = 120 + + // absent is what a cell shows before there is anything to put in it. + absent = "—" + + // minNoteWidth keeps a wrapped error readable even when the columns before + // it have eaten most of the line. + minNoteWidth = 40 +) -// terminalStatuses are the states a land request settles on, in the gateway's -// customer-facing vocabulary. -var terminalStatuses = map[string]bool{"landed": true, "error": true, "cancelled": true} +// terminalStatuses are the states a land request settles on. They are keyed off +// the gateway's own vocabulary so this tool cannot quietly drift from it. +var terminalStatuses = map[string]bool{ + string(entity.RequestStatusLanded): true, + string(entity.RequestStatusError): true, + string(entity.RequestStatusCancelled): true, +} func main() { cfg := parseFlags() @@ -147,19 +175,45 @@ func run(ctx context.Context, cfg config) error { tag := time.Now().Format("0102-150405") fmt.Printf("Creating %d pull request(s) in %s — %s\n\n", cfg.count, cfg.repo, shape(cfg)) - created, requests, err := createAndEnqueue(ctx, gh, client, cfg, strategy, tag, baseSHA) + // Every row is known before anything is created: one per pull request, 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 := newTracker(newRows(cfg)) + t.note("starting") + + // Statuses are read on their own clock, concurrently with creation. A run + // that only started polling once every pull request 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, client, cfg.queue) + } + + created, err := createAndEnqueue(ctx, gh, client, cfg, strategy, tag, baseSHA, t) if err != nil { return err } + t.seal() if !cfg.land { + t.note("created %d pull request(s), not enqueued", len(created)) fmt.Printf("\nEnqueue them with:\n make land PRS=\"%s\"\n", strings.Join(urlsOf(created), " ")) return nil } if !cfg.watch { + t.note("enqueued, not watching") return nil } - return watch(ctx, client, cfg.queue, requests) + + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.settled: + } + return t.conclude() } func shape(cfg config) string { @@ -186,7 +240,70 @@ func urlsOf(cs []change) []string { return out } -// createAndEnqueue opens the pull requests and puts them on the queue. +// row is one land request and everything shown about it. A row exists from the +// first draw, before the pull request it will carry has been opened, so the +// table never changes shape while the run is in progress. +type row struct { + // changes are the pull requests the request carries, in caller order. A + // stacked run puts every change on one row. + changes []change + + // sqid is empty until the gateway accepts the request. + sqid string + // submitted is when the gateway accepted it, and starts the elapsed clock. + submitted time.Time + // settled is when a terminal status was first observed, and stops it. + settled time.Time + + // trail is the ordered set of statuses the gateway recorded for the request. + trail []string + status string + note string + done bool +} + +// newRows allocates the rows the run will fill: one per pull request, or a +// single row for a stack, which lands as one request. +func newRows(cfg config) []*row { + n := cfg.count + if cfg.stacked { + n = 1 + } + rows := make([]*row, n) + for i := range rows { + rows[i] = &row{} + } + return rows +} + +// elapsed is how long the request has been with the queue: absent until it is +// accepted, running while it is in flight, and frozen once it settles. +func (rw *row) elapsed() string { + if rw.submitted.IsZero() { + return absent + } + end := time.Now() + if !rw.settled.IsZero() { + end = rw.settled + } + return fmt.Sprintf("%ds", int(end.Sub(rw.submitted).Seconds())) +} + +// stage is the path the request has taken, as the gateway recorded it. The +// waiting marker covers the gap between acceptance and the first recorded +// event, so an accepted request is never shown as though nothing happened. +func (rw *row) stage() string { + if len(rw.trail) > 0 { + return strings.Join(rw.trail, " → ") + } + if rw.sqid != "" { + return "…" + } + return absent +} + +// createAndEnqueue opens the pull requests 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 pull request is // enqueued the moment it exists, so the queue is already working on it while @@ -205,27 +322,36 @@ func createAndEnqueue( cfg config, strategy mergestrategypb.Strategy, tag, baseSHA string, -) ([]change, []*request, error) { + t *tracker, +) ([]change, error) { created := make([]change, 0, cfg.count) - requests := make([]*request, 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. + target := t.rows[0] + if !cfg.stacked { + target = t.rows[i-1] + } + branch := fmt.Sprintf("%s/%s/%d", cfg.prefix, tag, i) + t.note("creating branch %s", branch) if err := gh.createBranch(ctx, branch, parentSHA); err != nil { - return nil, nil, fmt.Errorf("create branch %s: %w", branch, err) + return nil, fmt.Errorf("create branch %s: %w", branch, err) } path := fmt.Sprintf("demo/%s-%d.txt", tag, i) body := fmt.Sprintf("change %d of run %s\n", i, tag) + t.note("committing %s", path) headSHA, err := gh.commitFile(ctx, branch, path, body, fmt.Sprintf("demo change %d (run %s)", i, tag)) if err != nil { - return nil, nil, fmt.Errorf("commit to %s: %w", branch, err) + return nil, fmt.Errorf("commit to %s: %w", branch, err) } + t.note("opening pull request for %s", branch) number, url, err := gh.openPR(ctx, fmt.Sprintf("demo change %d (run %s)", i, tag), branch, parentBranch) if err != nil { - return nil, nil, fmt.Errorf("open pull request for %s: %w", branch, err) + return nil, fmt.Errorf("open pull request for %s: %w", branch, err) } c := change{ @@ -236,7 +362,7 @@ func createAndEnqueue( }.String(), } created = append(created, c) - fmt.Printf(" created #%-5d %s\n", number, url) + t.update(func() { target.changes = append(target.changes, c) }) if cfg.stacked { // The next change builds on this one, so it sees this change's @@ -247,54 +373,39 @@ func createAndEnqueue( if !cfg.land { continue } - req, err := enqueue(ctx, client, cfg, strategy, []change{c}) + t.note("enqueuing #%d", number) + sqid, err := enqueue(ctx, client, cfg, strategy, []change{c}) if err != nil { - return nil, nil, err + return nil, err } - requests = append(requests, req) - fmt.Printf(" enqueued %-6s %s\n", req.changes, req.sqid) + t.update(func() { target.sqid, target.submitted = sqid, time.Now() }) } // The stack goes in as one request, which is only possible now that every // change in it exists. if cfg.stacked && cfg.land { - req, err := enqueue(ctx, client, cfg, strategy, created) + t.note("enqueuing the stack") + sqid, err := enqueue(ctx, client, cfg, strategy, created) if err != nil { - return nil, nil, err + return nil, err } - requests = append(requests, req) - fmt.Printf(" enqueued %-6s %s\n", req.changes, req.sqid) + t.update(func() { t.rows[0].sqid, t.rows[0].submitted = sqid, time.Now() }) } - return created, requests, nil + return created, nil } -// request is one land request being watched. -type request struct { - sqid string - // changes names the pull requests the request carries, for the table. - changes string - // submitted is when the gateway accepted it, so elapsed time is per request - // rather than per run. - submitted time.Time - - status string - note string - done bool -} - -// enqueue submits one land request carrying the given changes, in order. +// enqueue submits one land request carrying the given changes, in order, and +// returns the identifier the gateway assigned it. func enqueue( ctx context.Context, client pb.SubmitQueueGatewayClient, cfg config, strategy mergestrategypb.Strategy, changes []change, -) (*request, error) { +) (string, error) { uris := make([]string, 0, len(changes)) - labels := make([]string, 0, len(changes)) for _, c := range changes { uris = append(uris, c.uri) - labels = append(labels, fmt.Sprintf("#%d", c.number)) } resp, err := client.Land(ctx, &pb.LandRequest{ @@ -303,138 +414,480 @@ func enqueue( Strategy: strategy, }) if err != nil { - return nil, fmt.Errorf("land %s failed: %w", strings.Join(labels, ","), err) + return "", fmt.Errorf("land %s failed: %w", labelsOf(changes), err) } - return &request{ - sqid: resp.Sqid, - changes: strings.Join(labels, ","), - submitted: time.Now(), - status: "submitted", - }, nil + return resp.Sqid, nil } -// watch polls every request until they have all settled, redrawing a table as -// statuses change so the pipeline is visible while it runs. -func watch(ctx context.Context, client pb.SubmitQueueGatewayClient, queue string, requests []*request) error { - if len(requests) == 0 { - return nil +// labelsOf names the pull requests on a row the way they are shown. +func labelsOf(cs []change) string { + labels := make([]string, 0, len(cs)) + for _, c := range cs { + labels = append(labels, fmt.Sprintf("#%d", c.number)) + } + return strings.Join(labels, ",") +} + +// tracker owns the rows for the duration of the run. Two goroutines touch +// them — creation fills in pull requests and identifiers, polling fills in +// statuses — and both draw the same table, so the mutex is what keeps one from +// redrawing halfway through the other's update. +type tracker struct { + mu sync.Mutex + rows []*row + r *renderer + status string + // sealed records that every request that will be enqueued has been. Without + // it, polling would find nothing outstanding before creation had begun and + // call the run finished. + sealed bool + + // settled closes once every request has reached a terminal status. + settled chan struct{} + once sync.Once +} + +func newTracker(rows []*row) *tracker { + return &tracker{rows: rows, r: newRenderer(), settled: make(chan struct{})} +} + +// note replaces the line under the table and redraws. +func (t *tracker) note(format string, args ...any) { + t.mu.Lock() + defer t.mu.Unlock() + t.status = fmt.Sprintf(format, args...) + t.r.draw(t.rows, t.status) +} + +// update applies a change to the rows and redraws with it. +func (t *tracker) update(fn func()) { + t.mu.Lock() + defer t.mu.Unlock() + fn() + t.r.draw(t.rows, t.status) +} + +// conclude draws the verdict and reports whether everything landed. It reads +// the rows under the lock because a poll may still be applying its last round. +func (t *tracker) conclude() error { + t.mu.Lock() + defer t.mu.Unlock() + t.status = outcome(t.rows) + t.r.draw(t.rows, t.status) + return summarize(t.rows) +} + +// seal declares that nothing further will be enqueued, which is what lets an +// otherwise-finished run conclude. +func (t *tracker) seal() { + t.mu.Lock() + defer t.mu.Unlock() + t.sealed = true + t.signalLocked() +} + +// signalLocked closes settled once there is nothing left to wait for. +func (t *tracker) signalLocked() { + if !t.sealed { + return + } + for _, rw := range t.rows { + if rw.sqid == "" || !rw.done { + return + } } - fmt.Printf("\n") - r := newRenderer(len(requests) + 2) - r.draw(requests) + t.once.Do(func() { close(t.settled) }) +} +// poll re-reads statuses until the run finishes or the context ends. +func (t *tracker) poll(ctx context.Context, client pb.SubmitQueueGatewayClient, queue string) { ticker := time.NewTicker(pollInterval) defer ticker.Stop() for { select { case <-ctx.Done(): - return ctx.Err() + return + case <-t.settled: + return case <-ticker.C: } + t.refresh(ctx, client, queue) + } +} + +// refresh re-reads every request that has been accepted but has not settled. +// +// The reads happen outside the lock. Holding it across a round of RPCs would +// stall creation behind the network, and creation racing ahead is the whole +// point of enqueuing each pull request the moment it exists. +func (t *tracker) refresh(ctx context.Context, client pb.SubmitQueueGatewayClient, queue string) { + t.mu.Lock() + outstanding := make([]*row, 0, len(t.rows)) + for _, rw := range t.rows { + if rw.sqid != "" && !rw.done { + outstanding = append(outstanding, rw) + } + } + total := len(t.rows) + t.mu.Unlock() + + type reading struct { + rw *row + trail []string + status string + note string + } + 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 := client.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. + continue + } + trail, status, note := digest(resp.Events) + readings = append(readings, reading{rw: rw, trail: trail, status: status, note: note}) + } - changed, pending := refresh(ctx, client, queue, requests) - if changed || r.inPlace { - // On a terminal the elapsed column moves every tick, so redraw - // regardless; piped output only reprints when something happened. - r.draw(requests) + t.mu.Lock() + defer t.mu.Unlock() + + settled := 0 + for _, got := range readings { + 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 + // the elapsed column is measured end to end against one clock. + got.rw.done, got.rw.settled = true, time.Now() } - if pending == 0 { - return summarize(requests) + } + for _, rw := range t.rows { + if rw.done { + settled++ } } + + t.status = fmt.Sprintf("%d of %d settled", settled, total) + t.r.draw(t.rows, t.status) + t.signalLocked() } -// refresh re-reads every unsettled request, reporting whether anything moved -// and how many are still in flight. -func refresh(ctx context.Context, client pb.SubmitQueueGatewayClient, queue string, requests []*request) (changed bool, pending int) { - for _, req := range requests { - if req.done { +// digest reduces a request's recorded history to the trail worth showing, the +// status it currently holds, and the error the latest event carried. A status +// recorded more than once in a row is one step in the trail, not several. +func digest(events []*pb.HistoryEvent) (trail []string, status, note string) { + if len(events) == 0 { + return nil, "", "" + } + for _, e := range events { + if e == nil || e.Status == "" { continue } - resp, err := client.GetRequestSummaryByID(ctx, &pb.GetRequestSummaryByIDRequest{Sqid: req.sqid, Queue: queue}) - if err != nil || resp.Request == nil { - // A summary that is not readable yet is normal right after Land; - // the next tick picks it up. - pending++ + if len(trail) > 0 && trail[len(trail)-1] == e.Status { continue } - if resp.Request.Status != req.status || resp.Request.LastError != req.note { - changed = true - } - req.status = resp.Request.Status - req.note = resp.Request.LastError - req.done = terminalStatuses[req.status] - if !req.done { - pending++ + trail = append(trail, e.Status) + } + if last := events[len(events)-1]; last != nil { + status, note = last.Status, last.LastError + } + if status == "" && len(trail) > 0 { + status = trail[len(trail)-1] + } + return trail, status, note +} + +// outcome is the one-line verdict shown under the finished table. +func outcome(rows []*row) string { + landed := 0 + for _, rw := range rows { + if rw.status == string(entity.RequestStatusLanded) { + landed++ } } - return changed, pending + if landed == len(rows) { + return fmt.Sprintf("all %d request(s) landed", len(rows)) + } + return fmt.Sprintf("%d of %d request(s) did not land", len(rows)-landed, len(rows)) } // summarize fails the run if anything did not land, so a scripted demo notices. -func summarize(requests []*request) error { +func summarize(rows []*row) error { var failed []string - for _, req := range requests { - if req.status != "landed" { - failed = append(failed, fmt.Sprintf("%s=%s", req.sqid, req.status)) + for _, rw := range rows { + if rw.status != string(entity.RequestStatusLanded) { + failed = append(failed, fmt.Sprintf("%s=%s", rw.sqid, rw.status)) } } if len(failed) > 0 { sort.Strings(failed) - return fmt.Errorf("%d of %d request(s) did not land: %s", len(failed), len(requests), strings.Join(failed, ", ")) + return fmt.Errorf("%d of %d request(s) did not land: %s", len(failed), len(rows), strings.Join(failed, ", ")) } - fmt.Printf("\nAll %d request(s) landed.\n", len(requests)) return nil } // renderer draws the status table, redrawing in place on a terminal and // appending a fresh block otherwise, so piping the output to a file stays // readable instead of filling with escape codes. +// +// Column widths only ever grow, so a value that turns out to be wider than the +// header does not make the table jitter as rows fill in. type renderer struct { - lines int inPlace bool - drawn bool + + wRequest int + wChanges int + wElapsed int + wStage int + + // lastLines is how many lines the previous draw actually emitted, which is + // how far the cursor has to move back to overwrite them. + lastLines int + drawn bool + + // lastBody is the signature of the previous table, so piped output can skip + // a redundant reprint when a step moved but the table did not. + lastBody string } -func newRenderer(lines int) *renderer { +func newRenderer() *renderer { info, err := os.Stdout.Stat() tty := err == nil && info.Mode()&os.ModeCharDevice != 0 - return &renderer{lines: lines, inPlace: tty} + return &renderer{ + inPlace: tty, + wRequest: len("REQUEST"), + wChanges: len("CHANGES"), + wElapsed: len("ELAPSED"), + wStage: len("STAGE"), + } } -func (r *renderer) draw(requests []*request) { - if r.inPlace && r.drawn { - // Move back over the previous table and overwrite it in place. - fmt.Printf("\033[%dA", r.lines) - } - r.line(fmt.Sprintf(" %-24s %-10s %-12s %8s", "REQUEST", "CHANGES", "STATUS", "ELAPSED")) - r.line(" " + strings.Repeat("-", 24) + " " + strings.Repeat("-", 10) + " " + strings.Repeat("-", 12) + " " + strings.Repeat("-", 8)) - for _, req := range requests { - line := fmt.Sprintf(" %-24s %-10s %-12s %7ds", req.sqid, req.changes, req.status, int(time.Since(req.submitted).Seconds())) - if req.note != "" { - line += " " + truncate(req.note, 60) +func (r *renderer) draw(rows []*row, status string) { + body := r.body(rows) + + if !r.inPlace { + sig := signature(rows) + if sig == r.lastBody { + // Nothing in the table moved; the step that prompted this draw is a + // terminal affordance and has no place in a log. + return } - r.line(line) + r.lastBody = sig + fmt.Println(strings.Join(body, "\n")) + fmt.Println() + return } + + if r.drawn { + fmt.Printf("\033[%dA", r.lastLines) + } + for _, line := range body { + fmt.Printf("\033[K%s\n", line) + } + fmt.Printf("\033[K\n") + fmt.Printf("\033[K ▸ %s\n", truncate(status, maxLineWidth-4)) + // Every draw emits the body, one blank line, and the status line; moving + // back by exactly this many lines is what keeps the redraw from drifting. + r.lastLines = len(body) + 2 r.drawn = true } -// line writes one table row, clearing whatever the previous draw left on it -// when redrawing in place. -func (r *renderer) line(s string) { +// body renders the header and one line per row. +func (r *renderer) body(rows []*row) []string { + r.fit(rows) + + lines := make([]string, 0, len(rows)+2) + lines = append(lines, + fmt.Sprintf(" %-*s %-*s %*s %s", + r.wRequest, "REQUEST", r.wChanges, "CHANGES", r.wElapsed, "ELAPSED", "STAGE"), + fmt.Sprintf(" %s %s %s %s", + rule(r.wRequest), rule(r.wChanges), rule(r.wElapsed), rule(r.wStage))) + + for _, rw := range rows { + lines = append(lines, r.rowLine(rw)) + lines = append(lines, r.noteLines(rw)...) + } + return lines +} + +// fit grows the columns to hold what the rows now contain. Widths never shrink, +// so the table does not shift under a value that has already been printed — but +// on a terminal the stage column stays inside the line, since its rule would +// otherwise wrap on a long trail and take the redraw with it. +func (r *renderer) fit(rows []*row) { + for _, rw := range rows { + r.wRequest = max(r.wRequest, utf8.RuneCountInString(rw.sqid)) + _, visible := r.changesCell(rw) + r.wChanges = max(r.wChanges, visible) + r.wStage = max(r.wStage, utf8.RuneCountInString(rw.stage())) + } if r.inPlace { - fmt.Printf("\033[K%s\n", s) - return + r.wStage = min(r.wStage, max(len("STAGE"), maxLineWidth-r.prefixWidth())) + } +} + +// prefixWidth is the space every row spends before the stage column. +func (r *renderer) prefixWidth() int { + return 2 + r.wRequest + 2 + r.wChanges + 2 + r.wElapsed + 2 +} + +func (r *renderer) rowLine(rw *row) string { + sqid := rw.sqid + if sqid == "" { + sqid = absent + } + changes, visible := r.changesCell(rw) + + prefix := fmt.Sprintf(" %-*s %s %*s ", + r.wRequest, sqid, pad(changes, visible, r.wChanges), r.wElapsed, rw.elapsed()) + + tail := rw.stage() + if r.inPlace { + // Only the tail can overflow, and unlike the changes cell it never holds + // escape sequences, so it is the one part safe to cut. The budget comes + // from the column widths rather than the rendered prefix, which counts a + // hyperlink's escape bytes that take up no space on screen. + tail = truncate(tail, maxLineWidth-r.prefixWidth()) } - fmt.Println(s) + return prefix + tail +} + +// noteLines renders a request's error under its row, wrapped and indented to +// the stage column. An error is the one thing in the table worth reading in +// full — truncating it to the width of a cell hides the part that says what +// went wrong — so it gets as many lines as it needs instead of an ellipsis. +func (r *renderer) noteLines(rw *row) []string { + if rw.note == "" { + return nil + } + + indent := r.prefixWidth() + // A piped run spends most of the line on URLs, so the wrap width is floored + // rather than allowed to collapse to nothing. + width := max(minNoteWidth, maxLineWidth-indent-2) + + wrapped := wrap(rw.note, width) + lines := make([]string, 0, len(wrapped)) + for i, text := range wrapped { + marker := " " + if i == 0 { + marker = "↳ " + } + lines = append(lines, strings.Repeat(" ", indent)+marker+text) + } + return lines +} + +// wrap breaks text into lines no wider than width, splitting on spaces and +// hard-splitting any single token too long to fit on a line of its own. +func wrap(s string, width int) []string { + if width < 1 { + return nil + } + + var lines []string + current := "" + flush := func() { + if current != "" { + lines = append(lines, current) + current = "" + } + } + + for _, word := range strings.Fields(s) { + for utf8.RuneCountInString(word) > width { + flush() + runes := []rune(word) + lines = append(lines, string(runes[:width])) + word = string(runes[width:]) + } + switch { + case current == "": + current = word + case utf8.RuneCountInString(current)+1+utf8.RuneCountInString(word) <= width: + current += " " + word + default: + flush() + current = word + } + } + flush() + return lines +} + +// changesCell renders the pull requests on a row and reports the width they +// occupy on screen. The two differ on a terminal, where a hyperlink is mostly +// escape bytes that take up no space. +func (r *renderer) changesCell(rw *row) (string, int) { + if len(rw.changes) == 0 { + return absent, utf8.RuneCountInString(absent) + } + + parts := make([]string, 0, len(rw.changes)) + visible := 0 + for _, c := range rw.changes { + label := fmt.Sprintf("#%d", c.number) + if r.inPlace { + parts = append(parts, hyperlink(label, c.url)) + visible += len(label) + continue + } + // Piped output has nothing to click, so the address itself has to be + // readable — and copyable out of a log. + parts = append(parts, c.url) + visible += len(c.url) + } + separator := "," + if !r.inPlace { + separator = " " + } + return strings.Join(parts, separator), visible + len(separator)*(len(parts)-1) +} + +// hyperlink wraps text in an OSC 8 escape so terminals that understand it make +// the text clickable, and the rest simply show the text. +func hyperlink(text, url string) string { + if url == "" { + return text + } + return "\033]8;;" + url + "\033\\" + text + "\033]8;;\033\\" +} + +// pad right-pads a cell to a column width using its on-screen width, which is +// not its length whenever it carries escape sequences. +func pad(s string, visible, width int) string { + if visible >= width { + return s + } + return s + strings.Repeat(" ", width-visible) +} + +func rule(n int) string { + return strings.Repeat("─", n) +} + +// signature is what a piped run treats as the table having moved. The elapsed +// clock is left out on purpose: it advances every second, and a log that +// reprinted the table for that alone would say nothing while saying it often. +func signature(rows []*row) string { + var b strings.Builder + for _, rw := range rows { + fmt.Fprintf(&b, "%s|%s|%s|%s\n", rw.sqid, labelsOf(rw.changes), rw.stage(), rw.note) + } + return b.String() } func truncate(s string, n int) string { s = strings.ReplaceAll(s, "\n", " ") - if len(s) <= n { + if n < 1 { + return "" + } + if utf8.RuneCountInString(s) <= n { return s } - return s[:n-1] + "…" + return string([]rune(s)[:n-1]) + "…" } func parseStrategy(name string) (mergestrategypb.Strategy, error) { diff --git a/service/submitqueue/demo/pr/main_test.go b/service/submitqueue/demo/pr/main_test.go new file mode 100644 index 00000000..2ece441d --- /dev/null +++ b/service/submitqueue/demo/pr/main_test.go @@ -0,0 +1,658 @@ +// 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 ( + "context" + "fmt" + "io" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "google.golang.org/grpc" +) + +func TestDigest(t *testing.T) { + tests := []struct { + name string + events []*pb.HistoryEvent + wantTrail []string + wantStatus string + wantNote string + }{ + { + name: "no events yet", + }, + { + name: "one event", + events: []*pb.HistoryEvent{{Status: "accepted"}}, + wantTrail: []string{"accepted"}, + wantStatus: "accepted", + }, + { + name: "trail keeps the order it was recorded in", + events: []*pb.HistoryEvent{ + {Status: "accepted"}, {Status: "started"}, {Status: "batched"}, {Status: "landed"}, + }, + wantTrail: []string{"accepted", "started", "batched", "landed"}, + wantStatus: "landed", + }, + { + name: "a status recorded twice in a row is one step", + events: []*pb.HistoryEvent{ + {Status: "accepted"}, {Status: "started"}, {Status: "started"}, {Status: "batched"}, + }, + wantTrail: []string{"accepted", "started", "batched"}, + wantStatus: "batched", + }, + { + name: "a status revisited later is a step again", + events: []*pb.HistoryEvent{ + {Status: "speculating"}, {Status: "batched"}, {Status: "speculating"}, + }, + wantTrail: []string{"speculating", "batched", "speculating"}, + wantStatus: "speculating", + }, + { + name: "the error on the latest event is the one shown", + events: []*pb.HistoryEvent{ + {Status: "started", LastError: "transient"}, {Status: "error", LastError: "merge conflict"}, + }, + wantTrail: []string{"started", "error"}, + wantStatus: "error", + wantNote: "merge conflict", + }, + { + name: "events without a status do not become steps", + events: []*pb.HistoryEvent{{Status: ""}, {Status: "accepted"}, {Status: ""}}, + wantTrail: []string{"accepted"}, + wantStatus: "accepted", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + trail, status, note := digest(tt.events) + assert.Equal(t, tt.wantTrail, trail) + assert.Equal(t, tt.wantStatus, status) + assert.Equal(t, tt.wantNote, note) + }) + } +} + +func TestRowElapsed(t *testing.T) { + now := time.Now() + + tests := []struct { + name string + row row + want string + }{ + { + name: "absent before the gateway accepts it", + row: row{}, + want: absent, + }, + { + name: "running while in flight", + row: row{submitted: now.Add(-5 * time.Second)}, + want: "5s", + }, + { + name: "frozen once settled, however long ago that was", + row: row{submitted: now.Add(-90 * time.Second), settled: now.Add(-60 * time.Second)}, + want: "30s", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, tt.row.elapsed()) + }) + } +} + +// TestRowElapsedStopsAtSettle pins the behavior the clock exists for: a settled +// row reads the same however much later it is drawn, while an unsettled one +// does not. +func TestRowElapsedStopsAtSettle(t *testing.T) { + start := time.Now().Add(-time.Minute) + settled := row{submitted: start, settled: start.Add(10 * time.Second)} + inFlight := row{submitted: start} + + first := settled.elapsed() + time.Sleep(time.Millisecond) + assert.Equal(t, first, settled.elapsed()) + assert.NotEqual(t, first, inFlight.elapsed()) +} + +func TestRowStage(t *testing.T) { + tests := []struct { + name string + row row + want string + }{ + { + name: "nothing to report before the request exists", + row: row{}, + want: absent, + }, + { + name: "accepted but nothing recorded yet", + row: row{sqid: "demo-queue/17"}, + want: "…", + }, + { + name: "the states it passed through", + row: row{sqid: "demo-queue/17", trail: []string{"accepted", "started", "landed"}}, + want: "accepted → started → landed", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, tt.row.stage()) + }) + } +} + +func TestChangesCell(t *testing.T) { + one := []change{{number: 41, url: "https://github.com/o/r/pull/41"}} + two := []change{ + {number: 41, url: "https://github.com/o/r/pull/41"}, + {number: 421, url: "https://github.com/o/r/pull/421"}, + } + + t.Run("a terminal gets short clickable labels", func(t *testing.T) { + r := &renderer{inPlace: true} + text, visible := r.changesCell(&row{changes: two}) + + assert.Contains(t, text, "https://github.com/o/r/pull/41") + assert.Contains(t, text, "\033]8;;") + // "#41,#421" occupies eight columns however many escape bytes carry it. + assert.Equal(t, len("#41,#421"), visible) + assert.Greater(t, len(text), visible, "the escapes should not be counted as width") + }) + + t.Run("a pipe gets the addresses themselves", func(t *testing.T) { + r := &renderer{inPlace: false} + text, visible := r.changesCell(&row{changes: one}) + + assert.Equal(t, "https://github.com/o/r/pull/41", text) + assert.Equal(t, len(text), visible) + assert.NotContains(t, text, "\033") + }) + + t.Run("no pull requests yet", func(t *testing.T) { + r := &renderer{inPlace: true} + text, visible := r.changesCell(&row{}) + + assert.Equal(t, absent, text) + assert.Equal(t, 1, visible) + }) +} + +// TestPadCountsVisibleWidth guards the alignment trap: a hyperlinked cell is +// mostly escape bytes, so padding by length would push the columns apart. +func TestPadCountsVisibleWidth(t *testing.T) { + linked := hyperlink("#41", "https://github.com/o/r/pull/41") + + assert.Equal(t, linked+" ", pad(linked, len("#41"), 10)) + assert.Equal(t, "#41 ", pad("#41", 3, 10)) + assert.Equal(t, "#41", pad("#41", 3, 3), "a cell at the column width is not padded") + assert.Equal(t, "#41", pad("#41", 3, 2), "a cell wider than the column is left alone") +} + +func TestTruncate(t *testing.T) { + tests := []struct { + name string + in string + n int + want string + }{ + {name: "short enough to keep", in: "accepted", n: 20, want: "accepted"}, + {name: "exactly the limit", in: "accepted", n: 8, want: "accepted"}, + {name: "cut with a marker", in: "accepted → started", n: 10, want: "accepted …"}, + {name: "newlines flattened", in: "line\nbreak", n: 20, want: "line break"}, + {name: "no room at all", in: "accepted", n: 0, want: ""}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, truncate(tt.in, tt.n)) + }) + } +} + +// TestTruncateSplitsOnRunes checks that a cut lands between characters. The +// trail is joined with a multi-byte arrow, so cutting by bytes would leave +// mojibake in the middle of the table. +func TestTruncateSplitsOnRunes(t *testing.T) { + got := truncate("accepted → started → batched", 12) + assert.True(t, utf8ValidAndCounted(got, 12), "got %q", got) +} + +func utf8ValidAndCounted(s string, n int) bool { + return len([]rune(s)) <= n && strings.ToValidUTF8(s, "?") == s +} + +// TestDrawLineAccounting is the invariant the in-place redraw rests on: the +// cursor moves back exactly as far as the previous draw reached. Off by one and +// every subsequent draw leaves a stale row on screen. +func TestDrawLineAccounting(t *testing.T) { + r := newRenderer() + r.inPlace = true + rows := []*row{ + {sqid: "demo-queue/17", changes: []change{{number: 41, url: "https://github.com/o/r/pull/41"}}, + submitted: time.Now(), trail: []string{"accepted", "started"}}, + {sqid: "demo-queue/18", changes: []change{{number: 42, url: "https://github.com/o/r/pull/42"}}, + submitted: time.Now(), trail: []string{"accepted"}}, + {}, + } + + first := captureStdout(t, func() { r.draw(rows, "watching") }) + emitted := strings.Count(first, "\n") + assert.Equal(t, emitted, r.lastLines, "the first draw must record how far it reached") + assert.True(t, strings.HasPrefix(first, "\033[K"), "the first draw has nothing to move back over") + + second := captureStdout(t, func() { r.draw(rows, "still watching") }) + assert.True(t, strings.HasPrefix(second, fmt.Sprintf("\033[%dA", emitted)), + "the redraw must move back over exactly the %d lines it wrote, got %q", emitted, head(second, 12)) + assert.Equal(t, strings.Count(second, "\n"), r.lastLines) +} + +// TestDrawStaysWithinLineWidth checks the other half of the redraw contract: a +// line that wraps occupies two physical rows and desyncs the cursor for good. +func TestDrawStaysWithinLineWidth(t *testing.T) { + r := newRenderer() + r.inPlace = true + rows := []*row{{ + sqid: "demo-queue/17", + submitted: time.Now(), + trail: strings.Split(strings.Repeat("speculating ", 30), " "), + note: strings.Repeat("a very long error message ", 10), + }} + + out := captureStdout(t, func() { r.draw(rows, strings.Repeat("status ", 40)) }) + for _, line := range strings.Split(out, "\n") { + line = strings.ReplaceAll(line, "\033[K", "") + assert.LessOrEqual(t, len([]rune(line)), maxLineWidth, "line too wide: %q", line) + } +} + +// TestDrawPipedSkipsClockOnlyRedraws keeps a redirected run's log readable: the +// table is reprinted when it moves, not once a second because the clock did. +func TestDrawPipedSkipsClockOnlyRedraws(t *testing.T) { + r := newRenderer() + r.inPlace = false + rows := []*row{{sqid: "demo-queue/17", submitted: time.Now().Add(-5 * time.Second), trail: []string{"accepted"}}} + + first := captureStdout(t, func() { r.draw(rows, "watching") }) + require.NotEmpty(t, first) + assert.NotContains(t, first, "\033", "a redirected run must not emit escape codes") + + rows[0].submitted = time.Now().Add(-30 * time.Second) + assert.Empty(t, captureStdout(t, func() { r.draw(rows, "watching") }), + "only the clock moved, so there is nothing new to say") + + rows[0].trail = append(rows[0].trail, "started") + assert.NotEmpty(t, captureStdout(t, func() { r.draw(rows, "watching") }), + "the request moved, so the table should be reprinted") +} + +// TestFitGrowsColumnsOnly checks that a value wider than its header widens the +// column and that a later, narrower table does not pull it back in — a column +// that shrank would make the table jitter as rows fill in. +func TestFitGrowsColumnsOnly(t *testing.T) { + r := newRenderer() + wide := []*row{{sqid: "some-very-long-queue-name/1234"}} + r.fit(wide) + grown := r.wRequest + assert.Equal(t, len("some-very-long-queue-name/1234"), grown) + + r.fit([]*row{{}}) + assert.Equal(t, grown, r.wRequest) +} + +func TestNewRows(t *testing.T) { + assert.Len(t, newRows(config{count: 3}), 3, "independent changes are one request each") + assert.Len(t, newRows(config{count: 3, stacked: true}), 1, "a stack is a single request") +} + +func TestOutcome(t *testing.T) { + landed := &row{status: "landed"} + failed := &row{status: "error"} + + assert.Equal(t, "all 2 request(s) landed", outcome([]*row{landed, landed})) + assert.Equal(t, "1 of 2 request(s) did not land", outcome([]*row{landed, failed})) +} + +func TestSummarize(t *testing.T) { + assert.NoError(t, summarize([]*row{{sqid: "q/1", status: "landed"}})) + assert.Error(t, summarize([]*row{{sqid: "q/1", status: "landed"}, {sqid: "q/2", status: "error"}})) +} + +// TestRowLineAlignment is the column contract: on every row the stage begins at +// exactly the same screen column, whatever the cells before it contain. A +// hyperlinked row is the interesting one, since its changes cell is mostly +// escape bytes that occupy no width — measuring those as if they did would both +// shove the column sideways and eat the stage's room to render. +// fakeGateway answers history lookups from a table the test controls. Only the +// one method is reachable; the embedded interface satisfies the rest. +type fakeGateway struct { + pb.SubmitQueueGatewayClient + mu sync.Mutex + events map[string][]*pb.HistoryEvent +} + +func (f *fakeGateway) set(sqid string, statuses ...string) { + f.mu.Lock() + defer f.mu.Unlock() + if f.events == nil { + f.events = map[string][]*pb.HistoryEvent{} + } + events := make([]*pb.HistoryEvent, 0, len(statuses)) + for _, s := range statuses { + events = append(events, &pb.HistoryEvent{Status: s}) + } + f.events[sqid] = events +} + +func (f *fakeGateway) GetRequestHistoryByID( + _ context.Context, in *pb.GetRequestHistoryByIDRequest, _ ...grpc.CallOption, +) (*pb.GetRequestHistoryByIDResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + return &pb.GetRequestHistoryByIDResponse{Events: f.events[in.Sqid]}, nil +} + +func isClosed(ch <-chan struct{}) bool { + select { + case <-ch: + return true + default: + return false + } +} + +// TestTrackerSettlesOnlyWhenSealedAndTerminal covers the condition that ends a +// run. Polling starts while pull requests are still being created, so "nothing +// outstanding" is true before creation has begun — sealing is what separates +// that from actually being finished. +func TestTrackerSettlesOnlyWhenSealedAndTerminal(t *testing.T) { + tr := newTracker(newRows(config{count: 2})) + tr.r.inPlace = false + gw := &fakeGateway{} + ctx := context.Background() + + captureStdout(t, func() { + tr.update(func() { tr.rows[0].sqid = "demo-queue/1" }) + tr.seal() + }) + assert.False(t, isClosed(tr.settled), "a row that was never enqueued is not settled") + + captureStdout(t, func() { tr.update(func() { tr.rows[1].sqid = "demo-queue/2" }) }) + gw.set("demo-queue/1", "accepted", "started") + gw.set("demo-queue/2", "accepted") + captureStdout(t, func() { tr.refresh(ctx, gw, "demo-queue") }) + + assert.Equal(t, []string{"accepted", "started"}, tr.rows[0].trail) + assert.False(t, isClosed(tr.settled), "requests still in flight") + + gw.set("demo-queue/1", "accepted", "started", "landed") + gw.set("demo-queue/2", "accepted", "error") + captureStdout(t, func() { tr.refresh(ctx, gw, "demo-queue") }) + + assert.True(t, isClosed(tr.settled), "every request reached a terminal status") + assert.True(t, tr.rows[0].done) + assert.False(t, tr.rows[0].settled.IsZero(), "settling stops the clock") +} + +// TestTrackerSealBeforeEnqueueDoesNotSettle guards the ordering hazard the seal +// exists for: polling that ran before anything was enqueued must not conclude +// the run just because it found nothing outstanding. +func TestTrackerSealBeforeEnqueueDoesNotSettle(t *testing.T) { + tr := newTracker(newRows(config{count: 1})) + tr.r.inPlace = false + + captureStdout(t, func() { tr.refresh(context.Background(), &fakeGateway{}, "demo-queue") }) + assert.False(t, isClosed(tr.settled), "nothing has been enqueued yet") +} + +// TestTrackerPollsWhileCreating is the behavior the tracker exists for: a run +// that only polled after every pull request was created would show an empty +// trail for the whole creation phase. Here a row enqueued first picks up its +// trail while a later row has not been enqueued at all. +func TestTrackerPollsWhileCreating(t *testing.T) { + tr := newTracker(newRows(config{count: 3})) + tr.r.inPlace = false + gw := &fakeGateway{} + gw.set("demo-queue/1", "accepted", "started", "batched") + + captureStdout(t, func() { + tr.update(func() { tr.rows[0].sqid = "demo-queue/1" }) + tr.refresh(context.Background(), gw, "demo-queue") + }) + + assert.Equal(t, "accepted → started → batched", tr.rows[0].stage()) + assert.Equal(t, absent, tr.rows[2].stage(), "a row not yet enqueued has nothing to show") +} + +// TestTrackerConcurrentPollAndUpdate exercises the two writers against each +// other so the race detector has something to find. Creation fills in rows from +// one goroutine while polling reads and redraws from another. +func TestTrackerConcurrentPollAndUpdate(t *testing.T) { + tr := newTracker(newRows(config{count: 8})) + tr.r.inPlace = false + gw := &fakeGateway{} + ctx := context.Background() + + captureStdout(t, func() { + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + for i := range tr.rows { + sqid := fmt.Sprintf("demo-queue/%d", i) + gw.set(sqid, "accepted", "landed") + i := i + tr.update(func() { + tr.rows[i].changes = append(tr.rows[i].changes, change{number: 100 + i, url: "https://example.test/pull/1"}) + tr.rows[i].sqid, tr.rows[i].submitted = sqid, time.Now() + }) + } + tr.seal() + }() + + go func() { + defer wg.Done() + for range 20 { + tr.refresh(ctx, gw, "demo-queue") + } + }() + + wg.Wait() + tr.refresh(ctx, gw, "demo-queue") + }) + + assert.True(t, isClosed(tr.settled)) + require.NoError(t, tr.conclude()) +} + +func TestWrap(t *testing.T) { + tests := []struct { + name string + in string + width int + want []string + }{ + {name: "nothing to wrap", in: "short", width: 20, want: []string{"short"}}, + { + name: "breaks on spaces", + in: "queue name must not be empty", + width: 12, + want: []string{"queue name", "must not be", "empty"}, + }, + { + name: "a token longer than the line is split", + in: "aaaaaaaaaa bb", + width: 4, + want: []string{"aaaa", "aaaa", "aa", "bb"}, + }, + {name: "newlines are just whitespace", in: "one\ntwo", width: 20, want: []string{"one two"}}, + {name: "no width", in: "anything", width: 0}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := wrap(tt.in, tt.width) + assert.Equal(t, tt.want, got) + for _, line := range got { + assert.LessOrEqual(t, len([]rune(line)), tt.width) + } + }) + } +} + +// TestNoteLinesRenderErrorInFull is the point of wrapping rather than +// truncating: the interesting part of a pipeline error is usually at the end, +// so an ellipsis in the stage column hides exactly what the reader needs. +func TestNoteLinesRenderErrorInFull(t *testing.T) { + r := newRenderer() + r.inPlace = true + failed := &row{ + sqid: "demo-queue/1", + changes: []change{{number: 75, url: "https://github.com/behinddwalls/sq-demo/pull/75"}}, + submitted: time.Now(), + trail: []string{"accepted", "started", "validated", "batched", "error"}, + note: `speculator failed for queue demo-queue: score dependency "demo-queue/batch/1": ` + + `failed to resolve storage for queue "": queue name must not be empty`, + } + r.fit([]*row{failed}) + + lines := r.noteLines(failed) + require.NotEmpty(t, lines) + + var text strings.Builder + for i, line := range lines { + assert.LessOrEqual(t, len([]rune(line)), maxLineWidth, "a wrapped note still has to fit the line") + trimmed := strings.TrimLeft(line, " ") + if i == 0 { + assert.True(t, strings.HasPrefix(trimmed, "↳ "), "the first line is marked") + } + text.WriteString(strings.TrimPrefix(strings.TrimPrefix(trimmed, "↳ "), " ")) + text.WriteString(" ") + } + + assert.Contains(t, text.String(), "queue name must not be empty", + "the tail of the error is what says what went wrong; it must survive") + + // The row itself keeps only the trail, so the columns stay aligned. + assert.NotContains(t, r.rowLine(failed), "speculator failed") + assert.NotContains(t, r.rowLine(failed), "…") +} + +// TestNoteLinesIndentToStageColumn keeps a wrapped error visually attached to +// its row rather than looking like a new column. +func TestNoteLinesIndentToStageColumn(t *testing.T) { + r := newRenderer() + r.inPlace = true + rows := []*row{{sqid: "demo-queue/1", submitted: time.Now(), trail: []string{"error"}, note: "boom"}} + r.fit(rows) + + lines := r.noteLines(rows[0]) + require.Len(t, lines, 1) + assert.Equal(t, strings.Repeat(" ", r.prefixWidth())+"↳ boom", lines[0]) +} + +func TestRowLineAlignment(t *testing.T) { + r := newRenderer() + r.inPlace = true + rows := []*row{ + { + sqid: "demo-queue/17", + changes: []change{{number: 41, url: "https://github.com/behinddwalls/sq-demo/pull/41"}}, + submitted: time.Now(), + trail: []string{"accepted", "started", "batched", "speculating", "landed"}, + }, + {sqid: "demo-queue/1234", submitted: time.Now(), trail: []string{"accepted"}}, + {}, + } + r.fit(rows) + + for _, rw := range rows { + shown := []rune(visible(r.rowLine(rw))) + require.GreaterOrEqual(t, len(shown), r.prefixWidth()) + assert.Equal(t, rw.stage(), string(shown[r.prefixWidth():]), + "the stage should start at column %d and be rendered whole", r.prefixWidth()) + } +} + +// visible strips OSC 8 hyperlink sequences, leaving what the terminal draws. +func visible(s string) string { + for { + start := strings.Index(s, "\033]8;;") + if start < 0 { + return s + } + end := strings.Index(s[start:], "\033\\") + if end < 0 { + return s + } + s = s[:start] + s[start+end+len("\033\\"):] + } +} + +// captureStdout collects what fn writes to stdout. The renderer writes there +// directly, which is the thing under test. The pipe is drained as fn runs, so a +// test that draws more than the pipe buffer holds does not deadlock. +func captureStdout(t *testing.T, fn func()) string { + t.Helper() + + rd, wr, err := os.Pipe() + require.NoError(t, err) + + collected := make(chan string, 1) + go func() { + out, readErr := io.ReadAll(rd) + if readErr != nil { + collected <- "" + return + } + collected <- string(out) + }() + + original := os.Stdout + os.Stdout = wr + defer func() { os.Stdout = original }() + + fn() + require.NoError(t, wr.Close()) + return <-collected +} + +func head(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] +} From f7adf60f56b26ca918090147cdf8f32aa3ca5b50 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Sat, 8 Aug 2026 08:21:10 -0700 Subject: [PATCH 3/5] feat(messagequeue): level the queue's own logs separately MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The services run at debug, and the queue logs a line per message published, fetched, leased and acked. Reading a service's logs means scrolling past hundreds of `fetched messages {"count": 0}` to find the one line that matters — following a request through the pipeline is impractical at the level everything else is useful at. Turning the whole service down to info loses the output that is actually being read. The queue's chatter is the part worth silencing, not the service's. ### What? `Params.LogLevel` sets the minimum level for the queue's own logs, applied with `zap.IncreaseLevel` so the rest of the service keeps the level it was built with. Empty selects info, which is what stops the chatter by default. The four servers pass `QUEUE_LOG_LEVEL` through, and the compose files forward it, so the queue can be turned back up for a run that is chasing a message that never arrived: ``` QUEUE_LOG_LEVEL=debug make local-submitqueue-start ``` The level is a raw string on `Params`, parsed in the package, which keeps the environment lookup in the wiring layer where the other knobs live. `IncreaseLevel` can only raise the level, never lower it, so this cannot be used to make a quiet service verbose — an unparseable value is rejected at construction rather than silently ignored. ## Test Plan ✅ `bazel test //platform/extension/messagequeue/mysql:go_default_test` — the level is accepted, and a value that is not a level fails construction. ✅ `make test` — 98 targets. ✅ Passthrough verified rather than assumed, since compose forwards only what it declares: `QUEUE_LOG_LEVEL=debug docker compose -f service/submitqueue/docker-compose.yml config` shows the variable set on all three services. --- doc/howto/PROVIDER-E2E.md | 8 ++++ .../extension/messagequeue/mysql/BUILD.bazel | 1 + platform/extension/messagequeue/mysql/sql.go | 24 +++++++++++- .../extension/messagequeue/mysql/sql_test.go | 38 +++++++++++++++++++ service/runway/server/docker-compose.yml | 3 ++ service/runway/server/main.go | 1 + service/stovepipe/docker-compose.yml | 3 ++ service/stovepipe/server/main.go | 1 + service/submitqueue/docker-compose.yml | 9 +++++ .../gateway/server/docker-compose.yml | 3 ++ service/submitqueue/gateway/server/main.go | 1 + .../orchestrator/server/docker-compose.yml | 3 ++ .../submitqueue/orchestrator/server/main.go | 1 + 13 files changed, 94 insertions(+), 2 deletions(-) diff --git a/doc/howto/PROVIDER-E2E.md b/doc/howto/PROVIDER-E2E.md index 26b5a8ee..fe257d89 100644 --- a/doc/howto/PROVIDER-E2E.md +++ b/doc/howto/PROVIDER-E2E.md @@ -134,6 +134,14 @@ Runway logs each merge and each head-branch move: moved change head branch to its landed commit {"change": "you/repo#1", "branch": "refs/heads/feature-a", ...} ``` +The message queue logs a line per message published, fetched, leased and acked, which at debug level buries everything else a service says. It is levelled separately from the rest of the service, at info by default. To follow the queue itself — chasing a message that never arrived, or a partition that never got leased — turn it back up for the services you care about: + +```bash +QUEUE_LOG_LEVEL=debug make local-submitqueue-start +``` + +`QUEUE_LOG_LEVEL` takes any zap level name. It can only raise the queue's level above the one the service logger was built with, never lower it, so it cannot be used to make a quiet service verbose. + ## When it does not work **The push is rejected on the first try.** Branch protection on `main` — required status checks, or a linear-history or no-force-push rule — applies to the merger like anyone else. Either relax it on the scratch repo or add the token's identity to the bypass list. diff --git a/platform/extension/messagequeue/mysql/BUILD.bazel b/platform/extension/messagequeue/mysql/BUILD.bazel index 1a0a7f45..224f63de 100644 --- a/platform/extension/messagequeue/mysql/BUILD.bazel +++ b/platform/extension/messagequeue/mysql/BUILD.bazel @@ -26,6 +26,7 @@ go_library( "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_mock//gomock:go_default_library", "@org_uber_go_zap//:go_default_library", + "@org_uber_go_zap//zapcore:go_default_library", ], ) diff --git a/platform/extension/messagequeue/mysql/sql.go b/platform/extension/messagequeue/mysql/sql.go index 042ae9a5..de4e3b8d 100644 --- a/platform/extension/messagequeue/mysql/sql.go +++ b/platform/extension/messagequeue/mysql/sql.go @@ -22,6 +22,7 @@ import ( "github.com/uber-go/tally" "go.uber.org/zap" + "go.uber.org/zap/zapcore" extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" ) @@ -40,6 +41,16 @@ type Params struct { // Logger for debugging and observability (required) Logger *zap.Logger + // LogLevel is the minimum level for the queue's own logs, as a zap level + // name ("debug", "info", ...). Empty selects info. + // + // The queue logs a line per message published, fetched, leased and acked, + // which at debug buries everything else a service says. Levelling it here + // rather than at the service logger keeps the rest of that service's debug + // output intact. The level can only be raised above the one the supplied + // logger was built with, never lowered. + LogLevel string + // MetricsScope for metrics collection (required) MetricsScope tally.Scope @@ -55,8 +66,17 @@ func NewQueue(params Params) (extqueue.Queue, error) { return nil, fmt.Errorf("failed to ping database: %w", err) } - logger := params.Logger.Sugar().Named("queue_mysql") - logger.Infow("created SQL queue") + level := zapcore.InfoLevel + if params.LogLevel != "" { + parsed, err := zapcore.ParseLevel(params.LogLevel) + if err != nil { + return nil, fmt.Errorf("invalid queue log level %q: %w", params.LogLevel, err) + } + level = parsed + } + + logger := params.Logger.WithOptions(zap.IncreaseLevel(level)).Sugar().Named("queue_mysql") + logger.Infow("created SQL queue", "log_level", level.String()) // Create stores messageStore := newMessageStore(params.DB, logger, params.MetricsScope) diff --git a/platform/extension/messagequeue/mysql/sql_test.go b/platform/extension/messagequeue/mysql/sql_test.go index f04b0f36..6eadafbd 100644 --- a/platform/extension/messagequeue/mysql/sql_test.go +++ b/platform/extension/messagequeue/mysql/sql_test.go @@ -70,6 +70,44 @@ func TestNewQueue(t *testing.T) { require.NoError(t, mock.ExpectationsWereMet()) }) + t.Run("accepts a log level", func(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.MonitorPingsOption(true)) + require.NoError(t, err) + defer db.Close() + + mock.ExpectPing() + + q, err := NewQueue(Params{ + DB: db, + Logger: zaptest.NewLogger(t), + LogLevel: "debug", + MetricsScope: tally.NewTestScope("test", nil), + }) + + require.NoError(t, err) + require.NotNil(t, q) + assert.NoError(t, q.Close()) + + require.NoError(t, mock.ExpectationsWereMet()) + }) + + t.Run("error when the log level is not a level", func(t *testing.T) { + db, mock, err := sqlmock.New(sqlmock.MonitorPingsOption(true)) + require.NoError(t, err) + defer db.Close() + + mock.ExpectPing() + + q, err := NewQueue(Params{ + DB: db, + Logger: zaptest.NewLogger(t), + LogLevel: "loud", + MetricsScope: tally.NewTestScope("test", nil), + }) + + require.Error(t, err) + assert.Nil(t, q) + }) } func TestQueue_Publisher(t *testing.T) { diff --git a/service/runway/server/docker-compose.yml b/service/runway/server/docker-compose.yml index 427c4504..1706886d 100644 --- a/service/runway/server/docker-compose.yml +++ b/service/runway/server/docker-compose.yml @@ -49,6 +49,9 @@ services: - MERGER=${SQ_RUNWAY_MERGER:-} # Queue infrastructure connection - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + # Level for the queue's own logs; info by default so its per-message + # chatter does not bury the rest of the service at debug. + - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} - HOSTNAME=runway-dev depends_on: mysql-queue: diff --git a/service/runway/server/main.go b/service/runway/server/main.go index 350fdf54..dc998500 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -138,6 +138,7 @@ func run() error { mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{ DB: queueDB, Logger: logger, + LogLevel: os.Getenv("QUEUE_LOG_LEVEL"), MetricsScope: scope.SubScope("queue"), }) if err != nil { diff --git a/service/stovepipe/docker-compose.yml b/service/stovepipe/docker-compose.yml index 3c816c21..ab4a82e1 100644 --- a/service/stovepipe/docker-compose.yml +++ b/service/stovepipe/docker-compose.yml @@ -67,6 +67,9 @@ services: - PORT=:8080 - STORAGE_MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + # Level for the queue's own logs; info by default so its per-message + # chatter does not bury the rest of the service at debug. + - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} - HOSTNAME=stovepipe-dev depends_on: mysql-app: diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 6c58fabf..60176d33 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -241,6 +241,7 @@ func run() error { mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{ DB: queueDB, Logger: logger, + LogLevel: os.Getenv("QUEUE_LOG_LEVEL"), MetricsScope: scope.SubScope("queue"), }) if err != nil { diff --git a/service/submitqueue/docker-compose.yml b/service/submitqueue/docker-compose.yml index d3dd0b56..c7a11d77 100644 --- a/service/submitqueue/docker-compose.yml +++ b/service/submitqueue/docker-compose.yml @@ -73,6 +73,9 @@ services: - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + # Level for the queue's own logs; info by default so its per-message + # chatter does not bury the rest of the service at debug. + - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} # Path to YAML queue configuration baked into the image - QUEUE_CONFIG_PATH=/app/queues.yaml # Stable subscriber name for the request-log consumer @@ -101,6 +104,9 @@ services: - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + # Level for the queue's own logs; info by default so its per-message + # chatter does not bury the rest of the service at debug. + - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} - HOSTNAME=orchestrator-dev # Consumer-gate state shared with the host (see header comment) - CONSUMER_GATE_DIR=/var/submitqueue/consumergate @@ -129,6 +135,9 @@ services: - PORT=:8080 # Queue infrastructure connection (shared with the orchestrator) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + # Level for the queue's own logs; info by default so its per-message + # chatter does not bury the rest of the service at debug. + - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} - HOSTNAME=runway-dev # Consumer-gate state shared with the host (see header comment) - CONSUMER_GATE_DIR=/var/submitqueue/consumergate diff --git a/service/submitqueue/gateway/server/docker-compose.yml b/service/submitqueue/gateway/server/docker-compose.yml index a896f2c6..0ed86120 100644 --- a/service/submitqueue/gateway/server/docker-compose.yml +++ b/service/submitqueue/gateway/server/docker-compose.yml @@ -62,6 +62,9 @@ services: - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + # Level for the queue's own logs; info by default so its per-message + # chatter does not bury the rest of the service at debug. + - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} # Path to YAML queue configuration baked into the image - QUEUE_CONFIG_PATH=/app/queues.yaml # Stable subscriber name for the request-log consumer diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index a5556f04..ba0d91d5 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -245,6 +245,7 @@ func run() error { mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{ DB: queueDB, Logger: logger, + LogLevel: os.Getenv("QUEUE_LOG_LEVEL"), MetricsScope: scope.SubScope("queue"), }) if err != nil { diff --git a/service/submitqueue/orchestrator/server/docker-compose.yml b/service/submitqueue/orchestrator/server/docker-compose.yml index b9e20150..890478a1 100644 --- a/service/submitqueue/orchestrator/server/docker-compose.yml +++ b/service/submitqueue/orchestrator/server/docker-compose.yml @@ -62,6 +62,9 @@ services: - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true # Queue infrastructure connection (separate database) - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true + # Level for the queue's own logs; info by default so its per-message + # chatter does not bury the rest of the service at debug. + - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-} - HOSTNAME=orchestrator-dev depends_on: mysql-app: diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index a3a1d21a..8c4f91ff 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -157,6 +157,7 @@ func run() error { mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{ DB: queueDB, Logger: logger, + LogLevel: os.Getenv("QUEUE_LOG_LEVEL"), MetricsScope: scope.SubScope("queue"), }) if err != nil { From a240b6b4f8a808b34c63a43ca05d00b29ff57a34 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Sat, 8 Aug 2026 08:30:43 -0700 Subject: [PATCH 4/5] docs(demo): correct the sample trail to states the pipeline emits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The sample table showed `accepted → started → batched → speculating → landed`, which cannot happen. `RequestStatusSpeculating` is defined in the entity package and never published by any non-test code — along with `validating`, `batching`, `speculated`, `building`, `built`, `waitingpath`, `landing` and `processing`. Nine of the eighteen statuses are dead vocabulary today. Getting this wrong in the one place that shows what the tool looks like is worse than saying nothing, because it invites exactly the wrong conclusion: a request sitting on `batched` for minutes looks stuck, when its batch is in fact speculating and building normally with nothing to report it. ### What? The sample is the real thing the renderer produces for a trail the pipeline can actually emit, and the text says plainly what the vocabulary is limited to and how to read a request that rests on `batched`. The gap itself is CODEM-426. ## Test Plan ✅ Sample generated from the renderer rather than written by hand, so the column widths and rule lengths are the ones the tool produces. --- doc/howto/PROVIDER-E2E.md | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/doc/howto/PROVIDER-E2E.md b/doc/howto/PROVIDER-E2E.md index fe257d89..e432d1ae 100644 --- a/doc/howto/PROVIDER-E2E.md +++ b/doc/howto/PROVIDER-E2E.md @@ -106,15 +106,17 @@ The table is there from the start — one row per land request, drawn before the ``` REQUEST CHANGES ELAPSED STAGE - ───────────── ─────── ─────── ─────────────────────────────────────────────────── - demo-queue/12 #31 34s accepted → started → batched → speculating → landed - demo-queue/13 #32 31s accepted → started → batched → speculating + ───────────── ─────── ─────── ───────────────────────────────────────────────── + demo-queue/12 #31 34s accepted → started → validated → batched → landed + demo-queue/13 #32 31s accepted → started → validated → batched demo-queue/14 #33 28s accepted → started ▸ 1 of 3 settled ``` -Each row shows the states its request passed through, not just the one it is in. That comes from the gateway's history API rather than from sampling the current status, so a request that batches and speculates between two ticks still shows both. `CHANGES` links to the pull request: on a terminal `#31` is clickable, and in a redirected run it is written out as a full URL instead. `ELAPSED` runs from the moment the gateway accepted the request and stops when it settles, so a finished row keeps the time it took rather than counting on. +Each row shows the states its request passed through, not just the one it is in. That comes from the gateway's history API rather than from sampling the current status, so a transition between two polls is not missed. `CHANGES` links to the pull request: on a terminal `#31` is clickable, and in a redirected run it is written out as a full URL instead. `ELAPSED` runs from the moment the gateway accepted the request and stops when it settles, so a finished row keeps the time it took rather than counting on. + +The trail is only as detailed as what the pipeline reports, which today is `accepted`, `started`, `validated`, `batched` and then a terminal `landed`, `error` or `cancelled`. The finer-grained statuses the API defines — `speculating`, `building`, `landing` and the rest — are never published, so a request sits on `batched` for the whole of its active life even while its batch is speculating and building. Do not read that as the request being stuck. `STACKED=true` is the exception to the overlap: one request carries the whole chain, so it can only go in once every pull request in it exists. That is the atomic-stack path — the whole set reaches `main` in a single push, and the table shows it as the single row it is. From ff81ffbf7058a8eba5e5f7703879f40a4d457c87 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Mon, 10 Aug 2026 14:18:23 -0700 Subject: [PATCH 5/5] feat(demo): spread each change over a sharded file tree MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Every demo change wrote exactly one file, and every one of those files went into a single flat `demo/` directory. Two things follow from that, and neither is what the demo wants to show. The directory grows without bound as runs accumulate, until it is the kind of directory nobody wants in a repository. And a pull request that adds one line to one file is not much of a pull request — it exercises none of the handling a real change needs, and reads as a toy in a walkthrough whose whole purpose is to look like the real thing. ### What? A change now writes several files, each at a path sharded into two levels of hex buckets under `demo/` — `demo/c2/91/--.txt`. The buckets are the leading bytes of the leaf name's SHA-256, so the spread is uniform without any coordination and stable across runs, and no directory becomes a dumping ground however many times the demo is run. Uniqueness is preserved, and it is load-bearing rather than incidental: the leaf carries the run tag, the change index and the file index, so no two files in a run and no two runs against one repository can name the same path. Independent changes that collided on content would make the run measure conflict handling instead of the throughput it exists to show. Two unrelated changes landing in the same bucket is expected and harmless — the bucket is only a directory, and it is the leaf that has to be distinct. How many files a change touches varies rather than being fixed, so a run does not produce a row of identically shaped pull requests. `-files` sets the floor (default 3, exposed as `FILES` on `make demo-pr`) and the actual count runs a little above it. The variation is derived from the run tag and the change index rather than from a clock, so replaying a tag reproduces the same run — a demo that cannot be reproduced is hard to talk about once something in it goes wrong. Each file is committed separately, so a change arrives as a multi-file, multi-commit pull request. That is closer to a real change, and it means the demo exercises replaying a range of commits rather than always handing the merger a single one. ## Test Plan ✅ `bazel test //service/submitqueue/demo/pr:go_default_test` — five new cases over the two pure functions: paths are unique across every change and file of two runs, the layout is the demo root plus hex buckets plus the leaf, 80 files spread over more than 50 distinct buckets (a layout that put everything in one directory would pass the uniqueness test alone), the file count respects its floor and clamps a non-positive one, and the count both varies across changes and reproduces for a given tag. ✅ Cross-checked the generated layout against an independent recomputation of the same hashes in shell, rather than only against the implementation's own output. Not yet exercised against GitHub: the path and count logic is covered above, but the multi-commit loop against the live API — several sequential commits on one branch, and the last SHA as the head the change URI pins — will first run on the next `make demo-pr`. --- Makefile | 4 +- doc/howto/PROVIDER-E2E.md | 5 +- service/submitqueue/demo/pr/main.go | 81 +++++++++++++++++++++--- service/submitqueue/demo/pr/main_test.go | 81 ++++++++++++++++++++++++ 4 files changed, 159 insertions(+), 12 deletions(-) diff --git a/Makefile b/Makefile index adf42144..6d4a5eed 100644 --- a/Makefile +++ b/Makefile @@ -55,6 +55,7 @@ export SQ_PROVIDER_CONFIG_DIR ?= $(REPO_ROOT)/service/submitqueue/demo/provider/ # Defaults for `make land` / `make demo-pr` against the provider demo stack. DEMO_REPO ?= behinddwalls/sq-demo COUNT ?= 3 +FILES ?= 3 STACKED ?= false LAND ?= true WATCH ?= true @@ -152,10 +153,11 @@ clean-proto: ## Clean generated proto files @rm -f $(foreach p,$(PROTO_PACKAGES),$(p)/protopb/*.pb.go $(p)/protopb/*.pb.yarpc.go) @echo "Proto clean complete!" -demo-pr: ## Create N PRs in the demo repo, enqueue each as it is created, and watch (COUNT=3; needs GITHUB_TOKEN) +demo-pr: ## Create N PRs in the demo repo, enqueue each as it is created, and watch (COUNT=3 FILES=3; needs GITHUB_TOKEN) @$(BAZEL) run //service/submitqueue/demo/pr -- \ -repo $(DEMO_REPO) \ -count $(COUNT) \ + -files $(FILES) \ -stacked=$(STACKED) \ -gateway $(GATEWAY_ADDR) \ -queue $(QUEUE) \ diff --git a/doc/howto/PROVIDER-E2E.md b/doc/howto/PROVIDER-E2E.md index e432d1ae..5c9a4542 100644 --- a/doc/howto/PROVIDER-E2E.md +++ b/doc/howto/PROVIDER-E2E.md @@ -96,6 +96,7 @@ Opening pull requests by hand gets old fast. `demo-pr` creates them, enqueues th ```bash make demo-pr # 3 independent PRs, each enqueued as it is created make demo-pr COUNT=8 # more traffic +make demo-pr FILES=8 # wider changes, more files per PR make demo-pr STACKED=true # one stack, enqueued as a single request make demo-pr LAND=false # create only, print the land command ``` @@ -120,7 +121,9 @@ The trail is only as detailed as what the pipeline reports, which today is `acce `STACKED=true` is the exception to the overlap: one request carries the whole chain, so it can only go in once every pull request in it exists. That is the atomic-stack path — the whole set reaches `main` in a single push, and the table shows it as the single row it is. -It talks to GitHub over the REST API with the same `GITHUB_TOKEN`, so it needs no clone and no git binary. Each run tags its branches with a timestamp so repeated runs do not collide, and each change edits its own file so independent changes do not conflict by accident. +It talks to GitHub over the REST API with the same `GITHUB_TOKEN`, so it needs no clone and no git binary. Each run tags its branches with a timestamp so repeated runs do not collide, and every file a change writes is at a path no other change uses, so independent changes do not conflict by accident. + +A change touches several files rather than one, each committed separately, so it arrives as a multi-file, multi-commit pull request — closer to a real change, and enough to exercise replaying a range of commits. `FILES` sets the floor (default 3); the actual count varies a little above it, derived from the run tag so replaying a tag reproduces the same run. Paths are sharded into two levels of hex buckets under `demo/` (`demo/c2/91/--.txt`), which keeps the tree from degenerating into one enormous directory as runs accumulate. The command exits non-zero if any request settles anywhere other than `landed`, so it works in a script. Piped to a file it prints a fresh table whenever a request moves — and not when only the clock did — instead of redrawing in place. diff --git a/service/submitqueue/demo/pr/main.go b/service/submitqueue/demo/pr/main.go index 228c26ec..7d401124 100644 --- a/service/submitqueue/demo/pr/main.go +++ b/service/submitqueue/demo/pr/main.go @@ -47,6 +47,7 @@ package main import ( "bytes" "context" + "crypto/sha256" "encoding/base64" "encoding/json" "flag" @@ -106,6 +107,7 @@ type config struct { repo string base string count int + files int stacked bool prefix string land bool @@ -123,6 +125,7 @@ func parseFlags() config { flag.StringVar(&c.repo, "repo", "behinddwalls/sq-demo", "scratch repository as owner/name") flag.StringVar(&c.base, "base", "main", "branch the changes target") flag.IntVar(&c.count, "count", 3, "how many pull requests to create") + flag.IntVar(&c.files, "files", 3, "fewest files each pull request touches; the actual count varies a little above it") flag.BoolVar(&c.stacked, "stacked", false, "chain the pull requests and enqueue them as one stack") flag.StringVar(&c.prefix, "prefix", "demo", "branch name prefix") flag.BoolVar(&c.land, "land", true, "enqueue each pull request as it is created") @@ -302,6 +305,51 @@ func (rw *row) stage() string { return absent } +// shardDirs is how many nested bucket directories a path carries under the demo +// root. Two levels of 256 buckets spread a run's files widely enough that no +// directory becomes a dumping ground, while staying shallow enough to read in a +// diff. +const shardDirs = 2 + +// 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. +// +// The directories are the leading bytes of the leaf's SHA-256, so files land in +// buckets that are uniform without any coordination and stable across runs. Two +// unrelated changes sharing a bucket is expected and harmless: the bucket is +// only a directory, and it is the leaf that has to be distinct. +func changeFilePath(tag string, change, file int) string { + leaf := fmt.Sprintf("%s-%d-%d.txt", tag, change, file) + sum := sha256.Sum256([]byte(leaf)) + + parts := make([]string, 0, shardDirs+2) + parts = append(parts, "demo") + for i := 0; i < shardDirs; i++ { + parts = append(parts, fmt.Sprintf("%02x", sum[i])) + } + parts = append(parts, leaf) + return strings.Join(parts, "/") +} + +// 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 pull requests. +// +// 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) +} + // createAndEnqueue opens the pull requests and puts them on the queue, filling // in the tracker's rows as it goes and reporting each step beneath the table. // @@ -311,10 +359,13 @@ func (rw *row) stage() string { // carries the whole chain, so it can only be submitted once the chain is // complete. // -// Each change edits its own file. 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. +// 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 pull request 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, gh *githubClient, @@ -340,12 +391,22 @@ func createAndEnqueue( return nil, fmt.Errorf("create branch %s: %w", branch, err) } - path := fmt.Sprintf("demo/%s-%d.txt", tag, i) - body := fmt.Sprintf("change %d of run %s\n", i, tag) - t.note("committing %s", path) - headSHA, err := gh.commitFile(ctx, branch, path, body, fmt.Sprintf("demo change %d (run %s)", i, tag)) - if err != nil { - return nil, fmt.Errorf("commit to %s: %w", branch, err) + // Each file is its own commit, so the pull request arrives as a range of + // commits rather than a single edit. The last one is the head the change + // URI pins. + var headSHA string + fileCount := changeFileCount(tag, i, cfg.files) + for k := 1; k <= fileCount; k++ { + path := changeFilePath(tag, i, k) + body := fmt.Sprintf("change %d of run %s\nfile %d of %d\n", i, tag, k, fileCount) + t.note("committing %s (%d/%d)", path, k, fileCount) + + message := fmt.Sprintf("demo change %d (run %s): file %d of %d", i, tag, k, fileCount) + sha, err := gh.commitFile(ctx, branch, path, body, message) + if err != nil { + return nil, fmt.Errorf("commit %s to %s: %w", path, branch, err) + } + headSHA = sha } t.note("opening pull request for %s", branch) diff --git a/service/submitqueue/demo/pr/main_test.go b/service/submitqueue/demo/pr/main_test.go index 2ece441d..4b1c9988 100644 --- a/service/submitqueue/demo/pr/main_test.go +++ b/service/submitqueue/demo/pr/main_test.go @@ -656,3 +656,84 @@ func head(s string, n int) string { } return s[:n] } + +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, 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_ShardsUnderTheDemoRoot(t *testing.T) { + path := changeFilePath("0810-1203", 1, 1) + + parts := strings.Split(path, "/") + require.Len(t, parts, shardDirs+2, "demo root, %d bucket dirs, and the leaf", shardDirs) + assert.Equal(t, "demo", parts[0]) + for _, bucket := range parts[1 : len(parts)-1] { + assert.Len(t, bucket, 2, "each bucket is one hex byte") + assert.Regexp(t, "^[0-9a-f]{2}$", bucket) + } + assert.Equal(t, "0810-1203-1-1.txt", parts[len(parts)-1]) +} + +func TestChangeFilePath_SpreadsAcrossManyBuckets(t *testing.T) { + // A layout that puts everything in one directory would satisfy the + // uniqueness test above while defeating the point of sharding. + buckets := make(map[string]struct{}) + for change := 1; change <= 20; change++ { + for file := 1; file <= 4; file++ { + parts := strings.Split(changeFilePath("0810-1203", change, file), "/") + buckets[strings.Join(parts[1:len(parts)-1], "/")] = struct{}{} + } + } + assert.Greater(t, len(buckets), 50, "80 files should land in many distinct buckets") +} + +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") +}