diff --git a/Makefile b/Makefile index 7d29deeb..6d4a5eed 100644 --- a/Makefile +++ b/Makefile @@ -52,7 +52,13 @@ 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 +FILES ?= 3 +STACKED ?= false +LAND ?= true +WATCH ?= true QUEUE ?= demo-queue STRATEGY ?= SQUASH_REBASE GATEWAY_ADDR ?= localhost:8081 @@ -147,6 +153,17 @@ 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 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) \ + -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..5c9a4542 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,44 @@ 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 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 +``` + +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. + +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 ELAPSED STAGE + ───────────── ─────── ─────── ───────────────────────────────────────────────── + 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 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. + +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. + ## Watching it work ```bash @@ -101,6 +139,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/demo/pr/BUILD.bazel b/service/submitqueue/demo/pr/BUILD.bazel new file mode 100644 index 00000000..28729f35 --- /dev/null +++ b/service/submitqueue/demo/pr/BUILD.bazel @@ -0,0 +1,35 @@ +load("@rules_go//go:def.bzl", "go_binary", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["main.go"], + importpath = "github.com/uber/submitqueue/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", + "//submitqueue/entity: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"], +) + +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 new file mode 100644 index 00000000..7d401124 --- /dev/null +++ b/service/submitqueue/demo/pr/main.go @@ -0,0 +1,1069 @@ +// 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 table 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. +// +// 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. +package main + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "flag" + "fmt" + "net/http" + "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" +) + +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. 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() + 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 + files 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.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") + 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)) + + // 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 + } + + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.settled: + } + return t.conclude() +} + +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 +} + +// 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 +} + +// 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. +// +// 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. +// +// 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, + client pb.SubmitQueueGatewayClient, + cfg config, + strategy mergestrategypb.Strategy, + tag, baseSHA string, + t *tracker, +) ([]change, error) { + created := make([]change, 0, cfg.count) + + parentBranch, parentSHA := cfg.base, baseSHA + for i := 1; i <= cfg.count; i++ { + // A stack is one request, so every change lands on the single row. + 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, fmt.Errorf("create branch %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) + number, url, err := gh.openPR(ctx, fmt.Sprintf("demo change %d (run %s)", i, tag), branch, parentBranch) + if err != nil { + return 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) + 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 + // content and its pull request is based on this branch. + parentBranch, parentSHA = branch, headSHA + continue + } + if !cfg.land { + continue + } + t.note("enqueuing #%d", number) + sqid, err := enqueue(ctx, client, cfg, strategy, []change{c}) + if err != nil { + return nil, err + } + 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 { + t.note("enqueuing the stack") + sqid, err := enqueue(ctx, client, cfg, strategy, created) + if err != nil { + return nil, err + } + t.update(func() { t.rows[0].sqid, t.rows[0].submitted = sqid, time.Now() }) + } + return created, nil +} + +// 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, +) (string, error) { + uris := make([]string, 0, len(changes)) + for _, c := range changes { + uris = append(uris, c.uri) + } + + resp, err := client.Land(ctx, &pb.LandRequest{ + Queue: cfg.queue, + Change: &changepb.Change{Uris: uris}, + Strategy: strategy, + }) + if err != nil { + return "", fmt.Errorf("land %s failed: %w", labelsOf(changes), err) + } + return resp.Sqid, 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 + } + } + 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 + 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}) + } + + 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() + } + } + 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() +} + +// 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 + } + if len(trail) > 0 && trail[len(trail)-1] == e.Status { + continue + } + 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++ + } + } + 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(rows []*row) error { + var failed []string + 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(rows), strings.Join(failed, ", ")) + } + 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 { + inPlace 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() *renderer { + info, err := os.Stdout.Stat() + tty := err == nil && info.Mode()&os.ModeCharDevice != 0 + return &renderer{ + inPlace: tty, + wRequest: len("REQUEST"), + wChanges: len("CHANGES"), + wElapsed: len("ELAPSED"), + wStage: len("STAGE"), + } +} + +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.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 +} + +// 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 { + 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()) + } + 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 n < 1 { + return "" + } + if utf8.RuneCountInString(s) <= n { + return s + } + return string([]rune(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/pr/main_test.go b/service/submitqueue/demo/pr/main_test.go new file mode 100644 index 00000000..4b1c9988 --- /dev/null +++ b/service/submitqueue/demo/pr/main_test.go @@ -0,0 +1,739 @@ +// 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] +} + +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") +} 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 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 {