mirror of
https://github.com/lukaszraczylo/graphql-monitoring-proxy.git
synced 2026-10-12 05:32:36 +00:00
feat(circuit): progressive backoff and configurable trip conditions
The circuit breaker ignored the trip-on-timeout, trip-on-4xx, and trip-on-5xx flags and never cleared its counts, so the failure ratio became lifetime-cumulative. The backoff-multiplier and max-backoff settings did nothing. Circuit-open surfaced as HTTP 500. Failure classification now honours the trip flags, the breaker uses a rolling count window, and repeated trips extend the open state up to the configured maximum. A helper maps circuit-open to HTTP 503. The request body is copied once per request instead of once per retry. Defaults reproduce the prior behaviour: the breaker is off by default and a backoff multiplier of one keeps a constant open window.
This commit is contained in:
@@ -4,16 +4,21 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/gofiber/fiber/v3"
|
||||||
libpack_cache "github.com/lukaszraczylo/graphql-monitoring-proxy/cache"
|
libpack_cache "github.com/lukaszraczylo/graphql-monitoring-proxy/cache"
|
||||||
libpack_cache_memory "github.com/lukaszraczylo/graphql-monitoring-proxy/cache/memory"
|
libpack_cache_memory "github.com/lukaszraczylo/graphql-monitoring-proxy/cache/memory"
|
||||||
libpack_logger "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
|
libpack_logger "github.com/lukaszraczylo/graphql-monitoring-proxy/logging"
|
||||||
libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring"
|
libpack_monitoring "github.com/lukaszraczylo/graphql-monitoring-proxy/monitoring"
|
||||||
"github.com/sony/gobreaker"
|
"github.com/sony/gobreaker"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/suite"
|
"github.com/stretchr/testify/suite"
|
||||||
|
"github.com/valyala/fasthttp"
|
||||||
)
|
)
|
||||||
|
|
||||||
// CircuitBreakerTestSuite is a test suite for circuit breaker functionality
|
// CircuitBreakerTestSuite is a test suite for circuit breaker functionality
|
||||||
@@ -69,6 +74,9 @@ func (suite *CircuitBreakerTestSuite) TearDownTest() {
|
|||||||
cb = nil
|
cb = nil
|
||||||
// Circuit breaker metrics are now managed by cbMetrics
|
// Circuit breaker metrics are now managed by cbMetrics
|
||||||
cbMetrics = nil
|
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
|
// 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")
|
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
|
// Start the test suite
|
||||||
func TestCircuitBreakerSuite(t *testing.T) {
|
func TestCircuitBreakerSuite(t *testing.T) {
|
||||||
suite.Run(t, new(CircuitBreakerTestSuite))
|
suite.Run(t, new(CircuitBreakerTestSuite))
|
||||||
|
|||||||
@@ -5,6 +5,8 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/sony/gobreaker"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Error codes for structured error responses
|
// Error codes for structured error responses
|
||||||
@@ -143,3 +145,28 @@ func GetStatusCode(err error) int {
|
|||||||
|
|
||||||
return 500
|
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
|
||||||
|
}
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"github.com/sony/gobreaker"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -260,3 +261,62 @@ func Test_IsRetryable_GetStatusCode_Wrapped(t *testing.T) {
|
|||||||
assert.False(t, IsRetryable(nil))
|
assert.False(t, IsRetryable(nil))
|
||||||
assert.Equal(t, 200, GetStatusCode(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))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ import (
|
|||||||
"net/url"
|
"net/url"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"go.opentelemetry.io/otel/trace"
|
"go.opentelemetry.io/otel/trace"
|
||||||
@@ -49,12 +50,121 @@ const (
|
|||||||
defaultMaxRequestsInHalfOpen = 10 // Default maximum requests in half-open state
|
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
|
// Global circuit breaker
|
||||||
var (
|
var (
|
||||||
cb *gobreaker.CircuitBreaker
|
cb *gobreaker.CircuitBreaker
|
||||||
cbMutex sync.RWMutex
|
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.
|
// Package-level substring tables used by isConnectionError / isTimeoutError.
|
||||||
// Hoisted to avoid per-call slice allocations on the hot path. All entries
|
// 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.
|
// must be lower-case; callers lower-case the error string once before matching.
|
||||||
@@ -107,17 +217,26 @@ func initCircuitBreaker(config *config) {
|
|||||||
cbMutex.Lock()
|
cbMutex.Lock()
|
||||||
defer cbMutex.Unlock()
|
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
|
// Initialize circuit breaker metrics
|
||||||
InitializeCircuitBreakerMetrics(config.Monitoring)
|
InitializeCircuitBreakerMetrics(config.Monitoring)
|
||||||
|
|
||||||
// Create circuit breaker settings
|
// Create circuit breaker settings
|
||||||
cbSettings := gobreaker.Settings{
|
cbSettings := gobreaker.Settings{
|
||||||
Name: "graphql-proxy-circuit",
|
Name: "graphql-proxy-circuit",
|
||||||
MaxRequests: safeMaxRequests(config.CircuitBreaker.MaxRequestsInHalfOpen),
|
MaxRequests: safeMaxRequests(config.CircuitBreaker.MaxRequestsInHalfOpen),
|
||||||
Interval: 0, // No specific interval for counting failures
|
// 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,
|
Timeout: time.Duration(config.CircuitBreaker.Timeout) * time.Second,
|
||||||
ReadyToTrip: createTripFunc(config),
|
ReadyToTrip: createTripFunc(config),
|
||||||
OnStateChange: createStateChangeFunc(config),
|
OnStateChange: createStateChangeFunc(config),
|
||||||
|
IsSuccessful: circuitBreakerIsSuccessful,
|
||||||
}
|
}
|
||||||
|
|
||||||
// Initialize the circuit breaker
|
// 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
|
// createStateChangeFunc returns a function that handles circuit state changes
|
||||||
func createStateChangeFunc(config *config) func(name string, from gobreaker.State, to gobreaker.State) {
|
func createStateChangeFunc(config *config) func(name string, from gobreaker.State, to gobreaker.State) {
|
||||||
return 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 stateValue float64
|
||||||
var stateName string
|
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.
|
// createFasthttpClient creates and configures a fasthttp client with optimized settings.
|
||||||
// The client is configured based on the provided configuration settings, with careful
|
// The client is configured based on the provided configuration settings, with careful
|
||||||
// attention to performance and security considerations.
|
// attention to performance and security considerations.
|
||||||
@@ -437,35 +765,51 @@ func performProxyRequestCore(c fiber.Ctx, proxyURL string, cacheKey string) erro
|
|||||||
return performProxyRequestWithRetries(c, proxyURL)
|
return performProxyRequestWithRetries(c, proxyURL)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Execute request through circuit breaker
|
var err error
|
||||||
_, err := cb.Execute(func() (any, error) {
|
if circuitBackoffGateBlocks(cfg, cb, time.Now()) {
|
||||||
// Execute the request with retries
|
// Progressive backoff (C7): gobreaker's own Timeout has already
|
||||||
err := performProxyRequestWithRetries(c, proxyURL)
|
// elapsed -- it would otherwise let a half-open probe through -- but
|
||||||
// Check if the error or status code should trip the circuit breaker
|
// the extended per-trip backoff window has not. Reject exactly like
|
||||||
if err != nil {
|
// an open breaker, without calling cb.Execute: no backend call, and
|
||||||
// Log error that could potentially trip the circuit
|
// gobreaker's own Counts/generation are left untouched. gobreaker
|
||||||
cfg.Logger.Warning(&libpack_logger.LogMessage{
|
// stays idle in half-open until the extended window also elapses,
|
||||||
Message: "Error in circuit-protected request",
|
// at which point the Execute call below performs the real probe.
|
||||||
Pairs: map[string]any{
|
err = gobreaker.ErrOpenState
|
||||||
"path": c.Path(),
|
} else {
|
||||||
"error": err.Error(),
|
// Execute request through circuit breaker
|
||||||
},
|
_, err = cb.Execute(func() (any, error) {
|
||||||
})
|
// Execute the request with retries
|
||||||
return nil, err
|
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
|
// Request was successful
|
||||||
statusCode := c.Response().StatusCode()
|
cfg.Monitoring.Increment(libpack_monitoring.MetricsCircuitSuccessful, nil)
|
||||||
if cfg.CircuitBreaker.TripOn5xx && statusCode >= 500 && statusCode < 600 {
|
return nil, nil
|
||||||
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
|
|
||||||
})
|
|
||||||
|
|
||||||
// If the circuit is open, implement graceful degradation
|
// If the circuit is open, implement graceful degradation
|
||||||
if err == gobreaker.ErrOpenState {
|
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
|
// 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
|
// Additional safety check inside retry loop
|
||||||
if c == nil {
|
if c == nil {
|
||||||
return retry.Unrecoverable(errFiberCtxNilDuringRetry)
|
return retry.Unrecoverable(errFiberCtxNilDuringRetry)
|
||||||
@@ -503,8 +847,10 @@ func executeProxyAttempt(c fiber.Ctx, proxyURL string) error {
|
|||||||
// Get connection pool manager for stats tracking
|
// Get connection pool manager for stats tracking
|
||||||
poolMgr := GetConnectionPoolManager()
|
poolMgr := GetConnectionPoolManager()
|
||||||
|
|
||||||
// Execute the proxy request
|
// Execute the proxy request. req was built once by
|
||||||
proxyErr := doProxyRequestWithTimeout(c, proxyURL, cfg.Client.FastProxyClient)
|
// 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 {
|
if proxyErr != nil {
|
||||||
// Check if this is a connection error
|
// Check if this is a connection error
|
||||||
if isConnectionError(proxyErr) {
|
if isConnectionError(proxyErr) {
|
||||||
@@ -576,6 +922,17 @@ func performProxyRequestWithEnhancedRetries(c fiber.Ctx, proxyURL string, backen
|
|||||||
return errFiberCtxNil
|
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 attempts uint
|
||||||
var initialDelay time.Duration
|
var initialDelay time.Duration
|
||||||
var maxDelayTime time.Duration
|
var maxDelayTime time.Duration
|
||||||
@@ -595,7 +952,7 @@ func performProxyRequestWithEnhancedRetries(c fiber.Ctx, proxyURL string, backen
|
|||||||
|
|
||||||
return retry.Do(
|
return retry.Do(
|
||||||
func() error {
|
func() error {
|
||||||
return executeProxyAttempt(c, proxyURL)
|
return executeProxyAttempt(c, req)
|
||||||
},
|
},
|
||||||
retry.Attempts(attempts),
|
retry.Attempts(attempts),
|
||||||
retry.DelayType(retry.BackOffDelay),
|
retry.DelayType(retry.BackOffDelay),
|
||||||
@@ -773,24 +1130,22 @@ func handleCircuitOpenGracefulDegradation(c fiber.Ctx, cacheKey string) error {
|
|||||||
return ErrCircuitOpen
|
return ErrCircuitOpen
|
||||||
}
|
}
|
||||||
|
|
||||||
// doProxyRequestWithTimeout performs a proxy request with proper timeout handling
|
// doProxyRequestWithTimeout performs a proxy request with proper timeout
|
||||||
func doProxyRequestWithTimeout(c fiber.Ctx, proxyURL string, client *fasthttp.Client) error {
|
// 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
|
// Calculate timeout from client configuration
|
||||||
clientTimeout := time.Duration(cfg.Client.ClientTimeout) * time.Second
|
clientTimeout := time.Duration(cfg.Client.ClientTimeout) * time.Second
|
||||||
if clientTimeout <= 0 {
|
if clientTimeout <= 0 {
|
||||||
clientTimeout = 30 * time.Second
|
clientTimeout = 30 * time.Second
|
||||||
}
|
}
|
||||||
|
|
||||||
// Acquire request and response objects
|
// Acquire a fresh response object for this attempt
|
||||||
req := fasthttp.AcquireRequest()
|
|
||||||
resp := fasthttp.AcquireResponse()
|
resp := fasthttp.AcquireResponse()
|
||||||
defer fasthttp.ReleaseRequest(req)
|
|
||||||
defer fasthttp.ReleaseResponse(resp)
|
defer fasthttp.ReleaseResponse(resp)
|
||||||
|
|
||||||
// Copy the original request
|
|
||||||
c.Request().CopyTo(req)
|
|
||||||
req.SetRequestURI(proxyURL)
|
|
||||||
|
|
||||||
// Perform the request with timeout
|
// Perform the request with timeout
|
||||||
err := client.DoTimeout(req, resp, clientTimeout)
|
err := client.DoTimeout(req, resp, clientTimeout)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
@@ -344,3 +345,62 @@ func (suite *Tests) Test_proxyRetryPolicyByStatusCode() {
|
|||||||
suite.Greater(hits.Load(), int32(1), "5xx should trigger retries")
|
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")
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user