diff --git a/refresh_coordinator.go b/refresh_coordinator.go index 80cb187..d270f0d 100644 --- a/refresh_coordinator.go +++ b/refresh_coordinator.go @@ -672,7 +672,6 @@ func (rc *RefreshCoordinator) isInCooldown(sessionID string) bool { } return &attemptState{ windowStartNano: nowNano, - attempts: 1, } }) return false @@ -686,7 +685,7 @@ func (rc *RefreshCoordinator) isInCooldown(sessionID string) bool { } next := *s next.windowStartNano = nowNano - next.attempts = 1 + next.attempts = 0 return &next }) return false @@ -939,20 +938,28 @@ func (cb *RefreshCircuitBreaker) AllowRequest() bool { return true case 1: // open lastFail := atomic.LoadInt64(&cb.lastFailureNano) - if time.Duration(time.Now().UnixNano()-lastFail) > cb.config.OpenDuration { - // Transition to half-open; first CAS winner gets the probe. - if atomic.CompareAndSwapInt32(&cb.state, 1, 2) { - return true - } - } - return false - case 2: // half-open - return true + return cb.claimProbe(lastFail) + case 2: // half-open: one probe at a time; a stale probe is replaced + return cb.claimProbe(atomic.LoadInt64(&cb.lastFailureNano)) default: return false } } +// claimProbe admits one probe when OpenDuration has passed since lastFail. +// The CAS on lastFailureNano makes the claim exclusive and restarts the clock. +func (cb *RefreshCircuitBreaker) claimProbe(lastFail int64) bool { + now := time.Now().UnixNano() + if time.Duration(now-lastFail) <= cb.config.OpenDuration { + return false + } + if !atomic.CompareAndSwapInt64(&cb.lastFailureNano, lastFail, now) { + return false + } + atomic.StoreInt32(&cb.state, 2) + return true +} + // RecordSuccess records a successful operation. Lock-free. func (cb *RefreshCircuitBreaker) RecordSuccess() { switch atomic.LoadInt32(&cb.state) { diff --git a/refresh_window_halfopen_test.go b/refresh_window_halfopen_test.go new file mode 100644 index 0000000..ced95df --- /dev/null +++ b/refresh_window_halfopen_test.go @@ -0,0 +1,93 @@ +package traefikoidc + +import ( + "sync" + "sync/atomic" + "testing" + "time" +) + +// After a window reset the tracker must allow the full MaxRefreshAttempts, +// the same as a fresh tracker. +func TestRefreshWindowResetAllowsFullQuota(t *testing.T) { + cfg := DefaultRefreshCoordinatorConfig() + cfg.MaxRefreshAttempts = 3 + cfg.RefreshAttemptWindow = 50 * time.Millisecond + cfg.RefreshCooldownPeriod = time.Hour + rc := NewRefreshCoordinator(cfg, GetSingletonNoOpLogger()) + defer rc.Shutdown() + + pass := func() int { + n := 0 + for i := 0; i < 10; i++ { + if rc.isInCooldown("s") { + break + } + rc.recordRefreshAttempt("s") + n++ + } + return n + } + if got := pass(); got != 3 { + t.Fatalf("fresh window passes=%d want 3", got) + } + // Window expiry alone cannot reset while in cooldown, so use a new + // session that never hits the cap: record 1, wait out the window. + rc.recordRefreshAttempt("w") + time.Sleep(80 * time.Millisecond) + n := 0 + for i := 0; i < 10; i++ { + if rc.isInCooldown("w") { + break + } + rc.recordRefreshAttempt("w") + n++ + } + if n != 3 { + t.Fatalf("post-reset window passes=%d want 3", n) + } +} + +// Half-open admits a single probe until it reports. +func TestRefreshBreakerHalfOpenSingleProbe(t *testing.T) { + cb := &RefreshCircuitBreaker{config: RefreshCircuitBreakerConfig{ + MaxFailures: 1, OpenDuration: 20 * time.Millisecond, HalfOpenRequests: 1, + }} + cb.RecordFailure() + time.Sleep(40 * time.Millisecond) + var admitted int32 + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if cb.AllowRequest() { + atomic.AddInt32(&admitted, 1) + } + }() + } + wg.Wait() + if admitted != 1 { + t.Fatalf("half-open admitted %d, want 1", admitted) + } + cb.RecordSuccess() + if !cb.AllowRequest() { + t.Fatal("breaker must close after probe success") + } +} + +// A probe that never reports must not wedge the breaker. +func TestRefreshBreakerHalfOpenProbeTimeout(t *testing.T) { + cb := &RefreshCircuitBreaker{config: RefreshCircuitBreakerConfig{ + MaxFailures: 1, OpenDuration: 20 * time.Millisecond, HalfOpenRequests: 1, + }} + cb.RecordFailure() + time.Sleep(40 * time.Millisecond) + if !cb.AllowRequest() || cb.AllowRequest() { + t.Fatal("want exactly one probe") + } + time.Sleep(40 * time.Millisecond) + if !cb.AllowRequest() { + t.Fatal("stale probe must allow a new one") + } +}