From eec076f812224060c696e2574060a2c14c164f95 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Tue, 3 Nov 2026 00:46:46 +0000 Subject: [PATCH] fix(resilience): recover from unhealthy and close half-open on successes The health check promoted to healthy only when the fail counter was above zero. The first success zeroed that counter, so a threshold of 2 or more never recovered. It now promotes on the status and the success streak. The breaker closed when admitted probes reached HalfOpenMaxRequests, not when that many succeeded. A new counter tracks successes. --- internal/cache/resilience/circuit_breaker.go | 9 ++- .../cache/resilience/health_check_backend.go | 7 ++- .../cache/resilience/recovery_bounds_test.go | 56 +++++++++++++++++++ 3 files changed, 66 insertions(+), 6 deletions(-) create mode 100644 internal/cache/resilience/recovery_bounds_test.go diff --git a/internal/cache/resilience/circuit_breaker.go b/internal/cache/resilience/circuit_breaker.go index ea1e7fb..68f8413 100644 --- a/internal/cache/resilience/circuit_breaker.go +++ b/internal/cache/resilience/circuit_breaker.go @@ -81,6 +81,7 @@ type CircuitBreaker struct { stateMu sync.RWMutex timeMu sync.RWMutex halfOpenRequests atomic.Int32 + halfOpenSuccesses atomic.Int32 consecutiveFailures atomic.Int32 state atomic.Int32 } @@ -170,12 +171,12 @@ func (cb *CircuitBreaker) RecordSuccess() { case StateHalfOpen: // If we've had enough successful requests, close the circuit - successfulRequests := cb.halfOpenRequests.Load() - // #nosec G115 -- HalfOpenMaxRequests is a small config value that fits in int32 - if successfulRequests >= int32(cb.config.HalfOpenMaxRequests) { + successfulRequests := cb.halfOpenSuccesses.Add(1) + if int64(successfulRequests) >= int64(cb.config.HalfOpenMaxRequests) { cb.setState(StateClosed) cb.consecutiveFailures.Store(0) cb.halfOpenRequests.Store(0) + cb.halfOpenSuccesses.Store(0) } } } @@ -216,6 +217,7 @@ func (cb *CircuitBreaker) RecordFailure() { func (cb *CircuitBreaker) openCircuit() { cb.setState(StateOpen) cb.halfOpenRequests.Store(0) + cb.halfOpenSuccesses.Store(0) cb.timeMu.Lock() cb.nextRetryTime = time.Now().Add(cb.config.Timeout) @@ -251,6 +253,7 @@ func (cb *CircuitBreaker) Reset() { cb.totalRequests.Store(0) cb.totalFailures.Store(0) cb.halfOpenRequests.Store(0) + cb.halfOpenSuccesses.Store(0) cb.rejectedRequests.Store(0) cb.stateTransitions.Store(0) diff --git a/internal/cache/resilience/health_check_backend.go b/internal/cache/resilience/health_check_backend.go index 981bb1e..cff7f1a 100644 --- a/internal/cache/resilience/health_check_backend.go +++ b/internal/cache/resilience/health_check_backend.go @@ -168,11 +168,12 @@ func (h *HealthCheckBackend) IsHealthy() bool { func (h *HealthCheckBackend) recordResult(success bool) { // #nosec G115 -- threshold config values are small integers that fit in int32 if success { - fails := h.consecutiveFails.Swap(0) + h.consecutiveFails.Store(0) oks := h.consecutiveOK.Add(1) - // Check if we should transition to healthy - if fails > 0 && oks >= int32(h.config.HealthyThreshold) { + // Promote on a success streak. Keyed on the status, not on the fail + // counter, which the first success of the streak already zeroed. + if HealthStatus(h.status.Load()) != HealthHealthy && oks >= int32(h.config.HealthyThreshold) { oldStatus := HealthStatus(h.status.Swap(int32(HealthHealthy))) if oldStatus != HealthHealthy && h.config.OnStatusChange != nil { h.config.OnStatusChange(oldStatus, HealthHealthy) diff --git a/internal/cache/resilience/recovery_bounds_test.go b/internal/cache/resilience/recovery_bounds_test.go new file mode 100644 index 0000000..011809a --- /dev/null +++ b/internal/cache/resilience/recovery_bounds_test.go @@ -0,0 +1,56 @@ +package resilience + +import ( + "testing" + "time" +) + +// With HealthyThreshold >= 2 an Unhealthy backend must recover once it +// answers HealthyThreshold times in a row. +func TestHealthCheckRecoversWithThresholdTwo(t *testing.T) { + h := &HealthCheckBackend{config: &HealthCheckConfig{HealthyThreshold: 2, UnhealthyThreshold: 3}} + for i := 0; i < 6; i++ { + h.recordResult(false) + } + if HealthStatus(h.status.Load()) != HealthUnhealthy { + t.Fatalf("precondition: status=%v, want unhealthy", HealthStatus(h.status.Load())) + } + for i := 0; i < 5; i++ { + h.recordResult(true) + } + if got := HealthStatus(h.status.Load()); got != HealthHealthy { + t.Fatalf("status after 5 successes = %v, want healthy", got) + } +} + +// Half-open closes after HalfOpenMaxRequests successes, not admissions. +func TestHalfOpenClosesOnSuccessesNotAdmissions(t *testing.T) { + cb := NewCircuitBreaker(&CircuitBreakerConfig{ + MaxFailures: 1, Timeout: time.Millisecond, HalfOpenMaxRequests: 3, ResetTimeout: time.Second, + }) + cb.RecordFailure() + time.Sleep(20 * time.Millisecond) + for i := 0; i < 3; i++ { + if !cb.AllowRequest() { + t.Fatalf("probe %d rejected", i) + } + } + cb.RecordSuccess() + if cb.GetState() != StateHalfOpen { + t.Fatalf("closed after 1 of 3 probe successes; state=%v", cb.GetState()) + } + cb.RecordFailure() + if cb.GetState() != StateOpen { + t.Fatalf("a failing probe must reopen; state=%v", cb.GetState()) + } + time.Sleep(20 * time.Millisecond) + for i := 0; i < 3; i++ { + cb.AllowRequest() + } + for i := 0; i < 3; i++ { + cb.RecordSuccess() + } + if cb.GetState() != StateClosed { + t.Fatalf("state after 3 successes = %v, want closed", cb.GetState()) + } +}