From e70bb3ae451f6636b8759bf7cf23c3aa2df05a93 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Fri, 4 Sep 2026 22:20:32 +0000 Subject: [PATCH] fix(runway): ISS-004 retry transient Git failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Intent: - Prevent temporary Git remote and checkout failures from being dead-lettered on their first delivery. - Keep every other Git failure fast-failing, so a deterministic error is not replayed through the retry budget. Changes: - Add structured Git command errors and a Git classifier that opts a failure into retryability only on a known diagnostic/operation pair. - Surface a cancelled context at the Git execution boundary, so cancellation reaches the generic classifier instead of dying as an opaque "signal: killed". - Derive the Git subcommand through one guarded helper and wire the classifier into the Runway primary consumer. Reproduction: - A merge delivery runs `git fetch origin` or `git push origin ...` while the remote temporarily resets the connection, producing a wrapped `*exec.ExitError`. - Previously Runway registered only generic and MySQL classifiers, so the error stayed non-retryable and the consumer rejected it to the DLQ after one attempt. - With this change the structured Git error is classified as a retryable dependency failure, so the consumer nacks it for redelivery. Retryability is an allowlist. Git has no typed status to read, so the classifier pairs the subcommand with the diagnostic: a transport fragment counts only against a command that talks to the remote, and a lock fragment counts against any command that writes to the checkout. Only a recognised pair is retryable. Every other Git failure, including a diagnostic the package has never seen, is a permanent infrastructure failure attributed to the remote or to this service, so a deleted target branch, an empty squash commit or a rejected push still dead-letters on the first delivery rather than re-running the fetch, reset and cherry-picks behind it on every attempt. `os/exec` reports a context-killed child as a bare `*exec.ExitError` reading "signal: killed", with neither `context.Canceled` nor `context.DeadlineExceeded` anywhere in the chain. `gitexec.CommandFailure` reads `ctx.Err()` and surfaces it, which is what lets the generic classifier recognise a cancelled merge rather than seeing an unexplained Git failure. --- Generated by the 🪄 [pr-create](https://sg.uberinternal.com/code.uber.internal/uber-code/devexp-agent-marketplace/-/blob/claude-code/plugins/dev/uber-dev/skills/pr-create/SKILL.md) skill in devexp-agent-marketplace --- platform/errs/README.md | 8 +- platform/errs/git/BUILD.bazel | 24 ++ platform/errs/git/git.go | 150 ++++++++++++ platform/errs/git/git_test.go | 213 ++++++++++++++++++ platform/git/exec/BUILD.bazel | 5 +- platform/git/exec/command_error.go | 89 ++++++++ platform/git/exec/gitexec.go | 2 +- platform/git/exec/gitexec_test.go | 133 +++++++++++ runway/extension/merger/git/BUILD.bazel | 1 + runway/extension/merger/git/git_merger.go | 34 ++- .../extension/merger/git/git_merger_test.go | 9 +- service/runway/server/BUILD.bazel | 11 + service/runway/server/main.go | 17 +- service/runway/server/main_test.go | 147 ++++++++++++ 14 files changed, 823 insertions(+), 20 deletions(-) create mode 100644 platform/errs/git/BUILD.bazel create mode 100644 platform/errs/git/git.go create mode 100644 platform/errs/git/git_test.go create mode 100644 platform/git/exec/command_error.go create mode 100644 service/runway/server/main_test.go diff --git a/platform/errs/README.md b/platform/errs/README.md index b86c3b104..063c212fd 100644 --- a/platform/errs/README.md +++ b/platform/errs/README.md @@ -89,7 +89,7 @@ One operational consequence worth knowing before relying on any of this: **retry ## Adding a Backend-Specific Classifier -Backend classifiers live alongside the extension they classify, under `platform/errs//`. The canonical examples are `platform/errs/mysql` (MySQL driver errors), `platform/errs/http` (rejected status codes and transport failures from clients built on `platform/http`), `platform/errs/yarpc` (YARPC status codes), and `platform/errs/generic` (transport-agnostic concerns such as `context.Canceled`). +Backend classifiers live alongside the extension they classify, under `platform/errs//`. The canonical examples are `platform/errs/mysql` (MySQL driver errors), `platform/errs/http` (rejected status codes and transport failures from clients built on `platform/http`), `platform/errs/git` (structured Git process failures), `platform/errs/yarpc` (YARPC status codes), and `platform/errs/generic` (transport-agnostic concerns such as `context.Canceled`). A classifier: @@ -122,6 +122,7 @@ Servers wire each classifier into the consumer's `ErrorProcessor`. Order matters 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" yarpcerrs "github.com/uber/submitqueue/platform/errs/yarpc" @@ -130,6 +131,7 @@ import ( c := consumer.New(logger, scope, registry, errs.NewClassifierProcessor( genericerrs.Classifier, + giterrs.Classifier, httperrs.Classifier, yarpcerrs.Classifier, mysqlerrs.Classifier, @@ -143,7 +145,9 @@ Classifiers are not installed globally. A host that wants YARPC statuses classif The YARPC classifier reads the typed status code rather than matching its rendered message. Cancellation is retryable caller-side infrastructure; transient or ambiguous server codes (`Unknown`, `DeadlineExceeded`, `ResourceExhausted`, `Aborted`, `Internal`, and `Unavailable`) are retryable dependency failures; request verdicts and permanent server failures are non-retryable dependency failures. A deadline may expire after a mutating RPC succeeded, so this classification relies on the repository-wide requirement that queue-driven operations are idempotent. -Tests follow the same shape: assert per-node behaviour against `Classifier.Classify(node)` directly, and assert end-to-end behaviour by running `errs.NewClassifierProcessor(Classifier).Process(err)` and checking the helpers (`IsRetryable`, `IsUserError`, …) on the result. See `platform/errs/mysql/mysql_test.go`, `platform/errs/yarpc/yarpc_test.go`, and `platform/errs/generic/generic_test.go`. +The Git classifier reads `gitexec.CommandError`, which preserves the Git subcommand and the underlying `os/exec` error through contextual wrapping. Git has no typed status to read — a connection reset and a deleted branch both leave `fetch` at a non-zero exit — so the classifier pairs the subcommand with the diagnostic git printed: a transport fragment counts only against a command that talks to the remote, and a lock fragment counts against any command that writes to the checkout. Only a recognised pair is retryable; every other Git failure, including a diagnostic the package has never seen, is a permanent infrastructure failure attributed to the remote or to this service. The direction is deliberate — an unlisted transient failure costs one lost retry, while a permanent failure defaulting to retryable would replay a deterministic error through the whole retry budget before dead-lettering anyway — and it is what makes the fragment lists safe to extend as Git's wording drifts between versions. Cancellation is not the Git classifier's to report: `os/exec` kills a context-cancelled child and reports only `signal: killed`, so `gitexec.CommandFailure` puts `context.Canceled` back in the chain and the generic classifier recognises it there. + +Tests follow the same shape: assert per-node behaviour against `Classifier.Classify(node)` directly, and assert end-to-end behaviour by running `errs.NewClassifierProcessor(Classifier).Process(err)` and checking the helpers (`IsRetryable`, `IsUserError`, …) on the result. See `platform/errs/mysql/mysql_test.go`, `platform/errs/git/git_test.go`, `platform/errs/yarpc/yarpc_test.go`, and `platform/errs/generic/generic_test.go`. ## Overriding Classification from a Controller diff --git a/platform/errs/git/BUILD.bazel b/platform/errs/git/BUILD.bazel new file mode 100644 index 000000000..c8c9421d3 --- /dev/null +++ b/platform/errs/git/BUILD.bazel @@ -0,0 +1,24 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["git.go"], + importpath = "github.com/uber/submitqueue/platform/errs/git", + visibility = ["//visibility:public"], + deps = [ + "//platform/errs:go_default_library", + "//platform/git/exec:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["git_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/errs:go_default_library", + "//platform/errs/generic:go_default_library", + "//platform/git/exec:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + ], +) diff --git a/platform/errs/git/git.go b/platform/errs/git/git.go new file mode 100644 index 000000000..d691ae96f --- /dev/null +++ b/platform/errs/git/git.go @@ -0,0 +1,150 @@ +// 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 git provides an errs.Classifier for failures from Git processes. +// +// Git has no typed status to read: it reports almost everything as a non-zero +// exit and a line of prose, so a connection reset and a deleted branch both +// leave `git fetch` looking identical to a caller that only checks the code. +// The subcommand that was run and the diagnostic git printed are therefore the +// only signals available, and the classification pairs them: a fragment is +// evidence of a transient failure only for the operations it can actually +// arise from, so a transport fault counts against a command that talks to the +// remote and lock contention counts against any command that writes to the +// checkout. +// +// Only a recognised pair is retryable. Everything else is a permanent +// infrastructure failure, including a diagnostic this package has never seen. +// The direction is deliberate: an unlisted transient failure costs one lost +// retry, while a permanent failure that defaulted to retryable would replay a +// deterministic error — a deleted target branch, an empty squash commit, a +// rejected push — through the whole retry budget, re-running the fetch, reset +// and cherry-picks behind it each time, before dead-lettering anyway. +// +// Git's wording drifts between versions, so the fragment lists are expected to +// grow. Adding one is cheap and safe; the cost of a missing fragment is bounded +// at a single lost retry, which is what makes the allowlist maintainable. +// +// Cancellation is deliberately absent. A git process killed because its +// context ended dies with "signal: killed" and no trace of the cancellation, +// so it is gitexec.CommandFailure — not this classifier — that puts +// context.Canceled back in the chain, leaving the generic classifier to +// recognise it as it does for every other cancelled operation. +package git + +import ( + "strings" + + "github.com/uber/submitqueue/platform/errs" + gitexec "github.com/uber/submitqueue/platform/git/exec" +) + +// Classifier recognises Git process failures, reporting a known transient +// diagnostic on an operation it can arise from as retryable and every other +// Git failure as permanent. See the package doc for why the default runs that +// way. +// +// The classifier is stateless; this package-level singleton is the canonical +// handle. Pass it as one of the variadic classifiers to +// errs.NewClassifierProcessor; the resulting processor is what gets handed to +// consumer.New. +var Classifier errs.Classifier = classifier{} + +type classifier struct{} + +// remoteOperations are the Git subcommands that exchange data with the +// configured remote. They attribute their failures to that remote, and they +// are the only operations a transport fragment can legitimately describe. +var remoteOperations = map[string]bool{ + "clone": true, + "fetch": true, + "ls-remote": true, + "pull": true, + "push": true, +} + +// transientTransportFragments are diagnostics that mean the exchange with the +// remote did not complete, weighed only for a remoteOperations subcommand. +// A rejected push or a failed authentication is the remote answering, not +// failing to answer, and stays permanent. +var transientTransportFragments = []string{ + "502 bad gateway", + "503 service unavailable", + "504 gateway timeout", + "broken pipe", + "connection refused", + "connection reset by peer", + "connection timed out", + "could not resolve host", + "early eof", + "network is unreachable", + "no route to host", + "operation timed out", + "remote end hung up unexpectedly", + "rpc failed", + "ssh_exchange_identification", + "temporary failure in name resolution", + "transfer closed with outstanding read data remaining", + "unexpected disconnect while reading sideband packet", +} + +// transientCheckoutFragments are diagnostics that mean another process held +// the checkout, weighed for every subcommand: a remote operation writes refs +// and the index too, so it can lose the same race a local one can. +var transientCheckoutFragments = []string{ + ".lock': file exists", + "cannot lock ref", + "index.lock", + "resource temporarily unavailable", +} + +// Classify inspects a single node. Per the errs.Classifier contract, this must +// not call errors.Is / errors.As — the classifier-processor owns the chain +// walk. +func (classifier) Classify(err error) errs.Verdict { + commandErr, ok := err.(*gitexec.CommandError) + if !ok { + // The only Unknown this classifier returns, and it means "not my + // node" rather than "no opinion on this failure". Returning a verdict + // here would claim every error the walk passes — a MySQL driver error + // among them — before its own classifier were asked. + return errs.Unknown + } + + diagnostic := strings.ToLower(commandErr.Diagnostic()) + remote := remoteOperations[commandErr.Operation()] + + transient := containsAny(diagnostic, transientCheckoutFragments) || + (remote && containsAny(diagnostic, transientTransportFragments)) + + switch { + case transient && remote: + return errs.InfraDependencyRetryable + case transient: + return errs.InfraRetryable + case remote: + return errs.InfraDependency + default: + return errs.Infra + } +} + +func containsAny(diagnostic string, fragments []string) bool { + for _, fragment := range fragments { + if strings.Contains(diagnostic, fragment) { + return true + } + } + return false +} diff --git a/platform/errs/git/git_test.go b/platform/errs/git/git_test.go new file mode 100644 index 000000000..5076499b2 --- /dev/null +++ b/platform/errs/git/git_test.go @@ -0,0 +1,213 @@ +// 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 git + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + genericerrs "github.com/uber/submitqueue/platform/errs/generic" + + "github.com/uber/submitqueue/platform/errs" + gitexec "github.com/uber/submitqueue/platform/git/exec" +) + +func gitError(operation, diagnostic string) error { + return gitexec.NewCommandError(operation, diagnostic, errors.New("exit status 128")) +} + +func TestClassifier(t *testing.T) { + tests := []struct { + name string + err error + want errs.Verdict + }{ + { + name: "transport fault on fetch is a retryable dependency failure", + err: gitError("fetch", "fatal: unable to access 'https://host/r.git/': Connection reset by peer"), + want: errs.InfraDependencyRetryable, + }, + { + name: "hung-up remote on push is a retryable dependency failure", + err: gitError("push", "fatal: the remote end hung up unexpectedly"), + want: errs.InfraDependencyRetryable, + }, + { + name: "unresolvable host on ls-remote is a retryable dependency failure", + err: gitError("ls-remote", "fatal: Could not resolve host: github.example.com"), + want: errs.InfraDependencyRetryable, + }, + { + name: "checkout contention on a local commit is a retryable local failure", + err: gitError("commit", "fatal: Unable to create '/checkout/.git/index.lock': File exists."), + want: errs.InfraRetryable, + }, + { + name: "checkout contention during fetch is attributed to the remote it ran against", + err: gitError("fetch", "error: cannot lock ref 'refs/remotes/origin/main'"), + want: errs.InfraDependencyRetryable, + }, + { + name: "transport fragment on a local operation is not evidence of a transient failure", + err: gitError("merge", "error: could not resolve host mentioned in a commit message"), + want: errs.Infra, + }, + { + name: "unknown revision is a permanent local failure", + err: gitError("rev-parse", "fatal: ambiguous argument 'origin/main': unknown revision or path not in the working tree."), + want: errs.Infra, + }, + { + name: "empty squash commit is a permanent local failure", + err: gitError("commit", "exit status 1"), + want: errs.Infra, + }, + { + name: "failure with no diagnostic at all is a permanent local failure", + err: gitError("cat-file", ""), + want: errs.Infra, + }, + { + name: "path outside the repository is a permanent local failure", + err: gitError("clean", "fatal: '/etc': '/etc' is outside repository at '/checkout'"), + want: errs.Infra, + }, + { + name: "non-fast-forward push is a permanent dependency failure", + err: gitError("push", "! [rejected] main -> main (fetch first)"), + want: errs.InfraDependency, + }, + { + name: "authentication failure is a permanent dependency failure", + err: gitError("fetch", "fatal: Authentication failed for 'https://host/r.git/'"), + want: errs.InfraDependency, + }, + { + // Git prints this trailer under permanent failures too — a missing + // remote, absent access rights — so it is not evidence of a + // transient one. + name: "generic remote trailer is a permanent dependency failure", + err: gitError("fetch", "fatal: 'origin' does not appear to be a git repository\nfatal: Could not read from remote repository."), + want: errs.InfraDependency, + }, + { + name: "unknown subcommand is a permanent local failure", + err: gitError("bisect", "fatal: something went wrong"), + want: errs.Infra, + }, + { + name: "non-Git error is not this classifier's node", + err: errors.New("anything"), + want: errs.Unknown, + }, + { + name: "nil is not this classifier's node", + err: nil, + want: errs.Unknown, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, Classifier.Classify(tt.err)) + }) + } +} + +func TestClassifier_FragmentsMatchRegardlessOfCase(t *testing.T) { + for _, fragment := range transientTransportFragments { + t.Run(fragment, func(t *testing.T) { + shouted := "fatal: " + strings.ToUpper(fragment) + assert.Equal(t, errs.InfraDependencyRetryable, Classifier.Classify(gitError("fetch", shouted))) + }) + } + for _, fragment := range transientCheckoutFragments { + t.Run(fragment, func(t *testing.T) { + shouted := "fatal: " + strings.ToUpper(fragment) + assert.Equal(t, errs.InfraRetryable, Classifier.Classify(gitError("commit", shouted))) + }) + } +} + +func TestClassifier_AppliedViaProcessor(t *testing.T) { + tests := []struct { + name string + err error + wantRetryable bool + wantDependency bool + wantUnwrapped bool + }{ + { + name: "wrapped transport fault is a retryable dependency", + err: fmt.Errorf("reset checkout: %w", gitError("fetch", "fatal: Connection reset by peer")), + wantRetryable: true, + wantDependency: true, + }, + { + name: "wrapped checkout contention is retryable locally", + err: fmt.Errorf("apply change: %w", gitError("cherry-pick", "fatal: Unable to create '.git/index.lock': File exists.")), + wantRetryable: true, + }, + { + name: "wrapped unknown revision is non-retryable", + err: fmt.Errorf("resolve tip: %w", gitError("rev-parse", "fatal: ambiguous argument 'origin/main'")), + }, + { + name: "wrapped rejected push is a non-retryable dependency", + err: fmt.Errorf("promote: %w", gitError("push", "! [rejected] main -> main (fetch first)")), + wantDependency: true, + }, + { + name: "explicit user wrap wins over the classifier", + err: errs.NewUserError(gitError("fetch", "fatal: Connection reset by peer")), + wantUnwrapped: true, + }, + { + name: "unrecognised error is returned unchanged", + err: errors.New("anything"), + wantUnwrapped: true, + }, + } + + processor := errs.NewClassifierProcessor(Classifier) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := processor.Process(tt.err) + assert.Equal(t, tt.wantRetryable, errs.IsRetryable(got)) + assert.Equal(t, tt.wantDependency, errs.IsDependencyError(got)) + if tt.wantUnwrapped { + assert.Same(t, tt.err, got) + } + }) + } +} + +// A git process killed because its context ended reports only "signal: killed", +// so gitexec.CommandFailure is what puts the cancellation back in the chain. +// This classifier must stay out of the way for the generic one to see it. +func TestClassifier_LeavesCancellationToTheGenericClassifier(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := gitexec.CommandFailure(ctx, []string{"fetch", "origin"}, "signal: killed", errors.New("signal: killed")) + assert.Equal(t, errs.Unknown, Classifier.Classify(err), "a cancelled command is not a CommandError node") + + processed := errs.NewClassifierProcessor(genericerrs.Classifier, Classifier).Process(err) + assert.True(t, errs.IsRetryable(processed), "the generic classifier recognises the cancellation") +} diff --git a/platform/git/exec/BUILD.bazel b/platform/git/exec/BUILD.bazel index f5c575674..a834acc9b 100644 --- a/platform/git/exec/BUILD.bazel +++ b/platform/git/exec/BUILD.bazel @@ -2,7 +2,10 @@ load("@rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", - srcs = ["gitexec.go"], + srcs = [ + "command_error.go", + "gitexec.go", + ], importpath = "github.com/uber/submitqueue/platform/git/exec", visibility = ["//visibility:public"], ) diff --git a/platform/git/exec/command_error.go b/platform/git/exec/command_error.go new file mode 100644 index 000000000..326c086ed --- /dev/null +++ b/platform/git/exec/command_error.go @@ -0,0 +1,89 @@ +// 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 gitexec + +import ( + "context" + "fmt" +) + +// CommandError preserves the failed Git operation and its process error for +// backend-specific classification after callers add contextual wrapping. +type CommandError struct { + operation string + message string + cause error +} + +// NewCommandError records a failed Git operation without assigning retry +// policy. Callers supply the rendered diagnostic they want Error to expose. +func NewCommandError(operation, message string, cause error) *CommandError { + return &CommandError{ + operation: operation, + message: message, + cause: cause, + } +} + +// Error returns the command diagnostic supplied by the execution boundary. +func (e *CommandError) Error() string { + if e.message != "" { + return e.message + } + if e.cause != nil { + return e.cause.Error() + } + return "git command failed" +} + +// Unwrap returns the process error reported by os/exec. +func (e *CommandError) Unwrap() error { + return e.cause +} + +// Operation returns the Git subcommand, such as fetch or cherry-pick. +func (e *CommandError) Operation() string { + return e.operation +} + +// Diagnostic returns Git's rendered failure output. +func (e *CommandError) Diagnostic() string { + return e.message +} + +// Operation returns the Git subcommand in args, or "" when args is empty. +func Operation(args []string) string { + if len(args) == 0 { + return "" + } + return args[0] +} + +// CommandFailure builds the error for a failed git invocation. +// +// A context that has ended takes precedence over whatever git reported. When +// os/exec kills a child because its context was cancelled, Wait reports only +// the death — a bare *exec.ExitError reading "signal: killed" — and neither +// context.Canceled nor context.DeadlineExceeded appears anywhere in the +// chain, so nothing downstream can tell an interrupted command apart from one +// that genuinely failed. Reading ctx.Err() here is the only place that +// distinction still exists; surfacing it puts cancellation back in the chain +// where the generic classifier can recognise it. +func CommandFailure(ctx context.Context, args []string, message string, cause error) error { + if ctxErr := ctx.Err(); ctxErr != nil { + return fmt.Errorf("git %s: %s: %w", Operation(args), message, ctxErr) + } + return NewCommandError(Operation(args), message, cause) +} diff --git a/platform/git/exec/gitexec.go b/platform/git/exec/gitexec.go index 0f7a6d8ad..95c515fe3 100644 --- a/platform/git/exec/gitexec.go +++ b/platform/git/exec/gitexec.go @@ -161,7 +161,7 @@ func Output(ctx context.Context, git, dir string, args ...string) (string, error if message == "" { message = err.Error() } - return "", fmt.Errorf("git %s: %s", strings.Join(args, " "), message) + return "", fmt.Errorf("git %s: %w", strings.Join(args, " "), CommandFailure(ctx, args, message, err)) } return strings.TrimSpace(string(out)), nil } diff --git a/platform/git/exec/gitexec_test.go b/platform/git/exec/gitexec_test.go index 1658b06d9..610b0dd98 100644 --- a/platform/git/exec/gitexec_test.go +++ b/platform/git/exec/gitexec_test.go @@ -15,7 +15,10 @@ package gitexec import ( + "context" + "errors" "os" + "os/exec" "strings" "testing" @@ -95,3 +98,133 @@ func TestEnv_PassthroughDeduplicatesWithTransport(t *testing.T) { func TestEnv_HomeNotInSharedTransportList(t *testing.T) { assert.NotContains(t, transportEnvNames, "HOME") } + +func TestCommandError(t *testing.T) { + cause := errors.New("exit status 128") + tests := []struct { + name string + err *CommandError + wantMessage string + wantCause error + }{ + { + name: "supplied diagnostic is rendered", + err: NewCommandError("fetch", "connection reset", cause), + wantMessage: "connection reset", + wantCause: cause, + }, + { + name: "cause is rendered when diagnostic is empty", + err: NewCommandError("reset", "", cause), + wantMessage: cause.Error(), + wantCause: cause, + }, + { + name: "fallback is rendered without diagnostic or cause", + err: NewCommandError("unknown", "", nil), + wantMessage: "git command failed", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.wantMessage, tt.err.Error()) + assert.Equal(t, tt.err.message, tt.err.Diagnostic()) + assert.Equal(t, tt.err.operation, tt.err.Operation()) + if tt.wantCause == nil { + assert.NoError(t, tt.err.Unwrap()) + } else { + assert.ErrorIs(t, tt.err, tt.wantCause) + } + }) + } +} + +func TestOutput_PreservesCommandFailure(t *testing.T) { + tests := []struct { + name string + executable string + args []string + wantOperation string + }{ + { + name: "non-zero process exit retains command provenance and cause", + executable: os.Args[0], + args: []string{"-test.run=["}, + wantOperation: "-test.run=[", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := Output(context.Background(), tt.executable, "", tt.args...) + require.Error(t, err) + + var commandErr *CommandError + require.ErrorAs(t, err, &commandErr) + assert.Equal(t, tt.wantOperation, commandErr.Operation()) + + var exitErr *exec.ExitError + assert.ErrorAs(t, err, &exitErr) + }) + } +} + +func TestOperation(t *testing.T) { + tests := []struct { + name string + args []string + want string + }{ + { + name: "first argument is the operation", + args: []string{"fetch", "origin"}, + want: "fetch", + }, + { + name: "empty arguments have no operation", + want: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, Operation(tt.args)) + }) + } +} + +func TestCommandFailure(t *testing.T) { + cause := errors.New("signal: killed") + + t.Run("a live context yields a classifiable command error", func(t *testing.T) { + err := CommandFailure(context.Background(), []string{"fetch", "origin"}, "connection reset", cause) + + var commandErr *CommandError + require.ErrorAs(t, err, &commandErr) + assert.Equal(t, "fetch", commandErr.Operation()) + assert.Equal(t, "connection reset", commandErr.Diagnostic()) + }) + + // os/exec reports a context-killed child as a bare *exec.ExitError reading + // "signal: killed", with the cancellation nowhere in the chain. Surfacing + // ctx.Err() is what lets the generic classifier recognise it. + t.Run("an ended context surfaces the cancellation instead", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := CommandFailure(ctx, []string{"fetch", "origin"}, "signal: killed", cause) + + assert.ErrorIs(t, err, context.Canceled) + var commandErr *CommandError + assert.NotErrorAs(t, err, &commandErr, "the git node must not shadow the cancellation") + }) + + t.Run("empty arguments do not panic", func(t *testing.T) { + err := CommandFailure(context.Background(), nil, "boom", cause) + + var commandErr *CommandError + require.ErrorAs(t, err, &commandErr) + assert.Empty(t, commandErr.Operation()) + }) +} diff --git a/runway/extension/merger/git/BUILD.bazel b/runway/extension/merger/git/BUILD.bazel index 9dbc3feb6..fcaf7ce67 100644 --- a/runway/extension/merger/git/BUILD.bazel +++ b/runway/extension/merger/git/BUILD.bazel @@ -49,6 +49,7 @@ go_test( "//api/base/mergestrategy/protopb:go_default_library", "//api/runway/messagequeue:go_default_library", "//api/runway/messagequeue/protopb:go_default_library", + "//platform/git/exec:go_default_library", "//platform/git/exectest:go_default_library", "//runway/extension/merger:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", diff --git a/runway/extension/merger/git/git_merger.go b/runway/extension/merger/git/git_merger.go index f0d6575f6..8fe7d87a6 100644 --- a/runway/extension/merger/git/git_merger.go +++ b/runway/extension/merger/git/git_merger.go @@ -694,7 +694,7 @@ func (m *gitMerger) applyMerge(ctx context.Context, rs resolvedStep) (applied, e // does not establish that anything collided. conflicted := m.hasUnmergedPaths(ctx) _, _ = m.run(ctx, nil, "merge", "--abort") - return applied{}, m.classifyMergeFailure(ref, o, conflicted) + return applied{}, m.classifyMergeFailure(ref, o, conflicted, err) } mergeSHA, err := m.headSHA(ctx) if err != nil { @@ -785,7 +785,7 @@ func (m *gitMerger) promote(ctx context.Context, req *runwaymq.MergeRequest, rs // and fixed by configuration rather than by rebasing, and any other way git // can exit non-zero — a missing object, an unreadable repository, a killed // process — which is infrastructure and should be retried, not made terminal. -func (m *gitMerger) classifyMergeFailure(ref changeRef, out []byte, conflicted bool) error { +func (m *gitMerger) classifyMergeFailure(ref changeRef, out []byte, conflicted bool, cause error) error { detail := strings.TrimSpace(string(out)) if strings.Contains(detail, "refusing to merge unrelated histories") { coremetrics.NamedCounter(m.metricsScope, "merge", "unrelated_histories", 1) @@ -794,7 +794,7 @@ func (m *gitMerger) classifyMergeFailure(ref changeRef, out []byte, conflicted b } if !conflicted { coremetrics.NamedCounter(m.metricsScope, "merge", "merge_errors", 1) - return fmt.Errorf("git merge %s: %s", ref.SHA, detail) + return fmt.Errorf("git merge %s: %w", ref.SHA, cause) } coremetrics.NamedCounter(m.metricsScope, "merge", "merge_conflicts", 1) return fmt.Errorf("%w: git merge %s: %s", merger.ErrConflict, ref.SHA, detail) @@ -892,7 +892,7 @@ func (m *gitMerger) cherryPickRange(ctx context.Context, base, head string) erro detail := strings.TrimSpace(string(out)) if !conflicted { coremetrics.NamedCounter(m.metricsScope, "merge", "cherry_pick_errors", 1) - return fmt.Errorf("git cherry-pick %s..%s: %w: %s", base, head, err, detail) + return fmt.Errorf("git cherry-pick %s..%s: %w", base, head, err) } coremetrics.NamedCounter(m.metricsScope, "merge", "cherry_pick_conflicts", 1) return fmt.Errorf("%w: git cherry-pick %s..%s: %s", merger.ErrConflict, base, head, detail) @@ -989,7 +989,8 @@ func (m *gitMerger) refetchTipSHA(ctx context.Context) (string, error) { // descendant. `git merge-base --is-ancestor` exits 0 for true, 1 for false; // any other exit is a real error. func (m *gitMerger) isAncestor(ctx context.Context, ancestor, descendant string) (bool, error) { - cmd := m.command(ctx, "merge-base", "--is-ancestor", ancestor, descendant) + args := []string{"merge-base", "--is-ancestor", ancestor, descendant} + cmd := m.command(ctx, args...) var stderr bytes.Buffer cmd.Stderr = &stderr err := cmd.Run() @@ -1000,7 +1001,12 @@ func (m *gitMerger) isAncestor(ctx context.Context, ancestor, descendant string) if errors.As(err, &exitErr) && exitErr.ExitCode() == 1 { return false, nil } - return false, fmt.Errorf("git merge-base --is-ancestor %s %s: %w: %s", ancestor, descendant, err, strings.TrimSpace(stderr.String())) + message := err.Error() + if detail := strings.TrimSpace(stderr.String()); detail != "" { + message += ": " + detail + } + return false, fmt.Errorf("git merge-base --is-ancestor %s %s: %w", + ancestor, descendant, gitexec.CommandFailure(ctx, args, message, err)) } // commitTreeSHA returns the tree SHA recorded in the commit object at ref. @@ -1038,7 +1044,11 @@ func (m *gitMerger) runAs(ctx context.Context, author authorIdent, stdin []byte, cmd.Stdout = &stdout cmd.Stderr = &stderr if err := cmd.Run(); err != nil { - return nil, fmt.Errorf("%w: %s", err, strings.TrimSpace(stderr.String())) + message := err.Error() + if detail := strings.TrimSpace(stderr.String()); detail != "" { + message += ": " + detail + } + return nil, gitexec.CommandFailure(ctx, args, message, err) } return stdout.Bytes(), nil } @@ -1056,7 +1066,15 @@ func (m *gitMerger) runCombinedAs(ctx context.Context, author authorIdent, stdin if stdin != nil { cmd.Stdin = bytes.NewReader(stdin) } - return cmd.CombinedOutput() + out, err := cmd.CombinedOutput() + if err != nil { + message := err.Error() + if detail := strings.TrimSpace(string(out)); detail != "" { + message += ": " + detail + } + return out, gitexec.CommandFailure(ctx, args, message, err) + } + return out, nil } // command builds a git command with the committer identity injected via -c diff --git a/runway/extension/merger/git/git_merger_test.go b/runway/extension/merger/git/git_merger_test.go index b1e6e2224..05a84edf7 100644 --- a/runway/extension/merger/git/git_merger_test.go +++ b/runway/extension/merger/git/git_merger_test.go @@ -35,6 +35,7 @@ import ( 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" + gitexec "github.com/uber/submitqueue/platform/git/exec" gitexectest "github.com/uber/submitqueue/platform/git/exectest" "github.com/uber/submitqueue/runway/extension/merger" ) @@ -210,6 +211,8 @@ func TestCherryPickRange_NonConflictFailureIsRetryable(t *testing.T) { require.Error(t, err) assert.False(t, errors.Is(err, merger.ErrConflict), "a non-conflict failure must stay retryable") assert.False(t, errors.Is(err, merger.ErrInvalidRequest)) + var commandErr *gitexec.CommandError + assert.ErrorAs(t, err, &commandErr) } func TestCherryPickRange_RealConflictIsErrConflict(t *testing.T) { @@ -623,10 +626,14 @@ func TestClassifyMergeFailure(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - err := m.classifyMergeFailure(ref, []byte(tt.out), tt.conflicted) + cause := errors.New("git exited") + err := m.classifyMergeFailure(ref, []byte(tt.out), tt.conflicted, cause) require.Error(t, err) assert.Equal(t, tt.wantConflict, errors.Is(err, merger.ErrConflict)) assert.Equal(t, tt.wantInvalid, errors.Is(err, merger.ErrInvalidRequest)) + if !tt.wantConflict && !tt.wantInvalid { + assert.ErrorIs(t, err, cause) + } }) } } diff --git a/service/runway/server/BUILD.bazel b/service/runway/server/BUILD.bazel index 113cd6e23..85fa29808 100644 --- a/service/runway/server/BUILD.bazel +++ b/service/runway/server/BUILD.bazel @@ -22,6 +22,7 @@ go_library( "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/errs/generic:go_default_library", + "//platform/errs/git:go_default_library", "//platform/errs/mysql:go_default_library", "//platform/extension/consumergate:go_default_library", "//platform/extension/consumergate/file:go_default_library", @@ -79,6 +80,7 @@ go_test( srcs = [ "checkout_test.go", "config_test.go", + "main_test.go", ], # Checkout provisioning runs real git, so the test uses the same pinned # runtime the merger does rather than whatever git the host happens to have. @@ -97,10 +99,19 @@ go_test( }, deps = [ "//api/base/mergestrategy/protopb:go_default_library", + "//platform/base/failure:go_default_library", + "//platform/base/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/extension/consumergate/noop:go_default_library", + "//platform/extension/messagequeue:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", + "//platform/git/exec:go_default_library", "//platform/git/exectest:go_default_library", "//runway/extension/merger/git: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_mock//gomock:go_default_library", "@org_uber_go_zap//zaptest:go_default_library", ], ) diff --git a/service/runway/server/main.go b/service/runway/server/main.go index dc9985001..39d1ac95c 100644 --- a/service/runway/server/main.go +++ b/service/runway/server/main.go @@ -37,6 +37,7 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" genericerrs "github.com/uber/submitqueue/platform/errs/generic" + giterrs "github.com/uber/submitqueue/platform/errs/git" mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql" "github.com/uber/submitqueue/platform/extension/consumergate" consumergatefile "github.com/uber/submitqueue/platform/extension/consumergate/file" @@ -163,13 +164,7 @@ func run() error { // group name just like a primary stage. gate := newConsumerGate(logger) - primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, - errs.NewClassifierProcessor( - genericerrs.Classifier, - mysqlerrs.Classifier, - ), - gate, - ) + primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, newPrimaryErrorProcessor(), gate) mergerFactory, err := newMergerFactory(ctx, logger, scope.SubScope("merger")) if err != nil { @@ -305,6 +300,14 @@ func run() error { return err } +func newPrimaryErrorProcessor() errs.ErrorProcessor { + return errs.NewClassifierProcessor( + genericerrs.Classifier, + giterrs.Classifier, + mysqlerrs.Classifier, + ) +} + // newMergerFactory builds the mergers for the server. // // MERGER pins every queue to one implementation explicitly, which is how a test diff --git a/service/runway/server/main_test.go b/service/runway/server/main_test.go new file mode 100644 index 000000000..ac297ba5e --- /dev/null +++ b/service/runway/server/main_test.go @@ -0,0 +1,147 @@ +// 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 main + +import ( + "context" + "errors" + "os" + "os/exec" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/base/failure" + entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" + 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" + "go.uber.org/mock/gomock" + "go.uber.org/zap/zaptest" +) + +const ( + testGitTopicKey consumer.TopicKey = "git-test" + testGitGroup = "git-test-group" +) + +type errorController struct { + err error +} + +func (c errorController) Process(context.Context, consumer.Delivery) error { + return c.err +} + +func (errorController) Name() string { + return "git-test" +} + +func (errorController) TopicKey() consumer.TopicKey { + return testGitTopicKey +} + +func (errorController) ConsumerGroup() string { + return testGitGroup +} + +func gitExitError(t *testing.T) error { + t.Helper() + err := exec.Command(os.Args[0], "-test.run=[").Run() + require.Error(t, err) + var exitErr *exec.ExitError + require.ErrorAs(t, err, &exitErr) + return exitErr +} + +func TestPrimaryConsumer_GitFailureDisposition(t *testing.T) { + exitErr := gitExitError(t) + tests := []struct { + name string + controller error + wantOutcome string + }{ + { + name: "temporary remote fetch failure is nacked for retry", + controller: gitexec.NewCommandError("fetch", "fatal: unable to access 'https://host/r.git/': Connection reset by peer", exitErr), + wantOutcome: "nack", + }, + { + name: "deterministic git failure is rejected to dead letter", + controller: gitexec.NewCommandError("rev-parse", "exit status 128: fatal: ambiguous argument 'origin/main': unknown revision", exitErr), + wantOutcome: "reject", + }, + { + name: "unknown error is rejected to dead letter", + controller: errors.New("unknown failure"), + wantOutcome: "reject", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + deliveryChannel := make(chan extqueue.Delivery, 1) + subscriber := queuemock.NewMockSubscriber(ctrl) + subscriber.EXPECT().Subscribe(gomock.Any(), gomock.Any(), gomock.Any()).Return(deliveryChannel, nil) + queue := queuemock.NewMockQueue(ctrl) + queue.EXPECT().Subscriber().Return(subscriber) + + registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{{ + Key: testGitTopicKey, + Name: "git-test", + Queue: queue, + Subscription: extqueue.DefaultSubscriptionConfig( + "git-test-worker", + testGitGroup, + ), + }}) + require.NoError(t, err) + + serviceConsumer := consumer.New( + zaptest.NewLogger(t).Sugar(), + tally.NoopScope, + registry, + newPrimaryErrorProcessor(), + consumergatenoop.New(), + ) + require.NoError(t, serviceConsumer.Register(errorController{err: tt.controller})) + require.NoError(t, serviceConsumer.Start(context.Background())) + + message := entityqueue.NewMessage("git-test-message", []byte("payload"), "partition", nil) + delivery := queuemock.NewMockDelivery(ctrl) + delivery.EXPECT().Message().Return(message).AnyTimes() + delivery.EXPECT().Attempt().Return(1).AnyTimes() + done := make(chan struct{}) + if tt.wantOutcome == "nack" { + delivery.EXPECT().Nack(gomock.Any(), gomock.Any()).DoAndReturn(func(context.Context, failure.Failure) error { + close(done) + return nil + }) + } else { + delivery.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(func(context.Context, failure.Failure) error { + close(done) + return nil + }) + } + + deliveryChannel <- delivery + <-done + require.NoError(t, serviceConsumer.Stop(30000)) + }) + } +}