diff --git a/platform/errs/README.md b/platform/errs/README.md index 47e7cfa0d..391054c4c 100644 --- a/platform/errs/README.md +++ b/platform/errs/README.md @@ -58,7 +58,7 @@ Two implementations ship in this package: ## 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) 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`), and `platform/errs/generic` (transport-agnostic concerns such as `context.Canceled`). A classifier: @@ -91,17 +91,21 @@ 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" + httperrs "github.com/uber/submitqueue/platform/errs/http" mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql" ) c := consumer.New(logger, scope, registry, errs.NewClassifierProcessor( genericerrs.Classifier, + httperrs.Classifier, mysqlerrs.Classifier, ), ) ``` +`httperrs` precedes `mysqlerrs` for a reason worth knowing before reordering the list: the MySQL classifier treats any `net.Error` as retryable infra, and the `*url.Error` an HTTP client returns satisfies `net.Error`. Whichever runs first claims that node, so with the order reversed an HTTP transport failure is classified as a MySQL one — retryable either way, but no longer attributed to the dependency it came from. This is the cross-extension ambiguity `NewClassifierProcessor` documents as deferred; registration order is the workaround. + 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` and `platform/errs/generic/generic_test.go`. ## Overriding Classification from a Controller diff --git a/platform/errs/http/BUILD.bazel b/platform/errs/http/BUILD.bazel new file mode 100644 index 000000000..354d8aa97 --- /dev/null +++ b/platform/errs/http/BUILD.bazel @@ -0,0 +1,25 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["http.go"], + importpath = "github.com/uber/submitqueue/platform/errs/http", + visibility = ["//visibility:public"], + deps = [ + "//platform/errs:go_default_library", + "//platform/http:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["http_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/errs:go_default_library", + "//platform/errs/generic:go_default_library", + "//platform/errs/mysql:go_default_library", + "//platform/http:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + ], +) diff --git a/platform/errs/http/http.go b/platform/errs/http/http.go new file mode 100644 index 000000000..946ec8ce1 --- /dev/null +++ b/platform/errs/http/http.go @@ -0,0 +1,125 @@ +// 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 http provides an errs.Classifier for failures returned by HTTP +// clients: a rejected status code (platform/http.StatusError) and a transport +// failure (*url.Error, which is what http.Client.Do returns). +// +// Wire it into any service whose extensions call an HTTP API — build runners, +// CI gateways, webhook senders. Without it every status code looks the same to +// the pipeline: unclassified, and therefore non-retryable, so a 502 from a proxy +// dead-letters the message on its first attempt rather than being retried. +// +// Order matters when wiring this alongside platform/errs/mysql. The MySQL +// classifier treats any net.Error as retryable infra, and *url.Error satisfies +// net.Error, so it will claim HTTP transport failures if it runs first. List +// this classifier before it to keep those failures attributed to the dependency +// they came from: +// +// errs.NewClassifierProcessor( +// genericerrs.Classifier, +// httperrs.Classifier, +// mysqlerrs.Classifier, +// ) +package http + +import ( + "context" + nethttp "net/http" + "net/url" + + "github.com/uber/submitqueue/platform/errs" + phttp "github.com/uber/submitqueue/platform/http" +) + +// Classifier implements errs.Classifier for HTTP client failures. It recognises: +// +// - *phttp.StatusError — dispatches on the status code. Codes that describe a +// server-side or overload condition (500, 502, 503, 504, other unassigned +// 5xx, 429, 408) are retryable dependency errors. Codes that describe a +// verdict on the request itself (4xx, 3xx, and the permanently broken 501 +// and 505) are non-retryable dependency errors. +// - *url.Error — the wrapper http.Client.Do puts around connection resets, DNS +// failures, TLS errors and timeouts. A retryable dependency error, except +// for our own context cancellation (see Classify). +// +// Everything else returns errs.Unknown so the classifier-processor walk can keep +// looking down the unwrap chain. +// +// The classifier never returns errs.User. A 400 or 403 says the request was +// rejected, not that a person did something wrong; only the controller knows +// whether the request was built from user input. Controllers express that by +// wrapping with errs.NewUserError, which short-circuits pass 1 of the +// classifier-processor before this classifier is consulted. +// +// 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{} + +// 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 { + if se, ok := err.(*phttp.StatusError); ok { + return classifyStatusCode(se.StatusCode) + } + + if ue, ok := err.(*url.Error); ok { + // A cancelled context is ours, not theirs — process shutdown, or a parent + // operation that went away — so decline it and let the generic classifier + // claim context.Canceled as plain retryable infra, keeping shutdowns out + // of this backend's dependency metrics. An expired deadline is theirs: + // the remote end did not answer in time, so it takes the verdict below. + // Declining that one would strand it, since generic matches only Canceled. + if ue.Err == context.Canceled { + return errs.Unknown + } + // Everything else at this layer is a failed exchange with the remote end, + // and none of those shapes says the request was invalid. + return errs.InfraDependencyRetryable + } + + return errs.Unknown +} + +// classifyStatusCode maps an HTTP status code to a Verdict. The split is whether +// the code describes the state of the server, which can change on its own, or a +// verdict on the request, which replaying only reproduces. +func classifyStatusCode(code int) errs.Verdict { + switch code { + case nethttp.StatusRequestTimeout, // 408 — the server stopped waiting; sending it again is reasonable. + nethttp.StatusTooManyRequests: // 429 — over a rate limit that resets with time. + return errs.InfraDependencyRetryable + + case nethttp.StatusNotImplemented, // 501 — the route will not appear because we retried. + nethttp.StatusHTTPVersionNotSupported: // 505 — a client/server mismatch to fix in config. + return errs.InfraDependency + } + + // Remaining 5xx: the server reported its own failure. Covers 500, 502, 503 + // and 504, the shapes a proxy or overloaded backend produces, plus any + // unassigned or vendor-specific 5xx, which follow the same convention. + if code >= nethttp.StatusInternalServerError { + return errs.InfraDependencyRetryable + } + + // 4xx other than the two above, 3xx the client was not configured to follow, + // and anything else a caller chose to reject — including a code that was + // never a response, such as 0: a verdict on the request, or on a malformed + // call. Neither changes on a second attempt. + return errs.InfraDependency +} diff --git a/platform/errs/http/http_test.go b/platform/errs/http/http_test.go new file mode 100644 index 000000000..41eabd8d9 --- /dev/null +++ b/platform/errs/http/http_test.go @@ -0,0 +1,202 @@ +// 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 http + +import ( + "context" + "errors" + "fmt" + "net" + nethttp "net/http" + "net/url" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/uber/submitqueue/platform/errs" + genericerrs "github.com/uber/submitqueue/platform/errs/generic" + mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql" + phttp "github.com/uber/submitqueue/platform/http" +) + +func TestClassifier_StatusCodes(t *testing.T) { + tests := []struct { + name string + code int + want errs.Verdict + }{ + // Server state: changes without us doing anything differently. + {"bad gateway", nethttp.StatusBadGateway, errs.InfraDependencyRetryable}, + {"service unavailable", nethttp.StatusServiceUnavailable, errs.InfraDependencyRetryable}, + {"gateway timeout", nethttp.StatusGatewayTimeout, errs.InfraDependencyRetryable}, + {"internal server error", nethttp.StatusInternalServerError, errs.InfraDependencyRetryable}, + {"unassigned 5xx", 599, errs.InfraDependencyRetryable}, + {"request timeout", nethttp.StatusRequestTimeout, errs.InfraDependencyRetryable}, + {"too many requests", nethttp.StatusTooManyRequests, errs.InfraDependencyRetryable}, + + // Verdicts on the request: replaying reproduces the same answer. + {"not implemented", nethttp.StatusNotImplemented, errs.InfraDependency}, + {"http version not supported", nethttp.StatusHTTPVersionNotSupported, errs.InfraDependency}, + {"bad request", nethttp.StatusBadRequest, errs.InfraDependency}, + {"unauthorized", nethttp.StatusUnauthorized, errs.InfraDependency}, + {"forbidden", nethttp.StatusForbidden, errs.InfraDependency}, + {"not found", nethttp.StatusNotFound, errs.InfraDependency}, + {"unprocessable entity", nethttp.StatusUnprocessableEntity, errs.InfraDependency}, + {"unfollowed redirect", nethttp.StatusFound, errs.InfraDependency}, + + // Never a response: a caller built this from something else. + {"zero code", 0, errs.InfraDependency}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, Classifier.Classify(phttp.NewStatusError(tt.code, nil))) + }) + } +} + +func TestClassifier_TransportFailures(t *testing.T) { + tests := []struct { + name string + err error + want errs.Verdict + }{ + { + name: "connection refused", + err: &url.Error{Op: "Get", URL: "http://api.example", Err: errors.New("connection refused")}, + want: errs.InfraDependencyRetryable, + }, + { + name: "dns failure", + err: &url.Error{Op: "Get", URL: "http://api.example", Err: &net.DNSError{Err: "no such host"}}, + want: errs.InfraDependencyRetryable, + }, + { + // Ours, not theirs: declining lets the walk reach context.Canceled, + // where the generic classifier calls it plain retryable infra. + name: "context cancelled", + err: &url.Error{Op: "Get", URL: "http://api.example", Err: context.Canceled}, + want: errs.Unknown, + }, + { + // Theirs, not ours: the remote end did not answer in time. + name: "context deadline exceeded", + err: &url.Error{Op: "Get", URL: "http://api.example", Err: context.DeadlineExceeded}, + want: errs.InfraDependencyRetryable, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, Classifier.Classify(tt.err)) + }) + } +} + +func TestClassifier_Unknown(t *testing.T) { + tests := []struct { + name string + err error + }{ + // Per-node contract: a wrapped StatusError must not match here. The + // classifier-processor walk reaches the inner node and asks again there. + {"wrapped status error", fmt.Errorf("get build x: %w", phttp.NewStatusError(502, nil))}, + {"plain error", errors.New("anything")}, + {"bare context.Canceled", context.Canceled}, + {"nil", nil}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, errs.Unknown, Classifier.Classify(tt.err)) + }) + } +} + +func TestClassifier_AppliedViaProcessor(t *testing.T) { + // The order services wire: generic first, this one before mysqlerrs. + processor := errs.NewClassifierProcessor(genericerrs.Classifier, Classifier, mysqlerrs.Classifier) + + t.Run("wrapped 502 becomes a retryable dependency error", func(t *testing.T) { + err := fmt.Errorf("get build org/pipeline/builds/1: %w", phttp.NewStatusError(nethttp.StatusBadGateway, []byte("proxy forward failed"))) + out := processor.Process(err) + assert.True(t, errs.IsRetryable(out)) + assert.True(t, errs.IsDependencyError(out)) + }) + + t.Run("wrapped 404 stays non-retryable", func(t *testing.T) { + err := fmt.Errorf("get build org/pipeline/builds/1: %w", phttp.NewStatusError(nethttp.StatusNotFound, nil)) + out := processor.Process(err) + assert.False(t, errs.IsRetryable(out)) + assert.True(t, errs.IsDependencyError(out)) + }) + + t.Run("transport failure is attributed to the dependency not mysql", func(t *testing.T) { + // mysqlerrs calls any net.Error retryable infra, and *url.Error is one, + // so it would claim this node and drop the dependency attribution if it + // were listed first. + err := fmt.Errorf("send: %w", &url.Error{Op: "Get", URL: "http://api.example", Err: errors.New("connection reset by peer")}) + out := processor.Process(err) + assert.True(t, errs.IsRetryable(out)) + assert.True(t, errs.IsDependencyError(out), "should be attributed to the HTTP dependency") + }) + + t.Run("our cancellation is retryable but not a dependency failure", func(t *testing.T) { + err := fmt.Errorf("send: %w", &url.Error{Op: "Get", URL: "http://api.example", Err: context.Canceled}) + out := processor.Process(err) + assert.True(t, errs.IsRetryable(out)) + assert.False(t, errs.IsDependencyError(out)) + }) + + t.Run("expired deadline is retryable without mysqlerrs claiming it", func(t *testing.T) { + err := fmt.Errorf("send: %w", &url.Error{Op: "Get", URL: "http://api.example", Err: context.DeadlineExceeded}) + out := processor.Process(err) + assert.True(t, errs.IsRetryable(out)) + assert.True(t, errs.IsDependencyError(out), "should be attributed to the HTTP dependency") + }) + + t.Run("a controller verdict wins over the classifier", func(t *testing.T) { + // Pass 1 of the processor short-circuits on the existing framework wrap, + // so a 502 a controller decided was fatal stays fatal. + err := errs.NewDependencyError(phttp.NewStatusError(nethttp.StatusBadGateway, nil)) + out := processor.Process(err) + assert.Same(t, err, out) + assert.False(t, errs.IsRetryable(out)) + }) +} + +// TestClassifier_WithoutMySQLClassifier covers a service with no MySQL +// dependency: no verdict here may rely on mysqlerrs' net.Error rule. +func TestClassifier_WithoutMySQLClassifier(t *testing.T) { + processor := errs.NewClassifierProcessor(genericerrs.Classifier, Classifier) + + tests := []struct { + name string + cause error + wantDependency bool + }{ + {name: "connection reset", cause: errors.New("connection reset by peer"), wantDependency: true}, + {name: "expired deadline", cause: context.DeadlineExceeded, wantDependency: true}, + {name: "our cancellation", cause: context.Canceled, wantDependency: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := fmt.Errorf("send: %w", &url.Error{Op: "Get", URL: "http://api.example", Err: tt.cause}) + out := processor.Process(err) + assert.True(t, errs.IsRetryable(out), "must not depend on mysqlerrs being wired") + assert.Equal(t, tt.wantDependency, errs.IsDependencyError(out)) + }) + } +} diff --git a/platform/extension/buildrunner/buildkite/client.go b/platform/extension/buildrunner/buildkite/client.go index d304029ce..a37f1f842 100644 --- a/platform/extension/buildrunner/buildkite/client.go +++ b/platform/extension/buildrunner/buildkite/client.go @@ -66,6 +66,11 @@ type BuildResponse struct { } // CreateBuild creates a new Buildkite build. +// +// POST /builds is not idempotent and a rejection carries its status code like any +// other, so a caller that retries a 502 can create a second build when the first +// was already accepted. Callers that cannot tolerate that need their own +// idempotency check. func (c *Client) CreateBuild(ctx context.Context, req CreateBuildRequest) (BuildResponse, error) { body, err := json.Marshal(req) if err != nil { @@ -105,7 +110,7 @@ func (c *Client) CancelBuild(ctx context.Context, number int) error { // Already terminal — no-op per BuildRunner.Cancel contract. return nil default: - return fmt.Errorf("unexpected status %d from cancel: %s", status, respBody) + return fmt.Errorf("cancel build: %w", phttp.NewStatusError(status, respBody)) } } @@ -122,7 +127,9 @@ func (c *Client) do(ctx context.Context, method, rawURL string, body []byte, out return ErrNotFound } if status < 200 || status >= 300 { - return fmt.Errorf("API returned status %d: %s", status, respBody) + // Typed rather than formatted so platform/errs/http can read the code + // and tell a transient 502 from a permanent 400. + return phttp.NewStatusError(status, respBody) } if out != nil { diff --git a/platform/extension/buildrunner/buildkite/client_test.go b/platform/extension/buildrunner/buildkite/client_test.go index c78e82ea4..aee60df86 100644 --- a/platform/extension/buildrunner/buildkite/client_test.go +++ b/platform/extension/buildrunner/buildkite/client_test.go @@ -38,6 +38,15 @@ func newTestClient(t *testing.T, handler http.Handler) *Client { return NewClient(c) } +// requireStatusError asserts err carries code as a *phttp.StatusError, the shape +// platform/errs/http needs to classify it. +func requireStatusError(t *testing.T, err error, code int) { + t.Helper() + var se *phttp.StatusError + require.ErrorAs(t, err, &se) + assert.Equal(t, code, se.StatusCode) +} + func buildJSON(t *testing.T, number int, state, webURL string) []byte { t.Helper() return buildJSONWithEnv(t, number, state, webURL, nil) @@ -86,6 +95,7 @@ func TestCreateBuild_ErrorStatus_ReturnsError(t *testing.T) { _, err := c.CreateBuild(context.Background(), CreateBuildRequest{}) require.Error(t, err) + requireStatusError(t, err, http.StatusInternalServerError) } // --- GetBuild --- @@ -150,7 +160,9 @@ func TestCancelBuild_ErrorStatus_ReturnsError(t *testing.T) { w.WriteHeader(http.StatusInternalServerError) })) - require.Error(t, c.CancelBuild(context.Background(), 5)) + err := c.CancelBuild(context.Background(), 5) + require.Error(t, err) + requireStatusError(t, err, http.StatusInternalServerError) } // --- EncodeBuildNumber / ParseBuildNumber --- diff --git a/platform/extension/buildrunner/githubactions/client.go b/platform/extension/buildrunner/githubactions/client.go index 874878264..71d467e49 100644 --- a/platform/extension/buildrunner/githubactions/client.go +++ b/platform/extension/buildrunner/githubactions/client.go @@ -106,6 +106,11 @@ type WorkflowRun struct { } // DispatchWorkflow dispatches the bound workflow. +// +// Dispatching is not idempotent and a rejection carries its status code like any +// other, so a caller that retries a 502 can start a second run when the first was +// already accepted. Callers that cannot tolerate that need their own idempotency +// check. func (c *Client) DispatchWorkflow(ctx context.Context, req DispatchWorkflowRequest) (DispatchWorkflowResponse, error) { body, err := json.Marshal(req) if err != nil { @@ -146,7 +151,7 @@ func (c *Client) CancelRun(ctx context.Context, runID int64) error { case http.StatusNotFound: return ErrNotFound default: - return fmt.Errorf("unexpected status %d from cancel", status) + return fmt.Errorf("cancel run: %w", phttp.NewStatusError(status, nil)) } } @@ -185,7 +190,9 @@ func (c *Client) do(ctx context.Context, method, rawURL string, body []byte, out return ErrNotFound } if status < 200 || status >= 300 { - return fmt.Errorf("API returned status %d: %s", status, respBody) + // Typed rather than formatted so platform/errs/http can read the code + // and tell a transient 502 from a permanent 400. + return phttp.NewStatusError(status, respBody) } if out != nil && len(respBody) > 0 { diff --git a/platform/extension/buildrunner/githubactions/client_test.go b/platform/extension/buildrunner/githubactions/client_test.go index 82f0ce52c..f0adb1f28 100644 --- a/platform/extension/buildrunner/githubactions/client_test.go +++ b/platform/extension/buildrunner/githubactions/client_test.go @@ -40,6 +40,15 @@ func newTestClient(t *testing.T, handler http.Handler) *Client { return NewClient(c, "uber", "submitqueue", "submitqueue-ci.yml") } +// requireStatusError asserts err carries code as a *phttp.StatusError, the shape +// platform/errs/http needs to classify it. +func requireStatusError(t *testing.T, err error, code int) { + t.Helper() + var se *phttp.StatusError + require.ErrorAs(t, err, &se) + assert.Equal(t, code, se.StatusCode) +} + // --- NewClient / accessors --- func TestNewClient_ExposesIdentity(t *testing.T) { @@ -89,6 +98,7 @@ func TestDispatchWorkflow_ErrorStatus_ReturnsError(t *testing.T) { _, err := c.DispatchWorkflow(context.Background(), DispatchWorkflowRequest{}) require.Error(t, err) + requireStatusError(t, err, http.StatusInternalServerError) } // --- GetRun --- @@ -147,6 +157,16 @@ func TestCancelRun_NotFound_ReturnsError(t *testing.T) { require.Error(t, c.CancelRun(context.Background(), 5)) } +func TestCancelRun_ErrorStatus_ReturnsError(t *testing.T) { + c := newTestClient(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadGateway) + })) + + err := c.CancelRun(context.Background(), 5) + require.Error(t, err) + requireStatusError(t, err, http.StatusBadGateway) +} + // --- EncodeRunID / ParseRunID --- func TestEncodeParseRunID_RoundTrip(t *testing.T) { diff --git a/platform/http/BUILD.bazel b/platform/http/BUILD.bazel index 724884adf..336d044e4 100644 --- a/platform/http/BUILD.bazel +++ b/platform/http/BUILD.bazel @@ -4,6 +4,7 @@ go_library( name = "go_default_library", srcs = [ "request.go", + "status.go", "transport.go", ], importpath = "github.com/uber/submitqueue/platform/http", @@ -14,6 +15,7 @@ go_test( name = "go_default_test", srcs = [ "request_test.go", + "status_test.go", "transport_test.go", ], embed = [":go_default_library"], diff --git a/platform/http/status.go b/platform/http/status.go new file mode 100644 index 000000000..097e0ad06 --- /dev/null +++ b/platform/http/status.go @@ -0,0 +1,68 @@ +// 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 http + +import "fmt" + +// _maxRenderedBodyBytes bounds how much of Body reaches Error(). The rendered +// string lands in a consumer's dead-letter record, whose column is finite, so an +// error page from a chatty gateway must not be able to fail that write. Body +// itself is kept whole. +const _maxRenderedBodyBytes = 1024 + +// StatusError reports a response whose status code the caller rejected. +// +// SendRequest does not build this error itself: which codes count as success +// varies by API — a 404 may be a sentinel, a 422 may be a no-op — so the caller +// still decides. What the caller should not do is report the rejection with a +// plain fmt.Errorf. The status code is the only thing that says whether a retry +// has any chance of working, and a formatted string throws it away. Returning +// this type keeps the code in the error chain, where platform/errs/http can read +// it and classify the failure. +// +// Use it for the "this status is a failure" branch of a response check: +// +// if status < 200 || status >= 300 { +// return phttp.NewStatusError(status, respBody) +// } +type StatusError struct { + // StatusCode is the HTTP status code from the response. + StatusCode int + // Body is the response body as read from the wire, or empty when the + // caller had no body to attach. Error() renders at most + // _maxRenderedBodyBytes of it. + Body string +} + +// NewStatusError returns a StatusError for the given code and response body. +// body may be nil. +func NewStatusError(statusCode int, body []byte) *StatusError { + return &StatusError{StatusCode: statusCode, Body: string(body)} +} + +// Error renders the status and, when present, the response body truncated to +// _maxRenderedBodyBytes. Callers are expected to wrap it with the operation that +// failed, giving messages like "get build org/pipeline/builds/123: unexpected +// status 502: proxy forward failed". +func (e *StatusError) Error() string { + if e.Body == "" { + return fmt.Sprintf("unexpected status %d", e.StatusCode) + } + body := e.Body + if len(body) > _maxRenderedBodyBytes { + body = body[:_maxRenderedBodyBytes] + "… (truncated)" + } + return fmt.Sprintf("unexpected status %d: %s", e.StatusCode, body) +} diff --git a/platform/http/status_test.go b/platform/http/status_test.go new file mode 100644 index 000000000..ce5b78361 --- /dev/null +++ b/platform/http/status_test.go @@ -0,0 +1,76 @@ +// 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 http + +import ( + "bytes" + "errors" + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewStatusError(t *testing.T) { + tests := []struct { + name string + code int + body []byte + want string + }{ + { + name: "with body", + code: 502, + body: []byte("proxy forward failed: relay-connection-failed"), + want: "unexpected status 502: proxy forward failed: relay-connection-failed", + }, + {name: "nil body", code: 500, want: "unexpected status 500"}, + {name: "empty body", code: 404, body: []byte{}, want: "unexpected status 404"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := NewStatusError(tt.code, tt.body) + require.NotNil(t, err) + assert.Equal(t, tt.code, err.StatusCode) + assert.Equal(t, tt.want, err.Error()) + }) + } +} + +func TestStatusError_RenderedBodyIsBounded(t *testing.T) { + body := bytes.Repeat([]byte("a"), _maxRenderedBodyBytes*4) + err := NewStatusError(502, body) + + assert.Len(t, err.Body, _maxRenderedBodyBytes*4, "Body keeps what the caller passed") + rendered := err.Error() + assert.Less(t, len(rendered), _maxRenderedBodyBytes*2, "rendering must not grow with the body") + assert.Contains(t, rendered, "(truncated)") + assert.True(t, strings.HasPrefix(rendered, "unexpected status 502: aaa")) +} + +func TestStatusError_SurvivesWrapping(t *testing.T) { + // Callers wrap with the operation that failed. The code has to stay + // reachable through the chain, otherwise the classifier cannot read it. + inner := NewStatusError(503, []byte("unavailable")) + wrapped := fmt.Errorf("get build org/pipeline/builds/1: %w", inner) + + var se *StatusError + require.True(t, errors.As(wrapped, &se)) + assert.Equal(t, 503, se.StatusCode) + assert.Equal(t, "get build org/pipeline/builds/1: unexpected status 503: unavailable", wrapped.Error()) +} diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index c76638292..a162adb0c 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "//platform/consumer:go_default_library", "//platform/errs:go_default_library", "//platform/errs/generic:go_default_library", + "//platform/errs/http:go_default_library", "//platform/errs/mysql:go_default_library", "//platform/extension/consumergate/noop:go_default_library", "//platform/extension/counter:go_default_library", diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 3b44500bf..2e049e79c 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -32,6 +32,7 @@ import ( "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" genericerrs "github.com/uber/submitqueue/platform/errs/generic" + httperrs "github.com/uber/submitqueue/platform/errs/http" mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql" consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop" "github.com/uber/submitqueue/platform/extension/counter" @@ -264,6 +265,7 @@ func run() error { primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, errs.NewClassifierProcessor( genericerrs.Classifier, + httperrs.Classifier, mysqlerrs.Classifier, ), consumergatenoop.New(), diff --git a/stovepipe/extension/buildrunner/buildkite/BUILD.bazel b/stovepipe/extension/buildrunner/buildkite/BUILD.bazel index 64c3dda60..30f17f9d0 100644 --- a/stovepipe/extension/buildrunner/buildkite/BUILD.bazel +++ b/stovepipe/extension/buildrunner/buildkite/BUILD.bazel @@ -6,7 +6,9 @@ go_library( importpath = "github.com/uber/submitqueue/stovepipe/extension/buildrunner/buildkite", visibility = ["//visibility:public"], deps = [ + "//platform/errs:go_default_library", "//platform/extension/buildrunner/buildkite:go_default_library", + "//platform/http:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/buildrunner:go_default_library", "@org_uber_go_zap//:go_default_library", @@ -18,6 +20,8 @@ go_test( srcs = ["buildkite_test.go"], embed = [":go_default_library"], deps = [ + "//platform/errs:go_default_library", + "//platform/errs/http:go_default_library", "//platform/extension/buildrunner/buildkite:go_default_library", "//platform/http:go_default_library", "//stovepipe/entity:go_default_library", diff --git a/stovepipe/extension/buildrunner/buildkite/buildkite.go b/stovepipe/extension/buildrunner/buildkite/buildkite.go index 06a9d7172..07e1e985c 100644 --- a/stovepipe/extension/buildrunner/buildkite/buildkite.go +++ b/stovepipe/extension/buildrunner/buildkite/buildkite.go @@ -33,11 +33,15 @@ package buildkite import ( "context" "encoding/json" + "errors" "fmt" + "net/http" "go.uber.org/zap" + "github.com/uber/submitqueue/platform/errs" platformbuildkite "github.com/uber/submitqueue/platform/extension/buildrunner/buildkite" + phttp "github.com/uber/submitqueue/platform/http" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/buildrunner" ) @@ -123,7 +127,16 @@ func (r *runner) Trigger(ctx context.Context, baseURI, headURI string, metadata resp, err := r.client.CreateBuild(ctx, req) if err != nil { - return entity.BuildID{}, fmt.Errorf("buildkite: create build: %w", err) + err = fmt.Errorf("buildkite: create build: %w", err) + // Don't retry on error codes that may have been answered after the build was created. + var se *phttp.StatusError + if errors.As(err, &se) { + switch se.StatusCode { + case http.StatusInternalServerError, http.StatusBadGateway, http.StatusGatewayTimeout: + return entity.BuildID{}, errs.NewDependencyError(err) + } + } + return entity.BuildID{}, err } r.logger.Debugw("triggered Buildkite build", diff --git a/stovepipe/extension/buildrunner/buildkite/buildkite_test.go b/stovepipe/extension/buildrunner/buildkite/buildkite_test.go index 93cabdbf2..0a79a7272 100644 --- a/stovepipe/extension/buildrunner/buildkite/buildkite_test.go +++ b/stovepipe/extension/buildrunner/buildkite/buildkite_test.go @@ -27,6 +27,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/platform/errs" + httperrs "github.com/uber/submitqueue/platform/errs/http" platformbuildkite "github.com/uber/submitqueue/platform/extension/buildrunner/buildkite" phttp "github.com/uber/submitqueue/platform/http" "github.com/uber/submitqueue/stovepipe/entity" @@ -120,6 +122,37 @@ func TestTrigger_BuildkiteError_ReturnsError(t *testing.T) { require.Error(t, err) } +// TestTrigger_RetryabilityByStatus pins which create failures may be replayed. A +// code that Buildkite could have answered after creating the build must not be, +// because Trigger has no dedup key and the retry would start a second build. +func TestTrigger_RetryabilityByStatus(t *testing.T) { + processor := errs.NewClassifierProcessor(httperrs.Classifier) + + tests := []struct { + status int + wantRetryable bool + }{ + {status: http.StatusInternalServerError}, + {status: http.StatusBadGateway}, + {status: http.StatusGatewayTimeout}, + {status: http.StatusServiceUnavailable, wantRetryable: true}, + {status: http.StatusTooManyRequests, wantRetryable: true}, + {status: http.StatusBadRequest}, + } + + for _, tt := range tests { + t.Run(http.StatusText(tt.status), func(t *testing.T) { + r := newTestRunner(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(tt.status) + })) + + _, err := r.Trigger(context.Background(), "", "github://repo/head/aaa", nil) + require.Error(t, err) + assert.Equal(t, tt.wantRetryable, errs.IsRetryable(processor.Process(err))) + }) + } +} + func TestTrigger_WithMetadata_SetsEnvVar(t *testing.T) { var capturedBody []byte r := newTestRunner(t, http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { diff --git a/stovepipe/extension/buildrunner/githubactions/BUILD.bazel b/stovepipe/extension/buildrunner/githubactions/BUILD.bazel index 77f605602..220e94199 100644 --- a/stovepipe/extension/buildrunner/githubactions/BUILD.bazel +++ b/stovepipe/extension/buildrunner/githubactions/BUILD.bazel @@ -6,7 +6,9 @@ go_library( importpath = "github.com/uber/submitqueue/stovepipe/extension/buildrunner/githubactions", visibility = ["//visibility:public"], deps = [ + "//platform/errs:go_default_library", "//platform/extension/buildrunner/githubactions:go_default_library", + "//platform/http:go_default_library", "//stovepipe/entity:go_default_library", "//stovepipe/extension/buildrunner:go_default_library", "@org_uber_go_zap//:go_default_library", @@ -18,6 +20,8 @@ go_test( srcs = ["githubactions_test.go"], embed = [":go_default_library"], deps = [ + "//platform/errs:go_default_library", + "//platform/errs/http:go_default_library", "//platform/extension/buildrunner/githubactions:go_default_library", "//platform/http:go_default_library", "//stovepipe/entity:go_default_library", diff --git a/stovepipe/extension/buildrunner/githubactions/githubactions.go b/stovepipe/extension/buildrunner/githubactions/githubactions.go index ea4d74015..770d0ae5b 100644 --- a/stovepipe/extension/buildrunner/githubactions/githubactions.go +++ b/stovepipe/extension/buildrunner/githubactions/githubactions.go @@ -33,11 +33,15 @@ package githubactions import ( "context" "encoding/json" + "errors" "fmt" + "net/http" "go.uber.org/zap" + "github.com/uber/submitqueue/platform/errs" platformgithubactions "github.com/uber/submitqueue/platform/extension/buildrunner/githubactions" + phttp "github.com/uber/submitqueue/platform/http" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/buildrunner" ) @@ -148,7 +152,16 @@ func (r *runner) Trigger(ctx context.Context, baseURI, headURI string, metadata Inputs: inputs, }) if err != nil { - return entity.BuildID{}, fmt.Errorf("github actions: dispatch workflow: %w", err) + err = fmt.Errorf("github actions: dispatch workflow: %w", err) + // Don't retry on error codes that may have been answered after the run was queued. + var se *phttp.StatusError + if errors.As(err, &se) { + switch se.StatusCode { + case http.StatusInternalServerError, http.StatusBadGateway, http.StatusGatewayTimeout: + return entity.BuildID{}, errs.NewDependencyError(err) + } + } + return entity.BuildID{}, err } if resp.WorkflowRunID <= 0 { return entity.BuildID{}, fmt.Errorf("github actions: dispatch workflow: response missing workflow_run_id (requires return_run_details support)") diff --git a/stovepipe/extension/buildrunner/githubactions/githubactions_test.go b/stovepipe/extension/buildrunner/githubactions/githubactions_test.go index 0f5596860..5adb1b2bd 100644 --- a/stovepipe/extension/buildrunner/githubactions/githubactions_test.go +++ b/stovepipe/extension/buildrunner/githubactions/githubactions_test.go @@ -28,6 +28,8 @@ import ( "github.com/stretchr/testify/require" "go.uber.org/zap" + "github.com/uber/submitqueue/platform/errs" + httperrs "github.com/uber/submitqueue/platform/errs/http" platformgithubactions "github.com/uber/submitqueue/platform/extension/buildrunner/githubactions" phttp "github.com/uber/submitqueue/platform/http" "github.com/uber/submitqueue/stovepipe/entity" @@ -112,6 +114,37 @@ func TestTrigger_DispatchError_ReturnsError(t *testing.T) { require.Error(t, err) } +// TestTrigger_RetryabilityByStatus pins which dispatch failures may be replayed. +// A code that GitHub could have answered after queueing the run must not be, +// because Trigger has no dedup key and the retry would queue a second run. +func TestTrigger_RetryabilityByStatus(t *testing.T) { + processor := errs.NewClassifierProcessor(httperrs.Classifier) + + tests := []struct { + status int + wantRetryable bool + }{ + {status: http.StatusInternalServerError}, + {status: http.StatusBadGateway}, + {status: http.StatusGatewayTimeout}, + {status: http.StatusServiceUnavailable, wantRetryable: true}, + {status: http.StatusTooManyRequests, wantRetryable: true}, + {status: http.StatusBadRequest}, + } + + for _, tt := range tests { + t.Run(http.StatusText(tt.status), func(t *testing.T) { + r := newTestRunner(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(tt.status) + })) + + _, err := r.Trigger(context.Background(), "", "github://repo/head/aaa", nil) + require.Error(t, err) + assert.Equal(t, tt.wantRetryable, errs.IsRetryable(processor.Process(err))) + }) + } +} + func TestTrigger_ErrorsWhenDispatchResponseHasNoRunID(t *testing.T) { r := newTestRunner(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _ = json.NewEncoder(w).Encode(platformgithubactions.DispatchWorkflowResponse{})