Let a tripped circuit breaker keep retrying instead of latching open

README.md

@@ -411,7 +411,7 @@ When both `WithAuthFunc` and `FetchWithHeaders` set the same header, `WithAuthFu
 
 ### Circuit breaker
 
-Wrap a fetcher with per-host circuit breakers to avoid hammering a failing registry. The breaker trips after 5 consecutive failures and resets with exponential backoff (30s initial, 5min max).
+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.
 
 ```go
 f := fetch.NewFetcher()

fetch/circuit_breaker.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/cenk/backoff"
+	"github.com/facebookgo/clock"
 	circuit "github.com/rubyist/circuitbreaker"
 )
 
@@ -25,6 +26,12 @@ type CircuitBreakerFetcher struct {
 	fetcher  *Fetcher
 	breakers map[string]*circuit.Breaker
 	mu       sync.RWMutex
+
+	// clock is the time source for the breakers and their backoff, though not
+	// for the failure-count window, which the breaker library keeps on the
+	// system clock. A nil clock means the system clock; tests replace it to
+	// advance time without waiting out a backoff interval.
+	clock clock.Clock
 }
 
 // NewCircuitBreakerFetcher creates a new circuit breaker wrapper for a fetcher.
@@ -53,16 +60,30 @@ func (cbf *CircuitBreakerFetcher) getBreaker(registry string) *circuit.Breaker {
 		return breaker
 	}
 
-	// Create new circuit breaker with exponential backoff
-	// Trips after 5 consecutive failures
+	breakerClock := cbf.clock
+	if breakerClock == nil {
+		breakerClock = clock.New()
+	}
+
+	// Create new circuit breaker with exponential backoff. It trips once
+	// cbThreshold failures land inside the breaker's rolling failure window.
 	expBackoff := backoff.NewExponentialBackOff()
 	expBackoff.InitialInterval = cbInitialInterval
 	expBackoff.MaxInterval = cbMaxInterval
 	expBackoff.Multiplier = 2.0
+	// Retry forever, which is what the breaker library itself defaults to.
+	// NewExponentialBackOff instead defaults MaxElapsedTime to 15 minutes, after
+	// which NextBackOff returns backoff.Stop and the breaker never half-opens
+	// again. Only a success resets the backoff, and the breaker no longer lets
+	// one through, so an outage lasting longer than MaxElapsedTime leaves the
+	// breaker open for the life of the process even after the registry recovers.
+	expBackoff.MaxElapsedTime = 0
+	expBackoff.Clock = breakerClock
 	expBackoff.Reset()
 
 	opts := &circuit.Options{
 		BackOff:    expBackoff,
+		Clock:      breakerClock,
 		ShouldTrip: circuit.ThresholdTripFunc(cbThreshold),
 	}
 	breaker = circuit.NewBreakerWithOptions(opts)
@@ -82,12 +103,8 @@ func (cbf *CircuitBreakerFetcher) FetchWithHeaders(ctx context.Context, fetchURL
 	registry := extractRegistry(fetchURL)
 	breaker := cbf.getBreaker(registry)
 
-	// Check if circuit is open
-	if !breaker.Ready() {
-		return nil, fmt.Errorf("circuit breaker open for registry %s: %w", registry, ErrUpstreamDown)
-	}
-
-	// Attempt fetch
+	// Attempt fetch. Call checks the breaker itself; checking it here as well
+	// would spend the probe this call is about to make. See breakerError.
 	var artifact *Artifact
 	var fetchErr error
 	err := breaker.Call(func() error {
@@ -99,7 +116,7 @@ func (cbf *CircuitBreakerFetcher) FetchWithHeaders(ctx context.Context, fetchURL
 	}, 0)
 
 	if err != nil {
-		return nil, err
+		return nil, breakerError(registry, err)
 	}
 
 	return artifact, fetchErr
@@ -115,10 +132,6 @@ func (cbf *CircuitBreakerFetcher) FetchObservedWithHeaders(ctx context.Context,
 	registry := extractRegistry(fetchURL)
 	breaker := cbf.getBreaker(registry)
 
-	if !breaker.Ready() {
-		return nil, fmt.Errorf("circuit breaker open for registry %s: %w", registry, ErrUpstreamDown)
-	}
-
 	var artifact *ObservedArtifact
 	var fetchErr error
 	err := breaker.Call(func() error {
@@ -130,7 +143,7 @@ func (cbf *CircuitBreakerFetcher) FetchObservedWithHeaders(ctx context.Context,
 	}, 0)
 
 	if err != nil {
-		return nil, err
+		return nil, breakerError(registry, err)
 	}
 
 	return artifact, fetchErr
@@ -141,10 +154,6 @@ func (cbf *CircuitBreakerFetcher) Head(ctx context.Context, headURL string) (siz
 	registry := extractRegistry(headURL)
 	breaker := cbf.getBreaker(registry)
 
-	if !breaker.Ready() {
-		return 0, "", fmt.Errorf("circuit breaker open for registry %s: %w", registry, ErrUpstreamDown)
-	}
-
 	var headErr error
 	err = breaker.Call(func() error {
 		size, contentType, headErr = cbf.fetcher.Head(ctx, headURL)
@@ -155,11 +164,21 @@ func (cbf *CircuitBreakerFetcher) Head(ctx context.Context, headURL string) (siz
 	}, 0)
 
 	if err != nil {
-		return 0, "", err
+		return 0, "", breakerError(registry, err)
 	}
 	return size, contentType, headErr
 }
 
+// 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.
+func breakerError(registry string, err error) error {
+	if errors.Is(err, circuit.ErrBreakerOpen) {
+		return fmt.Errorf("circuit breaker open for registry %s: %w", registry, ErrUpstreamDown)
+	}
+	return err
+}
+
 // extractRegistry extracts a registry identifier from a URL for circuit breaker grouping.
 func extractRegistry(rawURL string) string {
 	// Parse URL and extract host for circuit breaker grouping

fetch/circuit_breaker_recovery_test.go

@@ -0,0 +1,124 @@
+package fetch
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/facebookgo/clock"
+)
+
+// A tripped breaker has to keep probing however long the registry stays down.
+// backoff.NewExponentialBackOff defaults MaxElapsedTime to 15 minutes, and once
+// NextBackOff returns backoff.Stop the breaker never half-opens again: it stays
+// open for the life of the process even after the registry comes back, because
+// only a success resets the backoff and no call gets through to produce one.
+func TestCircuitBreakerRecoversAfterProlongedOutage(t *testing.T) {
+	var down atomic.Bool
+	down.Store(true)
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		if down.Load() {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			return
+		}
+		_, _ = w.Write([]byte("ok"))
+	}))
+	defer server.Close()
+
+	mockClock := clock.NewMock()
+	cbFetcher := NewCircuitBreakerFetcher(NewFetcher(WithMaxRetries(0), WithBaseDelay(0)))
+	cbFetcher.clock = mockClock
+
+	ctx := context.Background()
+	artifactURL := server.URL + "/test.tar.gz"
+	registry := extractRegistry(artifactURL)
+
+	for range cbThreshold {
+		if _, err := cbFetcher.Fetch(ctx, artifactURL); err == nil {
+			t.Fatal("expected a fetch against a failing registry to fail")
+		}
+	}
+	if state := cbFetcher.GetBreakerState()[registry]; state != "open" {
+		t.Fatalf("breaker state = %q, want open after %d failures", state, cbThreshold)
+	}
+
+	// The outage lasts an hour, with a probe on every retry that keeps failing.
+	// Each step is longer than the 5 minute maximum backoff interval, so every
+	// step lets one probe through.
+	for range 6 {
+		mockClock.Add(10 * time.Minute)
+		if _, err := cbFetcher.Fetch(ctx, artifactURL); err == nil {
+			t.Fatal("expected a fetch against a failing registry to fail")
+		}
+	}
+
+	down.Store(false)
+	mockClock.Add(cbMaxInterval * 2)
+
+	artifact, err := cbFetcher.Fetch(ctx, artifactURL)
+	if err != nil {
+		t.Fatalf("fetch after the registry recovered: %v", err)
+	}
+	_ = artifact.Body.Close()
+
+	if state := cbFetcher.GetBreakerState()[registry]; state != "closed" {
+		t.Errorf("breaker state = %q, want closed after a successful fetch", state)
+	}
+}
+
+// An open breaker lets exactly one request per backoff interval reach the
+// registry, and refuses the rest without contacting it. Checking Ready() before
+// Call() would spend the probe: Call() checks the breaker itself, and the first
+// check advances the backoff, so the second one sees an interval that has not
+// elapsed yet and refuses the very call it was meant to admit.
+func TestCircuitBreakerProbesOncePerBackoffInterval(t *testing.T) {
+	var requests atomic.Int64
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		requests.Add(1)
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer server.Close()
+
+	mockClock := clock.NewMock()
+	cbFetcher := NewCircuitBreakerFetcher(NewFetcher(WithMaxRetries(0), WithBaseDelay(0)))
+	cbFetcher.clock = mockClock
+
+	ctx := context.Background()
+	artifactURL := server.URL + "/test.tar.gz"
+
+	for range cbThreshold {
+		if _, err := cbFetcher.Fetch(ctx, artifactURL); err == nil {
+			t.Fatal("expected a fetch against a failing registry to fail")
+		}
+	}
+
+	// Each step is longer than the maximum backoff interval, so each one is
+	// entitled to a probe.
+	for step := range 5 {
+		mockClock.Add(2 * cbMaxInterval)
+
+		before := requests.Load()
+		if _, err := cbFetcher.Fetch(ctx, artifactURL); !errors.Is(err, ErrUpstreamDown) {
+			t.Fatalf("step %d: error = %v, want one wrapping ErrUpstreamDown", step, err)
+		}
+		if got := requests.Load() - before; got != 1 {
+			t.Fatalf("step %d: %d requests reached the registry, want 1 probe", step, got)
+		}
+
+		// A second call in the same interval waits for the next one.
+		before = requests.Load()
+		_, err := cbFetcher.Fetch(ctx, artifactURL)
+		if !errors.Is(err, ErrUpstreamDown) {
+			t.Errorf("step %d: refusal = %v, want one wrapping ErrUpstreamDown", step, err)
+		}
+		if got := requests.Load() - before; got != 0 {
+			t.Errorf("step %d: %d further requests reached the registry, want none", step, got)
+		}
+	}
+}

go.mod

@@ -4,6 +4,7 @@ go 1.25.6
 
 require (
 	github.com/cenk/backoff v2.2.1+incompatible
+	github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a
 	github.com/git-pkgs/pom v0.1.5
 	github.com/git-pkgs/purl v0.1.15
 	github.com/git-pkgs/spdx v0.3.0
@@ -13,7 +14,6 @@ require (
 )
 
 require (
-	github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a // indirect
 	github.com/github/go-spdx/v2 v2.7.0 // indirect
 	github.com/package-url/packageurl-go v0.1.6 // indirect
 	github.com/peterbourgon/g2s v0.0.0-20170223122336-d4e7ad98afea // indirect