From 3663689ab32ca5b001a71f89ab1e2e988e3a2e35 Mon Sep 17 00:00:00 2001 From: Ravi Agarwal Date: Thu, 29 Oct 2026 06:38:36 +0000 Subject: [PATCH 1/2] codeartifact: verify GOAWAY recovery and retain HTTP/2 diagnostics --- .github/workflows/ci.yml | 11 + README.md | 7 + internal/strategy/codeartifact.go | 5 +- internal/strategy/codeartifact_goaway_test.go | 423 ++++++++++++++++++ internal/strategy/codeartifact_http2.go | 80 ++++ internal/strategy/codeartifact_http2_test.go | 212 +++++++++ internal/strategy/codeartifact_stream.go | 13 +- .../codeartifact_uv_integration_test.go | 172 +++++++ 8 files changed, 920 insertions(+), 3 deletions(-) create mode 100644 internal/strategy/codeartifact_goaway_test.go create mode 100644 internal/strategy/codeartifact_http2.go create mode 100644 internal/strategy/codeartifact_http2_test.go create mode 100644 internal/strategy/codeartifact_uv_integration_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0a8d4bc2..168cf6b3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,6 +22,17 @@ jobs: - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 - name: Activate Hermit run: ./bin/hermit env --raw >> "$GITHUB_ENV" + - name: Install pinned uv for recovery tests + run: | + archive="$RUNNER_TEMP/cachew-test-uv.tar.gz" + curl --fail --location --retry 3 \ + https://github.com/astral-sh/uv/releases/download/0.13.4/uv-x86_64-unknown-linux-gnu.tar.gz \ + --output "$archive" + printf '%s %s\n' \ + 2b428396f12572d4429cf210ca234dd599a33b3af0958a4d5ed569d63e3f344f "$archive" \ + | sha256sum --check --strict + tar --extract --gzip --file "$archive" --directory "$RUNNER_TEMP" + echo "$RUNNER_TEMP/uv-x86_64-unknown-linux-gnu" >> "$GITHUB_PATH" - name: Run tests run: just test lint: diff --git a/README.md b/README.md index 5948335b..65108682 100644 --- a/README.md +++ b/README.md @@ -288,6 +288,13 @@ Failures before response headers use bounded statuses in those same metrics: `transport_connection`, or `transport_error` for other failures. Logs include only that status, never the origin URL or signed redirect URL. +Incomplete origin bodies abort the downstream transfer rather than completing a +partial response. Recognized HTTP/2 GOAWAY body failures use the bounded +`http2_goaway` error kind; stream-failure logs also include `http2_error_code` and +`http2_last_stream_id`. Those details are not metric labels, and peer-provided +debug text is never logged. Unknown or malformed GOAWAY metadata retains the +generic transport classification. + Authorization tokens are reused in-process. The default 12-hour token enters proactive refresh one hour before expiration; one request performs each refresh attempt under a detached, bounded context while concurrent requests continue diff --git a/internal/strategy/codeartifact.go b/internal/strategy/codeartifact.go index 05a90a57..d189b0cc 100644 --- a/internal/strategy/codeartifact.go +++ b/internal/strategy/codeartifact.go @@ -22,7 +22,10 @@ import ( "github.com/block/cachew/internal/packagepolicy" ) -const codeArtifactUsername = "aws" +const ( + codeArtifactUsername = "aws" + codeArtifactStatusClientCanceled = "client_canceled" +) const ( defaultCodeArtifactOriginHeaderTimeout = 30 * time.Second diff --git a/internal/strategy/codeartifact_goaway_test.go b/internal/strategy/codeartifact_goaway_test.go new file mode 100644 index 00000000..d75c13d6 --- /dev/null +++ b/internal/strategy/codeartifact_goaway_test.go @@ -0,0 +1,423 @@ +package strategy //nolint:testpackage // Exercise the real strategy and standard-library HTTP/2 transport. + +import ( + "archive/zip" + "bytes" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "fmt" + "io" + "log/slog" + "math/big" + "net" + "net/http" + "net/http/httptest" + "net/url" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + "golang.org/x/net/http2" + "golang.org/x/net/http2/hpack" +) + +const ( + goAwayIndexPath = "/pypi/repository/simple/cachew-goaway/" + goAwayWheelName = "cachew_goaway-1.2.3-py3-none-any.whl" + goAwayWheelPath = "/pypi/repository/files/" + goAwayWheelName + goAwayPrivateDebug = "private-origin?token=never-log-goaway-debug" + goAwayPartialSize = 32 << 10 +) + +type goAwayOriginRequest struct { + path string + streamID uint32 + failed bool +} + +type goAwayOrigin struct { + listener net.Listener + roots *x509.CertPool + ca tls.Certificate + index []byte + wheel []byte + metadata []byte + mu sync.Mutex + failures int + requests []goAwayOriginRequest + pending map[string]chan struct{} + conns []net.Conn + workers sync.WaitGroup +} + +func newGoAwayOrigin(t *testing.T, failures int) *goAwayOrigin { + t.Helper() + certificateServer := httptest.NewTLSServer(http.NotFoundHandler()) + tlsConfig := certificateServer.TLS.Clone() + tlsConfig.NextProtos = []string{http2.NextProtoTLS} + roots := x509.NewCertPool() + roots.AddCert(certificateServer.Certificate()) + certificateServer.Close() + listener, err := tls.Listen("tcp", "127.0.0.1:0", tlsConfig) + assert.NoError(t, err) + metadata := []byte("Metadata-Version: 2.3\nName: cachew-goaway\nVersion: 1.2.3\nRequires-Python: >=3.12\n\n") + var wheel bytes.Buffer + archive := zip.NewWriter(&wheel) + for _, file := range []struct{ name, content string }{ + {"cachew_goaway-1.2.3.dist-info/METADATA", string(metadata)}, + {"cachew_goaway-1.2.3.dist-info/WHEEL", "Wheel-Version: 1.0\nGenerator: cachew-test\nRoot-Is-Purelib: true\nTag: py3-none-any\n"}, + {"cachew_goaway-1.2.3.dist-info/RECORD", ""}, + } { + writer, createErr := archive.Create(file.name) + assert.NoError(t, createErr) + _, writeErr := io.WriteString(writer, file.content) + assert.NoError(t, writeErr) + } + assert.NoError(t, archive.Close()) + index := fmt.Sprintf(`{"meta":{"api-version":"1.0"},"name":"cachew-goaway","padding":"%s","files":[`+ + `{"filename":%q,"url":%q,"hashes":{"sha256":"%x"},"requires-python":">=3.12","core-metadata":{"sha256":"%x"}}]}`, + strings.Repeat("x", goAwayPartialSize), goAwayWheelName, "../../files/"+goAwayWheelName, + sha256.Sum256(wheel.Bytes()), sha256.Sum256(metadata)) + origin := &goAwayOrigin{ + listener: listener, roots: roots, ca: tlsConfig.Certificates[0], + index: []byte(index), wheel: wheel.Bytes(), metadata: metadata, + failures: failures, pending: make(map[string]chan struct{}), + } + accepted := make(chan struct{}) + origin.workers.Go(func() { + defer close(accepted) + for { + conn, acceptErr := listener.Accept() + if acceptErr != nil { + return + } + origin.mu.Lock() + origin.conns = append(origin.conns, conn) + origin.mu.Unlock() + origin.workers.Go(func() { + defer conn.Close() + origin.serve(t, conn) + }) + } + }) + t.Cleanup(func() { + assert.NoError(t, listener.Close()) + <-accepted + origin.mu.Lock() + for _, conn := range origin.conns { + _ = conn.Close() + } + origin.mu.Unlock() + origin.workers.Wait() + }) + return origin +} + +func (o *goAwayOrigin) serve(t *testing.T, conn net.Conn) { + t.Helper() + assert.NoError(t, conn.SetDeadline(time.Now().Add(30*time.Second))) + preface := make([]byte, len(http2.ClientPreface)) + if _, err := io.ReadFull(conn, preface); err != nil { + t.Errorf("read HTTP/2 client preface: %v", err) + return + } + assert.Equal(t, http2.ClientPreface, string(preface)) + framer := http2.NewFramer(conn, conn) + framer.ReadMetaHeaders = hpack.NewDecoder(4096, nil) + if err := framer.WriteSettings(); err != nil { + t.Errorf("write HTTP/2 settings: %v", err) + return + } + for { + frame, err := framer.ReadFrame() + if err != nil { + return + } + switch frame := frame.(type) { + case *http2.SettingsFrame: + if !frame.IsAck() { + if err := framer.WriteSettingsAck(); err != nil { + t.Errorf("acknowledge HTTP/2 settings: %v", err) + return + } + } + case *http2.PingFrame: + if !frame.Flags.Has(http2.FlagPingAck) { + if err := framer.WritePing(true, frame.Data); err != nil { + t.Errorf("acknowledge HTTP/2 ping: %v", err) + return + } + } + case *http2.MetaHeadersFrame: + path := frame.PseudoValue("path") + o.mu.Lock() + failed := path == goAwayIndexPath && o.failures > 0 + var written chan struct{} + if failed { + o.failures-- + written = make(chan struct{}, 1) + o.pending[path] = written + } + o.requests = append(o.requests, goAwayOriginRequest{path, frame.StreamID, failed}) + o.mu.Unlock() + if !o.respond(t, framer, frame.StreamID, path, written) { + return + } + } + } +} + +func (o *goAwayOrigin) respond(t *testing.T, framer *http2.Framer, streamID uint32, path string, written <-chan struct{}) bool { + t.Helper() + body, contentType := o.index, "application/vnd.pypi.simple.v1+json" + switch path { + case goAwayIndexPath: + case goAwayWheelPath: + body, contentType = o.wheel, "application/zip" + case goAwayWheelPath + ".metadata": + body, contentType = o.metadata, "text/plain" + default: + t.Errorf("unexpected origin request: %s", path) + return false + } + var headers bytes.Buffer + encoder := hpack.NewEncoder(&headers) + for _, field := range []hpack.HeaderField{ + {Name: ":status", Value: "200"}, + {Name: "content-type", Value: contentType}, + {Name: "cache-control", Value: testCodeArtifactCacheControl}, + } { + assert.NoError(t, encoder.WriteField(field)) + } + if err := framer.WriteHeaders(http2.HeadersFrameParam{ + StreamID: streamID, BlockFragment: headers.Bytes(), EndHeaders: true, + }); err != nil { + t.Errorf("write HTTP/2 response headers: %v", err) + return false + } + if written != nil { + body = body[:goAwayPartialSize] + } + for len(body) > 0 { + n := min(len(body), 16<<10) + if err := framer.WriteData(streamID, written == nil && n == len(body), body[:n]); err != nil { + t.Errorf("write HTTP/2 response data: %v", err) + return false + } + body = body[n:] + } + if written == nil { + return true + } + timer := time.NewTimer(5 * time.Second) + defer timer.Stop() + select { + case <-written: + case <-timer.C: + t.Error("Cachew did not flush the partial body before GOAWAY") + return false + case <-t.Context().Done(): + return false + } + assert.NoError(t, framer.WriteGoAway(streamID, http2.ErrCodeNo, []byte(goAwayPrivateDebug))) + return false +} + +func (o *goAwayOrigin) observeWrite(path string) { + o.mu.Lock() + defer o.mu.Unlock() + if pending := o.pending[path]; pending != nil { + select { + case pending <- struct{}{}: + default: + } + } +} + +func (o *goAwayOrigin) snapshot() []goAwayOriginRequest { + o.mu.Lock() + defer o.mu.Unlock() + return slices.Clone(o.requests) +} + +type goAwayLogBuffer struct { + mu sync.Mutex + data bytes.Buffer +} + +func (b *goAwayLogBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.data.Write(p) +} + +func (b *goAwayLogBuffer) snapshot() []byte { + b.mu.Lock() + defer b.mu.Unlock() + return bytes.Clone(b.data.Bytes()) +} + +type goAwayResponseWriter struct { + http.ResponseWriter + origin *goAwayOrigin + path string +} + +func (w goAwayResponseWriter) Write(p []byte) (int, error) { + n, err := w.ResponseWriter.Write(p) + if err != nil { + return n, err + } + if err := http.NewResponseController(w.ResponseWriter).Flush(); err != nil { + return n, err + } + if n > 0 { + w.origin.observeWrite(w.path) + } + return n, nil +} + +type goAwayProxy struct { + server *httptest.Server + ca []byte + prefix string + logs *goAwayLogBuffer + finished chan goAwayProxyRequest +} + +type goAwayProxyRequest struct { + path string + protoMajor int +} + +func newGoAwayProxy(t *testing.T, origin *goAwayOrigin, downstream string) *goAwayProxy { + t.Helper() + mux, _, _, strategy, _ := newTestCachingCodeArtifact(t, http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + t.Error("unexpected request to the unused HTTP origin") + })) + target, err := url.Parse("https://" + origin.listener.Addr().String()) + assert.NoError(t, err) + strategy.target = target + transport := strategy.client.Transport.(*http.Transport) + transport.TLSClientConfig = &tls.Config{RootCAs: origin.roots, MinVersion: tls.VersionTLS12} + t.Cleanup(transport.CloseIdleConnections) + logs := &goAwayLogBuffer{} + strategy.logger = slog.New(slog.NewJSONHandler(logs, nil)) + proxy := &goAwayProxy{ + prefix: strategy.prefix, ca: origin.ca.Certificate[0], logs: logs, finished: make(chan goAwayProxyRequest, 64), + } + proxy.server = httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + path := strings.TrimPrefix(r.URL.Path, strategy.prefix) + defer func() { proxy.finished <- goAwayProxyRequest{path: path, protoMajor: r.ProtoMajor} }() + mux.ServeHTTP(goAwayResponseWriter{ResponseWriter: w, origin: origin, path: path}, r) + })) + proxy.server.TLS = &tls.Config{ + Certificates: []tls.Certificate{goAwayProxyCertificate(t, origin.ca)}, MinVersion: tls.VersionTLS12, + } + proxy.server.EnableHTTP2 = downstream == "http2" + proxy.server.StartTLS() + t.Cleanup(proxy.server.Close) + return proxy +} + +func goAwayProxyCertificate(t *testing.T, ca tls.Certificate) tls.Certificate { + t.Helper() + parent, err := x509.ParseCertificate(ca.Certificate[0]) + assert.NoError(t, err) + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + assert.NoError(t, err) + leaf := &x509.Certificate{ + SerialNumber: big.NewInt(1), NotBefore: time.Now().Add(-time.Minute), NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + } + der, err := x509.CreateCertificate(rand.Reader, leaf, parent, &key.PublicKey, ca.PrivateKey) + assert.NoError(t, err) + return tls.Certificate{Certificate: [][]byte{der, ca.Certificate[0]}, PrivateKey: key} +} + +func (p *goAwayProxy) wait(t *testing.T, path string) int { + t.Helper() + timer := time.NewTimer(5 * time.Second) + defer timer.Stop() + select { + case got := <-p.finished: + assert.Equal(t, path, got.path) + return got.protoMajor + case <-timer.C: + t.Fatal("Cachew request did not finish") + case <-t.Context().Done(): + t.Fatal("test canceled before Cachew request finished") + } + return 0 +} + +func assertGoAwayDiagnostics(t *testing.T, proxy *goAwayProxy, origin *goAwayOrigin, count int) { + t.Helper() + logs := proxy.logs.snapshot() + assert.False(t, bytes.Contains(logs, []byte(goAwayPrivateDebug)), "GOAWAY debug data must stay private") + var failures []map[string]any + for _, entry := range codeArtifactFailureLogs(t, logs) { + if entry["msg"] == "CodeArtifact response stream failed" { + failures = append(failures, entry) + } + } + assert.Equal(t, count, len(failures)) + var requests []goAwayOriginRequest + for _, request := range origin.snapshot() { + if request.failed { + requests = append(requests, request) + } + } + assert.Equal(t, count, len(requests)) + for i, failure := range failures { + assert.Equal[any](t, "origin", failure["source"]) + assert.Equal[any](t, "source_read", failure["stage"]) + assert.Equal[any](t, "http2_goaway", failure["error_kind"]) + assert.Equal[any](t, "NO_ERROR", failure["http2_error_code"]) + assert.Equal[any](t, float64(requests[i].streamID), failure["http2_last_stream_id"]) + assert.Equal[any](t, float64(goAwayPartialSize), failure["bytes_read"]) + assert.Equal[any](t, float64(goAwayPartialSize), failure["bytes_written"]) + } +} + +func TestCodeArtifactRealGoAwayRecovery(t *testing.T) { + for _, downstream := range []string{"http1", "http2"} { + t.Run(downstream, func(t *testing.T) { + origin := newGoAwayOrigin(t, 1) + proxy := newGoAwayProxy(t, origin, downstream) + client := proxy.server.Client() + client.Timeout = 10 * time.Second + for attempt := range 3 { + request, err := http.NewRequestWithContext(t.Context(), http.MethodGet, + proxy.server.URL+proxy.prefix+goAwayIndexPath, nil) + assert.NoError(t, err) + response, err := client.Do(request) + assert.NoError(t, err) + body, readErr := io.ReadAll(response.Body) + assert.NoError(t, response.Body.Close()) + proxy.wait(t, goAwayIndexPath) + assert.Equal(t, downstream == "http2", response.ProtoMajor == 2) + assert.Equal(t, http.StatusOK, response.StatusCode) + if attempt == 0 { + assert.Error(t, readErr) + assert.True(t, len(body) > 0, "failure must occur after the client receives a partial body") + assert.True(t, len(body) < len(origin.index)) + } else { + assert.NoError(t, readErr) + assert.Equal(t, origin.index, body) + } + assert.Equal(t, min(attempt+1, 2), len(origin.snapshot()), + "failed fill must not be cached; complete fill must serve the warm request") + } + assertGoAwayDiagnostics(t, proxy, origin, 1) + }) + } +} diff --git a/internal/strategy/codeartifact_http2.go b/internal/strategy/codeartifact_http2.go new file mode 100644 index 00000000..890c5486 --- /dev/null +++ b/internal/strategy/codeartifact_http2.go @@ -0,0 +1,80 @@ +package strategy + +import ( + "strconv" + "strings" +) + +type codeArtifactGoAwayDetails struct { + code string + lastStreamID uint32 +} + +func codeArtifactGoAway(err error) (codeArtifactGoAwayDetails, bool) { + remaining := 32 + return codeArtifactFindGoAway(err, &remaining) +} + +func codeArtifactFindGoAway(err error, remaining *int) (codeArtifactGoAwayDetails, bool) { + if *remaining == 0 { + return codeArtifactGoAwayDetails{}, false + } + *remaining-- + if err == nil { + return codeArtifactGoAwayDetails{}, false + } + switch wrapped := err.(type) { //nolint:errorlint // Inspect unwrap methods without unbounded errors.As traversal. + case interface{ Unwrap() error }: + if inner := wrapped.Unwrap(); inner != nil { + return codeArtifactFindGoAway(inner, remaining) + } + case interface{ Unwrap() []error }: + for _, inner := range wrapped.Unwrap() { + if details, ok := codeArtifactFindGoAway(inner, remaining); ok { + return details, true + } + if *remaining == 0 { + break + } + } + return codeArtifactGoAwayDetails{}, false + } + return parseCodeArtifactGoAway(err.Error()) +} + +// net/http's bundled HTTP/2 transport exposes an unexported GOAWAY error type. +// Parse only its bounded metadata prefix, never retaining peer-controlled debug +// data or wrapper text (which can contain credential-bearing URLs). +func parseCodeArtifactGoAway(message string) (codeArtifactGoAwayDetails, bool) { + const ( + prefix = "http2: server sent GOAWAY and closed the connection; LastStreamID=" + codePrefix = ", ErrCode=" + debugPrefix = ", debug=\"" + ) + header, ok := strings.CutPrefix(message[:min(len(message), 128)], prefix) + if !ok { + return codeArtifactGoAwayDetails{}, false + } + streamID, rest, ok := strings.Cut(header, codePrefix) + if !ok { + return codeArtifactGoAwayDetails{}, false + } + code, _, ok := strings.Cut(rest, debugPrefix) + minimumLength := len(prefix) + len(streamID) + len(codePrefix) + len(code) + len(debugPrefix) + 1 + if !ok || len(message) < minimumLength || !strings.HasSuffix(message, "\"") { + return codeArtifactGoAwayDetails{}, false + } + id, err := strconv.ParseUint(streamID, 10, 31) + if err != nil || strconv.FormatUint(id, 10) != streamID { + return codeArtifactGoAwayDetails{}, false + } + switch code { + case "NO_ERROR", "PROTOCOL_ERROR", "INTERNAL_ERROR", "FLOW_CONTROL_ERROR", + "SETTINGS_TIMEOUT", "STREAM_CLOSED", "FRAME_SIZE_ERROR", "REFUSED_STREAM", + "CANCEL", "COMPRESSION_ERROR", "CONNECT_ERROR", "ENHANCE_YOUR_CALM", + "INADEQUATE_SECURITY", "HTTP_1_1_REQUIRED": + return codeArtifactGoAwayDetails{code: strings.Clone(code), lastStreamID: uint32(id)}, true + default: + return codeArtifactGoAwayDetails{}, false + } +} diff --git a/internal/strategy/codeartifact_http2_test.go b/internal/strategy/codeartifact_http2_test.go new file mode 100644 index 00000000..8fa9dd4f --- /dev/null +++ b/internal/strategy/codeartifact_http2_test.go @@ -0,0 +1,212 @@ +package strategy //nolint:testpackage // Verify private diagnostics without exposing parser internals. + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "strings" + "testing" + + "github.com/alecthomas/assert/v2" + "github.com/alecthomas/errors" + "go.opentelemetry.io/otel/codes" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "golang.org/x/net/http2" +) + +func TestCodeArtifactGoAwayDetails(t *testing.T) { + goAway := errors.New(`http2: server sent GOAWAY and closed the connection; LastStreamID=19999, ErrCode=NO_ERROR, debug="private-debug"`) + typed := http2.GoAwayError{LastStreamID: 19999, ErrCode: http2.ErrCodeNo, DebugData: "private-debug"} + for _, test := range []struct { + name string + err error + }{ + {name: "stdlib shape", err: goAway}, + {name: "xnet value", err: typed}, + {name: "xnet pointer", err: &typed}, + {name: "wrapped", err: errors.Errorf("private-wrapper: %w", goAway)}, + {name: "stack wrapped", err: errors.WithStack(goAway)}, + {name: "url wrapped", err: &url.Error{Op: "Get", URL: "https://private.example/?token=secret", Err: goAway}}, + {name: "joined first", err: errors.Join(goAway, io.EOF)}, + {name: "joined last", err: errors.Join(io.EOF, errors.Errorf("private-wrapper: %w", goAway))}, + } { + t.Run(test.name, func(t *testing.T) { + details, ok := codeArtifactGoAway(test.err) + assert.True(t, ok) + assert.Equal(t, codeArtifactGoAwayDetails{code: "NO_ERROR", lastStreamID: 19999}, details) + assert.Equal(t, "http2_goaway", codeArtifactBodyError(t.Context(), test.err)) + }) + } +} + +func TestCodeArtifactGoAwayCodesAndStreamIDs(t *testing.T) { + for code := http2.ErrCodeNo; code <= http2.ErrCodeHTTP11Required; code++ { + for _, streamID := range []uint32{0, 19999, 1<<31 - 1} { + t.Run(fmt.Sprintf("%s/%d", code, streamID), func(t *testing.T) { + err := http2.GoAwayError{LastStreamID: streamID, ErrCode: code} + details, ok := codeArtifactGoAway(err) + assert.True(t, ok) + assert.Equal(t, codeArtifactGoAwayDetails{code: code.String(), lastStreamID: streamID}, details) + }) + } + } + for _, code := range []http2.ErrCode{14, 255, 1<<32 - 1} { + details, ok := codeArtifactGoAway(http2.GoAwayError{LastStreamID: 19999, ErrCode: code}) + assert.False(t, ok) + assert.Equal(t, codeArtifactGoAwayDetails{}, details) + } +} + +func TestCodeArtifactGoAwayMalformedMetadata(t *testing.T) { + const valid = `http2: server sent GOAWAY and closed the connection; LastStreamID=19999, ErrCode=NO_ERROR, debug=""` + invalid := []string{ + "", + "GOAWAY", + "private-wrapper: " + valid, + strings.Replace(valid, "http2:", "HTTP2:", 1), + strings.Replace(valid, "LastStreamID=", "LastStreamID = ", 1), + strings.Replace(valid, ", ErrCode=", " ErrCode=", 1), + strings.Replace(valid, ", debug=", ", private=", 1), + strings.TrimSuffix(valid, `, debug=""`), + strings.TrimSuffix(valid, `"`), + strings.TrimSuffix(valid, `""`) + "private-debug", + valid + "private-suffix", + } + for _, streamID := range []string{"", "-1", "+1", "01", " 1", "1 ", "1.0", "0x1", "2147483648", "4294967295", strings.Repeat("9", 1024)} { + invalid = append(invalid, strings.Replace(valid, "19999", streamID, 1)) + } + for _, code := range []string{"", "0", "no_error", "NO_ERROR ", "UNKNOWN", "unknown error code 0xff", "NO_ERROR\nprivate", strings.Repeat("X", 1024)} { + invalid = append(invalid, strings.Replace(valid, "NO_ERROR", code, 1)) + } + for i, message := range invalid { + t.Run(strconv.Itoa(i), func(t *testing.T) { + details, ok := codeArtifactGoAway(errors.New(message)) + assert.False(t, ok) + assert.Equal(t, codeArtifactGoAwayDetails{}, details) + assert.Equal(t, "transport_error", codeArtifactBodyError(t.Context(), errors.New(message))) + }) + } + for _, err := range []error{ + nil, + codeArtifactDiagnosticWrapperError{message: valid, inner: errors.New("not GOAWAY")}, + errors.New(`private-debug="` + valid + `"`), + http2.StreamError{StreamID: 19999, Code: http2.ErrCodeCancel}, + } { + details, ok := codeArtifactGoAway(err) + assert.False(t, ok) + assert.Equal(t, codeArtifactGoAwayDetails{}, details) + } +} + +func TestCodeArtifactGoAwayBoundedTraversal(t *testing.T) { + goAway := http2.GoAwayError{LastStreamID: 19999, ErrCode: http2.ErrCodeNo} + var wrapped error = goAway + for range 31 { + wrapped = codeArtifactDiagnosticWrapperError{inner: wrapped} + } + _, ok := codeArtifactGoAway(wrapped) + assert.True(t, ok) + wrapped = codeArtifactDiagnosticWrapperError{inner: wrapped} + _, ok = codeArtifactGoAway(wrapped) + assert.False(t, ok) + + cycle := &codeArtifactDiagnosticCycleError{} + _, ok = codeArtifactGoAway(cycle) + assert.False(t, ok) + assert.Equal(t, 32, cycle.visits) + + joined := make([]error, 0, 33) + for range 32 { + joined = append(joined, errors.New("private")) + } + joined = append(joined, goAway) + _, ok = codeArtifactGoAway(errors.Join(joined...)) + assert.False(t, ok) +} + +func TestCodeArtifactGoAwayLogPrivacyAndCancellation(t *testing.T) { + debug := "https://private.example/token=secret\npayload\x00\"; LastStreamID=9000, ErrCode=CANCEL" + for _, test := range []struct { + name string + cancel bool + join bool + debugData string + }{ + {name: "peer debug", debugData: debug}, + {name: "large peer debug", debugData: strings.Repeat(debug, 16384)}, + {name: "canceled request", cancel: true, debugData: debug}, + {name: "joined cancellation", join: true, debugData: debug}, + } { + t.Run(test.name, func(t *testing.T) { + var logs bytes.Buffer + recorded := &recordingCodeArtifactMetrics{} + strategy := &CodeArtifact{ + logger: slog.New(slog.NewJSONHandler(&logs, nil)), + target: &url.URL{Scheme: "https", Host: "private.example"}, + metric: recorded, + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + spans := tracetest.NewSpanRecorder() + provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(spans)) + t.Cleanup(func() { assert.NoError(t, provider.Shutdown(context.Background())) }) + ctx, span := provider.Tracer("goaway-test").Start(ctx, "download") + if test.cancel { + cancel() + } + var fault error = http2.GoAwayError{LastStreamID: 19999, ErrCode: http2.ErrCodeNo, DebugData: test.debugData} + if test.join { + fault = errors.Join(fault, context.Canceled) + } + fault = &url.Error{Op: "Get", URL: "https://private.example/?token=secret", Err: fault} + request := httptest.NewRequest(http.MethodGet, "/pypi/private/simple/private/?token=secret", nil).WithContext(ctx) + reader, writer := io.Pipe() + t.Cleanup(func() { assert.NoError(t, reader.Close()) }) + assert.NoError(t, writer.CloseWithError(fault)) + assertCodeArtifactStreamAborted(t, func() { strategy.streamOriginBody(httptest.NewRecorder(), request, reader) }) + span.End() + + kind, status := "http2_goaway", codes.Error + if test.cancel || test.join { + kind, status = "client_canceled", codes.Unset + } + assert.Equal(t, [][3]string{{"source_read", kind, "pypi"}}, recorded.streamFailures) + assert.Equal(t, status, spans.Ended()[0].Status().Code) + var entry map[string]any + assert.NoError(t, json.Unmarshal(logs.Bytes(), &entry)) + assert.Equal[any](t, kind, entry["error_kind"]) + assert.Equal[any](t, test.cancel || test.join, entry["client_canceled"]) + assert.Equal[any](t, "NO_ERROR", entry["http2_error_code"]) + assert.Equal[any](t, float64(19999), entry["http2_last_stream_id"]) + for _, private := range []string{"private", "secret", "payload", "debug", "9000", "CANCEL"} { + assert.False(t, strings.Contains(logs.String(), private)) + } + }) + } +} + +type codeArtifactDiagnosticWrapperError struct { + message string + inner error +} + +func (e codeArtifactDiagnosticWrapperError) Error() string { return e.message } +func (e codeArtifactDiagnosticWrapperError) Unwrap() error { return e.inner } + +type codeArtifactDiagnosticCycleError struct { + visits int +} + +func (*codeArtifactDiagnosticCycleError) Error() string { panic("wrapper text must not be inspected") } +func (e *codeArtifactDiagnosticCycleError) Unwrap() error { + e.visits++ + return e +} diff --git a/internal/strategy/codeartifact_stream.go b/internal/strategy/codeartifact_stream.go index 761babd6..947ff63e 100644 --- a/internal/strategy/codeartifact_stream.go +++ b/internal/strategy/codeartifact_stream.go @@ -64,7 +64,11 @@ func (c *CodeArtifact) abortStream(w http.ResponseWriter, r *http.Request, if c.metric != nil { c.metric.recordStreamFailure(context.WithoutCancel(r.Context()), stage, kind, format) } - c.logger.ErrorContext(r.Context(), "CodeArtifact response stream failed", + logger := c.logger + if goAway, ok := codeArtifactGoAway(err); ok { + logger = logger.With("http2_error_code", goAway.code, "http2_last_stream_id", goAway.lastStreamID) + } + logger.ErrorContext(r.Context(), "CodeArtifact response stream failed", "source", "origin", "stage", stage, "error_kind", kind, "format", format, "bytes_read", reader.bytes, "bytes_written", written, "expected_bytes", expected, "duration_ms", time.Since(started).Milliseconds(), "client_canceled", clientCanceled) @@ -84,12 +88,17 @@ func codeArtifactBodyError(ctx context.Context, err error) string { case errors.Is(err, io.ErrShortWrite): return "short_write" case errors.Is(err, context.Canceled): - return "client_canceled" + return codeArtifactStatusClientCanceled case errors.Is(err, syscall.ECONNRESET): return "connection_reset" case errors.Is(err, syscall.EPIPE): return "broken_pipe" + case errors.Is(ctx.Err(), context.Canceled): + return codeArtifactStatusClientCanceled default: + if _, ok := codeArtifactGoAway(err); ok { + return "http2_goaway" + } return originTransportStatus(ctx, err) } } diff --git a/internal/strategy/codeartifact_uv_integration_test.go b/internal/strategy/codeartifact_uv_integration_test.go new file mode 100644 index 00000000..7f524693 --- /dev/null +++ b/internal/strategy/codeartifact_uv_integration_test.go @@ -0,0 +1,172 @@ +//go:build integration + +package strategy //nolint:testpackage // Run an actual resolver through the real CodeArtifact strategy. + +import ( + "bytes" + "context" + "encoding/pem" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/alecthomas/assert/v2" +) + +func (o *goAwayOrigin) recover() { + o.mu.Lock() + defer o.mu.Unlock() + o.failures = 0 +} + +func codeArtifactUVExecutable(t *testing.T) string { + t.Helper() + executable := os.Getenv("CACHEW_TEST_UV") + if executable == "" { + var err error + executable, err = exec.LookPath("uv") + if err != nil { + t.Fatal("integration tests require uv 0.13.4 on PATH or CACHEW_TEST_UV pointing to that executable") + } + } + absolute, err := filepath.Abs(executable) + assert.NoError(t, err) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + output, err := exec.CommandContext(ctx, absolute, "--version").CombinedOutput() + assert.NoError(t, err, "cannot run pinned uv executable: %s", output) + version := strings.Fields(string(output)) + if len(version) < 2 || version[0] != "uv" || version[1] != "0.13.4" { + t.Fatalf("integration tests require uv 0.13.4, got %q from %s", strings.TrimSpace(string(output)), absolute) + } + return absolute +} + +type codeArtifactUVResult struct { + stdout string + stderr string + err error +} + +func compileCodeArtifactUV(t *testing.T, ctx context.Context, executable string, proxy *goAwayProxy) codeArtifactUVResult { + t.Helper() + home := t.TempDir() + ctx, cancel := context.WithTimeout(ctx, 20*time.Second) + defer cancel() + command := exec.CommandContext(ctx, executable, + "--no-config", "pip", "compile", "-", + "--default-index", proxy.server.URL+proxy.prefix+"/pypi/repository/simple/", + "--python-version", "3.12", "--python-platform", "x86_64-unknown-linux-gnu", + "--no-python-downloads", "--only-binary", ":all:", + "--no-header", "--no-annotate", "--no-progress", "--color", "never", + ) + command.Dir = home + command.Env = []string{ + "PATH=" + os.Getenv("PATH"), + "HOME=" + home, + "XDG_CONFIG_HOME=" + filepath.Join(home, "config"), + "XDG_CACHE_HOME=" + filepath.Join(home, "cache"), + "UV_CACHE_DIR=" + filepath.Join(home, "uv-cache"), + "UV_PYTHON_DOWNLOADS=never", + "HTTP_PROXY=http://127.0.0.1:1", + "HTTPS_PROXY=http://127.0.0.1:1", + "NO_PROXY=127.0.0.1,localhost", + } + certificate := filepath.Join(home, "fixture-ca.pem") + assert.NoError(t, os.WriteFile(certificate, pem.EncodeToMemory(&pem.Block{ + Type: "CERTIFICATE", Bytes: proxy.ca, + }), 0o600)) + command.Env = append(command.Env, "SSL_CERT_FILE="+certificate) + command.Stdin = strings.NewReader("cachew-goaway\n") + var stdout, stderr bytes.Buffer + command.Stdout, command.Stderr = &stdout, &stderr + err := command.Run() + if ctx.Err() != nil { + t.Fatalf("uv exceeded the bounded test deadline: %v\n%s", ctx.Err(), stderr.String()) + } + return codeArtifactUVResult{stdout: stdout.String(), stderr: stderr.String(), err: err} +} + +func assertCodeArtifactUVResolved(t *testing.T, result codeArtifactUVResult) { + t.Helper() + assert.NoError(t, result.err, "uv failed to resolve the fixture package:\n%s", result.stderr) + assert.Equal(t, "cachew-goaway==1.2.3\n", result.stdout) +} + +func assertCodeArtifactUVRequests(t *testing.T, proxy *goAwayProxy, paths ...string) { + t.Helper() + expectedProtocol := 1 + if proxy.server.EnableHTTP2 { + expectedProtocol = 2 + } + for _, path := range paths { + assert.Equal(t, expectedProtocol, proxy.wait(t, path), "uv must exercise the expected downstream protocol") + } +} + +func TestCodeArtifactUVGoAwayRecovery(t *testing.T) { + executable := codeArtifactUVExecutable(t) + ctx, cancel := context.WithTimeout(t.Context(), 45*time.Second) + defer cancel() + for _, downstream := range []string{"http1", "http2"} { + t.Run(downstream, func(t *testing.T) { + origin := newGoAwayOrigin(t, 1) + proxy := newGoAwayProxy(t, origin, downstream) + + assertCodeArtifactUVResolved(t, compileCodeArtifactUV(t, ctx, executable, proxy)) + assertCodeArtifactUVRequests(t, proxy, goAwayIndexPath, goAwayIndexPath, goAwayWheelPath+".metadata") + requests := origin.snapshot() + assert.Equal(t, 3, len(requests)) + assert.Equal(t, goAwayOriginRequest{path: goAwayIndexPath, streamID: 1, failed: true}, requests[0]) + assert.Equal(t, goAwayOriginRequest{path: goAwayIndexPath, streamID: 1}, requests[1]) + assert.Equal(t, goAwayWheelPath+".metadata", requests[2].path) + assertGoAwayDiagnostics(t, proxy, origin, 1) + + assertCodeArtifactUVResolved(t, compileCodeArtifactUV(t, ctx, executable, proxy)) + assertCodeArtifactUVRequests(t, proxy, goAwayIndexPath, goAwayWheelPath+".metadata") + assert.Equal(t, requests, origin.snapshot(), "fresh uv cache must use the complete Cachew index and metadata fills") + }) + } +} + +func TestCodeArtifactUVGoAwayRetryExhaustion(t *testing.T) { + executable := codeArtifactUVExecutable(t) + ctx, cancel := context.WithTimeout(t.Context(), 45*time.Second) + defer cancel() + for _, downstream := range []string{"http1", "http2"} { + t.Run(downstream, func(t *testing.T) { + origin := newGoAwayOrigin(t, 100) + proxy := newGoAwayProxy(t, origin, downstream) + + failure := compileCodeArtifactUV(t, ctx, executable, proxy) + assert.Error(t, failure.err, "repeated origin truncation must fail resolution") + assert.Equal(t, "", failure.stdout, "failed resolution must not emit a successful lock") + for range 4 { + assertCodeArtifactUVRequests(t, proxy, goAwayIndexPath) + } + requests := origin.snapshot() + assert.Equal(t, 4, len(requests), "uv's default three retries must terminate after four failed attempts") + for _, request := range requests { + assert.Equal(t, goAwayIndexPath, request.path) + assert.True(t, request.failed) + } + assertGoAwayDiagnostics(t, proxy, origin, 4) + + origin.recover() + assertCodeArtifactUVResolved(t, compileCodeArtifactUV(t, ctx, executable, proxy)) + assertCodeArtifactUVRequests(t, proxy, goAwayIndexPath, goAwayWheelPath+".metadata") + requests = origin.snapshot() + assert.Equal(t, 6, len(requests), "recovery must fetch a fresh index and metadata, not a partial cached response") + assert.Equal(t, goAwayIndexPath, requests[4].path) + assert.False(t, requests[4].failed) + assert.Equal(t, goAwayWheelPath+".metadata", requests[5].path) + + assertCodeArtifactUVResolved(t, compileCodeArtifactUV(t, ctx, executable, proxy)) + assertCodeArtifactUVRequests(t, proxy, goAwayIndexPath, goAwayWheelPath+".metadata") + assert.Equal(t, requests, origin.snapshot(), "only the complete recovery response may be a warm Cachew hit") + }) + } +} From 694dd755bfb106ef8b26525f9cd909a396f14f3b Mon Sep 17 00:00:00 2001 From: Ravi Agarwal Date: Sat, 7 Nov 2026 07:54:39 +0000 Subject: [PATCH 2/2] docs(codeartifact): scope GOAWAY recovery guarantees --- README.md | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 65108682..4dffa17a 100644 --- a/README.md +++ b/README.md @@ -288,12 +288,12 @@ Failures before response headers use bounded statuses in those same metrics: `transport_connection`, or `transport_error` for other failures. Logs include only that status, never the origin URL or signed redirect URL. -Incomplete origin bodies abort the downstream transfer rather than completing a -partial response. Recognized HTTP/2 GOAWAY body failures use the bounded -`http2_goaway` error kind; stream-failure logs also include `http2_error_code` and -`http2_last_stream_id`. Those details are not metric labels, and peer-provided -debug text is never logged. Unknown or malformed GOAWAY metadata retains the -generic transport classification. +Incomplete streamed origin bodies abort the downstream transfer rather than +completing a partial response. Recognized HTTP/2 GOAWAY body errors use the bounded +`http2_goaway` classification, subject to existing error and cancellation precedence. +Stream-failure logs may also include `http2_error_code` and `http2_last_stream_id`; +these fields are not metric labels, and these logs omit peer-provided debug text. +Unknown or malformed GOAWAY metadata retains the generic transport classification. Authorization tokens are reused in-process. The default 12-hour token enters proactive refresh one hour before expiration; one request performs each refresh