diff --git a/doc/rfc/runway/workflow.md b/doc/rfc/runway/workflow.md index 23e3fb084..0fa6d5767 100644 --- a/doc/rfc/runway/workflow.md +++ b/doc/rfc/runway/workflow.md @@ -79,7 +79,7 @@ Together these guarantee the client's correlation id always resolves: the primar Runway has no persistent state — no request store, no job store, no database. Idempotency is achieved through the VCS contract: merge detects already-pushed changes (revisions reachable from HEAD) and treats them as already-landed. Merge-conflict check is read-only and naturally idempotent. -A committing merge is also atomic against the merge target: it updates the target at most once per request, and afterwards either every step of the request is reachable from the target or the target is unchanged. A retried redelivery therefore either replays cleanly against the same unchanged target or finds its work already landed. +Each step of a committing merge lands atomically and in order, and the merge stops at the first step that fails: ordering cannot be guaranteed past a failure, so later steps are not attempted, and the steps before it stay landed and are reported in the `FAILED` result. A retried redelivery therefore finds the steps that landed already on the target and skips them. ## Ownership by service diff --git a/runway/controller/merge/merge.go b/runway/controller/merge/merge.go index 0d9c918e1..8636dca8f 100644 --- a/runway/controller/merge/merge.go +++ b/runway/controller/merge/merge.go @@ -125,11 +125,17 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er "queue_name", request.GetQueueName(), ) } - result = &runwaymq.MergeResult{ + failed := &runwaymq.MergeResult{ Id: request.GetId(), Outcome: runwaypb.Outcome_FAILED, Reason: err.Error(), } + // A merger that lands step by step reports the steps that landed before + // the failure; they stay landed, so the caller must see them. + if result != nil { + failed.Steps = result.GetSteps() + } + result = failed } // Echo the request's queue name so the consumer can route the result by diff --git a/runway/controller/merge/merge_test.go b/runway/controller/merge/merge_test.go index 9e2ec49d1..488935d15 100644 --- a/runway/controller/merge/merge_test.go +++ b/runway/controller/merge/merge_test.go @@ -217,6 +217,61 @@ func TestProcess_MergeConflict(t *testing.T) { assert.NotEmpty(t, result.Reason) } +func TestProcess_PartialLandIsReportedOnFailure(t *testing.T) { + ctrl := gomock.NewController(t) + + partial := &runwaymq.MergeResult{ + Id: testID, + Outcome: runwaypb.Outcome_FAILED, + Steps: []*runwaymq.StepResult{ + {StepId: "step-1", Outputs: []*runwaymq.StepOutput{{Id: "abc123"}}}, + {StepId: "step-2", Reason: "conflict"}, + }, + } + m := mergermock.NewMockMerger(ctrl) + m.EXPECT().Merge(gomock.Any(), gomock.Any()).Return(partial, fmt.Errorf("step-2: %w", merger.ErrConflict)) + + factory := mergermock.NewMockFactory(ctrl) + factory.EXPECT().For(merger.Config{QueueName: testQueue}).Return(m, nil) + + var gotPayload []byte + pub := queuemock.NewMockPublisher(ctrl) + pub.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, _ string, msg entityqueue.Message) error { + gotPayload = msg.Payload + return nil + }, + ) + q := queuemock.NewMockQueue(ctrl) + q.EXPECT().Publisher().Return(pub).AnyTimes() + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: runwaymq.TopicKeyMergeSignal, Name: "merge-signal", Queue: q}, + }) + require.NoError(t, err) + + controller := newController(t, factory, registry) + + req := &runwaymq.MergeRequest{ + Id: testID, + QueueName: testQueue, + Steps: []*runwaymq.MergeStep{{StepId: "step-1"}, {StepId: "step-2"}, {StepId: "step-3"}}, + } + delivery := newDelivery(t, ctrl, requestPayload(t, req)) + + require.NoError(t, controller.Process(context.Background(), delivery)) + + result := &runwaymq.MergeResult{} + require.NoError(t, runwaymq.Unmarshal(gotPayload, result)) + assert.Equal(t, runwaypb.Outcome_FAILED, result.Outcome) + assert.NotEmpty(t, result.Reason) + require.Len(t, result.Steps, 2) + assert.Equal(t, "step-1", result.Steps[0].StepId) + require.Len(t, result.Steps[0].Outputs, 1) + assert.Equal(t, "abc123", result.Steps[0].Outputs[0].Id) + assert.Equal(t, "step-2", result.Steps[1].StepId) + assert.NotEmpty(t, result.Steps[1].Reason) +} + func TestProcess_InvalidRequest(t *testing.T) { ctrl := gomock.NewController(t) diff --git a/runway/extension/merger/github/BUILD.bazel b/runway/extension/merger/github/BUILD.bazel new file mode 100644 index 000000000..182759e8b --- /dev/null +++ b/runway/extension/merger/github/BUILD.bazel @@ -0,0 +1,44 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "changeref.go", + "classifier.go", + "client.go", + "github_merger.go", + ], + importpath = "github.com/uber/submitqueue/runway/extension/merger/github", + visibility = ["//visibility:public"], + deps = [ + "//api/base/mergestrategy/protopb:go_default_library", + "//api/runway/messagequeue:go_default_library", + "//api/runway/messagequeue/protopb:go_default_library", + "//platform/base/change/github:go_default_library", + "//platform/errs:go_default_library", + "//platform/http:go_default_library", + "//platform/metrics:go_default_library", + "//runway/extension/merger:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["github_merger_test.go"], + embed = [":go_default_library"], + deps = [ + "//api/base/change/protopb:go_default_library", + "//api/base/mergestrategy/protopb:go_default_library", + "//api/runway/messagequeue:go_default_library", + "//api/runway/messagequeue/protopb:go_default_library", + "//platform/errs:go_default_library", + "//platform/http:go_default_library", + "//runway/extension/merger:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//zaptest:go_default_library", + ], +) diff --git a/runway/extension/merger/github/README.md b/runway/extension/merger/github/README.md new file mode 100644 index 000000000..17efa02af --- /dev/null +++ b/runway/extension/merger/github/README.md @@ -0,0 +1,51 @@ +# github merger + +A `merger.Merger` that lands `github://` changes through the GitHub REST API instead of a local checkout. GitHub performs the merge itself, so branch rules, the pull request's merged state, merge commit attribution and the rebasing of a stack's remaining pull requests are all GitHub's. It is constructed by the wiring layer (see [`service/runway`](../../../../service/runway)) with an HTTP client, the repository it serves (host, owner, repo), the trunk branch and the default strategy. + +## Model + +A request is an ordered list of steps, and a step's change is an ordered list of URIs — a stack, bottom first. This merger maps that directly onto GitHub's [stacked pull requests](https://docs.github.com/en/pull-requests/get-started/about-stacked-prs): **one step is one GitHub stack**, and a single pull request is the one-element case. + +A step is landed with one call to the [asynchronous merge API](https://docs.github.com/en/rest/pulls/pulls#merge-a-pull-request-asynchronously) on the step's top pull request. For a stacked pull request GitHub merges it together with every unmerged pull request below it, as one operation that either lands all of them or none, ordered bottom-up in the resulting history. Pull requests above the step's top stay open and are rebased onto the trunk by GitHub. The merge runs with `direct_merge`: Runway is the queue, so handing the pull requests to GitHub's own merge queue would let a second queue reorder what this one decided. + +Each step's strategy maps onto a GitHub merge method — `REBASE` to `rebase`, `SQUASH_REBASE` to `squash`, `MERGE` to `merge` — and `DEFAULT` resolves to the configured default first. `PROMOTE` advances a ref to an existing revision, which no pull request merge expresses, so it is an invalid request here; the git merger serves it. + +Each URI's output is the merge commit GitHub records for its pull request: the squash commit, the merge commit, or, for a rebase, the last commit the rebase created for that pull request. That is one output per URI, where the git merger reports one per created commit under `REBASE`. It is read from the pull request's `merged` issue event, because API version 2026-03-10 no longer reports `merge_commit_sha` on a merged pull request. GitHub reports a stack merge settled a moment before every pull request in it shows its merge, so the merger re-reads until each is recorded. + +## What a step must be + +The URIs of a step must be something GitHub will land as one stack onto the target, and anything else is refused as an invalid request: + +- Every URI names a pull request in the configured repository on the configured host, at the head commit the pull request still has. A head that moved is stale; a closed, unmerged pull request or a draft cannot be landed. +- Pull requests that are already merged must be a prefix of the list, since a stack merges bottom-up; they are skipped. +- One remaining pull request must be based on the target. A stacked pull request listed on its own is based on the one below it, and merging it would land that one too. +- Several remaining pull requests must be the unmerged bottom of one GitHub stack based on the target, in the stack's order and with no gap. Separate pull requests that each target the trunk, a stack listed out of order or with a pull request missing, and pull requests from two stacks are all refused. The merger never creates or edits a stack on the author's behalf. + +These checks run at the mergeability check, so SubmitQueue rejects such a request at validation rather than after batching it. `Merge` runs them again, before submitting anything, because a stack can be edited between validation and landing. + +## Mergeability check + +`CheckMergeability` writes nothing. After the checks above it reads GitHub's `mergeable` verdict for every remaining pull request, re-reading while GitHub reports it as still being computed; a pull request that is not mergeable, or whose `mergeable_state` is `dirty`, is a conflict. + +GitHub computes that verdict for each pull request against its own base, so a check sees conflicts within a stack and against the trunk, but not between the steps of one request. Those surface when the merge is attempted. + +## Atomicity + +Each step is one GitHub merge, so each step lands atomically. Steps land in request order, and `Merge` stops at the first that fails: ordering cannot be guaranteed past a failure, so later steps are not attempted, and the steps before it stay landed. The `FAILED` result lists a `StepResult` per landed step with its outputs, then one for the failed step with its reason. + +## Idempotency and redelivery + +A redelivered request converges instead of merging twice. Already-merged pull requests are skipped and report their recorded merge commit, so a step that fully landed before a crash is reported without another call. An already-merged pull request is trusted as landed on the target; no further check is made, since one would cost a call per pull request on every redelivery. A submission GitHub already has in flight answers with that request's id (HTTP 409); it is adopted and polled only when it merges the same head with the same method, and is an invalid request otherwise. + +## Failure classification + +- `merger.ErrInvalidRequest` (terminal): a malformed or foreign URI, an unsupported strategy, a stale head, a closed or draft pull request, a step that is not a stack based on the target, or GitHub refusing the merge as asked (HTTP 400/422, e.g. required checks not satisfied). +- `merger.ErrConflict` (terminal): a pull request that is not mergeable at the check, or a merge GitHub reports as failed. GitHub does not separate a conflict from other merge failures there; its message is the reason. +- `ErrMergePending` (retryable, via this package's `Classifier`): GitHub had not settled a merge, or a pull request's mergeability, within the poll budget. The work is still in flight on GitHub's side. +- Anything else is a plain error. HTTP rejections keep their status code (`platform/http.StatusError`) so `platform/errs/http` can retry a 5xx or 429 and dead-letter a 403 or 404. A 404 is not treated as terminal here: GitHub answers 404 for a repository the token cannot see, and that is a deployment fault, not a property of the request. + +## Auth and hosts + +The merger adds no credential and no base URL. Its HTTP client's transport roots relative paths at the API (for example `platform/http.NewClient("https://api.github.com")`) and authenticates them, so the wiring decides the scheme — a static token, an App installation token source, or anything an internal deployment injects. The token needs write access to the repository, and the right to bypass branch rules only when `BypassRules` is set. + +The stacks and asynchronous merge APIs exist on github.com. GitHub Enterprise Server has not shipped stacked pull requests yet; the host, repository and API base URL are configuration, so an Enterprise Server instance needs no code change once it does. diff --git a/runway/extension/merger/github/changeref.go b/runway/extension/merger/github/changeref.go new file mode 100644 index 000000000..2b380d7c1 --- /dev/null +++ b/runway/extension/merger/github/changeref.go @@ -0,0 +1,49 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package github + +import ( + "fmt" + + entitygithub "github.com/uber/submitqueue/platform/base/change/github" + "github.com/uber/submitqueue/runway/extension/merger" +) + +// changeRef is a change URI reduced to the pull request it names. +type changeRef struct { + // PRNumber is the pull request number in the merger's repository. + PRNumber int + // SHA is the head commit the URI pins the pull request to. + SHA string + // Label is a short human-readable name, "owner/repo#n". + Label string +} + +// resolveChange parses a change URI and rejects one that does not name a pull +// request in this merger's repository. Both failures are terminal. +func (m *githubMerger) resolveChange(uri string) (changeRef, error) { + cid, err := entitygithub.ParseChangeID(uri) + if err != nil { + return changeRef{}, fmt.Errorf("%w: invalid change URI %q: %v", merger.ErrInvalidRequest, uri, err) + } + if cid.Host != m.host || cid.Org != m.owner || cid.Repo != m.repo { + return changeRef{}, fmt.Errorf("%w: change URI %q is not in %s/%s/%s", merger.ErrInvalidRequest, uri, m.host, m.owner, m.repo) + } + return changeRef{ + PRNumber: cid.PRNumber, + SHA: cid.HeadCommitSHA, + Label: fmt.Sprintf("%s#%d", cid.OwnerRepo(), cid.PRNumber), + }, nil +} diff --git a/runway/extension/merger/github/classifier.go b/runway/extension/merger/github/classifier.go new file mode 100644 index 000000000..d4ae2f7d9 --- /dev/null +++ b/runway/extension/merger/github/classifier.go @@ -0,0 +1,33 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package github + +import "github.com/uber/submitqueue/platform/errs" + +// Classifier implements errs.Classifier for this merger's own sentinel: an +// ErrMergePending is a retryable dependency failure, since GitHub is still +// working and a redelivery converges on the same merge. HTTP status and +// transport failures are left to platform/errs/http. +var Classifier errs.Classifier = classifier{} + +type classifier struct{} + +// Classify inspects a single node; the classifier-processor walks the chain. +func (classifier) Classify(err error) errs.Verdict { + if err == ErrMergePending { + return errs.InfraDependencyRetryable + } + return errs.Unknown +} diff --git a/runway/extension/merger/github/client.go b/runway/extension/merger/github/client.go new file mode 100644 index 000000000..c46d26679 --- /dev/null +++ b/runway/extension/merger/github/client.go @@ -0,0 +1,272 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package github + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strconv" + + phttp "github.com/uber/submitqueue/platform/http" +) + +const githubAPIVersion = "2026-03-10" + +// pullStateOpen is the REST API state of a pull request that is neither merged +// nor closed. +const pullStateOpen = "open" + +// mergeableStateDirty is GitHub's mergeable_state for a pull request whose +// head conflicts with its base. +const mergeableStateDirty = "dirty" + +// Async merge request statuses. +const ( + mergeStatusPending = "pending" + mergeStatusMerged = "merged" + mergeStatusEnqueued = "enqueued" + mergeStatusFailed = "failed" +) + +// mergeActionDirect merges immediately instead of entering GitHub's own merge +// queue: Runway is the queue, so a second one would reorder what it decided. +const mergeActionDirect = "direct_merge" + +// pullRequest is the subset of a GitHub pull request the merger reads. +type pullRequest struct { + Number int `json:"number"` + State string `json:"state"` + Draft bool `json:"draft"` + Merged bool `json:"merged"` + Mergeable *bool `json:"mergeable"` + MergeableState string `json:"mergeable_state"` + Head gitRef `json:"head"` + Base gitRef `json:"base"` +} + +// gitRef is a branch name and the commit it points at. +type gitRef struct { + Ref string `json:"ref"` + SHA string `json:"sha"` +} + +// stack is a GitHub pull request stack; PullRequests are ordered bottom to top. +type stack struct { + Number int `json:"number"` + Open bool `json:"open"` + Base gitRef `json:"base"` + PullRequests []stackPull `json:"pull_requests"` +} + +// stackPull is one member of a stack. +type stackPull struct { + Number int `json:"number"` + State string `json:"state"` + MergedAt *string `json:"merged_at"` + Head gitRef `json:"head"` +} + +func (p stackPull) unmerged() bool { + return p.State == pullStateOpen && p.MergedAt == nil +} + +// asyncMergeRequest is the body of PUT .../merge-async. +type asyncMergeRequest struct { + SHA string `json:"sha"` + MergeMethod string `json:"merge_method"` + MergeAction string `json:"merge_action"` + BypassRules bool `json:"bypass_rules,omitempty"` +} + +// asyncMergeStatus is both the submit response and the poll response of the +// async merge API. +type asyncMergeStatus struct { + Status string `json:"status"` + Details struct { + Message string `json:"message"` + UUID string `json:"uuid"` + SHA string `json:"sha"` + ExpectedHeadSHA string `json:"expected_head_sha"` + MergeMethod string `json:"merge_method"` + } `json:"details"` +} + +// submitOutcome classifies a merge-async submission. +type submitOutcome int + +const ( + // submitAccepted means a merge request is in flight and must be polled; + // that includes adopting one an earlier delivery already submitted. + submitAccepted submitOutcome = iota + 1 + // submitDone means GitHub answered with a final status straight away. + submitDone + // submitRejected means GitHub refused the merge as asked (closed, draft, + // failed validation); retrying the same request cannot change that. + submitRejected + // submitConflicting means a merge request is already in flight for the pull + // request but for a different head or merge method, so it is not this one. + submitConflicting +) + +// client is a thin wrapper over the GitHub REST endpoints the merger needs, +// bound to one repository. Base URL and auth come from the injected +// *http.Client's transport. +type client struct { + httpClient *http.Client + owner string + repo string +} + +func (c *client) getPull(ctx context.Context, number int) (pullRequest, error) { + var pr pullRequest + if err := c.getJSON(ctx, c.repoPath("pulls", strconv.Itoa(number)), &pr); err != nil { + return pullRequest{}, fmt.Errorf("get pull request #%d: %w", number, err) + } + return pr, nil +} + +// issueEvent is one entry of a pull request's issue event timeline. +type issueEvent struct { + Event string `json:"event"` + CommitID string `json:"commit_id"` +} + +// issueEventsPageSize is the largest page the issue events endpoint serves. +const issueEventsPageSize = 100 + +// getMergeCommit returns the commit a merged pull request landed as, and false +// while GitHub has not recorded it yet. API version 2026-03-10 dropped +// merge_commit_sha from the pull request resource; the "merged" issue event is +// where that version still reports it. +func (c *client) getMergeCommit(ctx context.Context, number int) (string, bool, error) { + var mergeCommit string + for page := 1; ; page++ { + var events []issueEvent + path := fmt.Sprintf("%s?per_page=%d&page=%d", c.repoPath("issues", strconv.Itoa(number), "events"), issueEventsPageSize, page) + if err := c.getJSON(ctx, path, &events); err != nil { + return "", false, fmt.Errorf("list events of pull request #%d: %w", number, err) + } + for _, e := range events { + if e.Event == "merged" && e.CommitID != "" { + mergeCommit = e.CommitID + } + } + if len(events) < issueEventsPageSize { + return mergeCommit, mergeCommit != "", nil + } + } +} + +// getStackForPull returns the stack containing the pull request, and false +// when it belongs to none. +func (c *client) getStackForPull(ctx context.Context, number int) (stack, bool, error) { + var stacks []stack + path := c.repoPath("stacks") + "?pull_request=" + strconv.Itoa(number) + if err := c.getJSON(ctx, path, &stacks); err != nil { + return stack{}, false, fmt.Errorf("list stacks for pull request #%d: %w", number, err) + } + for _, s := range stacks { + for _, p := range s.PullRequests { + if p.Number == number { + return s, true, nil + } + } + } + return stack{}, false, nil +} + +// submitMerge asks GitHub to merge the pull request (and, for a stacked one, +// every unmerged pull request below it). A 409 carries the request already in +// flight; it is adopted only when it merges the same head with the same method, +// so a redelivered merge converges on its own GitHub request without taking +// over a different one. +func (c *client) submitMerge(ctx context.Context, number int, req asyncMergeRequest) (asyncMergeStatus, submitOutcome, error) { + body, err := json.Marshal(req) + if err != nil { + return asyncMergeStatus{}, 0, fmt.Errorf("marshal merge request: %w", err) + } + path := c.repoPath("pulls", strconv.Itoa(number), "merge-async") + status, respBody, err := phttp.SendRequest(ctx, c.httpClient, http.MethodPut, path, body, setHeaders) + if err != nil { + return asyncMergeStatus{}, 0, fmt.Errorf("merge pull request #%d: %w", number, err) + } + + var out asyncMergeStatus + switch status { + case http.StatusOK, http.StatusAccepted, http.StatusConflict: + if err := json.Unmarshal(respBody, &out); err != nil { + return asyncMergeStatus{}, 0, fmt.Errorf("merge pull request #%d: unmarshal response: %w", number, err) + } + case http.StatusBadRequest, http.StatusUnprocessableEntity: + _ = json.Unmarshal(respBody, &out) + return out, submitRejected, nil + default: + return asyncMergeStatus{}, 0, fmt.Errorf("merge pull request #%d: %w", number, phttp.NewStatusError(status, respBody)) + } + + switch { + case status == http.StatusConflict && out.Details.UUID != "": + if out.Details.ExpectedHeadSHA != req.SHA || out.Details.MergeMethod != req.MergeMethod { + return out, submitConflicting, nil + } + return out, submitAccepted, nil + case status == http.StatusConflict: + return asyncMergeStatus{}, 0, fmt.Errorf("merge pull request #%d: %w", number, phttp.NewStatusError(status, respBody)) + case out.Status == mergeStatusPending && out.Details.UUID != "": + return out, submitAccepted, nil + default: + return out, submitDone, nil + } +} + +func (c *client) getMergeStatus(ctx context.Context, number int, uuid string) (asyncMergeStatus, error) { + var out asyncMergeStatus + path := c.repoPath("pulls", strconv.Itoa(number), "merge-async", uuid) + if err := c.getJSON(ctx, path, &out); err != nil { + return asyncMergeStatus{}, fmt.Errorf("get merge status %s for pull request #%d: %w", uuid, number, err) + } + return out, nil +} + +func (c *client) getJSON(ctx context.Context, path string, out any) error { + status, respBody, err := phttp.SendRequest(ctx, c.httpClient, http.MethodGet, path, nil, setHeaders) + if err != nil { + return err + } + if status != http.StatusOK { + return phttp.NewStatusError(status, respBody) + } + if err := json.Unmarshal(respBody, out); err != nil { + return fmt.Errorf("unmarshal response: %w", err) + } + return nil +} + +func (c *client) repoPath(segments ...string) string { + path := "/repos/" + url.PathEscape(c.owner) + "/" + url.PathEscape(c.repo) + for _, s := range segments { + path += "/" + url.PathEscape(s) + } + return path +} + +func setHeaders(req *http.Request) { + req.Header.Set("Accept", "application/vnd.github+json") + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-GitHub-Api-Version", githubAPIVersion) +} diff --git a/runway/extension/merger/github/github_merger.go b/runway/extension/merger/github/github_merger.go new file mode 100644 index 000000000..3c6501801 --- /dev/null +++ b/runway/extension/merger/github/github_merger.go @@ -0,0 +1,503 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package github implements merger.Merger on top of the GitHub REST API. Each +// request step is one GitHub pull request stack (a single pull request being +// the one-element case) and is landed by GitHub's asynchronous merge API, so +// branch rules, the merged pull request state and stack rebasing stay +// GitHub's. See README.md for the model and its guarantees. +package github + +import ( + "context" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "github.com/uber-go/tally" + "go.uber.org/zap" + + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" + coremetrics "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/runway/extension/merger" +) + +// ErrMergePending signals that GitHub had not settled a merge, or a pull +// request's mergeability, within the merger's poll budget. The work is still in +// flight on GitHub's side, so the delivery should be retried: a redelivery +// adopts the in-flight request or finds the pull requests already merged. +var ErrMergePending = errors.New("github merge still pending") + +const ( + defaultPollInterval = 2 * time.Second + defaultMaxPollDuration = 10 * time.Minute +) + +// Params holds the dependencies for the GitHub Merger. +type Params struct { + // HTTPClient sends every API call. Its transport resolves relative paths + // against the API root (e.g. platform/http.NewClient("https://api.github.com")) + // and authenticates them; the merger adds neither. + HTTPClient *http.Client + // Host is the GitHub instance change URIs must name (e.g. "github.com"). + Host string + // Owner and Repo name the repository this merger lands into. + Owner string + Repo string + // Target is the trunk branch every stack must be based on (e.g. "main"). + Target string + // DefaultStrategy resolves a step whose strategy is DEFAULT. Must be + // REBASE, SQUASH_REBASE or MERGE. + DefaultStrategy mergestrategypb.Strategy + // BypassRules asks GitHub to merge past branch rules. The token must + // belong to an actor allowed to bypass them. + BypassRules bool + // PollInterval is the wait between status reads. Zero uses the default; a + // negative value polls without waiting. + PollInterval time.Duration + // MaxPollDuration bounds how long one call waits for GitHub to settle + // before returning ErrMergePending. Zero uses the default. + MaxPollDuration time.Duration + // Logger is the structured logger. + Logger *zap.SugaredLogger + // MetricsScope is the metrics scope for instrumentation. + MetricsScope tally.Scope +} + +type githubMerger struct { + client *client + host string + owner string + repo string + target string + defaultStrategy mergestrategypb.Strategy + bypassRules bool + pollInterval time.Duration + maxPollDuration time.Duration + logger *zap.SugaredLogger + metricsScope tally.Scope +} + +var _ merger.Merger = (*githubMerger)(nil) + +// stepPull is one URI of a step together with the pull request it names. +type stepPull struct { + ref changeRef + pr pullRequest +} + +// New constructs a GitHub-backed Merger for one repository. +func New(params Params) (merger.Merger, error) { + if params.HTTPClient == nil { + return nil, errors.New("github merger: HTTPClient is required") + } + for name, v := range map[string]string{"host": params.Host, "owner": params.Owner, "repo": params.Repo, "target": params.Target} { + if v == "" { + return nil, fmt.Errorf("github merger: %s is required", name) + } + } + if _, ok := mergeMethods[params.DefaultStrategy]; !ok { + return nil, fmt.Errorf("github merger: default strategy must be REBASE, SQUASH_REBASE or MERGE, got %v", params.DefaultStrategy) + } + pollInterval := params.PollInterval + if pollInterval == 0 { + pollInterval = defaultPollInterval + } + maxPollDuration := params.MaxPollDuration + if maxPollDuration <= 0 { + maxPollDuration = defaultMaxPollDuration + } + return &githubMerger{ + client: &client{httpClient: params.HTTPClient, owner: params.Owner, repo: params.Repo}, + host: params.Host, + owner: params.Owner, + repo: params.Repo, + target: params.Target, + defaultStrategy: params.DefaultStrategy, + bypassRules: params.BypassRules, + pollInterval: pollInterval, + maxPollDuration: maxPollDuration, + logger: params.Logger.Named("github_merger"), + metricsScope: params.MetricsScope.SubScope("github_merger"), + }, nil +} + +// mergeMethods maps the strategies this merger supports onto GitHub merge +// methods. PROMOTE has no pull request equivalent. +var mergeMethods = map[mergestrategypb.Strategy]string{ + mergestrategypb.Strategy_REBASE: "rebase", + mergestrategypb.Strategy_SQUASH_REBASE: "squash", + mergestrategypb.Strategy_MERGE: "merge", +} + +// CheckMergeability verifies, without writing anything, that every step is a +// GitHub stack based on the target and that none of its pull requests +// conflicts with its base. +func (m *githubMerger) CheckMergeability(ctx context.Context, req *runwaymq.MergeRequest) (ret *runwaymq.MergeResult, retErr error) { + op := coremetrics.Begin(m.metricsScope, "check", coremetrics.LongLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := m.validateRequest(req); err != nil { + return nil, err + } + results := make([]*runwaymq.StepResult, 0, len(req.GetSteps())) + for _, step := range req.GetSteps() { + pulls, err := m.loadValidStepPulls(ctx, step) + if err != nil { + return nil, err + } + for _, p := range unmergedPulls(pulls) { + if err := m.checkPullMergeable(ctx, step, p); err != nil { + return nil, err + } + } + results = append(results, &runwaymq.StepResult{StepId: step.GetStepId()}) + } + return successResult(req, results), nil +} + +// Merge lands each step's stack in request order, one GitHub merge per step, +// and reports the merge commit of every pull request as that URI's output. +func (m *githubMerger) Merge(ctx context.Context, req *runwaymq.MergeRequest) (ret *runwaymq.MergeResult, retErr error) { + op := coremetrics.Begin(m.metricsScope, "merge", coremetrics.LongLatencyBuckets) + defer func() { op.Complete(retErr) }() + + if err := m.validateRequest(req); err != nil { + return nil, err + } + results := make([]*runwaymq.StepResult, 0, len(req.GetSteps())) + for _, step := range req.GetSteps() { + outputs, err := m.mergeStep(ctx, step) + if err != nil { + if !merger.IsTerminal(err) { + return nil, err + } + // Ordering cannot be guaranteed past a failed step, so later steps + // are not attempted; the ones before it stay landed. + results = append(results, &runwaymq.StepResult{StepId: step.GetStepId(), Reason: err.Error()}) + return &runwaymq.MergeResult{ + Id: req.GetId(), + Outcome: runwaypb.Outcome_FAILED, + Reason: err.Error(), + Steps: results, + }, err + } + results = append(results, &runwaymq.StepResult{StepId: step.GetStepId(), Outputs: outputs}) + } + return successResult(req, results), nil +} + +// loadValidStepPulls loads a step's pull requests and rejects the step unless +// GitHub would land them as one stack onto the target. +func (m *githubMerger) loadValidStepPulls(ctx context.Context, step *runwaymq.MergeStep) ([]stepPull, error) { + pulls, err := m.loadStepPulls(ctx, step) + if err != nil { + return nil, err + } + if err := m.validateStepIsGitHubStack(ctx, step, pulls); err != nil { + return nil, err + } + return pulls, nil +} + +func (m *githubMerger) mergeStep(ctx context.Context, step *runwaymq.MergeStep) ([]*runwaymq.StepOutput, error) { + pulls, err := m.loadValidStepPulls(ctx, step) + if err != nil { + return nil, err + } + pending := unmergedPulls(pulls) + if len(pending) > 0 { + top := pending[len(pending)-1] + method := mergeMethods[m.resolveStrategy(step)] + m.logger.Infow("merging github stack", + "step_id", step.GetStepId(), + "top_pull", top.ref.PRNumber, + "pull_count", len(pending), + "merge_method", method, + ) + if err := m.mergeStack(ctx, step, top, method); err != nil { + return nil, err + } + } + return m.awaitMergeCommits(ctx, pulls) +} + +// awaitMergeCommits returns each pull request's merge commit, re-reading the +// ones GitHub has not recorded yet: a stack merge settles before every pull +// request in it shows its merge, so a lagging read is waited out rather than +// failed. Only the merge record is read — a stack merge may rewrite the heads +// of pull requests that a lagging read still shows open, so revalidating them +// here would reject a merge that already happened. +func (m *githubMerger) awaitMergeCommits(ctx context.Context, pulls []stepPull) ([]*runwaymq.StepOutput, error) { + deadline := time.Now().Add(m.maxPollDuration) + outputs := make([]*runwaymq.StepOutput, 0, len(pulls)) + for _, p := range pulls { + for { + sha, ok, err := m.client.getMergeCommit(ctx, p.ref.PRNumber) + if err != nil { + return nil, err + } + if ok { + outputs = append(outputs, &runwaymq.StepOutput{Id: sha}) + break + } + if !time.Now().Before(deadline) { + return nil, fmt.Errorf("%w: pull request %s not yet reported merged", ErrMergePending, p.ref.Label) + } + if err := m.waitPollInterval(ctx); err != nil { + return nil, err + } + } + } + return outputs, nil +} + +// mergeStack submits the async merge for the top pull request, which GitHub +// applies to every unmerged pull request below it as one operation, and waits +// for it to settle. +func (m *githubMerger) mergeStack(ctx context.Context, step *runwaymq.MergeStep, top stepPull, method string) error { + status, outcome, err := m.client.submitMerge(ctx, top.ref.PRNumber, asyncMergeRequest{ + SHA: top.ref.SHA, + MergeMethod: method, + MergeAction: mergeActionDirect, + BypassRules: m.bypassRules, + }) + if err != nil { + return err + } + if outcome == submitRejected { + return fmt.Errorf("%w: step %q: GitHub refused to merge %s: %s", merger.ErrInvalidRequest, step.GetStepId(), top.ref.Label, status.Details.Message) + } + if outcome == submitConflicting { + return fmt.Errorf("%w: step %q: %s already has a merge in flight for head %s by %s, not head %s by %s", + merger.ErrInvalidRequest, step.GetStepId(), top.ref.Label, status.Details.ExpectedHeadSHA, status.Details.MergeMethod, top.ref.SHA, method) + } + + deadline := time.Now().Add(m.maxPollDuration) + for status.Status == mergeStatusPending { + if !time.Now().Before(deadline) { + return fmt.Errorf("%w: merge %s of %s", ErrMergePending, status.Details.UUID, top.ref.Label) + } + if err := m.waitPollInterval(ctx); err != nil { + return err + } + uuid := status.Details.UUID + status, err = m.client.getMergeStatus(ctx, top.ref.PRNumber, uuid) + if err != nil { + return err + } + if status.Details.UUID == "" { + status.Details.UUID = uuid + } + } + + switch status.Status { + case mergeStatusMerged: + return nil + case mergeStatusFailed: + // GitHub does not separate a conflict from a rule or head-moved + // failure here; all are terminal, and its message is the reason. + return fmt.Errorf("%w: step %q: GitHub failed to merge %s: %s", merger.ErrConflict, step.GetStepId(), top.ref.Label, status.Details.Message) + default: + return fmt.Errorf("unexpected merge status %q for %s", status.Status, top.ref.Label) + } +} + +// checkPullMergeable reads GitHub's mergeability verdict for one pull request +// against its own base, re-reading while GitHub is still computing it. +func (m *githubMerger) checkPullMergeable(ctx context.Context, step *runwaymq.MergeStep, p stepPull) error { + pr := p.pr + deadline := time.Now().Add(m.maxPollDuration) + for pr.Mergeable == nil { + if !time.Now().Before(deadline) { + return fmt.Errorf("%w: mergeability of %s", ErrMergePending, p.ref.Label) + } + if err := m.waitPollInterval(ctx); err != nil { + return err + } + var err error + pr, err = m.client.getPull(ctx, p.ref.PRNumber) + if err != nil { + return err + } + } + if !*pr.Mergeable || pr.MergeableState == mergeableStateDirty { + return fmt.Errorf("%w: step %q: %s conflicts with %s", merger.ErrConflict, step.GetStepId(), p.ref.Label, pr.Base.Ref) + } + return nil +} + +// validateRequest checks the parts of a request that need no API call. +func (m *githubMerger) validateRequest(req *runwaymq.MergeRequest) error { + if len(req.GetSteps()) == 0 { + return fmt.Errorf("%w: request has no steps", merger.ErrInvalidRequest) + } + for _, step := range req.GetSteps() { + if len(step.GetChange().GetUris()) == 0 { + return fmt.Errorf("%w: step %q has no change URIs", merger.ErrInvalidRequest, step.GetStepId()) + } + if _, ok := mergeMethods[m.resolveStrategy(step)]; !ok { + return fmt.Errorf("%w: step %q: strategy %v is not supported by the GitHub merger", merger.ErrInvalidRequest, step.GetStepId(), step.GetStrategy()) + } + for _, uri := range step.GetChange().GetUris() { + if _, err := m.resolveChange(uri); err != nil { + return err + } + } + } + return nil +} + +func (m *githubMerger) resolveStrategy(step *runwaymq.MergeStep) mergestrategypb.Strategy { + if step.GetStrategy() == mergestrategypb.Strategy_DEFAULT { + return m.defaultStrategy + } + return step.GetStrategy() +} + +// loadStepPulls reads the pull request behind every URI of a step and rejects +// one that can no longer be landed as the URI describes it. +func (m *githubMerger) loadStepPulls(ctx context.Context, step *runwaymq.MergeStep) ([]stepPull, error) { + uris := step.GetChange().GetUris() + pulls := make([]stepPull, 0, len(uris)) + for _, uri := range uris { + ref, err := m.resolveChange(uri) + if err != nil { + return nil, err + } + pr, err := m.client.getPull(ctx, ref.PRNumber) + if err != nil { + return nil, err + } + if pr.Merged { + // No head check: landing a stack can rewrite the heads of the + // pull requests it merged, and a merged one is final either way. + pulls = append(pulls, stepPull{ref: ref, pr: pr}) + continue + } + if pr.Head.SHA != ref.SHA { + return nil, fmt.Errorf("%w: step %q: %s head moved from %s to %s", merger.ErrInvalidRequest, step.GetStepId(), ref.Label, ref.SHA, pr.Head.SHA) + } + if pr.State != pullStateOpen { + return nil, fmt.Errorf("%w: step %q: %s is closed without being merged", merger.ErrInvalidRequest, step.GetStepId(), ref.Label) + } + if pr.Draft { + return nil, fmt.Errorf("%w: step %q: %s is a draft", merger.ErrInvalidRequest, step.GetStepId(), ref.Label) + } + pulls = append(pulls, stepPull{ref: ref, pr: pr}) + } + return pulls, nil +} + +// validateStepIsGitHubStack rejects a step whose URIs GitHub would not land as +// one stack onto the target. Merged pull requests must form a prefix — GitHub +// merges bottom-up — and are skipped. The unmerged rest must be the bottom of a +// GitHub stack based on the target, in order and with no gap: merging the top +// lands everything below it, so a pull request the step does not name would +// otherwise land with it, and independent pull requests listed together would +// never land as one. +func (m *githubMerger) validateStepIsGitHubStack(ctx context.Context, step *runwaymq.MergeStep, pulls []stepPull) error { + pending := unmergedPulls(pulls) + if len(pending) == 0 { + return nil + } + for _, p := range pulls[:len(pulls)-len(pending)] { + if !p.pr.Merged { + return fmt.Errorf("%w: step %q: %s is unmerged but listed below merged pull requests", merger.ErrInvalidRequest, step.GetStepId(), p.ref.Label) + } + } + + if len(pending) == 1 { + if base := pending[0].pr.Base.Ref; base != m.target { + return fmt.Errorf("%w: step %q: %s targets %q, not %q; a stacked pull request must be listed with the pull requests below it", merger.ErrInvalidRequest, step.GetStepId(), pending[0].ref.Label, base, m.target) + } + return nil + } + + top := pending[len(pending)-1] + s, ok, err := m.client.getStackForPull(ctx, top.ref.PRNumber) + if err != nil { + return err + } + if !ok { + return fmt.Errorf("%w: step %q lists %s, which are not a GitHub stack", merger.ErrInvalidRequest, step.GetStepId(), pullLabels(pending)) + } + if s.Base.Ref != m.target { + return fmt.Errorf("%w: step %q: stack #%d is based on %q, not %q", merger.ErrInvalidRequest, step.GetStepId(), s.Number, s.Base.Ref, m.target) + } + var stackPending []int + for _, sp := range s.PullRequests { + if sp.unmerged() { + stackPending = append(stackPending, sp.Number) + } + } + for i, p := range pending { + if i >= len(stackPending) || stackPending[i] != p.ref.PRNumber { + return fmt.Errorf("%w: step %q lists %s, but the unmerged bottom of stack #%d is %s", merger.ErrInvalidRequest, step.GetStepId(), pullLabels(pending), s.Number, numbersLabel(stackPending)) + } + } + return nil +} + +func (m *githubMerger) waitPollInterval(ctx context.Context) error { + if m.pollInterval < 0 { + return ctx.Err() + } + timer := time.NewTimer(m.pollInterval) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +func unmergedPulls(pulls []stepPull) []stepPull { + for i, p := range pulls { + if !p.pr.Merged { + return pulls[i:] + } + } + return nil +} + +func pullLabels(pulls []stepPull) string { + labels := make([]string, 0, len(pulls)) + for _, p := range pulls { + labels = append(labels, p.ref.Label) + } + return strings.Join(labels, ", ") +} + +func numbersLabel(numbers []int) string { + labels := make([]string, 0, len(numbers)) + for _, n := range numbers { + labels = append(labels, "#"+strconv.Itoa(n)) + } + return "[" + strings.Join(labels, ", ") + "]" +} + +func successResult(req *runwaymq.MergeRequest, steps []*runwaymq.StepResult) *runwaymq.MergeResult { + return &runwaymq.MergeResult{ + Id: req.GetId(), + Outcome: runwaypb.Outcome_SUCCEEDED, + Steps: steps, + } +} diff --git a/runway/extension/merger/github/github_merger_test.go b/runway/extension/merger/github/github_merger_test.go new file mode 100644 index 000000000..1b178cc90 --- /dev/null +++ b/runway/extension/merger/github/github_merger_test.go @@ -0,0 +1,813 @@ +// Copyright (c) 2026 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package github + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "go.uber.org/zap/zaptest" + + changepb "github.com/uber/submitqueue/api/base/change/protopb" + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" + "github.com/uber/submitqueue/platform/errs" + phttp "github.com/uber/submitqueue/platform/http" + "github.com/uber/submitqueue/runway/extension/merger" +) + +const ( + testHost = "github.com" + testOwner = "uber" + testRepo = "submitqueue" + testTarget = "main" +) + +func headSHA(n int) string { return fmt.Sprintf("%040x", n) } + +func mergeSHA(n int) string { return fmt.Sprintf("%040x", 0xa000+n) } + +func uri(n int) string { + return fmt.Sprintf("github://%s/%s/%s/pull/%d/%s", testHost, testOwner, testRepo, n, headSHA(n)) +} + +// fakeGitHub serves the endpoints the merger calls from in-memory state. +type fakeGitHub struct { + mu sync.Mutex + pulls map[int]*pullRequest + stacks []stack + + // submitStatus overrides the merge-async response code (default 202). + submitStatus int + // submitBody overrides the merge-async response body. + submitBody string + // pollsUntilDone is how many status reads stay pending. + pollsUntilDone int + // finalStatus is what a settled merge reports (default merged). + finalStatus string + // mergeableNullReads is how many pull reads report mergeable as null. + mergeableNullReads int + // unmergeable names pull requests reported as conflicting. + unmergeable map[int]bool + // pullErrStatus makes every pull read fail with this code. + pullErrStatus int + // mergedReadLag is how many event reads of a pull request landed by a merge + // still lack its merged event, as github.com does briefly after a stack + // merge settles. + mergedReadLag int + // eventPadding is how many other events precede a merged event, to push it + // past the first page. + eventPadding int + // failMerges names top pull requests whose merge settles as failed. + failMerges map[int]bool + + mergeCommits map[int]string + lagging map[int]int + submits []submitCall + polls int +} + +type submitCall struct { + number int + req asyncMergeRequest +} + +func newFakeGitHub() *fakeGitHub { + return &fakeGitHub{pulls: make(map[int]*pullRequest), mergeCommits: make(map[int]string), lagging: make(map[int]int)} +} + +// addPull adds an open pull request with the given base branch. +func (f *fakeGitHub) addPull(n int, base string) { + f.pulls[n] = &pullRequest{ + Number: n, + State: pullStateOpen, + Head: gitRef{Ref: "branch-" + strconv.Itoa(n), SHA: headSHA(n)}, + Base: gitRef{Ref: base}, + } +} + +// addStack adds open pull requests forming a GitHub stack on the target, +// bottom to top. +func (f *fakeGitHub) addStack(number int, prs ...int) { + s := stack{Number: number, Open: true, Base: gitRef{Ref: testTarget}} + base := testTarget + for _, n := range prs { + f.addPull(n, base) + base = "branch-" + strconv.Itoa(n) + s.PullRequests = append(s.PullRequests, stackPull{Number: n, State: pullStateOpen, Head: gitRef{Ref: base, SHA: headSHA(n)}}) + } + f.stacks = append(f.stacks, s) +} + +func (f *fakeGitHub) markMerged(n int) { + pr := f.pulls[n] + pr.Merged = true + pr.State = "closed" + f.mergeCommits[n] = mergeSHA(n) + mergedAt := "2026-10-07T00:00:00Z" + for si := range f.stacks { + for pi := range f.stacks[si].PullRequests { + if f.stacks[si].PullRequests[pi].Number == n { + f.stacks[si].PullRequests[pi].State = "closed" + f.stacks[si].PullRequests[pi].MergedAt = &mergedAt + } + } + } +} + +// landStack merges top and every unmerged pull request below it, as GitHub's +// async merge does for a stacked pull request. +func (f *fakeGitHub) landStack(top int) { + for _, s := range f.stacks { + for i, p := range s.PullRequests { + if p.Number != top { + continue + } + for _, below := range s.PullRequests[:i+1] { + if !f.pulls[below.Number].Merged { + f.markMerged(below.Number) + } + } + return + } + } + f.markMerged(top) +} + +func (f *fakeGitHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + + prefix := "/repos/" + testOwner + "/" + testRepo + "/" + parts := strings.Split(strings.TrimPrefix(r.URL.Path, prefix), "/") + switch { + case r.Method == http.MethodGet && parts[0] == "stacks": + n, _ := strconv.Atoi(r.URL.Query().Get("pull_request")) + var out []stack + for _, s := range f.stacks { + for _, p := range s.PullRequests { + if p.Number == n { + out = append(out, s) + } + } + } + writeJSON(w, http.StatusOK, out) + + case r.Method == http.MethodGet && parts[0] == "issues" && len(parts) == 3 && parts[2] == "events": + n, _ := strconv.Atoi(parts[1]) + var events []issueEvent + for i := 0; i < f.eventPadding; i++ { + events = append(events, issueEvent{Event: "labeled"}) + } + if sha, ok := f.mergeCommits[n]; ok { + if f.lagging[n] > 0 { + f.lagging[n]-- + } else { + events = append(events, issueEvent{Event: "merged", CommitID: sha}) + } + } + page, _ := strconv.Atoi(r.URL.Query().Get("page")) + perPage, _ := strconv.Atoi(r.URL.Query().Get("per_page")) + start := min((page-1)*perPage, len(events)) + writeJSON(w, http.StatusOK, events[start:min(start+perPage, len(events))]) + + case parts[0] == "pulls" && len(parts) == 2 && r.Method == http.MethodGet: + if f.pullErrStatus != 0 { + w.WriteHeader(f.pullErrStatus) + return + } + n, _ := strconv.Atoi(parts[1]) + pr, ok := f.pulls[n] + if !ok { + w.WriteHeader(http.StatusNotFound) + return + } + out := *pr + if !out.Merged { + if f.mergeableNullReads > 0 { + f.mergeableNullReads-- + } else { + mergeable := !f.unmergeable[n] + out.Mergeable = &mergeable + if !mergeable { + out.MergeableState = mergeableStateDirty + } + } + } + writeJSON(w, http.StatusOK, out) + + case parts[0] == "pulls" && len(parts) == 3 && r.Method == http.MethodPut: + n, _ := strconv.Atoi(parts[1]) + var req asyncMergeRequest + _ = json.NewDecoder(r.Body).Decode(&req) + f.submits = append(f.submits, submitCall{number: n, req: req}) + if f.submitStatus != 0 { + w.WriteHeader(f.submitStatus) + _, _ = w.Write([]byte(f.submitBody)) + return + } + if f.pollsUntilDone == 0 { + f.settle(w, n) + return + } + writeJSON(w, http.StatusAccepted, map[string]any{"status": mergeStatusPending, "details": map[string]string{"uuid": "uuid-1"}}) + + case parts[0] == "pulls" && len(parts) == 4 && r.Method == http.MethodGet: + n, _ := strconv.Atoi(parts[1]) + f.polls++ + if f.polls < f.pollsUntilDone { + writeJSON(w, http.StatusOK, map[string]any{"status": mergeStatusPending}) + return + } + f.settle(w, n) + + default: + w.WriteHeader(http.StatusNotFound) + } +} + +func (f *fakeGitHub) settle(w http.ResponseWriter, top int) { + status := f.finalStatus + if status == "" { + status = mergeStatusMerged + } + if f.failMerges[top] { + status = mergeStatusFailed + } + if status == mergeStatusMerged { + f.landStack(top) + for n, pr := range f.pulls { + if pr.Merged { + f.lagging[n] = f.mergedReadLag + } + } + } + writeJSON(w, http.StatusOK, map[string]any{"status": status, "details": map[string]string{"message": "merge " + status, "sha": mergeSHA(top)}}) +} + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(v) +} + +func newTestMerger(t *testing.T, gh *fakeGitHub) merger.Merger { + t.Helper() + srv := httptest.NewServer(gh) + t.Cleanup(srv.Close) + httpClient, err := phttp.NewClient(srv.URL) + require.NoError(t, err) + m, err := New(Params{ + HTTPClient: httpClient, + Host: testHost, + Owner: testOwner, + Repo: testRepo, + Target: testTarget, + DefaultStrategy: mergestrategypb.Strategy_SQUASH_REBASE, + PollInterval: -1, + Logger: zaptest.NewLogger(t).Sugar(), + MetricsScope: tally.NoopScope, + }) + require.NoError(t, err) + return m +} + +func step(id string, strategy mergestrategypb.Strategy, uris ...string) *runwaymq.MergeStep { + return &runwaymq.MergeStep{StepId: id, Change: &changepb.Change{Uris: uris}, Strategy: strategy} +} + +func request(steps ...*runwaymq.MergeStep) *runwaymq.MergeRequest { + return &runwaymq.MergeRequest{Id: "req-1", QueueName: "q", Steps: steps} +} + +func outputIDs(r *runwaymq.StepResult) []string { + var ids []string + for _, o := range r.GetOutputs() { + ids = append(ids, o.GetId()) + } + return ids +} + +func TestMerge(t *testing.T) { + tests := []struct { + name string + setup func(*fakeGitHub) + req *runwaymq.MergeRequest + wantErr error + wantOutputs map[string][]string + wantSubmits []submitCall + // wantFailedStep, on an error, names the step reported as failed after + // the steps in wantOutputs landed; empty means no result is returned. + wantFailedStep string + }{ + { + name: "single pull request on trunk", + setup: func(f *fakeGitHub) { f.addPull(1, testTarget) }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1))), + wantOutputs: map[string][]string{"s1": {mergeSHA(1)}}, + wantSubmits: []submitCall{{number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "rebase", MergeAction: mergeActionDirect}}}, + }, + { + name: "stack lands with one merge on its top pull request", + setup: func(f *fakeGitHub) { f.addStack(7, 1, 2, 3) }, + req: request(step("s1", mergestrategypb.Strategy_DEFAULT, uri(1), uri(2), uri(3))), + wantOutputs: map[string][]string{"s1": {mergeSHA(1), mergeSHA(2), mergeSHA(3)}}, + wantSubmits: []submitCall{{number: 3, req: asyncMergeRequest{SHA: headSHA(3), MergeMethod: "squash", MergeAction: mergeActionDirect}}}, + }, + { + name: "bottom of a stack lands without the pull requests above it", + setup: func(f *fakeGitHub) { f.addStack(7, 1, 2, 3) }, + req: request(step("s1", mergestrategypb.Strategy_MERGE, uri(1), uri(2))), + wantOutputs: map[string][]string{"s1": {mergeSHA(1), mergeSHA(2)}}, + wantSubmits: []submitCall{{number: 2, req: asyncMergeRequest{SHA: headSHA(2), MergeMethod: "merge", MergeAction: mergeActionDirect}}}, + }, + { + name: "redelivery skips the already-merged bottom of a stack", + setup: func(f *fakeGitHub) { + f.addStack(7, 1, 2) + f.markMerged(1) + f.pulls[2].Base.Ref = testTarget + }, + req: request(step("s1", mergestrategypb.Strategy_SQUASH_REBASE, uri(1), uri(2))), + wantOutputs: map[string][]string{"s1": {mergeSHA(1), mergeSHA(2)}}, + wantSubmits: []submitCall{{number: 2, req: asyncMergeRequest{SHA: headSHA(2), MergeMethod: "squash", MergeAction: mergeActionDirect}}}, + }, + { + name: "fully merged step reports outputs without merging", + setup: func(f *fakeGitHub) { + f.addStack(7, 1, 2) + f.landStack(2) + }, + req: request(step("s1", mergestrategypb.Strategy_SQUASH_REBASE, uri(1), uri(2))), + wantOutputs: map[string][]string{"s1": {mergeSHA(1), mergeSHA(2)}}, + }, + { + name: "merge commits not yet recorded after the merge settles are re-read", + setup: func(f *fakeGitHub) { + f.addStack(7, 1, 2) + f.mergedReadLag = 2 + }, + req: request(step("s1", mergestrategypb.Strategy_SQUASH_REBASE, uri(1), uri(2))), + wantOutputs: map[string][]string{"s1": {mergeSHA(1), mergeSHA(2)}}, + wantSubmits: []submitCall{{number: 2, req: asyncMergeRequest{SHA: headSHA(2), MergeMethod: "squash", MergeAction: mergeActionDirect}}}, + }, + { + name: "merge commit past the first page of events", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.eventPadding = issueEventsPageSize + 5 + }, + req: request(step("s1", mergestrategypb.Strategy_MERGE, uri(1))), + wantOutputs: map[string][]string{"s1": {mergeSHA(1)}}, + wantSubmits: []submitCall{{number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "merge", MergeAction: mergeActionDirect}}}, + }, + { + name: "pending merge is polled until it lands", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.pollsUntilDone = 3 + }, + req: request(step("s1", mergestrategypb.Strategy_SQUASH_REBASE, uri(1))), + wantOutputs: map[string][]string{"s1": {mergeSHA(1)}}, + wantSubmits: []submitCall{{number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "squash", MergeAction: mergeActionDirect}}}, + }, + { + name: "existing in-flight request is adopted", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.pollsUntilDone = 2 + f.submitStatus = http.StatusConflict + f.submitBody = `{"status":"pending","details":{"uuid":"uuid-0","expected_head_sha":"` + headSHA(1) + `","merge_method":"squash"}}` + }, + req: request(step("s1", mergestrategypb.Strategy_SQUASH_REBASE, uri(1))), + wantOutputs: map[string][]string{"s1": {mergeSHA(1)}}, + wantSubmits: []submitCall{{number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "squash", MergeAction: mergeActionDirect}}}, + }, + { + name: "steps land in order", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.addStack(7, 2, 3) + }, + req: request( + step("s1", mergestrategypb.Strategy_REBASE, uri(1)), + step("s2", mergestrategypb.Strategy_SQUASH_REBASE, uri(2), uri(3)), + ), + wantOutputs: map[string][]string{"s1": {mergeSHA(1)}, "s2": {mergeSHA(2), mergeSHA(3)}}, + wantSubmits: []submitCall{ + {number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "rebase", MergeAction: mergeActionDirect}}, + {number: 3, req: asyncMergeRequest{SHA: headSHA(3), MergeMethod: "squash", MergeAction: mergeActionDirect}}, + }, + }, + { + name: "failed merge is a conflict", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.finalStatus = mergeStatusFailed + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1))), + wantErr: merger.ErrConflict, + wantFailedStep: "s1", + wantSubmits: []submitCall{{number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "rebase", MergeAction: mergeActionDirect}}}, + }, + { + name: "refused merge is an invalid request", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.submitStatus = http.StatusUnprocessableEntity + f.submitBody = `{"details":{"message":"required status checks"}}` + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1))), + wantErr: merger.ErrInvalidRequest, + wantFailedStep: "s1", + wantSubmits: []submitCall{{number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "rebase", MergeAction: mergeActionDirect}}}, + }, + { + name: "head moved since the URI was minted", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.pulls[1].Head.SHA = headSHA(99) + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1))), + wantErr: merger.ErrInvalidRequest, + wantFailedStep: "s1", + }, + { + name: "closed pull request", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.pulls[1].State = "closed" + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1))), + wantErr: merger.ErrInvalidRequest, + wantFailedStep: "s1", + }, + { + name: "stack unstacked after validation is refused before merging", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.addPull(2, testTarget) + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1), uri(2))), + wantErr: merger.ErrInvalidRequest, + wantFailedStep: "s1", + }, + { + name: "invalid later step stops the request after earlier steps land", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.addPull(2, "feature") + }, + req: request( + step("s1", mergestrategypb.Strategy_REBASE, uri(1)), + step("s2", mergestrategypb.Strategy_REBASE, uri(2)), + ), + wantErr: merger.ErrInvalidRequest, + wantOutputs: map[string][]string{"s1": {mergeSHA(1)}}, + wantFailedStep: "s2", + wantSubmits: []submitCall{{number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "rebase", MergeAction: mergeActionDirect}}}, + }, + { + name: "merge GitHub fails after a landed step is a conflict", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.addPull(2, testTarget) + f.failMerges = map[int]bool{2: true} + }, + req: request( + step("s1", mergestrategypb.Strategy_REBASE, uri(1)), + step("s2", mergestrategypb.Strategy_REBASE, uri(2)), + ), + wantErr: merger.ErrConflict, + wantOutputs: map[string][]string{"s1": {mergeSHA(1)}}, + wantFailedStep: "s2", + wantSubmits: []submitCall{ + {number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "rebase", MergeAction: mergeActionDirect}}, + {number: 2, req: asyncMergeRequest{SHA: headSHA(2), MergeMethod: "rebase", MergeAction: mergeActionDirect}}, + }, + }, + { + name: "steps after a failed one are not attempted", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.addPull(2, testTarget) + f.addPull(3, testTarget) + f.failMerges = map[int]bool{2: true} + }, + req: request( + step("s1", mergestrategypb.Strategy_REBASE, uri(1)), + step("s2", mergestrategypb.Strategy_REBASE, uri(2)), + step("s3", mergestrategypb.Strategy_REBASE, uri(3)), + ), + wantErr: merger.ErrConflict, + wantOutputs: map[string][]string{"s1": {mergeSHA(1)}}, + wantFailedStep: "s2", + wantSubmits: []submitCall{ + {number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "rebase", MergeAction: mergeActionDirect}}, + {number: 2, req: asyncMergeRequest{SHA: headSHA(2), MergeMethod: "rebase", MergeAction: mergeActionDirect}}, + }, + }, + { + name: "in-flight merge for a different head is not adopted", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.submitStatus = http.StatusConflict + f.submitBody = `{"status":"pending","details":{"uuid":"uuid-0","expected_head_sha":"` + headSHA(99) + `","merge_method":"squash"}}` + }, + req: request(step("s1", mergestrategypb.Strategy_SQUASH_REBASE, uri(1))), + wantErr: merger.ErrInvalidRequest, + wantFailedStep: "s1", + wantSubmits: []submitCall{{number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "squash", MergeAction: mergeActionDirect}}}, + }, + { + name: "in-flight merge with a different method is not adopted", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.submitStatus = http.StatusConflict + f.submitBody = `{"status":"pending","details":{"uuid":"uuid-0","expected_head_sha":"` + headSHA(1) + `","merge_method":"merge"}}` + }, + req: request(step("s1", mergestrategypb.Strategy_SQUASH_REBASE, uri(1))), + wantErr: merger.ErrInvalidRequest, + wantFailedStep: "s1", + wantSubmits: []submitCall{{number: 1, req: asyncMergeRequest{SHA: headSHA(1), MergeMethod: "squash", MergeAction: mergeActionDirect}}}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gh := newFakeGitHub() + tt.setup(gh) + m := newTestMerger(t, gh) + + result, err := m.Merge(context.Background(), tt.req) + assert.Equal(t, tt.wantSubmits, gh.submits) + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + if tt.wantFailedStep == "" { + assert.Nil(t, result) + return + } + assert.Equal(t, runwaypb.Outcome_FAILED, result.GetOutcome()) + steps := result.GetSteps() + require.Len(t, steps, len(tt.wantOutputs)+1) + for _, sr := range steps[:len(steps)-1] { + assert.Equal(t, tt.wantOutputs[sr.GetStepId()], outputIDs(sr), sr.GetStepId()) + } + failed := steps[len(steps)-1] + assert.Equal(t, tt.wantFailedStep, failed.GetStepId()) + assert.NotEmpty(t, failed.GetReason()) + assert.Empty(t, failed.GetOutputs()) + return + } + require.NoError(t, err) + assert.Equal(t, runwaypb.Outcome_SUCCEEDED, result.GetOutcome()) + assert.Equal(t, tt.req.GetId(), result.GetId()) + require.Len(t, result.GetSteps(), len(tt.req.GetSteps())) + for _, sr := range result.GetSteps() { + assert.Equal(t, tt.wantOutputs[sr.GetStepId()], outputIDs(sr), sr.GetStepId()) + } + }) + } +} + +func TestMergeErrorClassification(t *testing.T) { + t.Run("server error stays a retryable status error", func(t *testing.T) { + gh := newFakeGitHub() + gh.addPull(1, testTarget) + gh.pullErrStatus = http.StatusBadGateway + m := newTestMerger(t, gh) + + _, err := m.Merge(context.Background(), request(step("s1", mergestrategypb.Strategy_REBASE, uri(1)))) + require.Error(t, err) + assert.False(t, merger.IsTerminal(err)) + var se *phttp.StatusError + require.ErrorAs(t, err, &se) + assert.Equal(t, http.StatusBadGateway, se.StatusCode) + }) + + t.Run("merge outlasting the poll budget is pending and retryable", func(t *testing.T) { + gh := newFakeGitHub() + gh.addPull(1, testTarget) + gh.pollsUntilDone = 1 << 30 + srv := httptest.NewServer(gh) + t.Cleanup(srv.Close) + httpClient, err := phttp.NewClient(srv.URL) + require.NoError(t, err) + m, err := New(Params{ + HTTPClient: httpClient, + Host: testHost, + Owner: testOwner, + Repo: testRepo, + Target: testTarget, + DefaultStrategy: mergestrategypb.Strategy_REBASE, + PollInterval: -1, + MaxPollDuration: time.Nanosecond, + Logger: zaptest.NewLogger(t).Sugar(), + MetricsScope: tally.NoopScope, + }) + require.NoError(t, err) + + _, err = m.Merge(context.Background(), request(step("s1", mergestrategypb.Strategy_REBASE, uri(1)))) + require.ErrorIs(t, err, ErrMergePending) + assert.False(t, merger.IsTerminal(err)) + assert.True(t, errs.IsRetryable(errs.NewClassifierProcessor(Classifier).Process(err))) + }) +} + +func TestCheckMergeability(t *testing.T) { + tests := []struct { + name string + setup func(*fakeGitHub) + req *runwaymq.MergeRequest + wantErr error + }{ + { + name: "single pull request on trunk is mergeable", + setup: func(f *fakeGitHub) { f.addPull(1, testTarget) }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1))), + }, + { + name: "stack in order is mergeable", + setup: func(f *fakeGitHub) { f.addStack(7, 1, 2, 3) }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1), uri(2), uri(3))), + }, + { + name: "mergeability still being computed is re-read", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.mergeableNullReads = 3 + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1))), + }, + { + name: "conflicting pull request is a conflict", + setup: func(f *fakeGitHub) { + f.addStack(7, 1, 2) + f.unmergeable = map[int]bool{2: true} + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1), uri(2))), + wantErr: merger.ErrConflict, + }, + { + name: "independent pull requests on trunk are not a stack", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.addPull(2, testTarget) + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1), uri(2))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "stack listed top first", + setup: func(f *fakeGitHub) { f.addStack(7, 1, 2) }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(2), uri(1))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "stack listed with a gap", + setup: func(f *fakeGitHub) { f.addStack(7, 1, 2, 3) }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1), uri(3))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "pull requests from two stacks", + setup: func(f *fakeGitHub) { + f.addStack(7, 1, 2) + f.addStack(8, 3, 4) + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1), uri(4))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "upper stacked pull request without the ones below it", + setup: func(f *fakeGitHub) { f.addStack(7, 1, 2) }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(2))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "stack based on another branch", + setup: func(f *fakeGitHub) { + f.addStack(7, 1, 2) + f.stacks[0].Base.Ref = "release" + f.pulls[1].Base.Ref = "release" + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1), uri(2))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "draft pull request", + setup: func(f *fakeGitHub) { + f.addPull(1, testTarget) + f.pulls[1].Draft = true + }, + req: request(step("s1", mergestrategypb.Strategy_REBASE, uri(1))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "change in another repository", + setup: func(f *fakeGitHub) {}, + req: request(step("s1", mergestrategypb.Strategy_REBASE, "github://github.com/uber/other/pull/1/"+headSHA(1))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "change on another host", + setup: func(f *fakeGitHub) {}, + req: request(step("s1", mergestrategypb.Strategy_REBASE, "github://github.example.com/uber/submitqueue/pull/1/"+headSHA(1))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "non-github change", + setup: func(f *fakeGitHub) {}, + req: request(step("s1", mergestrategypb.Strategy_REBASE, "git://github.com/uber/submitqueue/refs%2Fheads%2Fmain/"+headSHA(1))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "promote is unsupported", + setup: func(f *fakeGitHub) { f.addPull(1, testTarget) }, + req: request(step("s1", mergestrategypb.Strategy_PROMOTE, uri(1))), + wantErr: merger.ErrInvalidRequest, + }, + { + name: "request without steps", + setup: func(f *fakeGitHub) {}, + req: request(), + wantErr: merger.ErrInvalidRequest, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gh := newFakeGitHub() + tt.setup(gh) + m := newTestMerger(t, gh) + + result, err := m.CheckMergeability(context.Background(), tt.req) + assert.Empty(t, gh.submits, "a dry run must not merge") + if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + return + } + require.NoError(t, err) + assert.Equal(t, runwaypb.Outcome_SUCCEEDED, result.GetOutcome()) + for _, sr := range result.GetSteps() { + assert.Empty(t, sr.GetOutputs()) + } + }) + } +} + +func TestNewRejectsUnusableParams(t *testing.T) { + valid := Params{ + HTTPClient: http.DefaultClient, + Host: testHost, + Owner: testOwner, + Repo: testRepo, + Target: testTarget, + DefaultStrategy: mergestrategypb.Strategy_REBASE, + Logger: zaptest.NewLogger(t).Sugar(), + MetricsScope: tally.NoopScope, + } + tests := []struct { + name string + mutate func(*Params) + }{ + {name: "missing http client", mutate: func(p *Params) { p.HTTPClient = nil }}, + {name: "missing repo", mutate: func(p *Params) { p.Repo = "" }}, + {name: "missing target", mutate: func(p *Params) { p.Target = "" }}, + {name: "promote default", mutate: func(p *Params) { p.DefaultStrategy = mergestrategypb.Strategy_PROMOTE }}, + {name: "default default", mutate: func(p *Params) { p.DefaultStrategy = mergestrategypb.Strategy_DEFAULT }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := valid + tt.mutate(&p) + _, err := New(p) + require.Error(t, err) + }) + } + + _, err := New(valid) + require.NoError(t, err) +} diff --git a/runway/extension/merger/merger.go b/runway/extension/merger/merger.go index bf0f7b1ec..ae3c18c7d 100644 --- a/runway/extension/merger/merger.go +++ b/runway/extension/merger/merger.go @@ -53,11 +53,11 @@ type Merger interface { // CheckMergeability performs a dry-run merge without committing. The // returned MergeResult reports per-step mergeability; Outputs are empty. CheckMergeability(ctx context.Context, req *runwaymq.MergeRequest) (*runwaymq.MergeResult, error) - // Merge applies the ordered steps, commits the result to the remote, and - // reports per-step Outputs (the VCS-neutral revision identifiers produced). - // Merge is all-or-nothing against the target: it updates the target at - // most once per request, and afterwards either every step is reachable - // from the target or the target is unchanged. + // Merge applies the ordered steps, commits them to the remote, and reports + // per-step Outputs (the VCS-neutral revision identifiers produced). Each + // step lands atomically, in order, and Merge stops at the first that fails: + // later steps are not attempted, earlier ones stay landed. On a terminal + // error the result, when non-nil, lists the landed steps and the failed one. Merge(ctx context.Context, req *runwaymq.MergeRequest) (*runwaymq.MergeResult, error) } diff --git a/service/runway/README.md b/service/runway/README.md index dc2a46bd2..4cb4e0578 100644 --- a/service/runway/README.md +++ b/service/runway/README.md @@ -15,7 +15,7 @@ These topic keys and their wire contracts are owned by the queue's producer side ### Merger backend -The merge work is done by the [`merger`](../../runway/extension/merger) extension, resolved **per queue** — so one Runway can serve several repositories by giving each queue its own merge target. By default every queue gets the **noop** merger (always succeeds — for local dev and compose). Point `MERGE_CONFIG_PATH` at a merge configuration file to wire real **git** merge targets, or set `MERGE_CHECKOUT_PATH` to configure a single one from the environment (see Configuration). Setting `MERGER=git` makes Git configuration mandatory: startup fails unless one of those sources defines at least one Git target. +The merge work is done by the [`merger`](../../runway/extension/merger) extension, resolved **per queue** — so one Runway can serve several repositories by giving each queue its own merge target. By default every queue gets the **noop** merger (always succeeds — for local dev and compose). Point `MERGE_CONFIG_PATH` at a merge configuration file to wire real **git** merge targets or **github** (REST API, stack-aware) ones, or set `MERGE_CHECKOUT_PATH` to configure a single one from the environment (see Configuration). Setting `MERGER=git` makes Git configuration mandatory: startup fails unless one of those sources defines at least one Git target. Two queues naming the same checkout resolve to the *same* merger instance, which is what serializes them against each other: a git merger locks the working tree it owns, and two instances over one tree would reset it out from under each other mid-merge. Naming one checkout for two *different* targets is rejected at startup. @@ -25,7 +25,7 @@ Two queues naming the same checkout resolve to the *same* merger instance, which ```yaml defaults: - merger: {type: noop} # noop | git + merger: {type: noop} # noop | git | github queues: - name: demo-queue merger: @@ -35,8 +35,20 @@ queues: checkoutPath: /var/runway/checkouts/sq-sandbox defaultStrategy: SQUASH_REBASE tokenEnv: GITHUB_TOKEN + - name: oss-queue + merger: + type: github + owner: my-org + repo: my-repo + target: main + defaultStrategy: SQUASH_REBASE + tokenEnv: GITHUB_TOKEN + # apiBaseUrl: https://api.github.com (default) + # host: github.com (default; the host change URIs name) ``` +A `github` target needs no checkout: it lands each step as one GitHub stack through the REST API (see the [github merger](../../runway/extension/merger/github)). `tokenEnv` becomes a bearer token on its HTTP client; `httpTimeout`, `pollInterval`, `maxPollDuration` and `bypassRules` tune it. + The file holds **no secret**: `tokenEnv` names the environment variable carrying the credential, so the file stays committable and rotating the token needs no edit. Omitting `remoteUrl` means the checkout is provisioned by something else and is used as it stands. ### Checkout provisioning diff --git a/service/runway/server/BUILD.bazel b/service/runway/server/BUILD.bazel index 8cdde8575..8ce2fc7c3 100644 --- a/service/runway/server/BUILD.bazel +++ b/service/runway/server/BUILD.bazel @@ -23,6 +23,7 @@ go_library( "//platform/errs:go_default_library", "//platform/errs/generic:go_default_library", "//platform/errs/git:go_default_library", + "//platform/errs/http:go_default_library", "//platform/errs/mysql:go_default_library", "//platform/extension/consumergate:go_default_library", "//platform/extension/consumergate/file:go_default_library", @@ -30,6 +31,7 @@ go_library( "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", "//platform/git/exec:go_default_library", + "//platform/http:go_default_library", "//runway/controller:go_default_library", "//runway/controller/dlq:go_default_library", "//runway/controller/merge:go_default_library", @@ -37,6 +39,7 @@ go_library( "//runway/extension/merger:go_default_library", "//runway/extension/merger/fake:go_default_library", "//runway/extension/merger/git:go_default_library", + "//runway/extension/merger/github:go_default_library", "//runway/extension/merger/noop:go_default_library", "//service/messagequeue:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", @@ -44,6 +47,7 @@ go_library( "@in_gopkg_yaml_v3//:go_default_library", "@org_golang_google_grpc//:go_default_library", "@org_golang_google_grpc//reflection:go_default_library", + "@org_golang_x_oauth2//:go_default_library", "@org_uber_go_zap//:go_default_library", ], ) @@ -109,8 +113,10 @@ go_test( "//platform/extension/messagequeue/mock:go_default_library", "//platform/git/exec:go_default_library", "//platform/git/exectest:go_default_library", + "//platform/http:go_default_library", "//runway/controller/dlq:go_default_library", "//runway/extension/merger/git:go_default_library", + "//runway/extension/merger/github:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@com_github_uber_go_tally//:go_default_library", diff --git a/service/runway/server/config.go b/service/runway/server/config.go index d66d31c3d..a08f9e09b 100644 --- a/service/runway/server/config.go +++ b/service/runway/server/config.go @@ -19,6 +19,7 @@ import ( "os" "reflect" "strings" + "time" yamlv3 "gopkg.in/yaml.v3" @@ -27,8 +28,16 @@ import ( // Merger types selectable from configuration. const ( - mergerTypeNoop = "noop" - mergerTypeGit = "git" + mergerTypeNoop = "noop" + mergerTypeGit = "git" + mergerTypeGitHub = "github" +) + +// Defaults for a github merger. +const ( + defaultGitHubAPIBaseURL = "https://api.github.com" + defaultGitHubHost = "github.com" + defaultGitHubHTTPTimeout = 30 * time.Second ) // mergeConfig is the runway merge configuration file: which merger each queue @@ -61,10 +70,11 @@ type queueMergeConfig struct { Merger mergerConfig `yaml:"merger"` } -// mergerConfig selects a merger implementation and configures it. Fields other -// than Type apply only to the git merger. +// mergerConfig selects a merger implementation and configures it. Target, +// DefaultStrategy and TokenEnv apply to the git and github mergers; every other +// field applies only to the merger its comment names, git when it names none. type mergerConfig struct { - // Type selects the implementation: "noop" or "git". + // Type selects the implementation: "noop", "git" or "github". Type string `yaml:"type"` // RemoteURL is the repository the merger clones and pushes to. When empty // the checkout is taken as provisioned by something else and used as it @@ -104,6 +114,28 @@ type mergerConfig struct { // TokenUser is the username paired with the token in basic auth. Providers // each have their own convention; defaults to "x-access-token". TokenUser string `yaml:"tokenUser"` + + // APIBaseURL is the github merger's REST API root. Defaults to + // "https://api.github.com". + APIBaseURL string `yaml:"apiBaseUrl"` + // Host is the GitHub instance the github merger's change URIs name. + // Defaults to "github.com". + Host string `yaml:"host"` + // Owner and Repo name the repository the github merger lands into. + Owner string `yaml:"owner"` + Repo string `yaml:"repo"` + // BypassRules asks GitHub to merge past branch rules (github merger); the + // token's actor must be allowed to bypass them. + BypassRules bool `yaml:"bypassRules"` + // HTTPTimeout bounds each GitHub API call of the github merger (Go + // duration). Defaults to 30s. + HTTPTimeout string `yaml:"httpTimeout"` + // PollInterval is the github merger's wait between merge status reads (Go + // duration). Empty uses the merger's own default. + PollInterval string `yaml:"pollInterval"` + // MaxPollDuration bounds how long one delivery waits for GitHub to settle + // a merge (github merger, Go duration). Empty uses the merger's default. + MaxPollDuration string `yaml:"maxPollDuration"` } // loadMergeConfig reads and validates the merge configuration at path. @@ -198,10 +230,15 @@ func (m *mergerConfig) normalizeAndValidate(where string) error { case mergerTypeNoop: return nil case mergerTypeGit: + return m.normalizeAndValidateGit(where) + case mergerTypeGitHub: + return m.normalizeAndValidateGitHub(where) default: - return fmt.Errorf("%s: unknown merger type %q (want %q or %q)", where, m.Type, mergerTypeNoop, mergerTypeGit) + return fmt.Errorf("%s: unknown merger type %q (want %q, %q or %q)", where, m.Type, mergerTypeNoop, mergerTypeGit, mergerTypeGitHub) } +} +func (m *mergerConfig) normalizeAndValidateGit(where string) error { if m.CheckoutPath == "" { return fmt.Errorf("%s: git merger requires checkoutPath", where) } @@ -233,6 +270,45 @@ func (m *mergerConfig) normalizeAndValidate(where string) error { return nil } +func (m *mergerConfig) normalizeAndValidateGitHub(where string) error { + if m.Owner == "" || m.Repo == "" { + return fmt.Errorf("%s: github merger requires owner and repo", where) + } + if m.CheckoutPath != "" || m.RemoteURL != "" { + return fmt.Errorf("%s: github merger works through the API and takes no checkoutPath or remoteUrl", where) + } + if m.APIBaseURL == "" { + m.APIBaseURL = defaultGitHubAPIBaseURL + } + if m.Host == "" { + m.Host = defaultGitHubHost + } + if m.Target == "" { + m.Target = "main" + } + strategy, err := parseStrategy(m.DefaultStrategy) + if err != nil { + return fmt.Errorf("%s: %w", where, err) + } + if strategy == mergestrategypb.Strategy_PROMOTE { + return fmt.Errorf("%s: github merger cannot default to PROMOTE", where) + } + for name, v := range map[string]string{"httpTimeout": m.HTTPTimeout, "pollInterval": m.PollInterval, "maxPollDuration": m.MaxPollDuration} { + if _, err := parseOptionalDuration(v); err != nil { + return fmt.Errorf("%s: invalid %s: %w", where, name, err) + } + } + return nil +} + +// parseOptionalDuration parses a Go duration, reading an empty value as zero. +func parseOptionalDuration(v string) (time.Duration, error) { + if v == "" { + return 0, nil + } + return time.ParseDuration(v) +} + // firstDifference reports the first field in which two git mergers disagree, // rendered for an error message, and whether there was one. // diff --git a/service/runway/server/config_test.go b/service/runway/server/config_test.go index 684a676eb..0f62aa9cf 100644 --- a/service/runway/server/config_test.go +++ b/service/runway/server/config_test.go @@ -111,6 +111,43 @@ queues: assert.False(t, cfg.usesGit(), "a noop-only deployment must not require a git runtime") } +func TestLoadMergeConfig_GitHub(t *testing.T) { + path := writeConfig(t, ` +defaults: + merger: {type: noop} +queues: + - name: oss + merger: {type: github, owner: uber, repo: submitqueue} + - name: internal + merger: + type: github + apiBaseUrl: https://github.example.com/api/v3 + host: github.example.com + owner: org + repo: monorepo + target: trunk + defaultStrategy: SQUASH_REBASE + tokenEnv: GHE_TOKEN + pollInterval: 5s +`) + + cfg, err := loadMergeConfig(path) + require.NoError(t, err) + assert.False(t, cfg.usesGit(), "a github-only deployment must not require a git runtime") + + oss := cfg.Queues[0].Merger + assert.Equal(t, defaultGitHubAPIBaseURL, oss.APIBaseURL) + assert.Equal(t, defaultGitHubHost, oss.Host) + assert.Equal(t, "main", oss.Target) + assert.Equal(t, mergestrategypb.Strategy_REBASE, oss.strategy()) + + internal := cfg.Queues[1].Merger + assert.Equal(t, "https://github.example.com/api/v3", internal.APIBaseURL) + assert.Equal(t, "github.example.com", internal.Host) + assert.Equal(t, "trunk", internal.Target) + assert.Equal(t, mergestrategypb.Strategy_SQUASH_REBASE, internal.strategy()) +} + func TestLoadMergeConfig_EmptyFileIsNoop(t *testing.T) { cfg, err := loadMergeConfig(writeConfig(t, "")) require.NoError(t, err) @@ -434,6 +471,34 @@ defaults: remoteUrl: https://example.com/o/r.git checkoutPath: /var/checkouts/r defaultStrategy: FAST_FORWARD +`, + }, + { + name: "github without a repository", + contents: ` +defaults: + merger: {type: github, owner: uber} +`, + }, + { + name: "github with a checkout", + contents: ` +defaults: + merger: {type: github, owner: uber, repo: submitqueue, checkoutPath: /var/checkouts/r} +`, + }, + { + name: "github defaulting to promote", + contents: ` +defaults: + merger: {type: github, owner: uber, repo: submitqueue, defaultStrategy: PROMOTE} +`, + }, + { + name: "github with an unparseable poll interval", + contents: ` +defaults: + merger: {type: github, owner: uber, repo: submitqueue, pollInterval: soon} `, }, { diff --git a/service/runway/server/main.go b/service/runway/server/main.go index 64ed86859..3cd224899 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -38,12 +38,14 @@ import ( "github.com/uber/submitqueue/platform/errs" genericerrs "github.com/uber/submitqueue/platform/errs/generic" giterrs "github.com/uber/submitqueue/platform/errs/git" + httperrs "github.com/uber/submitqueue/platform/errs/http" mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql" "github.com/uber/submitqueue/platform/extension/consumergate" consumergatefile "github.com/uber/submitqueue/platform/extension/consumergate/file" consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + phttp "github.com/uber/submitqueue/platform/http" "github.com/uber/submitqueue/runway/controller" "github.com/uber/submitqueue/runway/controller/dlq" "github.com/uber/submitqueue/runway/controller/merge" @@ -51,9 +53,11 @@ import ( "github.com/uber/submitqueue/runway/extension/merger" "github.com/uber/submitqueue/runway/extension/merger/fake" gitmerger "github.com/uber/submitqueue/runway/extension/merger/git" + githubmerger "github.com/uber/submitqueue/runway/extension/merger/github" "github.com/uber/submitqueue/runway/extension/merger/noop" servicemq "github.com/uber/submitqueue/service/messagequeue" "go.uber.org/zap" + "golang.org/x/oauth2" "google.golang.org/grpc" "google.golang.org/grpc/reflection" ) @@ -321,9 +325,13 @@ func logQueueInitialized(logger *zap.Logger) { } func newPrimaryErrorProcessor() errs.ErrorProcessor { + // httperrs precedes mysqlerrs, which would otherwise claim HTTP transport + // failures as MySQL ones (see platform/errs/http). return errs.NewClassifierProcessor( genericerrs.Classifier, giterrs.Classifier, + githubmerger.Classifier, + httperrs.Classifier, mysqlerrs.Classifier, ) } @@ -514,6 +522,9 @@ type mergerBuilder struct { } func (b *mergerBuilder) build(cfg mergerConfig, where string) (merger.Factory, error) { + if cfg.Type == mergerTypeGitHub { + return b.buildGitHub(cfg, where) + } if cfg.Type != mergerTypeGit { return &noopMergerFactory{seq: b.seq}, nil } @@ -570,6 +581,69 @@ func (f *gitMergerFactory) For(_ merger.Config) (merger.Merger, error) { return f.merger, nil } +// buildGitHub constructs a github merger. Auth is the wiring's concern, not the +// merger's: it receives an HTTP client whose transport already roots requests +// at the API and attaches the credential, so a deployment needing a different +// scheme swaps the transport here. +func (b *mergerBuilder) buildGitHub(cfg mergerConfig, where string) (merger.Factory, error) { + httpClient, err := phttp.NewClient(cfg.APIBaseURL) + if err != nil { + return nil, fmt.Errorf("%s: invalid apiBaseUrl %q: %w", where, cfg.APIBaseURL, err) + } + timeout, _ := parseOptionalDuration(cfg.HTTPTimeout) + if timeout == 0 { + timeout = defaultGitHubHTTPTimeout + } + httpClient.Timeout = timeout + if cfg.TokenEnv != "" { + token, ok := cfg.token() + if !ok || token == "" { + return nil, fmt.Errorf("%s: github merger needs %s to be set", where, cfg.TokenEnv) + } + httpClient.Transport = &oauth2.Transport{ + Source: oauth2.StaticTokenSource(&oauth2.Token{AccessToken: token}), + Base: httpClient.Transport, + } + } + + pollInterval, _ := parseOptionalDuration(cfg.PollInterval) + maxPollDuration, _ := parseOptionalDuration(cfg.MaxPollDuration) + m, err := githubmerger.New(githubmerger.Params{ + HTTPClient: httpClient, + Host: cfg.Host, + Owner: cfg.Owner, + Repo: cfg.Repo, + Target: cfg.Target, + DefaultStrategy: cfg.strategy(), + BypassRules: cfg.BypassRules, + PollInterval: pollInterval, + MaxPollDuration: maxPollDuration, + Logger: b.logger.Sugar(), + MetricsScope: b.scope, + }) + if err != nil { + return nil, fmt.Errorf("%s: failed to build github merger: %w", where, err) + } + + b.logger.Info("github merger configured", + zap.String("api_base_url", cfg.APIBaseURL), + zap.String("repo", cfg.Owner+"/"+cfg.Repo), + zap.String("target", cfg.Target), + zap.String("default_strategy", cfg.strategy().String()), + ) + return &githubMergerFactory{merger: m}, nil +} + +// githubMergerFactory returns one github merger for every queue routed to it. +// The merger holds no per-queue state, so the queue's Config is not needed. +type githubMergerFactory struct { + merger merger.Merger +} + +func (f *githubMergerFactory) For(_ merger.Config) (merger.Merger, error) { + return f.merger, nil +} + // noopMergerFactory builds a noop merger per queue, bound to that queue's // config. The synthetic revision-id counter is held here rather than on the // merger so ids stay unique across every queue in the process. diff --git a/service/runway/server/main_test.go b/service/runway/server/main_test.go index e5804c9dd..fc986ad2e 100644 --- a/service/runway/server/main_test.go +++ b/service/runway/server/main_test.go @@ -17,6 +17,8 @@ package main import ( "context" "errors" + "fmt" + "net/http" "os" "os/exec" "testing" @@ -32,7 +34,9 @@ import ( extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" gitexec "github.com/uber/submitqueue/platform/git/exec" + phttp "github.com/uber/submitqueue/platform/http" "github.com/uber/submitqueue/runway/controller/dlq" + githubmerger "github.com/uber/submitqueue/runway/extension/merger/github" "go.uber.org/mock/gomock" "go.uber.org/zap/zaptest" ) @@ -88,6 +92,21 @@ func TestPrimaryConsumer_GitFailureDisposition(t *testing.T) { controller: gitexec.NewCommandError("rev-parse", "exit status 128: fatal: ambiguous argument 'origin/main': unknown revision", exitErr), wantOutcome: "reject", }, + { + name: "GitHub gateway failure is nacked for retry", + controller: fmt.Errorf("merge pull request #1: %w", phttp.NewStatusError(http.StatusBadGateway, nil)), + wantOutcome: "nack", + }, + { + name: "GitHub permission failure is rejected to dead letter", + controller: fmt.Errorf("get pull request #1: %w", phttp.NewStatusError(http.StatusForbidden, nil)), + wantOutcome: "reject", + }, + { + name: "GitHub merge still pending is nacked for retry", + controller: fmt.Errorf("%w: merge uuid-1 of uber/submitqueue#1", githubmerger.ErrMergePending), + wantOutcome: "nack", + }, { name: "unknown error is rejected to dead letter", controller: errors.New("unknown failure"),