From b3cf80cd9b3653cd3c3378ccc74102ce9ae05477 Mon Sep 17 00:00:00 2001 From: Preetam Dwivedi Date: Tue, 28 Jul 2026 14:57:55 -0700 Subject: [PATCH] feat(runway): wire the git merger into the server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary ### Why? The git merger exists but nothing constructs it — the server still builds the noop factory, so a deployed Runway acknowledges every merge as an instant success. This is the change that makes Runway actually merge. It is deliberately last in the stack and deliberately opt-in: the switch is the presence of `MERGE_CHECKOUT_PATH`. Unset, the server keeps wiring noop, which is what local development, the compose stack, and the e2e suite depend on — none of them have a git checkout to hand the merger. ### What? `newMergerFactory` now reads the environment and returns either backend. With `MERGE_CHECKOUT_PATH` set it builds a git merger from the `MERGE_*` / `GIT_*` variables — remote, target branch, default strategy, committer identity, and the pinned git runtime — and logs the resolved configuration at startup. Without it, noop, with a log line saying so. A malformed configuration fails startup rather than degrading silently: `parseStrategy` rejects an unrecognized `MERGE_DEFAULT_STRATEGY`, and `DEFAULT` itself is rejected because it cannot be the value a `DEFAULT` step resolves to. Three settings govern the behaviors the merger cannot infer: - `MERGE_CHECK_STALENESS` (default on) verifies each change's provider ref still points at the commit its URI names before applying. - `MERGE_ALLOW_UNRELATED_HISTORIES` (default off) lets a MERGE step import a history that shares no ancestry with the target. It stays off because the refusal it lifts is a safeguard everywhere except a queue whose purpose is such imports. - `MERGE_FETCH_REFSPECS` supplies extra refspecs for a remote that will not serve an unadvertised commit by SHA. Normally empty. `gitMergerFactory` hands the same merger instance to every queue. The merger owns one checkout and serializes its own operations, so a second instance over the same directory would race; a deployment that lands multiple targets wires a per-queue map instead. This is also why the factory is built once at startup rather than per request. ## Test Plan ✅ `bazel build //service/runway/...` — wiring compiles ✅ `bazel test //runway/...` — all targets pass Not covered by automated tests: the git path only engages when `MERGE_CHECKOUT_PATH` points at a real checkout, so the env-to-`Params` mapping is exercised by the merger's own suite rather than through `main`. Watch the `git merger configured` startup log — with checkout, target, and default strategy — on the first deployment that sets the variable; its absence means the server silently fell back to noop. --- runway/README.md | 7 ++ service/runway/README.md | 14 ++++ service/runway/server/BUILD.bazel | 2 + service/runway/server/main.go | 124 ++++++++++++++++++++++++++++-- 4 files changed, 142 insertions(+), 5 deletions(-) diff --git a/runway/README.md b/runway/README.md index f8810559..68e1baa9 100644 --- a/runway/README.md +++ b/runway/README.md @@ -13,6 +13,13 @@ Runway is a single service (the domain *is* the service); its controllers live d Each controller deserializes the `MergeRequest`, obtains a `Merger` for the request's queue from the [`merger`](extension/merger) extension, applies the ordered steps, and publishes a `MergeResult` to the corresponding signal queue (`merge-conflict-check-signal` / `merge-signal`). `merge` commits and reports the produced revisions; `merge-conflict-check` is a dry run that reports mergeability with empty outputs. +## Merger extension + +[`extension/merger`](extension/merger) is the pluggable merge contract. Implementations: + +- [`git`](extension/merger/git) — a git-CLI backend that honors each step's strategy (REBASE, SQUASH_REBASE, MERGE, PROMOTE; DEFAULT resolves to a per-instance default). See its [README](extension/merger/git/README.md). +- `noop` — always-succeeds stub for local development. + ## Failure handling A merge outcome the controller can name is published as a `FAILED` result and acked, not retried: a merge conflict (`merger.ErrConflict`) or an invalid request (`merger.ErrInvalidRequest` — unknown strategy, malformed change URI, invalid PROMOTE composition). The `merger.IsTerminal` helper draws that line. Any other error is an infrastructure fault and is nacked for retry. diff --git a/service/runway/README.md b/service/runway/README.md index a2a7598d..3d7b46b3 100644 --- a/service/runway/README.md +++ b/service/runway/README.md @@ -13,6 +13,10 @@ Each controller applies the request's ordered steps via a `Merger` and publishes These topic keys and their wire contracts are owned by the queue's producer side and published under `api/runway/messagequeue/` (the external, cross-domain contract). +### Merger backend + +The merge work is done by the [`merger`](../../runway/extension/merger) extension. By default the server wires the **noop** merger (always succeeds — for local dev and compose). Set `MERGE_CHECKOUT_PATH` to wire the real **git** merger, built from the `MERGE_*` / `GIT_*` environment (see Configuration); the git merger owns the checkout at that path and pushes to the configured remote/target. + Because Runway only consumes queues and serves `Ping`, it needs a **queue** database but no application/storage database. ## Layout @@ -36,6 +40,16 @@ The Runway controllers themselves live under [`runway/controller/`](../../runway | `QUEUE_MYSQL_DSN` | yes | Queue database DSN | — | | `PORT` | no | gRPC listen address | `:8086` | | `HOSTNAME` | no | Subscriber name for the queue consumer | `runway-` | +| `MERGE_CHECKOUT_PATH` | no | Absolute path to the git checkout the merger owns. When unset, the noop merger is used. | — (noop) | +| `MERGE_REMOTE` | no | Git remote to fetch/push | `origin` | +| `MERGE_TARGET` | no | Destination branch on the remote | `main` | +| `MERGE_DEFAULT_STRATEGY` | no | Strategy a `DEFAULT` step resolves to: `REBASE`, `SQUASH_REBASE`, `MERGE`, or `PROMOTE` | `REBASE` | +| `MERGE_COMMITTER_NAME` | no | Committer name for service-created commits | `SubmitQueue Runway` | +| `MERGE_COMMITTER_EMAIL` | no | Committer email for service-created commits | `runway@submitqueue.invalid` | +| `GIT_EXECUTABLE` / `GIT_EXEC_PATH` / `GIT_TEMPLATE_DIR` | when `MERGE_CHECKOUT_PATH` set | Absolute paths to the pinned git runtime | — | +| `MERGE_CHECK_STALENESS` | no | Verify each change's provider ref still points at the commit its URI names before applying | `true` | +| `MERGE_ALLOW_UNRELATED_HISTORIES` | no | Let a `MERGE` step integrate a change sharing no ancestry with the target (repository imports). Leave off unless the queue exists to perform imports. | `false` | +| `MERGE_FETCH_REFSPECS` | no | Comma-separated extra refspecs fetched each cycle. Only needed for a remote that refuses to serve an unadvertised commit by SHA. | — | ## Running diff --git a/service/runway/server/BUILD.bazel b/service/runway/server/BUILD.bazel index 9cdf9231..b3e8d4f7 100644 --- a/service/runway/server/BUILD.bazel +++ b/service/runway/server/BUILD.bazel @@ -11,6 +11,7 @@ go_library( importpath = "github.com/uber/submitqueue/service/runway/server", visibility = ["//visibility:private"], deps = [ + "//api/base/mergestrategy/protopb:go_default_library", "//api/runway/messagequeue:go_default_library", "//api/runway/protopb:go_default_library", "//platform/consumer:go_default_library", @@ -27,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/git:go_default_library", "//runway/extension/merger/noop:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_uber_go_tally//:go_default_library", diff --git a/service/runway/server/main.go b/service/runway/server/main.go index cb121315..0030c620 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -22,12 +22,15 @@ import ( "net" "os" "os/signal" + "strconv" + "strings" "sync" "syscall" "time" _ "github.com/go-sql-driver/mysql" "github.com/uber-go/tally" + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" pb "github.com/uber/submitqueue/api/runway/protopb" "github.com/uber/submitqueue/platform/consumer" @@ -44,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" + gitmerger "github.com/uber/submitqueue/runway/extension/merger/git" "github.com/uber/submitqueue/runway/extension/merger/noop" "go.uber.org/zap" "google.golang.org/grpc" @@ -164,7 +168,10 @@ func run() error { gate, ) - mergerFactory := newMergerFactory() + mergerFactory, err := newMergerFactory(logger, scope.SubScope("merger")) + if err != nil { + return fmt.Errorf("failed to create merger factory: %w", err) + } mergeConflictCheckController := mergeconflictcheck.NewController(mergeconflictcheck.Params{ Logger: logger.Sugar(), @@ -295,10 +302,65 @@ func run() error { return err } -// newMergerFactory returns a merger.Factory for the example server. The noop -// implementation always succeeds; a real deployment wires a VCS-backed factory. -func newMergerFactory() merger.Factory { - return &noopMergerFactory{} +// 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. +func newMergerFactory(logger *zap.Logger, scope tally.Scope) (merger.Factory, error) { + checkoutPath := os.Getenv("MERGE_CHECKOUT_PATH") + if checkoutPath == "" { + logger.Info("MERGE_CHECKOUT_PATH not set; using noop merger") + return &noopMergerFactory{}, nil + } + + defaultStrategy, err := parseStrategy(os.Getenv("MERGE_DEFAULT_STRATEGY")) + if err != nil { + return nil, err + } + + target := envOr("MERGE_TARGET", "main") + m, err := gitmerger.NewMerger(gitmerger.Params{ + CheckoutPath: checkoutPath, + Remote: envOr("MERGE_REMOTE", "origin"), + Target: target, + DefaultStrategy: defaultStrategy, + Runtime: gitmerger.GitRuntime{ + Executable: os.Getenv("GIT_EXECUTABLE"), + ExecPath: os.Getenv("GIT_EXEC_PATH"), + TemplateDir: os.Getenv("GIT_TEMPLATE_DIR"), + }, + CommitterName: os.Getenv("MERGE_COMMITTER_NAME"), + CommitterEmail: os.Getenv("MERGE_COMMITTER_EMAIL"), + FetchRefspecs: splitRefspecs(os.Getenv("MERGE_FETCH_REFSPECS")), + CheckStaleness: envBool("MERGE_CHECK_STALENESS", true), + // Off by default: it lifts git's refusal to join two unrelated history + // graphs, which is a safeguard everywhere except a queue whose purpose + // is importing one repository's history into another. + AllowUnrelatedHistories: envBool("MERGE_ALLOW_UNRELATED_HISTORIES", false), + Logger: logger.Sugar(), + MetricsScope: scope, + }) + if err != nil { + return nil, fmt.Errorf("failed to build git merger: %w", err) + } + logger.Info("git merger configured", + zap.String("checkout", checkoutPath), + zap.String("target", target), + zap.String("default_strategy", defaultStrategy.String()), + ) + return &gitMergerFactory{merger: m}, nil +} + +// gitMergerFactory returns a single git-backed merger for every queue. The +// merger owns one checkout and serializes its own operations, so one instance +// is shared across queues. A deployment that lands multiple targets wires a +// factory with a per-queue map instead. +type gitMergerFactory struct { + merger merger.Merger +} + +func (f *gitMergerFactory) For(_ merger.Config) (merger.Merger, error) { + return f.merger, nil } type noopMergerFactory struct{} @@ -307,6 +369,58 @@ func (f *noopMergerFactory) For(_ merger.Config) (merger.Merger, error) { return noop.New(), 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. +func parseStrategy(name string) (mergestrategypb.Strategy, error) { + switch strings.ToUpper(strings.TrimSpace(name)) { + case "", "REBASE": + return mergestrategypb.Strategy_REBASE, nil + case "SQUASH_REBASE": + return mergestrategypb.Strategy_SQUASH_REBASE, nil + case "MERGE": + return mergestrategypb.Strategy_MERGE, nil + case "PROMOTE": + return mergestrategypb.Strategy_PROMOTE, nil + default: + return mergestrategypb.Strategy_DEFAULT, fmt.Errorf("invalid MERGE_DEFAULT_STRATEGY %q", name) + } +} + +// splitRefspecs parses the comma-separated MERGE_FETCH_REFSPECS value. Empty +// entries are dropped so a trailing comma is harmless. +func splitRefspecs(v string) []string { + var out []string + for _, part := range strings.Split(v, ",") { + if trimmed := strings.TrimSpace(part); trimmed != "" { + out = append(out, trimmed) + } + } + return out +} + +// envBool reads a boolean environment value, returning fallback when unset or +// unparseable. +func envBool(key string, fallback bool) bool { + v := strings.TrimSpace(os.Getenv(key)) + if v == "" { + return fallback + } + parsed, err := strconv.ParseBool(v) + if err != nil { + return fallback + } + return parsed +} + +// envOr returns the environment value for key, or fallback when unset. +func envOr(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} + // newTopicRegistry builds the TopicRegistry for Runway's merge queues. Inbound // topics (merge-conflict-check, merge) have subscriptions; outbound signal topics // are publish-only.