From 13f4eed971ae4ef37fed0d8becba7175e2aa14b6 Mon Sep 17 00:00:00 2001 From: Christoph Leygraf Date: Sun, 1 Nov 2026 18:14:09 +0100 Subject: [PATCH 1/4] Don't count a 4xx answer as a registry failure in the breaker The circuit breaker counted every non-nil fetch error as a failure except a 404, an artifact over the size limit and a request held by rate limiting. That conflates a registry that answered with one that is down: a 403 policy denial, a 401 or a 409 is the registry replying, and because the breaker is keyed by host, a run of refusals from one host opens the breaker for every registry that host serves. Return the status as a typed *StatusError from the fetcher instead of a formatted string, and exempt any 4xx StatusError in the breaker, so only transport errors, rate limiting and 5xx responses count as failures. The messages are unchanged, so callers that read the status out of them keep working. Head gets the same exemption. It never returned ErrArtifactTooLarge, which the shared predicate now also ignores. --- README.md | 2 + fetch/circuit_breaker.go | 40 +++-- fetch/circuit_breaker_client_error_test.go | 168 +++++++++++++++++++++ fetch/fetcher.go | 4 +- fetch/status_error.go | 21 +++ 5 files changed, 224 insertions(+), 11 deletions(-) create mode 100644 fetch/circuit_breaker_client_error_test.go create mode 100644 fetch/status_error.go diff --git a/README.md b/README.md index 79146e0..59ea5f2 100644 --- a/README.md +++ b/README.md @@ -508,6 +508,8 @@ When both `WithAuthFunc` and `FetchWithHeaders` set the same header, `WithAuthFu Wrap a fetcher with per-host circuit breakers to avoid hammering a failing registry. The breaker trips once 5 failures land inside its rolling 10 second failure window, then retries with exponential backoff (30s initial, 5min max). While it is open, one request per backoff interval is let through as a probe and the rest fail with `ErrUpstreamDown` without contacting the registry; a probe that succeeds closes the breaker again. Retries never give up, so a breaker recovers no matter how long the registry stayed down. +A registry that answers is up, whatever it answered: a 4xx response such as a 404, a 403 policy denial or a 409 does not count as a failure, so a run of refused downloads cannot open the breaker. Only transport errors, rate limiting and 5xx responses count. + ```go f := fetch.NewFetcher() cbf := fetch.NewCircuitBreakerFetcher(f) diff --git a/fetch/circuit_breaker.go b/fetch/circuit_breaker.go index 6ec1dbf..b6faba0 100644 --- a/fetch/circuit_breaker.go +++ b/fetch/circuit_breaker.go @@ -131,10 +131,10 @@ func (cbf *CircuitBreakerFetcher) FetchWithHeaders(ctx context.Context, fetchURL var fetchErr error err := breaker.Call(func() error { artifact, fetchErr = cbf.fetcher.FetchWithHeaders(ctx, fetchURL, headers) - if errors.Is(fetchErr, ErrNotFound) || errors.Is(fetchErr, ErrArtifactTooLarge) || errors.Is(fetchErr, client.ErrUpstreamHeld) { - return nil + if countsAgainstBreaker(fetchErr) { + return fetchErr } - return fetchErr + return nil }, 0) if err != nil { @@ -158,10 +158,10 @@ func (cbf *CircuitBreakerFetcher) FetchObservedWithHeaders(ctx context.Context, var fetchErr error err := breaker.Call(func() error { artifact, fetchErr = cbf.fetcher.FetchObservedWithHeaders(ctx, fetchURL, headers) - if errors.Is(fetchErr, ErrNotFound) || errors.Is(fetchErr, ErrArtifactTooLarge) || errors.Is(fetchErr, client.ErrUpstreamHeld) { - return nil + if countsAgainstBreaker(fetchErr) { + return fetchErr } - return fetchErr + return nil }, 0) if err != nil { @@ -179,10 +179,10 @@ func (cbf *CircuitBreakerFetcher) Head(ctx context.Context, headURL string) (siz var headErr error err = breaker.Call(func() error { size, contentType, headErr = cbf.fetcher.Head(ctx, headURL) - if errors.Is(headErr, ErrNotFound) || errors.Is(headErr, client.ErrUpstreamHeld) { - return nil + if countsAgainstBreaker(headErr) { + return headErr } - return headErr + return nil }, 0) if err != nil { @@ -191,6 +191,28 @@ func (cbf *CircuitBreakerFetcher) Head(ctx context.Context, headURL string) (siz return size, contentType, headErr } +// countsAgainstBreaker reports whether err counts as a failure of the registry. +// +// A registry that answers is up, whatever it answered: a 404, any other 4xx +// status such as a 403 policy denial or a 409, an artifact over the size limit +// and a request held by rate limiting are all answers, so none of them count +// against the breaker. Only what says the registry could not answer does: +// transport errors, 5xx responses and rate limiting, which asks for the backoff +// the breaker exists to provide. +func countsAgainstBreaker(err error) bool { + if err == nil { + return false + } + if errors.Is(err, ErrNotFound) || errors.Is(err, ErrArtifactTooLarge) || errors.Is(err, client.ErrUpstreamHeld) { + return false + } + var status *StatusError + if errors.As(err, &status) && status.Code >= 400 && status.Code < 500 { + return false + } + return true +} + // breakerError maps the breaker's own open-circuit error onto ErrUpstreamDown, // so that every refusal to contact a registry reports the same way to callers, // and passes errors from the fetch itself through untouched. diff --git a/fetch/circuit_breaker_client_error_test.go b/fetch/circuit_breaker_client_error_test.go new file mode 100644 index 0000000..ca5a876 --- /dev/null +++ b/fetch/circuit_breaker_client_error_test.go @@ -0,0 +1,168 @@ +package fetch + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" +) + +// A 4xx answer is the registry replying, not failing, so the breaker must stay +// closed however many of them arrive, and every answer must reach the caller. +func TestCircuitBreakerClientErrorsDoNotTrip(t *testing.T) { + const body = "denied by policy" + + for _, code := range []int{ + http.StatusBadRequest, + http.StatusUnauthorized, + http.StatusForbidden, + http.StatusConflict, + http.StatusGone, + } { + t.Run(http.StatusText(code), func(t *testing.T) { + hits := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits++ + w.WriteHeader(code) + _, _ = w.Write([]byte(body)) + })) + defer server.Close() + + cbFetcher := NewCircuitBreakerFetcher(NewFetcher(WithMaxRetries(0), WithBaseDelay(0))) + + // Well past the threshold. + const attempts = 20 + for range attempts { + _, err := cbFetcher.Fetch(context.Background(), server.URL+"/artifact.tar.gz") + var status *StatusError + if !errors.As(err, &status) { + t.Fatalf("want a StatusError passed through, got %v", err) + } + if status.Code != code || status.Body != body { + t.Fatalf("got status %d body %q, want %d body %q", status.Code, status.Body, code, body) + } + } + + if hits != attempts { + t.Errorf("breaker swallowed requests: want %d upstream hits, got %d", attempts, hits) + } + for registry, state := range cbFetcher.GetBreakerState() { + if state != "closed" { + t.Errorf("breaker for %s is %s, want closed", registry, state) + } + } + }) + } +} + +func TestCircuitBreakerHeadClientErrorDoesNotTrip(t *testing.T) { + hits := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits++ + w.WriteHeader(http.StatusForbidden) + })) + defer server.Close() + + cbFetcher := NewCircuitBreakerFetcher(NewFetcher(WithMaxRetries(0), WithBaseDelay(0))) + + const attempts = 20 + for range attempts { + _, _, err := cbFetcher.Head(context.Background(), server.URL+"/artifact.tar.gz") + var status *StatusError + if !errors.As(err, &status) || status.Code != http.StatusForbidden { + t.Fatalf("want a 403 StatusError passed through, got %v", err) + } + } + + if hits != attempts { + t.Errorf("breaker swallowed requests: want %d upstream hits, got %d", attempts, hits) + } + for registry, state := range cbFetcher.GetBreakerState() { + if state != "closed" { + t.Errorf("breaker for %s is %s, want closed", registry, state) + } + } +} + +// A registry that cannot answer still trips the breaker, so later requests are +// short-circuited instead of hammering it. +func TestCircuitBreakerServerErrorsStillTrip(t *testing.T) { + hits := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits++ + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer server.Close() + + cbFetcher := NewCircuitBreakerFetcher(NewFetcher(WithMaxRetries(0), WithBaseDelay(0))) + + const attempts = 20 + for range attempts { + if _, err := cbFetcher.Fetch(context.Background(), server.URL+"/artifact.tar.gz"); !errors.Is(err, ErrUpstreamDown) { + t.Fatalf("want ErrUpstreamDown, got %v", err) + } + } + + if hits >= attempts { + t.Errorf("breaker never opened: %d upstream hits for %d requests", hits, attempts) + } + for registry, state := range cbFetcher.GetBreakerState() { + if state != "open" { + t.Errorf("breaker for %s is %s, want open", registry, state) + } + } +} + +func TestCircuitBreakerTransportErrorsStillTrip(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + deadURL := server.URL + "/artifact.tar.gz" + server.Close() + + cbFetcher := NewCircuitBreakerFetcher(NewFetcher(WithMaxRetries(0), WithBaseDelay(0))) + + const attempts = 20 + for range attempts { + if _, err := cbFetcher.Fetch(context.Background(), deadURL); err == nil { + t.Fatal("want an error from a registry that cannot be reached, got nil") + } + } + + // Open now, so the next request fails without dialling. + if _, err := cbFetcher.Fetch(context.Background(), deadURL); !errors.Is(err, ErrUpstreamDown) { + t.Errorf("want ErrUpstreamDown once the breaker is open, got %v", err) + } + for registry, state := range cbFetcher.GetBreakerState() { + if state != "open" { + t.Errorf("breaker for %s is %s, want open", registry, state) + } + } +} + +// Callers that read the status out of the message keep working. +func TestStatusErrorMessage(t *testing.T) { + tests := []struct { + name string + err *StatusError + want string + }{ + { + name: "fetch keeps the status and body", + err: &StatusError{Code: http.StatusForbidden, Body: "denied by policy"}, + want: "unexpected status 403: denied by policy", + }, + { + name: "head has no body", + err: &StatusError{Code: http.StatusForbidden}, + want: "unexpected status 403", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.err.Error(); got != tt.want { + t.Errorf("message = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/fetch/fetcher.go b/fetch/fetcher.go index 74d656c..f847c67 100644 --- a/fetch/fetcher.go +++ b/fetch/fetcher.go @@ -290,7 +290,7 @@ func (f *Fetcher) doFetch(ctx context.Context, url string, headers http.Header, default: body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrBodySize)) _ = resp.Body.Close() - return nil, nil, fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body)) + return nil, nil, &StatusError{Code: resp.StatusCode, Body: string(body)} } } @@ -320,7 +320,7 @@ func (f *Fetcher) Head(ctx context.Context, url string) (size int64, contentType return 0, "", ErrNotFound } if resp.StatusCode != http.StatusOK { - return 0, "", fmt.Errorf("unexpected status %d", resp.StatusCode) + return 0, "", &StatusError{Code: resp.StatusCode} } size = -1 diff --git a/fetch/status_error.go b/fetch/status_error.go new file mode 100644 index 0000000..80a76c0 --- /dev/null +++ b/fetch/status_error.go @@ -0,0 +1,21 @@ +package fetch + +import "fmt" + +// StatusError reports an upstream response whose status the fetcher does not +// handle itself: not a 404, not rate limiting and not a server error. The +// registry answered, so the status and the body it sent are the artifact's +// answer rather than a failure to reach the registry. +type StatusError struct { + Code int + // Body is what the upstream sent with the status, truncated to + // maxErrBodySize. Head reads no body, so it is empty for a Head request. + Body string +} + +func (e *StatusError) Error() string { + if e.Body == "" { + return fmt.Sprintf("unexpected status %d", e.Code) + } + return fmt.Sprintf("unexpected status %d: %s", e.Code, e.Body) +} From d4bb933573ebe672d4305792432d03a2285c5518 Mon Sep 17 00:00:00 2001 From: Christoph Leygraf Date: Thu, 5 Nov 2026 07:39:31 +0100 Subject: [PATCH 2/4] Address review: Head sentinels, StatusError content type, test counters Head turned every non-200, non-404 status into a StatusError, so a 429 from Head was a 4xx StatusError and countsAgainstBreaker ignored it, where it counted before this branch. Map 429 to ErrRateLimited and 5xx to ErrUpstreamDown there as doFetch does, and cover both in the breaker tests. The StatusError doc comment, which excludes rate limiting and server errors, now holds for Head as well. StatusError carries the upstream Content-Type, so a caller can forward the body as the upstream labelled it rather than guessing the type from the first byte. git-pkgs/proxy#650 needs that to serve a JSON report that the fetcher truncated to maxErrBodySize without handing npm invalid JSON. The hits counters in the breaker tests are atomic.Int32: the handler goroutine writes them and the test goroutine reads them, and CI runs with -race. countsAgainstBreaker keeps one line of doc comment; the rest repeated the code and the README paragraph. --- fetch/circuit_breaker.go | 9 +-- fetch/circuit_breaker_client_error_test.go | 79 ++++++++++++++++++---- fetch/fetcher.go | 14 +++- fetch/status_error.go | 3 + 4 files changed, 83 insertions(+), 22 deletions(-) diff --git a/fetch/circuit_breaker.go b/fetch/circuit_breaker.go index b6faba0..13316af 100644 --- a/fetch/circuit_breaker.go +++ b/fetch/circuit_breaker.go @@ -191,14 +191,7 @@ func (cbf *CircuitBreakerFetcher) Head(ctx context.Context, headURL string) (siz return size, contentType, headErr } -// countsAgainstBreaker reports whether err counts as a failure of the registry. -// -// A registry that answers is up, whatever it answered: a 404, any other 4xx -// status such as a 403 policy denial or a 409, an artifact over the size limit -// and a request held by rate limiting are all answers, so none of them count -// against the breaker. Only what says the registry could not answer does: -// transport errors, 5xx responses and rate limiting, which asks for the backoff -// the breaker exists to provide. +// countsAgainstBreaker reports whether err means the registry could not answer. func countsAgainstBreaker(err error) bool { if err == nil { return false diff --git a/fetch/circuit_breaker_client_error_test.go b/fetch/circuit_breaker_client_error_test.go index ca5a876..1bd539f 100644 --- a/fetch/circuit_breaker_client_error_test.go +++ b/fetch/circuit_breaker_client_error_test.go @@ -5,6 +5,7 @@ import ( "errors" "net/http" "net/http/httptest" + "sync/atomic" "testing" ) @@ -12,6 +13,7 @@ import ( // closed however many of them arrive, and every answer must reach the caller. func TestCircuitBreakerClientErrorsDoNotTrip(t *testing.T) { const body = "denied by policy" + const contentType = "application/json" for _, code := range []int{ http.StatusBadRequest, @@ -21,9 +23,10 @@ func TestCircuitBreakerClientErrorsDoNotTrip(t *testing.T) { http.StatusGone, } { t.Run(http.StatusText(code), func(t *testing.T) { - hits := 0 + var hits atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - hits++ + hits.Add(1) + w.Header().Set("Content-Type", contentType) w.WriteHeader(code) _, _ = w.Write([]byte(body)) })) @@ -42,10 +45,13 @@ func TestCircuitBreakerClientErrorsDoNotTrip(t *testing.T) { if status.Code != code || status.Body != body { t.Fatalf("got status %d body %q, want %d body %q", status.Code, status.Body, code, body) } + if status.ContentType != contentType { + t.Fatalf("got content type %q, want %q", status.ContentType, contentType) + } } - if hits != attempts { - t.Errorf("breaker swallowed requests: want %d upstream hits, got %d", attempts, hits) + if got := hits.Load(); got != attempts { + t.Errorf("breaker swallowed requests: want %d upstream hits, got %d", attempts, got) } for registry, state := range cbFetcher.GetBreakerState() { if state != "closed" { @@ -57,9 +63,9 @@ func TestCircuitBreakerClientErrorsDoNotTrip(t *testing.T) { } func TestCircuitBreakerHeadClientErrorDoesNotTrip(t *testing.T) { - hits := 0 + var hits atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - hits++ + hits.Add(1) w.WriteHeader(http.StatusForbidden) })) defer server.Close() @@ -75,8 +81,8 @@ func TestCircuitBreakerHeadClientErrorDoesNotTrip(t *testing.T) { } } - if hits != attempts { - t.Errorf("breaker swallowed requests: want %d upstream hits, got %d", attempts, hits) + if got := hits.Load(); got != attempts { + t.Errorf("breaker swallowed requests: want %d upstream hits, got %d", attempts, got) } for registry, state := range cbFetcher.GetBreakerState() { if state != "closed" { @@ -88,9 +94,9 @@ func TestCircuitBreakerHeadClientErrorDoesNotTrip(t *testing.T) { // A registry that cannot answer still trips the breaker, so later requests are // short-circuited instead of hammering it. func TestCircuitBreakerServerErrorsStillTrip(t *testing.T) { - hits := 0 + var hits atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - hits++ + hits.Add(1) w.WriteHeader(http.StatusServiceUnavailable) })) defer server.Close() @@ -104,8 +110,8 @@ func TestCircuitBreakerServerErrorsStillTrip(t *testing.T) { } } - if hits >= attempts { - t.Errorf("breaker never opened: %d upstream hits for %d requests", hits, attempts) + if got := hits.Load(); got >= attempts { + t.Errorf("breaker never opened: %d upstream hits for %d requests", got, attempts) } for registry, state := range cbFetcher.GetBreakerState() { if state != "open" { @@ -114,6 +120,55 @@ func TestCircuitBreakerServerErrorsStillTrip(t *testing.T) { } } +// Head reports the statuses that mean "cannot answer" with the same sentinels +// as Fetch, so a rate limit or a server error still counts against the breaker. +func TestCircuitBreakerHeadUnavailableStillTrips(t *testing.T) { + tests := []struct { + name string + code int + want error + }{ + {name: "rate limited", code: http.StatusTooManyRequests, want: ErrRateLimited}, + {name: "server error", code: http.StatusServiceUnavailable, want: ErrUpstreamDown}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var hits atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + w.WriteHeader(tt.code) + })) + defer server.Close() + + cbFetcher := NewCircuitBreakerFetcher(NewFetcher(WithMaxRetries(0), WithBaseDelay(0))) + + if _, _, err := cbFetcher.Head(context.Background(), server.URL+"/artifact.tar.gz"); !errors.Is(err, tt.want) { + t.Fatalf("first Head = %v, want %v", err, tt.want) + } + + // Well past the threshold. Once the breaker is open the answer is + // ErrUpstreamDown for both statuses. + const attempts = 20 + for range attempts { + _, _, err := cbFetcher.Head(context.Background(), server.URL+"/artifact.tar.gz") + if !errors.Is(err, tt.want) && !errors.Is(err, ErrUpstreamDown) { + t.Fatalf("Head = %v, want %v or ErrUpstreamDown once the breaker is open", err, tt.want) + } + } + + if got := hits.Load(); got >= attempts { + t.Errorf("breaker never opened: %d upstream hits for %d requests", got, attempts) + } + for registry, state := range cbFetcher.GetBreakerState() { + if state != "open" { + t.Errorf("breaker for %s is %s, want open", registry, state) + } + } + }) + } +} + func TestCircuitBreakerTransportErrorsStillTrip(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) deadURL := server.URL + "/artifact.tar.gz" diff --git a/fetch/fetcher.go b/fetch/fetcher.go index f847c67..a60307a 100644 --- a/fetch/fetcher.go +++ b/fetch/fetcher.go @@ -290,7 +290,11 @@ func (f *Fetcher) doFetch(ctx context.Context, url string, headers http.Header, default: body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrBodySize)) _ = resp.Body.Close() - return nil, nil, &StatusError{Code: resp.StatusCode, Body: string(body)} + return nil, nil, &StatusError{ + Code: resp.StatusCode, + Body: string(body), + ContentType: resp.Header.Get("Content-Type"), + } } } @@ -319,8 +323,14 @@ func (f *Fetcher) Head(ctx context.Context, url string) (size int64, contentType if resp.StatusCode == http.StatusNotFound { return 0, "", ErrNotFound } + if resp.StatusCode == http.StatusTooManyRequests { + return 0, "", ErrRateLimited + } + if resp.StatusCode >= serverErrThreshold { + return 0, "", ErrUpstreamDown + } if resp.StatusCode != http.StatusOK { - return 0, "", &StatusError{Code: resp.StatusCode} + return 0, "", &StatusError{Code: resp.StatusCode, ContentType: resp.Header.Get("Content-Type")} } size = -1 diff --git a/fetch/status_error.go b/fetch/status_error.go index 80a76c0..c313002 100644 --- a/fetch/status_error.go +++ b/fetch/status_error.go @@ -11,6 +11,9 @@ type StatusError struct { // Body is what the upstream sent with the status, truncated to // maxErrBodySize. Head reads no body, so it is empty for a Head request. Body string + // ContentType is the Content-Type the upstream sent with the status, so a + // caller can forward the body as the upstream labelled it. + ContentType string } func (e *StatusError) Error() string { From 00a6aa9795aeb1c713c7ca959c2a63dfcc29b2de Mon Sep 17 00:00:00 2001 From: Christoph Leygraf Date: Fri, 6 Nov 2026 16:15:56 +0100 Subject: [PATCH 3/4] Exempt every StatusError from the breaker and cover 304 doFetch and Head report rate limiting and server errors as sentinels before building a StatusError, so a StatusError always means the registry answered. The 400..499 range check let a 304 from a conditional request count as a failure and could open the host-wide breaker. Also cover the observed fetch path, which shares the predicate. --- fetch/circuit_breaker.go | 2 +- fetch/circuit_breaker_client_error_test.go | 67 ++++++++++++++++++++++ fetch/status_error.go | 4 +- 3 files changed, 69 insertions(+), 4 deletions(-) diff --git a/fetch/circuit_breaker.go b/fetch/circuit_breaker.go index 13316af..0d5941b 100644 --- a/fetch/circuit_breaker.go +++ b/fetch/circuit_breaker.go @@ -200,7 +200,7 @@ func countsAgainstBreaker(err error) bool { return false } var status *StatusError - if errors.As(err, &status) && status.Code >= 400 && status.Code < 500 { + if errors.As(err, &status) { return false } return true diff --git a/fetch/circuit_breaker_client_error_test.go b/fetch/circuit_breaker_client_error_test.go index 1bd539f..fb3d594 100644 --- a/fetch/circuit_breaker_client_error_test.go +++ b/fetch/circuit_breaker_client_error_test.go @@ -60,6 +60,73 @@ func TestCircuitBreakerClientErrorsDoNotTrip(t *testing.T) { } }) } + + // A 304 answers a conditional request, so it is the registry answering too. + // The server sends neither a body nor a Content-Type with it. + t.Run(http.StatusText(http.StatusNotModified), func(t *testing.T) { + var hits atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + w.WriteHeader(http.StatusNotModified) + })) + defer server.Close() + + cbFetcher := NewCircuitBreakerFetcher(NewFetcher(WithMaxRetries(0), WithBaseDelay(0))) + headers := http.Header{"If-None-Match": []string{`"etag"`}} + + const attempts = 20 + for range attempts { + _, err := cbFetcher.FetchWithHeaders(context.Background(), server.URL+"/artifact.tar.gz", headers) + var status *StatusError + if !errors.As(err, &status) { + t.Fatalf("want a StatusError passed through, got %v", err) + } + if status.Code != http.StatusNotModified || status.Body != "" || status.ContentType != "" { + t.Fatalf("got status %d body %q content type %q, want 304 without body or content type", + status.Code, status.Body, status.ContentType) + } + } + + if got := hits.Load(); got != attempts { + t.Errorf("breaker swallowed requests: want %d upstream hits, got %d", attempts, got) + } + for registry, state := range cbFetcher.GetBreakerState() { + if state != "closed" { + t.Errorf("breaker for %s is %s, want closed", registry, state) + } + } + }) +} + +// The observed fetch path shares the predicate, so a repeated 4xx must reach +// the caller there as well and leave the breaker closed. +func TestCircuitBreakerObservedClientErrorsDoNotTrip(t *testing.T) { + var hits atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + w.WriteHeader(http.StatusForbidden) + })) + defer server.Close() + + cbFetcher := NewCircuitBreakerFetcher(NewFetcher(WithMaxRetries(0), WithBaseDelay(0))) + + const attempts = 20 + for range attempts { + _, err := cbFetcher.FetchObservedWithHeaders(context.Background(), server.URL+"/artifact.tar.gz", nil) + var status *StatusError + if !errors.As(err, &status) || status.Code != http.StatusForbidden { + t.Fatalf("want a 403 StatusError passed through, got %v", err) + } + } + + if got := hits.Load(); got != attempts { + t.Errorf("breaker swallowed requests: want %d upstream hits, got %d", attempts, got) + } + for registry, state := range cbFetcher.GetBreakerState() { + if state != "closed" { + t.Errorf("breaker for %s is %s, want closed", registry, state) + } + } } func TestCircuitBreakerHeadClientErrorDoesNotTrip(t *testing.T) { diff --git a/fetch/status_error.go b/fetch/status_error.go index c313002..37c888c 100644 --- a/fetch/status_error.go +++ b/fetch/status_error.go @@ -3,9 +3,7 @@ package fetch import "fmt" // StatusError reports an upstream response whose status the fetcher does not -// handle itself: not a 404, not rate limiting and not a server error. The -// registry answered, so the status and the body it sent are the artifact's -// answer rather than a failure to reach the registry. +// handle itself: not a 404, not rate limiting and not a server error. type StatusError struct { Code int // Body is what the upstream sent with the status, truncated to From f5e809f5be21cc63899c0fa7ec67d654a36049ed Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Sat, 7 Nov 2026 14:21:00 +0000 Subject: [PATCH 4/4] Simplify StatusError check in countsAgainstBreaker --- fetch/circuit_breaker.go | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/fetch/circuit_breaker.go b/fetch/circuit_breaker.go index 0d5941b..be68c7a 100644 --- a/fetch/circuit_breaker.go +++ b/fetch/circuit_breaker.go @@ -200,10 +200,7 @@ func countsAgainstBreaker(err error) bool { return false } var status *StatusError - if errors.As(err, &status) { - return false - } - return true + return !errors.As(err, &status) } // breakerError maps the breaker's own open-circuit error onto ErrUpstreamDown,