From b22fbd9ed121a14098768a9d5a4133b50713295d Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Thu, 6 Aug 2026 15:50:58 -0700 Subject: [PATCH] test(runway): end-to-end coverage for the merge queue round trip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? Runway had no integration or e2e coverage at all — `test/integration/` has no `runway/` subtree, and `test/e2e/` covered only `submitqueue` and `stovepipe`. Everything Runway does was proven in isolation: `git_merger_test.go` drives a real `git` binary against a real bare remote across every merge strategy, and the controllers are covered with mocks. What nothing proved is the service. A client publishes a `MergeRequest` to a real queue and a `MergeResult` has to come back on the signal topic under the client's correlation id. Topic-registry wiring, the protojson round trip over the wire, partition-key propagation, the ack-and-publish-FAILED path for expected outcomes, and DLQ reconciliation were all untested. The submitqueue e2e does boot the Runway container, but with the noop merger and only as a dependency of submitqueue's own flow. ### What? `test/e2e/runway/` is a compose-backed suite (queue MySQL + the Runway service) that drives Runway the way its real client does. Runway is consumer-only, so the suite publishes to an inbound merge topic and listens on the corresponding signal topic. Five cases: - **Merge happy path** — SUCCEEDED on `merge-signal` with the id echoed, every step attributable by step id in application order, a produced revision per step, and the partition key carried from request to result. - **Conflict-check happy path** — SUCCEEDED on `merge-conflict-check-signal` with no outputs, since a dry run commits nothing. - **Terminal failures** — conflict and invalid request, on both topic pairs: FAILED with a reason, answered by the controller rather than dead-lettered. - **DLQ reconcile** — an unexpected fault is rejected to the dead-letter topic and the reconciler resolves the correlation id, so the client never waits forever. - **Undecodable payload** — bytes that carry no correlation id signal nothing. Asserted with a same-partition sentinel rather than a sleep. Outcomes have to be steerable from the payload, so the service runs a new `runway/extension/merger/fake` whose result is driven by an `sq-fake=` marker in a change URI (`merge-conflict`, `merge-invalid`, `merge-error`), matching the convention the submitqueue fakes already use. It is selected by `MERGER=fake`; the git and noop paths are untouched. What a merge *does* stays out of the e2e — that is the git merger's own test. The marker helper moved from `submitqueue/core/fakemarker` to `platform/fakemarker`. Runway needs it and `{domain}/core/` is domain-internal; four importers were updated. One assertion is worth calling out. Checking that a request landed on the dead-letter topic races the queue's message GC, which reclaims the row once the reconciler acks it. The suite discriminates on the `dead-lettered: ` reason prefix instead — deterministic, and also the only thing on the wire that tells a client which component answered. ## Test Plan - ✅ `bazel test //test/e2e/runway:go_default_test` — 5 tests, ~13s. Verified with `-test.v` that every case and subtest actually runs. - ✅ `bazel test //test/e2e/submitqueue:go_default_test` and `//test/e2e/stovepipe:go_default_test` — confirms the `fakemarker` move. - ✅ `bazel test //runway/...` — including the new fake's unit tests. - ✅ `make test`, `make fmt`, license headers, gazelle/tidy idempotency. `//submitqueue/orchestrator/controller/{batch,cancel}:go_default_test` fail, but they fail identically on unmodified `main` and `bazel query` shows neither depends on anything this touches. --- .../core => platform}/fakemarker/BUILD.bazel | 2 +- .../fakemarker/fakemarker.go | 0 .../fakemarker/fakemarker_test.go | 0 runway/extension/merger/fake/BUILD.bazel | 29 ++ runway/extension/merger/fake/fake.go | 115 +++++++ runway/extension/merger/fake/fake_test.go | 151 ++++++++++ service/runway/server/BUILD.bazel | 1 + service/runway/server/docker-compose.yml | 4 + service/runway/server/main.go | 35 ++- .../extension/buildrunner/fake/BUILD.bazel | 2 +- .../extension/buildrunner/fake/fake.go | 2 +- .../extension/changeprovider/fake/BUILD.bazel | 2 +- .../extension/changeprovider/fake/fake.go | 2 +- .../extension/mergechecker/fake/BUILD.bazel | 2 +- .../extension/mergechecker/fake/fake.go | 2 +- submitqueue/extension/scorer/fake/BUILD.bazel | 2 +- submitqueue/extension/scorer/fake/fake.go | 2 +- test/e2e/runway/BUILD.bazel | 35 +++ test/e2e/runway/harness_test.go | 196 ++++++++++++ test/e2e/runway/suite_test.go | 282 ++++++++++++++++++ test/e2e/submitqueue/harness_test.go | 2 +- 21 files changed, 854 insertions(+), 14 deletions(-) rename {submitqueue/core => platform}/fakemarker/BUILD.bazel (86%) rename {submitqueue/core => platform}/fakemarker/fakemarker.go (100%) rename {submitqueue/core => platform}/fakemarker/fakemarker_test.go (100%) create mode 100644 runway/extension/merger/fake/BUILD.bazel create mode 100644 runway/extension/merger/fake/fake.go create mode 100644 runway/extension/merger/fake/fake_test.go create mode 100644 test/e2e/runway/BUILD.bazel create mode 100644 test/e2e/runway/harness_test.go create mode 100644 test/e2e/runway/suite_test.go diff --git a/submitqueue/core/fakemarker/BUILD.bazel b/platform/fakemarker/BUILD.bazel similarity index 86% rename from submitqueue/core/fakemarker/BUILD.bazel rename to platform/fakemarker/BUILD.bazel index 260ec1ef..504fc051 100644 --- a/submitqueue/core/fakemarker/BUILD.bazel +++ b/platform/fakemarker/BUILD.bazel @@ -3,7 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = ["fakemarker.go"], - importpath = "github.com/uber/submitqueue/submitqueue/core/fakemarker", + importpath = "github.com/uber/submitqueue/platform/fakemarker", visibility = ["//visibility:public"], deps = ["//platform/base/change:go_default_library"], ) diff --git a/submitqueue/core/fakemarker/fakemarker.go b/platform/fakemarker/fakemarker.go similarity index 100% rename from submitqueue/core/fakemarker/fakemarker.go rename to platform/fakemarker/fakemarker.go diff --git a/submitqueue/core/fakemarker/fakemarker_test.go b/platform/fakemarker/fakemarker_test.go similarity index 100% rename from submitqueue/core/fakemarker/fakemarker_test.go rename to platform/fakemarker/fakemarker_test.go diff --git a/runway/extension/merger/fake/BUILD.bazel b/runway/extension/merger/fake/BUILD.bazel new file mode 100644 index 00000000..4f3f61c8 --- /dev/null +++ b/runway/extension/merger/fake/BUILD.bazel @@ -0,0 +1,29 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["fake.go"], + importpath = "github.com/uber/submitqueue/runway/extension/merger/fake", + visibility = ["//visibility:public"], + deps = [ + "//api/runway/messagequeue:go_default_library", + "//api/runway/messagequeue/protopb:go_default_library", + "//platform/fakemarker:go_default_library", + "//runway/extension/merger:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["fake_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", + "//runway/extension/merger:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/runway/extension/merger/fake/fake.go b/runway/extension/merger/fake/fake.go new file mode 100644 index 00000000..618f11c9 --- /dev/null +++ b/runway/extension/merger/fake/fake.go @@ -0,0 +1,115 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package fake provides a merger.Merger whose outcome is driven by the request +// payload. With no marker it succeeds like the noop merger, behaving as a +// best-case stub for wiring and baselines. A failure can be injected end-to-end +// (e.g. from an e2e merge request) by embedding a marker token in a change URI +// of the form "sq-fake=": +// +// sq-fake=merge-conflict -> merger.ErrConflict +// sq-fake=merge-invalid -> merger.ErrInvalidRequest +// sq-fake=merge-error -> a plain (non-retryable) error +// +// The first token found across the request's steps, in order, decides the +// outcome for the whole request. This lets a single running stack exercise +// Runway's terminal-failure and dead-letter paths purely by varying request +// payloads. It is intended for examples and tests only, never production. +package fake + +import ( + "context" + "fmt" + "sync/atomic" + + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" + "github.com/uber/submitqueue/platform/fakemarker" + "github.com/uber/submitqueue/runway/extension/merger" +) + +// Recognized marker tokens. See the package doc for the convention. +const ( + tokenConflict = "merge-conflict" + tokenInvalid = "merge-invalid" + tokenError = "merge-error" +) + +var _ merger.Merger = (*Merger)(nil) + +// Merger is a Merger that succeeds unless a marker token in a change URI +// requests otherwise. +type Merger struct { + seq atomic.Uint64 +} + +// New returns a Merger that defaults to success and honors marker tokens +// embedded in change URIs. +func New() *Merger { return &Merger{} } + +// CheckMergeability reports the request as mergeable unless a recognized marker +// token asks for a failure. Outputs are empty, as for any dry run. +func (m *Merger) CheckMergeability(_ context.Context, req *runwaymq.MergeRequest) (*runwaymq.MergeResult, error) { + if err := injectedFailure(req); err != nil { + return nil, err + } + + steps := make([]*runwaymq.StepResult, len(req.GetSteps())) + for i, s := range req.GetSteps() { + steps[i] = &runwaymq.StepResult{StepId: s.GetStepId()} + } + return &runwaymq.MergeResult{ + Id: req.GetId(), + Outcome: runwaypb.Outcome_SUCCEEDED, + Steps: steps, + }, nil +} + +// Merge reports the request as merged unless a recognized marker token asks for +// a failure, producing one synthetic revision id per step. +func (m *Merger) Merge(_ context.Context, req *runwaymq.MergeRequest) (*runwaymq.MergeResult, error) { + if err := injectedFailure(req); err != nil { + return nil, err + } + + steps := make([]*runwaymq.StepResult, len(req.GetSteps())) + for i, s := range req.GetSteps() { + n := m.seq.Add(1) + steps[i] = &runwaymq.StepResult{ + StepId: s.GetStepId(), + Outputs: []*runwaymq.StepOutput{{Id: fmt.Sprintf("%040x", n)}}, + } + } + return &runwaymq.MergeResult{ + Id: req.GetId(), + Outcome: runwaypb.Outcome_SUCCEEDED, + Steps: steps, + }, nil +} + +// injectedFailure returns the error the request's marker token asks for, or nil +// when no step carries a recognized token. +func injectedFailure(req *runwaymq.MergeRequest) error { + for _, s := range req.GetSteps() { + switch fakemarker.Token(s.GetChange().GetUris()) { + case tokenConflict: + return fmt.Errorf("fake: marked conflicting on step %s: %w", s.GetStepId(), merger.ErrConflict) + case tokenInvalid: + return fmt.Errorf("fake: marked invalid on step %s: %w", s.GetStepId(), merger.ErrInvalidRequest) + case tokenError: + return fmt.Errorf("fake: marked failing on step %s", s.GetStepId()) + } + } + return nil +} diff --git a/runway/extension/merger/fake/fake_test.go b/runway/extension/merger/fake/fake_test.go new file mode 100644 index 00000000..868afaf9 --- /dev/null +++ b/runway/extension/merger/fake/fake_test.go @@ -0,0 +1,151 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package fake + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + changepb "github.com/uber/submitqueue/api/base/change/protopb" + strategypb "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/runway/extension/merger" +) + +const baseURI = "github://github.example.com/uber/repo/pull/1/abcdef0123456789abcdef0123456789abcdef01" + +// requestWith builds a two-step request whose second step carries the given +// URIs, so tests can prove the token is found on any step, not just the first. +func requestWith(uris ...string) *runwaymq.MergeRequest { + return &runwaymq.MergeRequest{ + Id: "queue-a/42", + QueueName: "queue-a", + Steps: []*runwaymq.MergeStep{ + { + StepId: "queue-a/1", + Change: &changepb.Change{Uris: []string{baseURI}}, + Strategy: strategypb.Strategy_REBASE, + }, + { + StepId: "queue-a/2", + Change: &changepb.Change{Uris: uris}, + Strategy: strategypb.Strategy_REBASE, + }, + }, + } +} + +func TestUnmarkedRequestSucceeds(t *testing.T) { + req := requestWith(baseURI) + + t.Run("check mergeability reports no outputs", func(t *testing.T) { + res, err := New().CheckMergeability(context.Background(), req) + require.NoError(t, err) + + assert.Equal(t, req.GetId(), res.GetId()) + assert.Equal(t, runwaypb.Outcome_SUCCEEDED, res.GetOutcome()) + require.Len(t, res.GetSteps(), 2) + assert.Equal(t, "queue-a/1", res.GetSteps()[0].GetStepId()) + assert.Empty(t, res.GetSteps()[0].GetOutputs()) + assert.Equal(t, "queue-a/2", res.GetSteps()[1].GetStepId()) + assert.Empty(t, res.GetSteps()[1].GetOutputs()) + }) + + t.Run("merge reports one output per step", func(t *testing.T) { + res, err := New().Merge(context.Background(), req) + require.NoError(t, err) + + assert.Equal(t, req.GetId(), res.GetId()) + assert.Equal(t, runwaypb.Outcome_SUCCEEDED, res.GetOutcome()) + require.Len(t, res.GetSteps(), 2) + require.Len(t, res.GetSteps()[0].GetOutputs(), 1) + require.Len(t, res.GetSteps()[1].GetOutputs(), 1) + assert.NotEqual(t, + res.GetSteps()[0].GetOutputs()[0].GetId(), + res.GetSteps()[1].GetOutputs()[0].GetId(), + "each step should produce a distinct revision id") + }) +} + +func TestMarkedRequestFails(t *testing.T) { + tests := []struct { + name string + token string + want error // nil means "an error that is neither terminal sentinel" + }{ + {name: "conflict", token: tokenConflict, want: merger.ErrConflict}, + {name: "invalid", token: tokenInvalid, want: merger.ErrInvalidRequest}, + {name: "plain error", token: tokenError, want: nil}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := requestWith(baseURI + "?sq-fake=" + tt.token) + + for _, call := range []struct { + name string + fn func(*runwaymq.MergeRequest) (*runwaymq.MergeResult, error) + }{ + {"CheckMergeability", func(r *runwaymq.MergeRequest) (*runwaymq.MergeResult, error) { + return New().CheckMergeability(context.Background(), r) + }}, + {"Merge", func(r *runwaymq.MergeRequest) (*runwaymq.MergeResult, error) { + return New().Merge(context.Background(), r) + }}, + } { + t.Run(call.name, func(t *testing.T) { + res, err := call.fn(req) + require.Error(t, err) + assert.Nil(t, res) + + if tt.want != nil { + assert.ErrorIs(t, err, tt.want) + assert.True(t, merger.IsTerminal(err), "sentinel failures must be terminal") + return + } + assert.False(t, merger.IsTerminal(err), + "an unmarked failure must not look terminal, so it dead-letters") + }) + } + }) + } +} + +func TestUnrecognizedTokenSucceeds(t *testing.T) { + req := requestWith(baseURI + "?sq-fake=some-other-fakes-token") + + res, err := New().Merge(context.Background(), req) + require.NoError(t, err) + assert.Equal(t, runwaypb.Outcome_SUCCEEDED, res.GetOutcome()) +} + +func TestFirstRecognizedTokenWins(t *testing.T) { + req := &runwaymq.MergeRequest{ + Id: "queue-a/42", + QueueName: "queue-a", + Steps: []*runwaymq.MergeStep{ + {StepId: "queue-a/1", Change: &changepb.Change{Uris: []string{baseURI + "?sq-fake=" + tokenConflict}}}, + {StepId: "queue-a/2", Change: &changepb.Change{Uris: []string{baseURI + "?sq-fake=" + tokenInvalid}}}, + }, + } + + _, err := New().Merge(context.Background(), req) + require.Error(t, err) + assert.ErrorIs(t, err, merger.ErrConflict) + assert.NotErrorIs(t, err, merger.ErrInvalidRequest) +} diff --git a/service/runway/server/BUILD.bazel b/service/runway/server/BUILD.bazel index b3e8d4f7..e48895a6 100644 --- a/service/runway/server/BUILD.bazel +++ b/service/runway/server/BUILD.bazel @@ -28,6 +28,7 @@ go_library( "//runway/controller/merge:go_default_library", "//runway/controller/mergeconflictcheck:go_default_library", "//runway/extension/merger:go_default_library", + "//runway/extension/merger/fake:go_default_library", "//runway/extension/merger/git:go_default_library", "//runway/extension/merger/noop:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", diff --git a/service/runway/server/docker-compose.yml b/service/runway/server/docker-compose.yml index 25bb7f8d..427c4504 100644 --- a/service/runway/server/docker-compose.yml +++ b/service/runway/server/docker-compose.yml @@ -43,6 +43,10 @@ services: - "8080" # Random ephemeral port to avoid conflicts environment: - PORT=:8080 + # Merger implementation. Empty (the default) resolves from the merge + # environment: git when MERGE_CHECKOUT_PATH is set, noop otherwise. The + # e2e suite sets SQ_RUNWAY_MERGER=fake to drive outcomes from the payload. + - MERGER=${SQ_RUNWAY_MERGER:-} # Queue infrastructure connection - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true - HOSTNAME=runway-dev diff --git a/service/runway/server/main.go b/service/runway/server/main.go index fdabcf29..4e36388e 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -47,6 +47,7 @@ import ( "github.com/uber/submitqueue/runway/controller/merge" "github.com/uber/submitqueue/runway/controller/mergeconflictcheck" "github.com/uber/submitqueue/runway/extension/merger" + "github.com/uber/submitqueue/runway/extension/merger/fake" gitmerger "github.com/uber/submitqueue/runway/extension/merger/git" "github.com/uber/submitqueue/runway/extension/merger/noop" "go.uber.org/zap" @@ -302,11 +303,27 @@ func run() error { return err } -// newMergerFactory returns a merger.Factory for the server. When -// MERGE_CHECKOUT_PATH is set it wires the git-backed merger built from the -// MERGE_* / GIT_* environment; otherwise it falls back to the noop merger so -// local development and compose runs need no git checkout. +// newMergerFactory returns a merger.Factory for the server. MERGER selects the +// implementation explicitly; when it is unset the choice falls back to the merge +// environment — MERGE_CHECKOUT_PATH wires the git-backed merger built from the +// MERGE_* / GIT_* environment, and its absence wires the noop merger so local +// development and compose runs need no git checkout. func newMergerFactory(logger *zap.Logger, scope tally.Scope) (merger.Factory, error) { + switch impl := strings.ToLower(strings.TrimSpace(os.Getenv("MERGER"))); impl { + case "fake": + // Marker-driven outcomes, for e2e tests that need Runway to fail on + // demand without a git checkout. Never production. + logger.Info("MERGER=fake; using marker-driven fake merger") + return &fakeMergerFactory{merger: fake.New()}, nil + case "noop": + logger.Info("MERGER=noop; using noop merger") + return &noopMergerFactory{}, nil + case "", "git": + // Fall through to the merge-environment default below. + default: + return nil, fmt.Errorf("invalid MERGER %q", impl) + } + checkoutPath := os.Getenv("MERGE_CHECKOUT_PATH") if checkoutPath == "" { logger.Info("MERGE_CHECKOUT_PATH not set; using noop merger") @@ -369,6 +386,16 @@ func (f *noopMergerFactory) For(_ merger.Config) (merger.Merger, error) { return noop.New(), nil } +// fakeMergerFactory shares one fake merger across queues so the synthetic +// revision ids it mints stay unique for the lifetime of the process. +type fakeMergerFactory struct { + merger merger.Merger +} + +func (f *fakeMergerFactory) For(_ merger.Config) (merger.Merger, error) { + return f.merger, nil +} + // parseStrategy maps the MERGE_DEFAULT_STRATEGY env value to a concrete merge // strategy, defaulting to REBASE when unset. DEFAULT is rejected because it // cannot itself be the default a step resolves to. diff --git a/submitqueue/extension/buildrunner/fake/BUILD.bazel b/submitqueue/extension/buildrunner/fake/BUILD.bazel index a13cc7d6..9e7ed80d 100644 --- a/submitqueue/extension/buildrunner/fake/BUILD.bazel +++ b/submitqueue/extension/buildrunner/fake/BUILD.bazel @@ -6,8 +6,8 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/extension/buildrunner/fake", visibility = ["//visibility:public"], deps = [ + "//platform/fakemarker:go_default_library", "//submitqueue/core/changeset:go_default_library", - "//submitqueue/core/fakemarker:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/buildrunner:go_default_library", ], diff --git a/submitqueue/extension/buildrunner/fake/fake.go b/submitqueue/extension/buildrunner/fake/fake.go index 9b99c39f..df9aa7d9 100644 --- a/submitqueue/extension/buildrunner/fake/fake.go +++ b/submitqueue/extension/buildrunner/fake/fake.go @@ -37,8 +37,8 @@ import ( "fmt" "strings" + "github.com/uber/submitqueue/platform/fakemarker" "github.com/uber/submitqueue/submitqueue/core/changeset" - "github.com/uber/submitqueue/submitqueue/core/fakemarker" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/buildrunner" ) diff --git a/submitqueue/extension/changeprovider/fake/BUILD.bazel b/submitqueue/extension/changeprovider/fake/BUILD.bazel index 14fc2bb5..7d9216cf 100644 --- a/submitqueue/extension/changeprovider/fake/BUILD.bazel +++ b/submitqueue/extension/changeprovider/fake/BUILD.bazel @@ -6,7 +6,7 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/extension/changeprovider/fake", visibility = ["//visibility:public"], deps = [ - "//submitqueue/core/fakemarker:go_default_library", + "//platform/fakemarker:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/changeprovider:go_default_library", ], diff --git a/submitqueue/extension/changeprovider/fake/fake.go b/submitqueue/extension/changeprovider/fake/fake.go index 1e30998e..f49c1f59 100644 --- a/submitqueue/extension/changeprovider/fake/fake.go +++ b/submitqueue/extension/changeprovider/fake/fake.go @@ -29,7 +29,7 @@ import ( "context" "fmt" - "github.com/uber/submitqueue/submitqueue/core/fakemarker" + "github.com/uber/submitqueue/platform/fakemarker" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/changeprovider" ) diff --git a/submitqueue/extension/mergechecker/fake/BUILD.bazel b/submitqueue/extension/mergechecker/fake/BUILD.bazel index 2528b872..d1c85e0c 100644 --- a/submitqueue/extension/mergechecker/fake/BUILD.bazel +++ b/submitqueue/extension/mergechecker/fake/BUILD.bazel @@ -6,7 +6,7 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/extension/mergechecker/fake", visibility = ["//visibility:public"], deps = [ - "//submitqueue/core/fakemarker:go_default_library", + "//platform/fakemarker:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/mergechecker:go_default_library", ], diff --git a/submitqueue/extension/mergechecker/fake/fake.go b/submitqueue/extension/mergechecker/fake/fake.go index 7d497074..60519720 100644 --- a/submitqueue/extension/mergechecker/fake/fake.go +++ b/submitqueue/extension/mergechecker/fake/fake.go @@ -30,7 +30,7 @@ import ( "context" "fmt" - "github.com/uber/submitqueue/submitqueue/core/fakemarker" + "github.com/uber/submitqueue/platform/fakemarker" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/mergechecker" ) diff --git a/submitqueue/extension/scorer/fake/BUILD.bazel b/submitqueue/extension/scorer/fake/BUILD.bazel index a50cfc2b..2d6635e7 100644 --- a/submitqueue/extension/scorer/fake/BUILD.bazel +++ b/submitqueue/extension/scorer/fake/BUILD.bazel @@ -6,8 +6,8 @@ go_library( importpath = "github.com/uber/submitqueue/submitqueue/extension/scorer/fake", visibility = ["//visibility:public"], deps = [ + "//platform/fakemarker:go_default_library", "//submitqueue/core/changeset:go_default_library", - "//submitqueue/core/fakemarker:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/scorer:go_default_library", ], diff --git a/submitqueue/extension/scorer/fake/fake.go b/submitqueue/extension/scorer/fake/fake.go index b5a76c78..d07e7beb 100644 --- a/submitqueue/extension/scorer/fake/fake.go +++ b/submitqueue/extension/scorer/fake/fake.go @@ -27,8 +27,8 @@ import ( "context" "fmt" + "github.com/uber/submitqueue/platform/fakemarker" "github.com/uber/submitqueue/submitqueue/core/changeset" - "github.com/uber/submitqueue/submitqueue/core/fakemarker" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/scorer" ) diff --git a/test/e2e/runway/BUILD.bazel b/test/e2e/runway/BUILD.bazel new file mode 100644 index 00000000..6b98e4ad --- /dev/null +++ b/test/e2e/runway/BUILD.bazel @@ -0,0 +1,35 @@ +load("@rules_go//go:def.bzl", "go_test") + +go_test( + name = "go_default_test", + srcs = [ + "harness_test.go", + "suite_test.go", + ], + data = [ + "//platform/extension/messagequeue/mysql/schema", + "//service/runway/server:docker-compose.yml", + "//service/runway/server:docker_test_context", + ], + tags = [ + "e2e", + "integration", + "requires-network", + ], + 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/base/messagequeue:go_default_library", + "//platform/extension/messagequeue:go_default_library", + "//platform/extension/messagequeue/mysql:go_default_library", + "//test/testutil:go_default_library", + "@com_github_go_sql_driver_mysql//:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_stretchr_testify//suite:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//zaptest:go_default_library", + ], +) diff --git a/test/e2e/runway/harness_test.go b/test/e2e/runway/harness_test.go new file mode 100644 index 00000000..1fdf43e3 --- /dev/null +++ b/test/e2e/runway/harness_test.go @@ -0,0 +1,196 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e_test + +// Helpers for the Runway e2e suite: the client side of the merge contract +// (publish a request, await the correlated result) plus the request builders +// the tests vary. + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/require" + changepb "github.com/uber/submitqueue/api/base/change/protopb" + strategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" +) + +// Topic names Runway binds its topic keys to. The suite is the client on the +// other side of that contract, so it addresses topics by the same keys the +// service registers (see service/runway/server/main.go newTopicRegistry). +var ( + topicMerge = runwaymq.TopicKeyMerge.String() + topicMergeSignal = runwaymq.TopicKeyMergeSignal.String() + topicCheck = runwaymq.TopicKeyMergeConflictCheck.String() + topicCheckSignal = runwaymq.TopicKeyMergeConflictCheckSignal.String() +) + +// reconciledPrefix is what the DLQ reconciler prepends to the reason of a +// result it republishes (see runway/controller/dlq). It is the only thing on +// the wire that distinguishes a failure the primary controller answered from +// one it abandoned to the dead-letter topic, so the suite reads it to tell the +// two paths apart. Asserting on the dead-letter row instead would race the +// queue's message GC, which reclaims the row once the reconciler acks it. +const reconciledPrefix = "dead-lettered: " + +// reconciled reports whether a result came from the DLQ reconciler rather than +// from the primary controller. +func reconciled(result *runwaymq.MergeResult) bool { + return strings.HasPrefix(result.GetReason(), reconciledPrefix) +} + +// baseURI is a well-formed change URI. Tests append a "sq-fake=" marker +// to steer the fake merger; without one it merges cleanly. +const baseURI = "github://github.example.com/uber/runway-e2e/pull/1/abcdef0123456789abcdef0123456789abcdef01" + +// Marker tokens the fake merger recognizes (runway/extension/merger/fake). +const ( + markerConflict = "merge-conflict" + markerInvalid = "merge-invalid" + markerError = "merge-error" +) + +// markedURI returns baseURI carrying the given fake-merger marker token. +func markedURI(token string) string { + return baseURI + "?sq-fake=" + token +} + +// observedResult is a MergeResult as it arrived on a signal topic, with the +// envelope fields the tests assert on alongside the payload. +type observedResult struct { + result *runwaymq.MergeResult + messageID string + partitionKey string +} + +// observer consumes one signal topic on behalf of the suite, standing in for +// the client that published the request. Deliveries are decoded and acked as +// they arrive and retained in arrival order, so a test can await its own +// correlation id without losing results belonging to another. +// +// Only the test goroutine touches an observer, so it needs no locking. +type observer struct { + topic string + ch <-chan extqueue.Delivery + seen []observedResult +} + +// mark returns a cursor into the results observed so far, for asserting later +// on exactly what arrived after some point (see newSince). +func (o *observer) mark() int { return len(o.seen) } + +// newSince returns the results observed after the given cursor. +func (o *observer) newSince(mark int) []observedResult { return o.seen[mark:] } + +// await blocks until a result carrying the given correlation id has arrived on +// the topic, returning it. Results for other ids are retained, so awaiting them +// later still works. There is no timeout by design: Bazel's test timeout is the +// only deadline. +func (o *observer) await(t *testing.T, ctx context.Context, id string) observedResult { + t.Helper() + + for i := range o.seen { + if o.seen[i].result.GetId() == id { + return o.seen[i] + } + } + + for { + delivery, ok := <-o.ch + require.True(t, ok, "delivery channel for %s closed while awaiting %s", o.topic, id) + require.NotNil(t, delivery, "nil delivery on %s", o.topic) + + msg := delivery.Message() + result := &runwaymq.MergeResult{} + require.NoError(t, runwaymq.Unmarshal(msg.Payload, result), + "failed to decode result on %s", o.topic) + require.NoError(t, delivery.Ack(ctx), "failed to ack result on %s", o.topic) + + o.seen = append(o.seen, observedResult{ + result: result, + messageID: msg.ID, + partitionKey: msg.PartitionKey, + }) + if result.GetId() == id { + return o.seen[len(o.seen)-1] + } + } +} + +// observe subscribes to a signal topic under the suite's own consumer group. +// Signal topics have no other consumer, so the suite reads them in full. +func (s *RunwayE2ESuite) observe(topic string) *observer { + t := s.T() + + config := extqueue.DefaultSubscriptionConfig("runway-e2e-observer", "runway-e2e-observer") + // The defaults (30s lease, 60s visibility) only slow teardown down here: + // the suite acks every delivery as it decodes it. + config.VisibilityTimeoutMs = 2000 + config.LeaseDurationMs = 3000 + config.LeaseRenewalIntervalMs = 1000 + + ch, err := s.queue.Subscriber().Subscribe(s.ctx, topic, config) + require.NoError(t, err, "failed to subscribe to %s", topic) + + return &observer{topic: topic, ch: ch} +} + +// publish sends a merge request to one of Runway's inbound topics, partitioned +// by queue name exactly as the orchestrator publishes it. +func (s *RunwayE2ESuite) publish(topic string, request *runwaymq.MergeRequest) { + t := s.T() + + payload, err := runwaymq.Marshal(request) + require.NoError(t, err, "failed to marshal merge request") + + s.publishRaw(topic, request.GetId(), request.GetQueueName(), payload) +} + +// publishRaw sends arbitrary bytes to a topic, for the malformed-payload case. +func (s *RunwayE2ESuite) publishRaw(topic, id, partitionKey string, payload []byte) { + t := s.T() + + msg := entityqueue.NewMessage(id, payload, partitionKey, nil) + require.NoError(t, s.queue.Publisher().Publish(s.ctx, topic, msg), + "failed to publish %s to %s", id, topic) + s.log.Logf("published %s to %s (partition %s)", id, topic, partitionKey) +} + +// mergeRequest builds a request with a unique correlation id. Ids must be +// unique across the suite: the queue deduplicates publishes on +// (topic, partition, id), and the correlation id is the message id on both the +// request and its result, so a reused id silently drops a publish. +func (s *RunwayE2ESuite) mergeRequest(queue string, steps ...*runwaymq.MergeStep) *runwaymq.MergeRequest { + s.seq++ + return &runwaymq.MergeRequest{ + Id: fmt.Sprintf("%s/%d", queue, s.seq), + QueueName: queue, + Steps: steps, + } +} + +// step builds one merge step applying the given URIs with REBASE. +func step(id string, uris ...string) *runwaymq.MergeStep { + return &runwaymq.MergeStep{ + StepId: id, + Change: &changepb.Change{Uris: uris}, + Strategy: strategypb.Strategy_REBASE, + } +} diff --git a/test/e2e/runway/suite_test.go b/test/e2e/runway/suite_test.go new file mode 100644 index 00000000..868bd992 --- /dev/null +++ b/test/e2e/runway/suite_test.go @@ -0,0 +1,282 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package e2e_test + +// Runway end-to-end tests. +// +// These tests use docker-compose from service/runway/server/docker-compose.yml. +// They are hermetic: the runway image is built from a staged context whose +// inputs (Bazel-built Linux binary, Dockerfile) are all declared data +// dependencies of the test target. +// +// Run with: +// +// make e2e-test +// +// or only this package: +// +// bazel test //test/e2e/runway:go_default_test +// +// The stack runs the Runway service plus a queue MySQL. Runway is consumer-only +// — it has no gateway — so the suite drives it the way its real client does: it +// publishes a MergeRequest to an inbound merge topic and listens on the +// corresponding signal topic for the MergeResult carrying its correlation id. +// That is the seam no other test covers: topic wiring, the protojson round-trip +// over the wire, partition-key propagation, the ack-and-publish-FAILED path for +// expected outcomes, and DLQ reconciliation. +// +// The service runs with MERGER=fake, so what a merge *does* is out of scope +// here; the fake's marker tokens only pick which outcome Runway has to carry +// back. Merge behavior proper — strategies, real conflicts, staleness, author +// attribution — is covered against a real git remote in +// runway/extension/merger/git/git_merger_test.go. + +import ( + "context" + "database/sql" + "testing" + + _ "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/stretchr/testify/suite" + "github.com/uber-go/tally" + runwaypb "github.com/uber/submitqueue/api/runway/messagequeue/protopb" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" + "github.com/uber/submitqueue/test/testutil" + "go.uber.org/zap/zaptest" +) + +type RunwayE2ESuite struct { + suite.Suite + ctx context.Context + log *testutil.TestLogger + stack *testutil.ComposeStack + queueDB *sql.DB + queue extqueue.Queue + + // Signal-topic observers, standing in for the client awaiting its results. + mergeSignal *observer + checkSignal *observer + + // seq mints unique correlation ids across the suite. + seq int +} + +func TestRunwayE2E(t *testing.T) { + suite.Run(t, new(RunwayE2ESuite)) +} + +func (s *RunwayE2ESuite) SetupSuite() { + t := s.T() + s.ctx = context.Background() + s.log = testutil.NewTestLogger(t) + + s.log.Logf("Starting Runway e2e test suite using docker-compose") + + // Outcomes have to be steerable from the request payload, so the service + // runs the marker-driven fake merger instead of noop or git. Set before the + // stack is created: ComposeStack snapshots the environment at construction. + t.Setenv("SQ_RUNWAY_MERGER", "fake") + + composeFile := testutil.Runfile("service/runway/server/docker-compose.yml") + s.stack = testutil.NewComposeStack(t, s.log, s.ctx, composeFile, "e2e-runway", + testutil.WithBuildContext(map[string]string{ + ".docker-bin/runway": "service/runway/server/runway_linux", + "service/runway/server/Dockerfile": "service/runway/server/Dockerfile", + })) + + err := s.stack.Up() + require.NoError(t, err, "failed to start compose stack") + + s.queueDB, err = s.stack.ConnectMySQLService("mysql-queue") + require.NoError(t, err, "failed to connect to queue MySQL") + + // Applied after the stack is up; the service connects lazily and its + // consumers retry, so the boot ordering is tolerated. + testutil.ApplySchema(t, s.log, s.queueDB, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema")) + + // The suite talks to the same queue backend Runway does, as its client. + s.queue, err = queueMySQL.NewQueue(queueMySQL.Params{ + DB: s.queueDB, + Logger: zaptest.NewLogger(t), + MetricsScope: tally.NoopScope, + }) + require.NoError(t, err, "failed to create queue client") + t.Cleanup(func() { s.queue.Close() }) + + s.mergeSignal = s.observe(topicMergeSignal) + s.checkSignal = s.observe(topicCheckSignal) + + s.log.Logf("Runway e2e test suite ready") +} + +func (s *RunwayE2ESuite) TearDownSuite() { + // Compose stack cleanup is handled automatically by t.Cleanup (registered in + // NewComposeStack). + s.log.Logf("Tearing down Runway e2e test suite") +} + +// TestMerge_HappyPath_PublishesMergedResult drives the committing merge: a +// two-step request in, a SUCCEEDED result out on the merge-signal topic. It +// asserts the whole envelope the client depends on — the correlation id echoed, +// every step attributable by its step id in application order, a produced +// revision per step, and the partition key carried from request to result so +// the client's per-queue ordering survives the round trip. +func (s *RunwayE2ESuite) TestMerge_HappyPath_PublishesMergedResult() { + t := s.T() + const queue = "e2e-runway/merge" + + request := s.mergeRequest(queue, + step("base", baseURI), + step("candidate", baseURI), + ) + s.publish(topicMerge, request) + + observed := s.mergeSignal.await(t, s.ctx, request.GetId()) + result := observed.result + + assert.Equal(t, runwaypb.Outcome_SUCCEEDED, result.GetOutcome()) + assert.Empty(t, result.GetReason(), "a successful merge carries no reason") + assert.Equal(t, queue, observed.partitionKey, + "the result must stay on the request's partition") + + require.Len(t, result.GetSteps(), 2) + assert.Equal(t, "base", result.GetSteps()[0].GetStepId()) + assert.Equal(t, "candidate", result.GetSteps()[1].GetStepId()) + for i, stepResult := range result.GetSteps() { + assert.NotEmpty(t, stepResult.GetOutputs(), + "step %d of a committing merge must report the revision it produced", i) + } +} + +// TestMergeConflictCheck_HappyPath_PublishesMergeableResult drives the dry-run +// check on its own topic pair. The distinguishing assertion against the +// committing merge is that a check commits nothing, so it reports no outputs. +func (s *RunwayE2ESuite) TestMergeConflictCheck_HappyPath_PublishesMergeableResult() { + t := s.T() + const queue = "e2e-runway/check" + + request := s.mergeRequest(queue, step("candidate", baseURI)) + s.publish(topicCheck, request) + + observed := s.checkSignal.await(t, s.ctx, request.GetId()) + result := observed.result + + assert.Equal(t, runwaypb.Outcome_SUCCEEDED, result.GetOutcome()) + assert.Equal(t, queue, observed.partitionKey) + require.Len(t, result.GetSteps(), 1) + assert.Equal(t, "candidate", result.GetSteps()[0].GetStepId()) + assert.Empty(t, result.GetSteps()[0].GetOutputs(), "a dry run produces no revisions") +} + +// TestTerminalFailure_AcksAndSignalsFailed covers the outcomes Runway treats as +// answers rather than faults: a conflict and an unapplicable request. Both must +// come back as a FAILED result on the signal topic, published by the controller +// that handled the request — not by the dead-letter reconciler. Dead-lettering +// an expected outcome would burn the retry budget on a request that can never +// succeed and resolve the client through the backstop instead of the answer. +func (s *RunwayE2ESuite) TestTerminalFailure_AcksAndSignalsFailed() { + tests := []struct { + name string + marker string + }{ + {name: "conflict", marker: markerConflict}, + {name: "invalid request", marker: markerInvalid}, + } + + topics := []struct { + name string + inbound string + observer func() *observer + }{ + {"merge", topicMerge, func() *observer { return s.mergeSignal }}, + {"conflict check", topicCheck, func() *observer { return s.checkSignal }}, + } + + for _, topic := range topics { + for _, tt := range tests { + s.Run(topic.name+"/"+tt.name, func() { + t := s.T() + queue := "e2e-runway/failed" + + request := s.mergeRequest(queue, step("candidate", markedURI(tt.marker))) + s.publish(topic.inbound, request) + + observed := topic.observer().await(t, s.ctx, request.GetId()) + result := observed.result + + assert.Equal(t, runwaypb.Outcome_FAILED, result.GetOutcome()) + assert.NotEmpty(t, result.GetReason(), + "a failed result must tell the client why") + assert.Equal(t, queue, observed.partitionKey) + assert.False(t, reconciled(result), + "an expected outcome must be answered by the controller, not dead-lettered") + }) + } + } +} + +// TestUnexpectedFailure_ReconcilesFromDLQ covers the backstop. An unexpected +// merger fault is not a terminal outcome, so the controller does not answer it; +// the consumer rejects the message to the dead-letter topic instead. Runway is +// the sole responder on the client's correlation id, so a request that stopped +// there would leave the client waiting forever — the DLQ reconciler is what +// resolves it, republishing a FAILED result to the same signal topic. +func (s *RunwayE2ESuite) TestUnexpectedFailure_ReconcilesFromDLQ() { + t := s.T() + const queue = "e2e-runway/dlq" + + request := s.mergeRequest(queue, step("candidate", markedURI(markerError))) + s.publish(topicMerge, request) + + observed := s.mergeSignal.await(t, s.ctx, request.GetId()) + + assert.Equal(t, runwaypb.Outcome_FAILED, observed.result.GetOutcome()) + assert.True(t, reconciled(observed.result), + "an unexpected fault should be resolved by the dead-letter reconciler") + assert.Equal(t, queue, observed.partitionKey, + "the reconciled result must stay on the request's partition") +} + +// TestUndecodablePayload_DropsWithoutSignalling covers the one request Runway +// cannot answer: bytes that do not decode carry no correlation id, so there is +// nothing to resolve and both the controller and the reconciler drop them. +// +// The absence is asserted without sleeping. The garbage and a following +// sentinel share a partition, and a partition is consumed in order, so the +// sentinel's result arriving proves the garbage was already processed. Anything +// the garbage had signalled would sit ahead of the sentinel on the same +// partition of the same signal topic. +func (s *RunwayE2ESuite) TestUndecodablePayload_DropsWithoutSignalling() { + t := s.T() + const queue = "e2e-runway/undecodable" + + mark := s.mergeSignal.mark() + + s.publishRaw(topicMerge, "e2e-runway/garbage", queue, []byte("{not a merge request")) + + sentinel := s.mergeRequest(queue, step("candidate", baseURI)) + s.publish(topicMerge, sentinel) + + observed := s.mergeSignal.await(t, s.ctx, sentinel.GetId()) + assert.Equal(t, runwaypb.Outcome_SUCCEEDED, observed.result.GetOutcome()) + + arrived := s.mergeSignal.newSince(mark) + require.Len(t, arrived, 1, + "the undecodable payload must not signal; only the sentinel should have arrived") + assert.Equal(t, sentinel.GetId(), arrived[0].result.GetId()) +} diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go index 1b6cde82..fc1a7c1e 100644 --- a/test/e2e/submitqueue/harness_test.go +++ b/test/e2e/submitqueue/harness_test.go @@ -51,7 +51,7 @@ func pollUntil(interval time.Duration, condition func() bool) { // land submits a request with the default REBASE strategy and returns its sqid. // URIs may carry "sq-fake=" markers to steer negative paths (see -// submitqueue/core/fakemarker); the happy path uses a plain change URI. +// platform/fakemarker); the happy path uses a plain change URI. func (s *E2EIntegrationSuite) land(queue string, uris ...string) string { t := s.T() resp, err := s.gatewayClient.Land(s.ctx, &gatewaypb.LandRequest{