diff --git a/circuit_breaker_test.go b/circuit_breaker_test.go index 23bd698..1f73c54 100644 --- a/circuit_breaker_test.go +++ b/circuit_breaker_test.go @@ -4,16 +4,21 @@ import ( "bytes" "errors" "fmt" + "net/http" + "net/http/httptest" "strings" "testing" "time" + "github.com/gofiber/fiber/v3" libpack_cache "github.com/lukaszraczylo/graphql-monitoring-proxy/cache" libpack_cache_memory "github.com/lukaszraczylo/graphql-monitoring-proxy/cache/memory" libpack_logger "github.com/lukaszraczylo/graphql-monitoring-proxy/logging" libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring" "github.com/sony/gobreaker" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/suite" + "github.com/valyala/fasthttp" ) // CircuitBreakerTestSuite is a test suite for circuit breaker functionality @@ -69,6 +74,9 @@ func (suite *CircuitBreakerTestSuite) TearDownTest() { cb = nil // Circuit breaker metrics are now managed by cbMetrics cbMetrics = nil + // Reset the C7 progressive-backoff tracker so a trip count/timestamp + // from one test can't leak into the next. + cbBackoff.reset() } // Helper function to check if a specific message appears in the logger output @@ -210,6 +218,522 @@ func (suite *CircuitBreakerTestSuite) TestExecuteFunctionBehavior() { suite.Equal(testErr.Error(), err.Error(), "Error message should match") } +// TestClassifyCircuitFailure is a pure-function table test for the C5/C6 +// gating logic: whether a proxy failure counts toward the circuit +// breaker's failure statistics is gated by the matching +// CIRCUIT_TRIP_ON_TIMEOUTS / CIRCUIT_TRIP_ON_4XX / CIRCUIT_TRIP_ON_5XX flag. +// Any other failure kind (e.g. a connection error) has no dedicated flag +// and always counts, regardless of flag values -- that's the pre-fix +// baseline (flags were parsed but never enforced) this preserves. +func TestClassifyCircuitFailure(t *testing.T) { + tests := []struct { + name string + err error + statusCode int + tripOnTimeouts bool + tripOn4xx bool + tripOn5xx bool + wantCounts bool + }{ + { + name: "timeout counts when TripOnTimeouts=true (library default)", + err: errors.New("timeout"), + tripOnTimeouts: true, + wantCounts: true, + }, + { + name: "timeout does not count when TripOnTimeouts=false", + err: errors.New("timeout"), + tripOnTimeouts: false, + wantCounts: false, + }, + { + name: "5xx counts when TripOn5xx=true (library default)", + err: fmt.Errorf("received non-200 response: %d", 500), + statusCode: 500, + tripOn5xx: true, + wantCounts: true, + }, + { + name: "5xx does not count when TripOn5xx=false", + err: fmt.Errorf("received non-200 response: %d", 503), + statusCode: 503, + tripOn5xx: false, + wantCounts: false, + }, + { + name: "4xx counts when TripOn4xx=true", + err: fmt.Errorf("received non-200 response: %d", 404), + statusCode: 404, + tripOn4xx: true, + wantCounts: true, + }, + { + name: "4xx does not count when TripOn4xx=false (library default)", + err: fmt.Errorf("received non-200 response: %d", 400), + statusCode: 400, + tripOn4xx: false, + wantCounts: false, + }, + { + name: "connection error always counts, regardless of any flag", + err: errors.New("connection refused"), + statusCode: 0, + wantCounts: true, + }, + { + name: "nil fiber context error always counts, regardless of any flag", + err: errFiberCtxNilDuringRetry, + statusCode: 0, + wantCounts: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + testCfg := &config{} + testCfg.CircuitBreaker.TripOnTimeouts = tt.tripOnTimeouts + testCfg.CircuitBreaker.TripOn4xx = tt.tripOn4xx + testCfg.CircuitBreaker.TripOn5xx = tt.tripOn5xx + + got := classifyCircuitFailure(testCfg, tt.err, tt.statusCode) + assert.Equal(t, tt.wantCounts, got) + }) + } +} + +// TestCircuitBreakerIsSuccessful checks the gobreaker Settings.IsSuccessful +// hook directly: a nil error is success, a *circuitBreakerOutcome defers to +// its pre-computed verdict, and any other non-nil error (one that never +// went through performProxyRequestCore's classification) is a failure -- +// matching gobreaker's own default. +func TestCircuitBreakerIsSuccessful(t *testing.T) { + assert.True(t, circuitBreakerIsSuccessful(nil)) + assert.False(t, circuitBreakerIsSuccessful(errors.New("boom"))) + assert.False(t, circuitBreakerIsSuccessful(&circuitBreakerOutcome{err: errors.New("boom"), countsAsFailure: true})) + assert.True(t, circuitBreakerIsSuccessful(&circuitBreakerOutcome{err: errors.New("boom"), countsAsFailure: false})) + + // Wrapping must not defeat classification. + wrapped := fmt.Errorf("outer: %w", &circuitBreakerOutcome{err: errors.New("boom"), countsAsFailure: false}) + assert.True(t, circuitBreakerIsSuccessful(wrapped)) +} + +// TestCircuitBreakerFlagGatingIntegration exercises the full +// performProxyRequest path (not just the pure classifier) for the two +// fast, non-retried outcome kinds -- 4xx and timeout -- confirming both +// halves of the C5/C6 contract: (1) the client-visible result is an error +// either way (only the breaker bookkeeping differs), and (2) the circuit +// only trips when the matching CIRCUIT_TRIP_ON_* flag is true. The 5xx +// case is covered by TestClassifyCircuitFailure above and by +// TestCachingAndCircuitBreakerInteraction (integration_test.go); an +// end-to-end flag-gated 5xx test would additionally need to wait out the +// full 7-attempt retry backoff (~25s per request), which isn't worth the +// wall-clock cost here. +func (suite *CircuitBreakerTestSuite) TestCircuitBreakerFlagGatingIntegration() { + runOnce := func(server *httptest.Server) error { + app := fiber.New() + reqCtx := &fasthttp.RequestCtx{} + reqCtx.Request.SetRequestURI("/graphql") + reqCtx.Request.Header.SetMethod("POST") + reqCtx.Request.Header.SetContentType("application/json") + reqCtx.Request.SetBody([]byte(`{"query":"query { test }"}`)) + ctx := app.AcquireCtx(reqCtx) + defer app.ReleaseCtx(ctx) + return performProxyRequest(ctx, server.URL) + } + + suite.Run("4xx", func() { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"errors":[{"message":"bad request"}]}`)) + })) + defer server.Close() + + for _, tripOn4xx := range []bool{true, false} { + suite.Run(fmt.Sprintf("TripOn4xx=%v", tripOn4xx), func() { + suite.outputBuffer.Reset() + cfg.CircuitBreaker.MaxFailures = 1 + cfg.CircuitBreaker.Timeout = 5 + cfg.CircuitBreaker.TripOn4xx = tripOn4xx + cfg.CircuitBreaker.ReturnCachedOnOpen = false + cfg.Client.ClientTimeout = 5 + cfg.Client.FastProxyClient = createFasthttpClient(cfg) + initCircuitBreaker(cfg) + + err := runOnce(server) + assert.Error(suite.T(), err, "client must still see the 4xx failure regardless of the flag") + + wantState := gobreaker.StateClosed + if tripOn4xx { + wantState = gobreaker.StateOpen + } + assert.Equal(suite.T(), wantState.String(), cb.State().String(), + "circuit state must match TripOn4xx=%v", tripOn4xx) + }) + } + }) + + suite.Run("timeout", func() { + // Handler sleeps well past the 1s client timeout below so every + // attempt is a genuine client-side timeout, not a race. + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(2 * time.Second) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"data":{"ok":true}}`)) + })) + defer server.Close() + + for _, tripOnTimeouts := range []bool{true, false} { + suite.Run(fmt.Sprintf("TripOnTimeouts=%v", tripOnTimeouts), func() { + suite.outputBuffer.Reset() + cfg.CircuitBreaker.MaxFailures = 1 + cfg.CircuitBreaker.Timeout = 5 + cfg.CircuitBreaker.TripOnTimeouts = tripOnTimeouts + cfg.CircuitBreaker.ReturnCachedOnOpen = false + cfg.Client.ClientTimeout = 1 + cfg.Client.FastProxyClient = createFasthttpClient(cfg) + initCircuitBreaker(cfg) + + err := runOnce(server) + assert.Error(suite.T(), err, "client must still see the timeout failure regardless of the flag") + + wantState := gobreaker.StateClosed + if tripOnTimeouts { + wantState = gobreaker.StateOpen + } + assert.Equal(suite.T(), wantState.String(), cb.State().String(), + "circuit state must match TripOnTimeouts=%v", tripOnTimeouts) + }) + } + }) +} + +// TestCircuitBreakerFailureRatioWindow covers the C12 fix: with +// Settings.Interval left at 0 (the pre-fix default), gobreaker never clears +// Counts while closed, so CIRCUIT_FAILURE_RATIO becomes lifetime-cumulative +// -- once enough historical successes have piled up, no realistic burst of +// new failures can move the ratio again. This test shrinks +// circuitBreakerCountWindow to a fraction of a second (restored via +// t.Cleanup) so the window boundary can be observed without sleeping for +// the real 60s default, then shows: (1) Counts actually reset for a new +// generation once the window elapses, instead of accumulating forever, and +// (2) a fresh in-window failure burst can trip the ratio check purely on +// its own -- which a lifetime-cumulative ratio, diluted by an earlier +// batch of successes, would not allow. +func (suite *CircuitBreakerTestSuite) TestCircuitBreakerFailureRatioWindow() { + originalWindow := circuitBreakerCountWindow + circuitBreakerCountWindow = 150 * time.Millisecond + suite.T().Cleanup(func() { circuitBreakerCountWindow = originalWindow }) + + // Disable the consecutive-failures trip path so only FailureRatio can + // trip the breaker in this test; the circuit stays closed throughout. + cfg.CircuitBreaker.MaxFailures = 1000 + cfg.CircuitBreaker.FailureRatio = 0.5 + cfg.CircuitBreaker.SampleSize = 4 + cfg.CircuitBreaker.Timeout = 5 + initCircuitBreaker(cfg) + + require := suite.Require() + testErr := errors.New("test error") + execute := func(fail bool) { + _, _ = cb.Execute(func() (any, error) { + if fail { + return nil, testErr + } + return "ok", nil + }) + } + + // Phase 1: a batch of mostly-successful requests with one failure -- + // well under the 50% ratio, so it must not trip, but it does + // accumulate Requests/TotalFailures in Counts. + for i := 0; i < 10; i++ { + execute(false) + } + execute(true) + require.Equal(gobreaker.StateClosed.String(), cb.State().String(), "low ratio must not trip") + require.Equal(uint32(11), cb.Counts().Requests, "phase 1 requests should be counted") + + // Wait past the (shrunk) window. gobreaker clears Counts lazily, on + // the next request, once Interval has elapsed while closed. + time.Sleep(circuitBreakerCountWindow + 100*time.Millisecond) + + // This request both triggers the lazy clear and starts the new + // generation's Counts. + execute(false) + assert.Equal(suite.T(), uint32(1), cb.Counts().Requests, + "Counts must reset for the new window instead of accumulating across it (lifetime-cumulative regression)") + + // Phase 2: within the fresh window, a small failure burst alone (2 + // successes + 2 failures = 50% ratio, matching SampleSize) is enough to + // trip. Under the pre-fix lifetime-cumulative behaviour, phase 1's 10 + // successes would still be diluting the ratio (13 requests, 3 failures + // = 23%) and this burst alone would not trip the circuit. + execute(false) + execute(true) + execute(true) + assert.Equal(suite.T(), gobreaker.StateOpen.String(), cb.State().String(), + "a fresh in-window failure burst must be able to trip the ratio check on its own") +} + +// TestCircuitBackoffResetsOnSuccessfulRecovery is a suite method (needs +// cfg.Logger from SetupTest, which createStateChangeFunc's logging calls +// into) that exercises the C7 OnStateChange wiring end-to-end against a +// real *gobreaker.CircuitBreaker: consecutiveTrips increments on Closed -> +// Open, and resets to 0 exactly on the eventual HalfOpen -> Closed +// transition (a successful recovery). +// +// This breaker is constructed directly (bypassing initCircuitBreaker, whose +// Settings.Timeout only accepts whole seconds via cfg.CircuitBreaker.Timeout) +// with a deliberately tiny real Timeout so the Open -> HalfOpen -> Closed +// sequence completes after a bounded few-millisecond sleep -- gobreaker's +// own clock is not injectable, so some real wait is unavoidable to advance +// it. That sleep only advances gobreaker's OWN internal timeout to exercise +// the OnStateChange plumbing; it is unrelated to, and far shorter than, the +// C7 multiplier-scaled backoff window itself, which is verified +// deterministically (via injected trip counts and an injected `now`, no +// sleeping) by TestCircuitBackoffGateBlocks and +// TestCircuitBackoffLongerClosedDoorWithHigherMultiplier below. +func (suite *CircuitBreakerTestSuite) TestCircuitBackoffResetsOnSuccessfulRecovery() { + cbMutex.Lock() + cbBackoff.reset() + cb = gobreaker.NewCircuitBreaker(gobreaker.Settings{ + Name: "recovery-test", + MaxRequests: 1, + Timeout: 5 * time.Millisecond, + ReadyToTrip: func(counts gobreaker.Counts) bool { return counts.ConsecutiveFailures >= 1 }, + OnStateChange: createStateChangeFunc(cfg), + IsSuccessful: circuitBreakerIsSuccessful, + }) + cbMutex.Unlock() + + require := suite.Require() + require.Equal(int64(0), cbBackoff.trips(), "fresh breaker must start with zero trips") + + // Closed -> Open: a single failure trips it immediately, no timing + // dependency at all. + _, _ = cb.Execute(func() (any, error) { + return nil, &circuitBreakerOutcome{err: errors.New("boom"), countsAsFailure: true} + }) + require.Equal(gobreaker.StateOpen.String(), cb.State().String(), "one failure must trip the breaker") + require.Equal(int64(1), cbBackoff.trips(), "the trip must be recorded") + require.False(cbBackoff.lastTrip().IsZero(), "lastTrip must be stamped") + + // Wait past gobreaker's own (tiny) Timeout so the next call observes the + // lazy Open -> HalfOpen transition, then succeed to close it. + time.Sleep(15 * time.Millisecond) + _, err := cb.Execute(func() (any, error) { return "ok", nil }) + suite.NoError(err, "the half-open probe must succeed") + suite.Equal(gobreaker.StateClosed.String(), cb.State().String(), "a successful probe must close the breaker") + suite.Equal(int64(0), cbBackoff.trips(), "trip count must reset to 0 after a successful recovery to Closed") +} + +// TestEffectiveCircuitBackoff is a pure-function table test of the C7 +// backoff formula: effective = base * multiplier^(trips-1), capped at +// maxBackoff. It is the primary proof of default-preservation: with the +// default BackoffMultiplier=1.0, effective must equal base exactly, for +// every trip count, with no dependency on wall-clock or a live breaker. +func TestEffectiveCircuitBackoff(t *testing.T) { + const base = 10 * time.Second + + t.Run("multiplier=1 (default) is constant=base for any trip count", func(t *testing.T) { + for _, trips := range []int64{0, 1, 2, 5, 100} { + got := effectiveCircuitBackoff(base, 1.0, 0, trips) + assert.Equal(t, base, got, "trips=%d must equal base exactly under the default multiplier", trips) + } + }) + + t.Run("multiplier=2 compounds from the second trip, capped at maxBackoff", func(t *testing.T) { + const maxBackoff = 100 * time.Second + tests := []struct { + trips int64 + want time.Duration + }{ + {trips: 0, want: base}, // defensive: never tripped + {trips: 1, want: base}, // first trip: exponent 0 -> multiplier^0=1 + {trips: 2, want: 2 * base}, // exponent 1 + {trips: 3, want: 4 * base}, // exponent 2 + {trips: 4, want: 8 * base}, // exponent 3, still under the 100s cap + {trips: 5, want: maxBackoff}, // exponent 4 -> 16*base=160s, capped at 100s + } + for _, tt := range tests { + got := effectiveCircuitBackoff(base, 2.0, maxBackoff, tt.trips) + assert.Equal(t, tt.want, got, "trips=%d", tt.trips) + } + }) + + t.Run("maxBackoff<=0 means uncapped", func(t *testing.T) { + got := effectiveCircuitBackoff(base, 2.0, 0, 6) // exponent 5 -> 32*base + assert.Equal(t, 32*base, got) + }) + + t.Run("maxBackoff<=0 with a high trip count is still bounded by the sanity ceiling", func(t *testing.T) { + // finding C-overflow: CIRCUIT_MAX_BACKOFF_TIMEOUT=0 (uncapped) plus a + // high enough consecutive-trip count makes + // base*multiplier^(trips-1) overflow math.Pow's float64 arithmetic + // (or the subsequent float64->time.Duration conversion) into a + // huge-but-still-positive Duration -- on this platform the + // out-of-range conversion saturates to math.MaxInt64 nanoseconds + // (~292 years), which is NOT less than base, so the pre-existing + // "effective < base" underflow guard alone does not catch it. The + // hard ceiling must, regardless of trip count. + got := effectiveCircuitBackoff(base, 2.0, 0, 1000) + assert.Equal(t, maxSaneCircuitBackoff, got, "an uncapped config must never exceed the hard sanity ceiling") + assert.LessOrEqual(t, got, maxSaneCircuitBackoff) + }) + + t.Run("misconfigured multiplier<1 is clamped to 1, never shrinks below base", func(t *testing.T) { + got := effectiveCircuitBackoff(base, 0.5, 0, 3) + assert.Equal(t, base, got) + }) + + t.Run("misconfigured maxBackoff below base never shrinks below base", func(t *testing.T) { + got := effectiveCircuitBackoff(base, 2.0, 1*time.Second, 3) + assert.Equal(t, base, got) + }) +} + +// TestCircuitBackoffGateBlocks is a deterministic table test of +// circuitBackoffGateBlocks: trip count and elapsed time are both injected +// (via repeated cbBackoff.recordTrip calls and an explicit `now` parameter) +// rather than produced by sleeping, so the pass/fail boundary is exact and +// cannot flake. The breaker passed in is a freshly constructed, never- +// tripped *gobreaker.CircuitBreaker (State()==Closed) -- circuitBackoffGateBlocks +// only cares that breaker.State() != StateOpen, which a fresh breaker +// satisfies trivially without needing to wait out any real timeout. +func TestCircuitBackoffGateBlocks(t *testing.T) { + t.Cleanup(func() { cbBackoff.reset() }) + + const base = 10 * time.Second + tripTime := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + + tests := []struct { + name string + trips int64 + multiplier float64 + maxBackoff time.Duration + elapsed time.Duration + wantBlock bool + }{ + { + name: "never tripped -> never blocks", + trips: 0, + elapsed: 0, + wantBlock: false, + }, + { + name: "first trip, multiplier=1, well within base -> blocks", + trips: 1, + multiplier: 1, + elapsed: 1 * time.Second, + wantBlock: true, + }, + { + name: "first trip, multiplier=1, base elapsed -> does not block", + trips: 1, + multiplier: 1, + elapsed: base, + wantBlock: false, + }, + { + name: "second trip, multiplier=2, base elapsed but effective (2x base) not -> blocks", + trips: 2, + multiplier: 2, + elapsed: base + time.Second, + wantBlock: true, + }, + { + name: "second trip, multiplier=2, effective (2x base) elapsed -> does not block", + trips: 2, + multiplier: 2, + elapsed: 2 * base, + wantBlock: false, + }, + { + name: "high trip count capped at maxBackoff, still within cap -> blocks", + trips: 10, + multiplier: 2, + maxBackoff: 20 * time.Second, + elapsed: 19 * time.Second, + wantBlock: true, + }, + { + name: "high trip count capped at maxBackoff, cap elapsed -> does not block", + trips: 10, + multiplier: 2, + maxBackoff: 20 * time.Second, + elapsed: 20 * time.Second, + wantBlock: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cbBackoff.reset() + for i := int64(0); i < tt.trips; i++ { + // Repeated recordTrip calls model consecutive trips; each + // overwrites lastTripUnixNano, so the final timestamp is + // exactly tripTime regardless of trip count. + cbBackoff.recordTrip(tripTime) + } + + testCfg := &config{} + testCfg.CircuitBreaker.Timeout = int(base / time.Second) + testCfg.CircuitBreaker.BackoffMultiplier = tt.multiplier + testCfg.CircuitBreaker.MaxBackoffTimeout = int(tt.maxBackoff / time.Second) + + breaker := gobreaker.NewCircuitBreaker(gobreaker.Settings{Name: "gate-test"}) + now := tripTime.Add(tt.elapsed) + + got := circuitBackoffGateBlocks(testCfg, breaker, now) + assert.Equal(t, tt.wantBlock, got) + }) + } +} + +// TestCircuitBackoffLongerClosedDoorWithHigherMultiplier directly proves +// the C7 requirement: for the SAME consecutive-trip count and the SAME +// elapsed time since the last trip (both injected, not slept), a higher +// BackoffMultiplier keeps the gate blocking while multiplier=1 (the +// default) has already let go -- i.e. progressive backoff strictly +// lengthens the closed-door window, and the default reproduces today's +// behaviour exactly. +func TestCircuitBackoffLongerClosedDoorWithHigherMultiplier(t *testing.T) { + t.Cleanup(func() { cbBackoff.reset() }) + + const base = 10 * time.Second + const trips = 3 // third consecutive trip: multiplier exponent = trips-1 = 2 + tripTime := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + // Past base (so a multiplier=1 breaker would already let a probe + // through) but well short of 2^2 * base = 40s. + elapsed := base + time.Second + now := tripTime.Add(elapsed) + + seed := func() { + cbBackoff.reset() + for i := 0; i < trips; i++ { + cbBackoff.recordTrip(tripTime) + } + } + + cfgDefault := &config{} + cfgDefault.CircuitBreaker.Timeout = int(base / time.Second) + cfgDefault.CircuitBreaker.BackoffMultiplier = 1.0 + + cfgProgressive := &config{} + cfgProgressive.CircuitBreaker.Timeout = int(base / time.Second) + cfgProgressive.CircuitBreaker.BackoffMultiplier = 2.0 + + seed() + blockedAtDefault := circuitBackoffGateBlocks(cfgDefault, gobreaker.NewCircuitBreaker(gobreaker.Settings{Name: "default"}), now) + + seed() + blockedAtProgressive := circuitBackoffGateBlocks(cfgProgressive, gobreaker.NewCircuitBreaker(gobreaker.Settings{Name: "progressive"}), now) + + assert.False(t, blockedAtDefault, "multiplier=1.0 (default): base has already elapsed, gate must match today's behaviour and not block") + assert.True(t, blockedAtProgressive, "multiplier=2.0: effective backoff has not yet elapsed at the same instant, gate must still block -- strictly longer closed-door time than the default") +} + // Start the test suite func TestCircuitBreakerSuite(t *testing.T) { suite.Run(t, new(CircuitBreakerTestSuite)) diff --git a/errors.go b/errors.go index a9e3ad0..2f2ca0a 100644 --- a/errors.go +++ b/errors.go @@ -5,6 +5,8 @@ import ( "errors" "fmt" "time" + + "github.com/sony/gobreaker" ) // Error codes for structured error responses @@ -143,3 +145,28 @@ func GetStatusCode(err error) int { return 500 } + +// StatusCodeForError maps a proxy-path error to the HTTP status code an +// error handler must return for it. It recognises the circuit-breaker +// sentinel errors (ErrCircuitOpen, defined in proxy.go, and +// gobreaker.ErrOpenState) as 503 Service Unavailable -- previously these +// surfaced as a generic 500 because nothing mapped them -- defers to a +// wrapped *ProxyError's StatusCode when present (via errors.As, so it also +// unwraps errors returned through circuitBreakerOutcome), and otherwise +// falls back to 500. A nil error maps to 200. +func StatusCodeForError(err error) int { + if err == nil { + return 200 + } + + if errors.Is(err, ErrCircuitOpen) || errors.Is(err, gobreaker.ErrOpenState) { + return 503 + } + + var proxyErr *ProxyError + if errors.As(err, &proxyErr) { + return proxyErr.StatusCode + } + + return 500 +} diff --git a/errors_test.go b/errors_test.go index 6b4ba43..8db2bde 100644 --- a/errors_test.go +++ b/errors_test.go @@ -6,6 +6,7 @@ import ( "net/http" "testing" + "github.com/sony/gobreaker" "github.com/stretchr/testify/assert" ) @@ -260,3 +261,62 @@ func Test_IsRetryable_GetStatusCode_Wrapped(t *testing.T) { assert.False(t, IsRetryable(nil)) assert.Equal(t, 200, GetStatusCode(nil)) } + +// Test_StatusCodeForError covers the C4 fix: circuit-open errors must map to +// 503 (previously fell through to a generic 500 because nothing mapped +// them), a wrapped *ProxyError's own status code must survive, and any +// other error defaults to 500. nil maps to 200. +func Test_StatusCodeForError(t *testing.T) { + tests := []struct { + name string + err error + wantStatus int + }{ + { + name: "nil error maps to 200", + err: nil, + wantStatus: 200, + }, + { + name: "circuit breaker open (proxy sentinel) maps to 503", + err: ErrCircuitOpen, + wantStatus: 503, + }, + { + name: "circuit breaker open (gobreaker sentinel) maps to 503", + err: gobreaker.ErrOpenState, + wantStatus: 503, + }, + { + name: "wrapped circuit breaker open still maps to 503", + err: fmt.Errorf("request failed: %w", ErrCircuitOpen), + wantStatus: 503, + }, + { + name: "wrapped gobreaker open state still maps to 503", + err: fmt.Errorf("request failed: %w", gobreaker.ErrOpenState), + wantStatus: 503, + }, + { + name: "generic error defaults to 500", + err: errors.New("boom"), + wantStatus: 500, + }, + { + name: "ProxyError with explicit status code is honoured", + err: NewProxyError(ErrCodeRateLimited, "too many requests", http.StatusTooManyRequests, false), + wantStatus: http.StatusTooManyRequests, + }, + { + name: "wrapped ProxyError status code survives wrapping", + err: fmt.Errorf("wrapped: %w", NewProxyError(ErrCodeBackendError, "boom", http.StatusBadGateway, true)), + wantStatus: http.StatusBadGateway, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.wantStatus, StatusCodeForError(tt.err)) + }) + } +} diff --git a/proxy.go b/proxy.go index e916ac5..43acac6 100644 --- a/proxy.go +++ b/proxy.go @@ -12,6 +12,7 @@ import ( "net/url" "strings" "sync" + "sync/atomic" "time" "go.opentelemetry.io/otel/trace" @@ -49,12 +50,121 @@ const ( defaultMaxRequestsInHalfOpen = 10 // Default maximum requests in half-open state ) +// circuitBreakerCountWindow bounds how long the circuit breaker accumulates +// Counts while CLOSED before clearing them (gobreaker Settings.Interval). +// gobreaker treats Interval<=0 as "never clear Counts during the closed +// state" (see sony/gobreaker Settings docs), which makes +// CIRCUIT_FAILURE_RATIO lifetime-cumulative instead of a recent-window +// ratio: after long uptime a handful of new failures can no longer move a +// ratio diluted by millions of past successes. A positive Interval makes +// gobreaker periodically clear Counts while closed, so the ratio measures a +// rolling window instead. +// +// The window must comfortably exceed the worst-case time a single logical +// request can take, not just typical request latency. gobreaker checks the +// interval lazily, both when a request starts (beforeRequest) and when it +// finishes (afterRequest); if the interval elapses *while* a request is +// in flight, afterRequest rolls Counts over to a new generation before +// recording that request's outcome, and then silently drops it (its +// generation no longer matches the one captured at the start). Each +// logical proxied request here can itself retry up to 7x with exponential +// backoff capped at 10s (performProxyRequestWithEnhancedRetries), i.e. +// ~25.5s worst case (0.5+1+2+4+8+10s) before it finally fails -- so with +// the library default MaxFailures=10, a run of consecutive failing +// requests can legitimately take ~4-5 minutes. A short window (e.g. the +// more typical 60s) reproduces exactly that: it was caught by +// TestCachingAndCircuitBreakerInteraction, where 3 sequential +// retry-exhausting failures (~76s total) spuriously left the circuit +// closed because the last failure's generation had already rolled over +// out from under it. 10 minutes keeps the ratio meaningfully "recent" +// (instead of lifetime-cumulative) while leaving multiples of headroom +// over that worst case. With that much headroom, a single request is very +// unlikely to straddle the window boundary; it is not impossible, since +// gobreaker checks the interval lazily against wall-clock time rather than +// reserving a slot up front. If one still does straddle it, gobreaker drops +// only that one outcome, which delays a trip by at most one request rather +// than breaking correctness. +// +// It is a package var (not const) purely so tests can shrink it to observe +// window-clearing behaviour without sleeping for the real duration. +var circuitBreakerCountWindow = 10 * time.Minute + // Global circuit breaker var ( cb *gobreaker.CircuitBreaker cbMutex sync.RWMutex ) +// circuitBackoffState tracks the progressive open-state backoff (finding +// C7) for the global circuit breaker cb. gobreaker v1.0.0's own open-state +// Timeout (cb.timeout) is fixed at construction and has no setter, so it +// cannot be varied per-trip from outside the library. This state instead +// backs a decorator layered IN FRONT of gobreaker -- see +// circuitBackoffGateBlocks and its use in performProxyRequestCore -- that +// never reaches into gobreaker's internals; gobreaker's own Timeout keeps +// governing its internal half-open transition unchanged. +// +// consecutiveTrips counts transitions to StateOpen since the last +// successful recovery (a transition to StateClosed); lastTripUnixNano is +// the UnixNano() of the most recent such transition (0 = never tripped). +// Both are written from createStateChangeFunc's callback, which gobreaker +// invokes synchronously while still holding its own internal mutex (see +// sony/gobreaker CircuitBreaker.setState) -- so that callback must never +// call back into the breaker (e.g. cb.State()), which would deadlock on +// gobreaker's own non-reentrant lock. circuitBackoffGateBlocks reads this +// state on every proxied request. Atomics (not a mutex) keep both +// directions lock-free and sidestep any lock-ordering dependency on +// gobreaker's internal lock. +type circuitBackoffState struct { + consecutiveTrips atomic.Int64 + lastTripUnixNano atomic.Int64 +} + +// recordTrip increments the consecutive-trip counter and stamps the current +// time. Called from createStateChangeFunc on transition to StateOpen. +func (s *circuitBackoffState) recordTrip(now time.Time) { + s.consecutiveTrips.Add(1) + s.lastTripUnixNano.Store(now.UnixNano()) +} + +// recordRecovery resets the consecutive-trip counter. Called from +// createStateChangeFunc on transition to StateClosed (a successful +// recovery), so the next trip after a clean close starts progressive +// backoff over again at the base timeout. +func (s *circuitBackoffState) recordRecovery() { + s.consecutiveTrips.Store(0) +} + +// reset clears all state. Called when a new circuit breaker is constructed +// (initCircuitBreaker) so a fresh breaker never inherits a stale trip count +// or timestamp left over from a previous one (e.g. across tests, or a +// config reload). +func (s *circuitBackoffState) reset() { + s.consecutiveTrips.Store(0) + s.lastTripUnixNano.Store(0) +} + +// trips returns the current consecutive-trip count (0 = never tripped, or +// recovered since the last trip). +func (s *circuitBackoffState) trips() int64 { + return s.consecutiveTrips.Load() +} + +// lastTrip returns the time of the most recent transition to StateOpen, or +// the zero time.Time if the breaker has never tripped. +func (s *circuitBackoffState) lastTrip() time.Time { + ns := s.lastTripUnixNano.Load() + if ns == 0 { + return time.Time{} + } + return time.Unix(0, ns) +} + +// cbBackoff is the progressive-backoff tracker for the global circuit +// breaker cb -- one breaker, one tracker, mirroring the existing cb/cbMutex +// global pattern. +var cbBackoff circuitBackoffState + // Package-level substring tables used by isConnectionError / isTimeoutError. // Hoisted to avoid per-call slice allocations on the hot path. All entries // must be lower-case; callers lower-case the error string once before matching. @@ -107,17 +217,26 @@ func initCircuitBreaker(config *config) { cbMutex.Lock() defer cbMutex.Unlock() + // A freshly constructed breaker must never inherit a stale + // consecutive-trip count or timestamp from a previous one (see + // circuitBackoffState.reset doc) -- e.g. across tests, or a config + // reload that reinitializes cb. + cbBackoff.reset() + // Initialize circuit breaker metrics InitializeCircuitBreakerMetrics(config.Monitoring) // Create circuit breaker settings cbSettings := gobreaker.Settings{ - Name: "graphql-proxy-circuit", - MaxRequests: safeMaxRequests(config.CircuitBreaker.MaxRequestsInHalfOpen), - Interval: 0, // No specific interval for counting failures + Name: "graphql-proxy-circuit", + MaxRequests: safeMaxRequests(config.CircuitBreaker.MaxRequestsInHalfOpen), + // Periodic reset while closed keeps CIRCUIT_FAILURE_RATIO windowed + // instead of lifetime-cumulative; see circuitBreakerCountWindow doc. + Interval: circuitBreakerCountWindow, Timeout: time.Duration(config.CircuitBreaker.Timeout) * time.Second, ReadyToTrip: createTripFunc(config), OnStateChange: createStateChangeFunc(config), + IsSuccessful: circuitBreakerIsSuccessful, } // Initialize the circuit breaker @@ -172,9 +291,88 @@ func createTripFunc(config *config) func(counts gobreaker.Counts) bool { } } +// circuitBreakerOutcome wraps a proxy failure with a pre-computed verdict on +// whether it should count against the circuit breaker's failure statistics. +// gobreaker.Settings.IsSuccessful only receives the error returned from the +// Execute closure and cannot see the per-request fiber context, so the +// classification (which needs the response status code) is made once, at +// the point where both are available -- see performProxyRequestCore -- and +// travels with the error. Error() and Unwrap() forward to the wrapped +// error, so retry policy, StatusCodeForError, and logging all see the +// original, unwrapped error. +type circuitBreakerOutcome struct { + err error + countsAsFailure bool +} + +func (e *circuitBreakerOutcome) Error() string { return e.err.Error() } +func (e *circuitBreakerOutcome) Unwrap() error { return e.err } + +// circuitBreakerIsSuccessful is gobreaker's IsSuccessful hook (see +// Settings.IsSuccessful). It treats nil errors as success and defers to the +// pre-computed circuitBreakerOutcome.countsAsFailure verdict for classified +// proxy failures. Any other non-nil error -- one that never went through +// performProxyRequestCore's classification -- is treated as a failure, +// matching gobreaker's own default (defaultIsSuccessful) and pre-fix +// behaviour. +func circuitBreakerIsSuccessful(err error) bool { + if err == nil { + return true + } + var outcome *circuitBreakerOutcome + if errors.As(err, &outcome) { + return !outcome.countsAsFailure + } + return false +} + +// classifyCircuitFailure decides whether a failed proxy attempt should count +// toward the circuit breaker's failure statistics, gated by +// CIRCUIT_TRIP_ON_TIMEOUTS / CIRCUIT_TRIP_ON_4XX / CIRCUIT_TRIP_ON_5XX. +// statusCode is the last response status observed for this request (0 when +// no HTTP response was ever received, e.g. a connection error). +// +// Every failure that isn't a classified timeout/4xx/5xx (connection errors, +// nil fiber context, etc.) always counts -- those outcomes have no +// dedicated flag and always tripped the breaker before this fix, so that +// baseline is preserved regardless of flag values. +func classifyCircuitFailure(config *config, err error, statusCode int) bool { + if isTimeoutError(err) { + return config.CircuitBreaker.TripOnTimeouts + } + + // doProxyRequestWithTimeout's non-200 error always carries the real, + // freshly-copied status in c.Response() at the moment it was produced + // (see the comment in executeProxyAttempt). Guarding on the message + // first means a stale statusCode left over from an earlier attempt + // (e.g. this attempt was a connection error, which never copies a + // response) can't be misread as an HTTP status classification. + if strings.Contains(err.Error(), "non-200 response") { + switch { + case statusCode >= 500 && statusCode < 600: + return config.CircuitBreaker.TripOn5xx + case statusCode >= 400 && statusCode < 500: + return config.CircuitBreaker.TripOn4xx + } + } + + return true +} + // createStateChangeFunc returns a function that handles circuit state changes func createStateChangeFunc(config *config) func(name string, from gobreaker.State, to gobreaker.State) { return func(name string, from gobreaker.State, to gobreaker.State) { + // Progressive backoff bookkeeping (C7). gobreaker invokes + // OnStateChange synchronously while still holding its own internal + // mutex, so this must only touch cbBackoff's atomics -- never call + // back into the breaker (e.g. cb.State()), which would deadlock. + switch to { + case gobreaker.StateOpen: + cbBackoff.recordTrip(time.Now()) + case gobreaker.StateClosed: + cbBackoff.recordRecovery() + } + var stateValue float64 var stateName string @@ -215,6 +413,136 @@ func createStateChangeFunc(config *config) func(name string, from gobreaker.Stat } } +// gobreakerDefaultTimeout mirrors sony/gobreaker's own fallback for +// Settings.Timeout<=0 ("If Timeout is less than or equal to 0, the timeout +// value of the CircuitBreaker is set to 60 seconds" -- gobreaker.go +// NewCircuitBreaker). Duplicated here, since *gobreaker.CircuitBreaker +// exposes no getter for its internal timeout, so circuitBreakerBaseTimeout's +// result always matches what gobreaker itself actually uses as its +// open-state Timeout. +const gobreakerDefaultTimeout = 60 * time.Second + +// circuitBreakerBaseTimeout returns the open-state Timeout gobreaker was +// constructed with (see initCircuitBreaker), applying the same <=0 +// fallback gobreaker applies internally. +func circuitBreakerBaseTimeout(config *config) time.Duration { + base := time.Duration(config.CircuitBreaker.Timeout) * time.Second + if base <= 0 { + return gobreakerDefaultTimeout + } + return base +} + +// maxSaneCircuitBackoff hard-caps effectiveCircuitBackoff's result +// independently of the configured maxBackoff. When an operator sets +// CIRCUIT_MAX_BACKOFF_TIMEOUT=0 (uncapped) together with a +// BackoffMultiplier>1, a high enough consecutive-trip count makes +// base*multiplier^(trips-1) overflow math.Pow's float64 arithmetic (or the +// subsequent float64->time.Duration conversion) into a garbage or absurdly +// large Duration. This ceiling makes sure the breaker can never block +// requests for longer than a sane maximum, no matter what maxBackoff is +// configured to. +const maxSaneCircuitBackoff = 24 * time.Hour + +// effectiveCircuitBackoff computes the progressive open-state backoff +// duration for the given consecutive-trip count (finding C7): +// +// effective = base * multiplier^(trips-1), capped at maxBackoff. +// +// trips-1, not trips, so the FIRST trip (trips<=1) always backs off by +// exactly base -- identical to gobreaker's own fixed Timeout for that first +// open period. The multiplier only compounds starting from the SECOND +// consecutive trip. trips<=0 is defensive (a breaker that never tripped +// never reaches this path) and also returns base. +// +// With the default BackoffMultiplier=1.0, effective==base unconditionally +// for every trip count -- an exact fast path below skips the float +// round-trip entirely, so this is byte-identical to pre-C7 behaviour, not +// merely float-equal. A misconfigured multiplier<1 is clamped to 1, and the +// result is never allowed to fall below base or above a maxBackoff that is +// itself (misconfigured) smaller than base -- effective is always >= base, +// so this can only ever ADD closed-door time on top of gobreaker's own +// Timeout, never remove it. See maxSaneCircuitBackoff for the absolute +// ceiling applied regardless of maxBackoff. +func effectiveCircuitBackoff(base time.Duration, multiplier float64, maxBackoff time.Duration, trips int64) time.Duration { + if base <= 0 || trips <= 1 { + return base + } + + if multiplier < 1 { + multiplier = 1 + } + if multiplier == 1 { + return base + } + + exponent := float64(trips - 1) + effective := time.Duration(float64(base) * math.Pow(multiplier, exponent)) + if effective < base { + // Overflow or float rounding pushed us below base; never shorten. + return base + } + + // Hard sanity ceiling, applied before the configured cap below and + // independent of it -- see maxSaneCircuitBackoff. + if effective > maxSaneCircuitBackoff { + effective = maxSaneCircuitBackoff + } + + if maxBackoff > 0 && effective > maxBackoff { + if maxBackoff < base { + return base // misconfigured cap below base; never shorten + } + return maxBackoff + } + + return effective +} + +// circuitBackoffGateBlocks reports whether the C7 progressive-backoff gate +// should reject a request IN FRONT of gobreaker, without calling +// breaker.Execute -- and therefore without calling the backend or touching +// gobreaker's own Counts/generation. +// +// gobreaker's own Timeout still fully governs its internal half-open +// transition (see sony/gobreaker CircuitBreaker.currentState); this gate +// only adds an extra rejection window on top of it. It blocks exactly when +// both hold: +// - the progressively-extended effective backoff (computed from the +// consecutive-trip count) has not yet elapsed since the last trip, and +// - gobreaker's own (shorter-or-equal) Timeout HAS already elapsed, i.e. +// breaker.State() has lazily moved off StateOpen and would otherwise +// let a half-open probe through. +// +// When gobreaker itself is still StateOpen, calling Execute would reject +// with ErrOpenState anyway, so no gate is needed there. When there have +// been no trips yet (trips<=0), this always returns false without touching +// breaker.State() at all, keeping the common (never-tripped) case +// allocation- and lock-free on the hot path. now is a parameter (not +// time.Now() read internally) so tests can drive this deterministically +// without wall-clock sleeps. +func circuitBackoffGateBlocks(config *config, breaker *gobreaker.CircuitBreaker, now time.Time) bool { + trips := cbBackoff.trips() + if trips <= 0 { + return false + } + + lastTrip := cbBackoff.lastTrip() + if lastTrip.IsZero() { + return false + } + + base := circuitBreakerBaseTimeout(config) + maxBackoff := time.Duration(config.CircuitBreaker.MaxBackoffTimeout) * time.Second + effective := effectiveCircuitBackoff(base, config.CircuitBreaker.BackoffMultiplier, maxBackoff, trips) + + if now.Sub(lastTrip) >= effective { + return false // extended window elapsed; defer to gobreaker as normal + } + + return breaker.State() != gobreaker.StateOpen +} + // createFasthttpClient creates and configures a fasthttp client with optimized settings. // The client is configured based on the provided configuration settings, with careful // attention to performance and security considerations. @@ -437,35 +765,51 @@ func performProxyRequestCore(c fiber.Ctx, proxyURL string, cacheKey string) erro return performProxyRequestWithRetries(c, proxyURL) } - // Execute request through circuit breaker - _, err := cb.Execute(func() (any, error) { - // Execute the request with retries - err := performProxyRequestWithRetries(c, proxyURL) - // Check if the error or status code should trip the circuit breaker - if err != nil { - // Log error that could potentially trip the circuit - cfg.Logger.Warning(&libpack_logger.LogMessage{ - Message: "Error in circuit-protected request", - Pairs: map[string]any{ - "path": c.Path(), - "error": err.Error(), - }, - }) - return nil, err - } + var err error + if circuitBackoffGateBlocks(cfg, cb, time.Now()) { + // Progressive backoff (C7): gobreaker's own Timeout has already + // elapsed -- it would otherwise let a half-open probe through -- but + // the extended per-trip backoff window has not. Reject exactly like + // an open breaker, without calling cb.Execute: no backend call, and + // gobreaker's own Counts/generation are left untouched. gobreaker + // stays idle in half-open until the extended window also elapses, + // at which point the Execute call below performs the real probe. + err = gobreaker.ErrOpenState + } else { + // Execute request through circuit breaker + _, err = cb.Execute(func() (any, error) { + // Execute the request with retries + err := performProxyRequestWithRetries(c, proxyURL) + if err != nil { + // Log error that could potentially trip the circuit + cfg.Logger.Warning(&libpack_logger.LogMessage{ + Message: "Error in circuit-protected request", + Pairs: map[string]any{ + "path": c.Path(), + "error": err.Error(), + }, + }) + // Classify the failure so CIRCUIT_TRIP_ON_TIMEOUTS/_4XX/_5XX gate + // whether it counts toward the breaker's failure statistics + // (see circuitBreakerIsSuccessful). The error returned to the + // caller is unchanged either way -- only the breaker bookkeeping + // is affected. + countsAsFailure := classifyCircuitFailure(cfg, err, c.Response().StatusCode()) + if countsAsFailure { + // Mirrors MetricsCircuitSuccessful below: incremented once per + // Execute call, only for outcomes that actually count against + // the breaker's failure statistics, so a single request is + // never double-counted. + cfg.Monitoring.Increment(libpack_monitoring.MetricsCircuitFailed, nil) + } + return nil, &circuitBreakerOutcome{err: err, countsAsFailure: countsAsFailure} + } - // Check if non-2xx responses should trip the circuit - statusCode := c.Response().StatusCode() - if cfg.CircuitBreaker.TripOn5xx && statusCode >= 500 && statusCode < 600 { - err := fmt.Errorf("received 5xx status code: %d", statusCode) - cfg.Monitoring.Increment(libpack_monitoring.MetricsCircuitFailed, nil) - return nil, err - } - - // Request was successful - cfg.Monitoring.Increment(libpack_monitoring.MetricsCircuitSuccessful, nil) - return nil, nil - }) + // Request was successful + cfg.Monitoring.Increment(libpack_monitoring.MetricsCircuitSuccessful, nil) + return nil, nil + }) + } // If the circuit is open, implement graceful degradation if err == gobreaker.ErrOpenState { @@ -494,7 +838,7 @@ func performProxyRequestWithRetries(c fiber.Ctx, proxyURL string) error { } // executeProxyAttempt performs a single proxy attempt with error handling -func executeProxyAttempt(c fiber.Ctx, proxyURL string) error { +func executeProxyAttempt(c fiber.Ctx, req *fasthttp.Request) error { // Additional safety check inside retry loop if c == nil { return retry.Unrecoverable(errFiberCtxNilDuringRetry) @@ -503,8 +847,10 @@ func executeProxyAttempt(c fiber.Ctx, proxyURL string) error { // Get connection pool manager for stats tracking poolMgr := GetConnectionPoolManager() - // Execute the proxy request - proxyErr := doProxyRequestWithTimeout(c, proxyURL, cfg.Client.FastProxyClient) + // Execute the proxy request. req was built once by + // performProxyRequestWithEnhancedRetries and is reused unmodified across + // every attempt (see the comment there for why that's safe). + proxyErr := doProxyRequestWithTimeout(c, req, cfg.Client.FastProxyClient) if proxyErr != nil { // Check if this is a connection error if isConnectionError(proxyErr) { @@ -576,6 +922,17 @@ func performProxyRequestWithEnhancedRetries(c fiber.Ctx, proxyURL string, backen return errFiberCtxNil } + // Build the outbound request once, before the retry loop, instead of + // re-copying c.Request() (headers + body) on every attempt. A retry + // replays the exact same request, so the copy-in is invariant across + // attempts; only the response is fresh per attempt (doProxyRequestWithTimeout + // acquires/releases its own *fasthttp.Response each call). This previously + // ran c.Request().CopyTo(req) up to 7x per proxied request. + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + c.Request().CopyTo(req) + req.SetRequestURI(proxyURL) + var attempts uint var initialDelay time.Duration var maxDelayTime time.Duration @@ -595,7 +952,7 @@ func performProxyRequestWithEnhancedRetries(c fiber.Ctx, proxyURL string, backen return retry.Do( func() error { - return executeProxyAttempt(c, proxyURL) + return executeProxyAttempt(c, req) }, retry.Attempts(attempts), retry.DelayType(retry.BackOffDelay), @@ -773,24 +1130,22 @@ func handleCircuitOpenGracefulDegradation(c fiber.Ctx, cacheKey string) error { return ErrCircuitOpen } -// doProxyRequestWithTimeout performs a proxy request with proper timeout handling -func doProxyRequestWithTimeout(c fiber.Ctx, proxyURL string, client *fasthttp.Client) error { +// doProxyRequestWithTimeout performs a proxy request with proper timeout +// handling. req is the already-built outbound request (see +// performProxyRequestWithEnhancedRetries); only the response is +// acquired/released per call, so each retry attempt gets fresh +// per-attempt response state while reusing the same request. +func doProxyRequestWithTimeout(c fiber.Ctx, req *fasthttp.Request, client *fasthttp.Client) error { // Calculate timeout from client configuration clientTimeout := time.Duration(cfg.Client.ClientTimeout) * time.Second if clientTimeout <= 0 { clientTimeout = 30 * time.Second } - // Acquire request and response objects - req := fasthttp.AcquireRequest() + // Acquire a fresh response object for this attempt resp := fasthttp.AcquireResponse() - defer fasthttp.ReleaseRequest(req) defer fasthttp.ReleaseResponse(resp) - // Copy the original request - c.Request().CopyTo(req) - req.SetRequestURI(proxyURL) - // Perform the request with timeout err := client.DoTimeout(req, resp, clientTimeout) if err != nil { diff --git a/proxy_test.go b/proxy_test.go index b6549c9..e01070d 100644 --- a/proxy_test.go +++ b/proxy_test.go @@ -1,6 +1,7 @@ package main import ( + "io" "net/http" "net/http/httptest" "sync/atomic" @@ -344,3 +345,62 @@ func (suite *Tests) Test_proxyRetryPolicyByStatusCode() { suite.Greater(hits.Load(), int32(1), "5xx should trigger retries") }) } + +// Test_MultiAttemptRetryReusesRequestBody locks in the P3 fix: the outbound +// fasthttp.Request (headers + body) is built once, before the retry loop in +// performProxyRequestWithEnhancedRetries, and reused unmodified across every +// attempt -- instead of being re-copied from c.Request() up to 7x per +// proxied request (doProxyRequestWithTimeout previously ran +// c.Request().CopyTo(req) on every attempt). This verifies the hoist didn't +// corrupt, truncate, or drop the body on any attempt: every retried hit +// must still observe the exact original request body. +func (suite *Tests) Test_MultiAttemptRetryReusesRequestBody() { + originalCoalescing := cfg.RequestCoalescing.Enable + originalCB := cfg.CircuitBreaker.Enable + cfg.RequestCoalescing.Enable = false + cfg.CircuitBreaker.Enable = false + defer func() { + cfg.RequestCoalescing.Enable = originalCoalescing + cfg.CircuitBreaker.Enable = originalCB + }() + + const wantBody = `{"query":"query { test }"}` + + var hits atomic.Int32 + var bodyMismatch atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := hits.Add(1) + body, _ := io.ReadAll(r.Body) + if string(body) != wantBody { + bodyMismatch.Store(true) + } + if n < 3 { + // Fail the first two attempts so the request is retried at + // least twice, exercising the reused request across attempts. + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte(`{"errors":[{"message":"error"}]}`)) + return + } + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"data":{"ok":true}}`)) + })) + defer server.Close() + + cfg.Server.HostGraphQL = server.URL + cfg.Client.ClientTimeout = 5 + cfg.Client.FastProxyClient = createFasthttpClient(cfg) + + reqCtx := &fasthttp.RequestCtx{} + reqCtx.Request.SetRequestURI("/graphql") + reqCtx.Request.Header.SetMethod("POST") + reqCtx.Request.Header.Set("Content-Type", "application/json") + reqCtx.Request.SetBody([]byte(wantBody)) + + ctx := suite.app.AcquireCtx(reqCtx) + err := proxyTheRequest(ctx, cfg.Server.HostGraphQL) + suite.app.ReleaseCtx(ctx) + + suite.Nil(err, "request should eventually succeed after retries") + suite.GreaterOrEqual(hits.Load(), int32(3), "must have retried at least twice before succeeding") + suite.False(bodyMismatch.Load(), "every retry attempt must see the exact original request body, not a stale or corrupted copy") +}