feat(runway): wire the git merger into the server

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.

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-<unix_ts>`       |
+| `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
 

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",

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.