diff --git a/.design/priority-routing.md b/.design/priority-routing.md index c8a8fa993..3fdea3817 100644 --- a/.design/priority-routing.md +++ b/.design/priority-routing.md @@ -58,8 +58,8 @@ Three properties fall out: The breaker is a three-state machine (`Closed → Open → HalfOpen`): - **Closed** — normal. `Allow()` returns true, failures are counted. -- **Open** — too many consecutive failures (`FailureThreshold`, default 3). `Allow()` returns false. After `OpenDuration` (default 30 s) the next `Allow()` call lazily flips to HalfOpen. -- **HalfOpen** — exactly one probe is permitted. Success → Closed, failure → Open with a fresh timer. +- **Open** — too many consecutive failures (`FailureThreshold`, default 3). `Allow()` returns false. After `OpenDuration` (default 60 s) the next `Allow()` call lazily flips to HalfOpen. +- **HalfOpen** — exactly one probe is permitted. Success → Closed, failure → Open with exponential backoff (60 s → 120 s → 240 s → 5 min cap). Success resets backoff to the base duration. Recovery requires **no separate scheduler**. Selection re-evaluates the tier list every request, and the breaker's lazy state transition admits one probe naturally. Active probing was considered and rejected for v1 — for hot rules it's redundant, and for cold rules there is no one to serve anyway. diff --git a/internal/loadbalance/breaker.go b/internal/loadbalance/breaker.go index 8553b4ba7..9c3034e9e 100644 --- a/internal/loadbalance/breaker.go +++ b/internal/loadbalance/breaker.go @@ -17,7 +17,8 @@ const ( // Default circuit breaker tunables. const ( DefaultBreakerFailureThreshold = 3 - DefaultBreakerOpenDuration = 30 * time.Second + DefaultBreakerOpenDuration = 60 * time.Second + DefaultBreakerMaxOpenDuration = 5 * time.Minute ) // Breaker is a simple three-state circuit breaker for a single service. @@ -37,9 +38,11 @@ type Breaker struct { consecFails int openedAt time.Time halfOpenInFlight bool + halfOpenFails int // consecutive half-open probe failures for backoff FailureThreshold int OpenDuration time.Duration + MaxOpenDuration time.Duration } // NewBreaker creates a breaker with the supplied thresholds. Zero values @@ -55,6 +58,7 @@ func NewBreaker(failureThreshold int, openDuration time.Duration) *Breaker { state: BreakerClosed, FailureThreshold: failureThreshold, OpenDuration: openDuration, + MaxOpenDuration: DefaultBreakerMaxOpenDuration, } } @@ -69,7 +73,7 @@ func (b *Breaker) Allow() bool { case BreakerClosed: return true case BreakerOpen: - if time.Since(b.openedAt) >= b.OpenDuration { + if time.Since(b.openedAt) >= b.currentOpenDuration() { b.state = BreakerHalfOpen b.halfOpenInFlight = true return true @@ -91,11 +95,13 @@ func (b *Breaker) RecordSuccess() { defer b.mu.Unlock() b.state = BreakerClosed b.consecFails = 0 + b.halfOpenFails = 0 b.halfOpenInFlight = false } // RecordFailure increments failure tracking and trips the breaker when -// the threshold is reached. A failure during HalfOpen immediately re-opens. +// the threshold is reached. A failure during HalfOpen immediately re-opens +// and increases the backoff for the next open window. func (b *Breaker) RecordFailure() { b.mu.Lock() defer b.mu.Unlock() @@ -103,6 +109,7 @@ func (b *Breaker) RecordFailure() { if b.state == BreakerHalfOpen { b.state = BreakerOpen b.openedAt = time.Now() + b.halfOpenFails++ b.halfOpenInFlight = false return } @@ -113,12 +120,26 @@ func (b *Breaker) RecordFailure() { } } +// currentOpenDuration returns the backoff-adjusted open duration. +// Each consecutive half-open failure doubles the wait (e.g. 60s → 120s → 240s → cap). +// Must be called with b.mu held. +func (b *Breaker) currentOpenDuration() time.Duration { + d := b.OpenDuration + for i := 0; i < b.halfOpenFails; i++ { + d *= 2 + if d >= b.MaxOpenDuration { + return b.MaxOpenDuration + } + } + return d +} + // State returns the current breaker state. Intended for introspection / UI. func (b *Breaker) State() BreakerState { b.mu.Lock() defer b.mu.Unlock() // Apply the lazy Open→HalfOpen transition for read consistency. - if b.state == BreakerOpen && time.Since(b.openedAt) >= b.OpenDuration { + if b.state == BreakerOpen && time.Since(b.openedAt) >= b.currentOpenDuration() { return BreakerHalfOpen } return b.state diff --git a/internal/loadbalance/breaker_test.go b/internal/loadbalance/breaker_test.go index af7b15ad5..461807188 100644 --- a/internal/loadbalance/breaker_test.go +++ b/internal/loadbalance/breaker_test.go @@ -73,6 +73,90 @@ func TestBreakerSuccessResetsCounter(t *testing.T) { } } +// windForward moves a breaker's openedAt backwards in time so the open +// window appears to have elapsed, without any real time.Sleep. +func windForward(b *Breaker, d time.Duration) { + b.mu.Lock() + b.openedAt = b.openedAt.Add(-d) + b.mu.Unlock() +} + +func TestBreakerExponentialBackoff(t *testing.T) { + base := 100 * time.Millisecond + b := NewBreaker(1, base) + b.MaxOpenDuration = 500 * time.Millisecond + + b.RecordFailure() // → open + + // 1st window: base (100 ms). Fast-forward past it. + windForward(b, base) + if !b.Allow() { + t.Fatal("should allow probe after 1st open window") + } + b.RecordFailure() // half-open → open, halfOpenFails=1 + + // 2nd window: 200 ms. Advance only 150 ms — too early. + windForward(b, 150*time.Millisecond) + if b.Allow() { + t.Fatal("should NOT allow probe yet (backoff doubled to 200 ms)") + } + // Advance the remaining 50 ms. + windForward(b, 50*time.Millisecond) + if !b.Allow() { + t.Fatal("should allow probe after 2nd backoff window") + } + b.RecordFailure() // halfOpenFails=2 + + // 3rd window: 400 ms. Advance only 300 ms — too early. + windForward(b, 300*time.Millisecond) + if b.Allow() { + t.Fatal("should NOT allow probe yet (backoff at 400 ms)") + } + windForward(b, 100*time.Millisecond) + if !b.Allow() { + t.Fatal("should allow probe after 3rd backoff window") + } + b.RecordFailure() // halfOpenFails=3, next would be 800 ms → capped at 500 ms + + // 4th window: capped at 500 ms. Advance only 450 ms — too early. + windForward(b, 450*time.Millisecond) + if b.Allow() { + t.Fatal("should NOT allow probe yet (capped at 500 ms)") + } + windForward(b, 50*time.Millisecond) + if !b.Allow() { + t.Fatal("should allow probe after cap-limited window") + } +} + +func TestBreakerBackoffResetsOnSuccess(t *testing.T) { + base := 100 * time.Millisecond + b := NewBreaker(1, base) + b.MaxOpenDuration = 5 * time.Second + + // Build up backoff: trip → probe fail → trip → probe fail. + b.RecordFailure() + windForward(b, base) + b.Allow() + b.RecordFailure() // halfOpenFails=1, next window=200 ms + + windForward(b, 200*time.Millisecond) + b.Allow() + + // Probe succeeds → everything resets. + b.RecordSuccess() + if b.State() != BreakerClosed { + t.Fatal("should be closed after success") + } + + // Trip again — backoff should be back to base, not 400 ms. + b.RecordFailure() + windForward(b, base) + if !b.Allow() { + t.Fatal("after success+retrip, should probe at base duration, not backed-off") + } +} + func TestBreakerStoreLazyCreation(t *testing.T) { store := NewBreakerStore(2, time.Second) b1 := store.Get("svc:a") diff --git a/internal/loadbalance/health_monitor.go b/internal/loadbalance/health_monitor.go index 926d7a1c8..86f2a04c9 100644 --- a/internal/loadbalance/health_monitor.go +++ b/internal/loadbalance/health_monitor.go @@ -317,6 +317,20 @@ func (hm *HealthMonitor) GetAllHealth() map[string]*ServiceHealth { return result } +// HasAuthError returns true if the service is currently marked with an auth +// error (401/403). Auth errors are permanent until the key is fixed, so +// callers that manage their own transient-failure recovery (e.g. circuit +// breakers) can still exclude these. +func (hm *HealthMonitor) HasAuthError(serviceID string) bool { + health := hm.getHealth(serviceID) + if health == nil { + return false + } + health.mutex.RLock() + defer health.mutex.RUnlock() + return health.AuthError +} + // ResetHealth manually resets a service's health to healthy func (hm *HealthMonitor) ResetHealth(serviceID string) { hm.recoverService(serviceID) diff --git a/internal/server/load_balancer.go b/internal/server/load_balancer.go index b19b5a438..bb572226a 100644 --- a/internal/server/load_balancer.go +++ b/internal/server/load_balancer.go @@ -76,6 +76,13 @@ func (lb *LoadBalancer) SelectService(rule *typ.Rule) (*loadbalance.Service, err return nil, fmt.Errorf("no active services for rule %s", rule.RequestModel) } + // Tier tactic has its own circuit-breaker-based health management with + // fast recovery (30 s). The full HealthFilter would hide services for the + // monitor's recovery window (5 min), defeating the breaker's quick + // failover/recovery. However auth errors (401/403) are permanent — a + // revoked key never self-heals — so those must still be filtered out. + isTier := rule.LBTactic.Type == loadbalance.TacticTier + // Filter healthy services using health filter. When every active service is // currently marked unhealthy (e.g. a transient 429 on a single-service rule, // or all services inside the recovery window at once), fall back to the full @@ -83,11 +90,19 @@ func (lb *LoadBalancer) SelectService(rule *typ.Rule) (*loadbalance.Service, err // is strictly better than a hard "no service available": the service may have // already recovered, and if it really is still failing the caller gets the // real upstream error (e.g. 429) rather than a confusing routing error. - healthyServices := lb.healthFilter.Filter(activeServices) - if len(healthyServices) == 0 { - logrus.Warnf("[load_balancer] all %d active services for rule %s are unhealthy; "+ - "falling back to active set", len(activeServices), rule.RequestModel) - healthyServices = activeServices + var healthyServices []*loadbalance.Service + if isTier { + healthyServices = lb.healthFilter.FilterAuthErrors(activeServices) + if len(healthyServices) == 0 { + healthyServices = activeServices + } + } else { + healthyServices = lb.healthFilter.Filter(activeServices) + if len(healthyServices) == 0 { + logrus.Warnf("[load_balancer] all %d active services for rule %s are unhealthy; "+ + "falling back to active set", len(activeServices), rule.RequestModel) + healthyServices = activeServices + } } // For single healthy service rules, return it directly diff --git a/internal/servertest/health_filter_test.go b/internal/servertest/health_filter_test.go index a1049cfaf..82ccd1907 100644 --- a/internal/servertest/health_filter_test.go +++ b/internal/servertest/health_filter_test.go @@ -1,6 +1,7 @@ package servertest import ( + "fmt" "testing" "time" @@ -335,3 +336,426 @@ func TestHealthFilter_InactiveServices(t *testing.T) { assert.Equal(t, "provider-active", service.Provider) } } + +// --- Tier tactic vs health filter interaction tests --- + +// newTierTestLB creates a LoadBalancer with a health monitor that has a long +// recovery timeout (simulating the production 5-min window) so we can verify +// that tier rules bypass it while non-tier rules honour it. +func newTierTestLB(t *testing.T) (*server.LoadBalancer, *loadbalance.HealthMonitor) { + t.Helper() + appConfig, err := config.NewAppConfig(config.WithConfigDir(t.TempDir())) + require.NoError(t, err) + + healthConfig := loadbalance.HealthMonitorConfig{ + ConsecutiveErrorThreshold: 3, + RecoveryTimeoutSeconds: 600, // 10 min — effectively "never recovers during this test" + } + hm := loadbalance.NewHealthMonitor(healthConfig) + hf := typ.NewHealthFilter(hm) + lb := server.NewLoadBalancer(appConfig.GetGlobalConfig(), hf) + t.Cleanup(lb.Stop) + return lb, hm +} + +func tierService(provider, model string, tier int) *loadbalance.Service { + return &loadbalance.Service{ + Provider: provider, + Model: model, + Tier: tier, + Active: true, + Weight: 1, + } +} + +// TestTierTactic_BypassesHealthFilter verifies that a tier-based rule still +// sees all active services even when the HealthMonitor marks T0 as unhealthy. +// Before the fix, the health filter would hide T0 for ~5 min, blocking the +// tier tactic's 30-second breaker recovery. +func TestTierTactic_BypassesHealthFilter(t *testing.T) { + lb, hm := newTierTestLB(t) + + primary := tierService("tier-bypass-p1", "m1", 0) + backup := tierService("tier-bypass-p2", "m1", 1) + rule := &typ.Rule{ + UUID: uuid.New().String(), + RequestModel: "test-model", + LBTactic: typ.Tactic{ + Type: loadbalance.TacticTier, + Params: typ.DefaultTierParams(), + }, + Services: []*loadbalance.Service{primary, backup}, + Active: true, + } + + // Mark T0 as unhealthy via the HealthMonitor (rate limited). + hm.ReportRateLimit(primary.ServiceID()) + assert.False(t, hm.IsHealthy(primary.ServiceID())) + + // Even though the HealthMonitor says T0 is unhealthy, the tier tactic + // should still see it (breaker is closed) and pick it. + svc, err := lb.SelectService(rule) + require.NoError(t, err) + require.NotNil(t, svc) + assert.Equal(t, primary.Provider, svc.Provider, + "tier tactic should bypass health filter and pick T0") +} + +// TestNonTierTactic_StillUsesHealthFilter confirms that the fix only +// bypasses the health filter for tier rules — other tactics still respect it. +func TestNonTierTactic_StillUsesHealthFilter(t *testing.T) { + lb, hm := newTierTestLB(t) + + rule := &typ.Rule{ + UUID: uuid.New().String(), + RequestModel: "test-model", + LBTactic: typ.Tactic{ + Type: loadbalance.TacticRandom, + Params: nil, + }, + Services: []*loadbalance.Service{ + {Provider: "hf-rand-p1", Model: "m1", Active: true, Weight: 1}, + {Provider: "hf-rand-p2", Model: "m1", Active: true, Weight: 1}, + }, + Active: true, + } + + hm.ReportRateLimit(rule.Services[0].ServiceID()) + + counts := map[string]int{} + for i := 0; i < 20; i++ { + svc, err := lb.SelectService(rule) + require.NoError(t, err) + require.NotNil(t, svc) + counts[svc.Provider]++ + } + assert.Equal(t, 0, counts["hf-rand-p1"], + "random tactic should still respect health filter") + assert.Equal(t, 20, counts["hf-rand-p2"]) +} + +// TestTierTactic_BreakerFallbackWhileHealthFilterWouldBlock demonstrates the +// end-to-end scenario: T0 is both HealthMonitor-unhealthy (long timeout) and +// breaker-open (short timeout). The tier tactic should fall to T1 via the +// breaker — not because the health filter hid T0. +func TestTierTactic_BreakerFallbackWhileHealthFilterWouldBlock(t *testing.T) { + lb, hm := newTierTestLB(t) + + primary := tierService("brk-hf-p1", "m1", 0) + backup := tierService("brk-hf-p2", "m1", 1) + rule := &typ.Rule{ + UUID: uuid.New().String(), + RequestModel: "test-model", + LBTactic: typ.Tactic{ + Type: loadbalance.TacticTier, + Params: typ.DefaultTierParams(), + }, + Services: []*loadbalance.Service{primary, backup}, + Active: true, + } + + // Trip BOTH the health monitor and the circuit breaker for T0. + hm.ReportRateLimit(primary.ServiceID()) + store := loadbalance.DefaultBreakerStore() + for i := 0; i < loadbalance.DefaultBreakerFailureThreshold; i++ { + store.RecordFailure(primary.ServiceID()) + } + defer store.RecordSuccess(primary.ServiceID()) + + // Breaker is open → tier tactic should fall to T1. + svc, err := lb.SelectService(rule) + require.NoError(t, err) + require.NotNil(t, svc) + assert.Equal(t, backup.Provider, svc.Provider, + "breaker-open T0 should fall to T1") + + // Now recover the breaker (simulating 30 s elapsed). The health monitor + // still says "unhealthy" (10-min window), but the tier tactic bypasses + // the filter, sees T0, checks the breaker, and routes back to T0. + store.RecordSuccess(primary.ServiceID()) + assert.False(t, hm.IsHealthy(primary.ServiceID()), + "health monitor should still say unhealthy") + + svc, err = lb.SelectService(rule) + require.NoError(t, err) + require.NotNil(t, svc) + assert.Equal(t, primary.Provider, svc.Provider, + "breaker recovered T0 should be picked even though health monitor says unhealthy") +} + +// TestTierTactic_MultiTierWaterfallWithUnhealthyServices tests a 3-tier +// setup where tiers are selectively tripped via breakers while the health +// monitor marks everything unhealthy. The tier tactic should waterfall +// through breakers, not be blocked by the health filter. +func TestTierTactic_MultiTierWaterfallWithUnhealthyServices(t *testing.T) { + lb, hm := newTierTestLB(t) + + t0 := tierService("waterfall-p0", "m1", 0) + t1 := tierService("waterfall-p1", "m1", 1) + t2 := tierService("waterfall-p2", "m1", 2) + rule := &typ.Rule{ + UUID: uuid.New().String(), + RequestModel: "test-model", + LBTactic: typ.Tactic{ + Type: loadbalance.TacticTier, + Params: typ.DefaultTierParams(), + }, + Services: []*loadbalance.Service{t0, t1, t2}, + Active: true, + } + + // Mark ALL services as unhealthy in HealthMonitor. + for _, svc := range rule.Services { + hm.ReportRateLimit(svc.ServiceID()) + } + + store := loadbalance.DefaultBreakerStore() + // Trip T0 and T1 breakers; leave T2 breaker closed. + for _, svc := range []*loadbalance.Service{t0, t1} { + for i := 0; i < loadbalance.DefaultBreakerFailureThreshold; i++ { + store.RecordFailure(svc.ServiceID()) + } + } + defer func() { + store.RecordSuccess(t0.ServiceID()) + store.RecordSuccess(t1.ServiceID()) + }() + + svc, err := lb.SelectService(rule) + require.NoError(t, err) + require.NotNil(t, svc) + assert.Equal(t, t2.Provider, svc.Provider, + "should waterfall to T2 via breakers despite health monitor blocking all") + + // Recover T1 breaker — traffic should go to T1 (not stay on T2). + store.RecordSuccess(t1.ServiceID()) + svc, err = lb.SelectService(rule) + require.NoError(t, err) + assert.Equal(t, t1.Provider, svc.Provider, + "T1 breaker recovery should route back to T1") + + // Recover T0 breaker — traffic should return to T0. + store.RecordSuccess(t0.ServiceID()) + svc, err = lb.SelectService(rule) + require.NoError(t, err) + assert.Equal(t, t0.Provider, svc.Provider, + "T0 breaker recovery should route back to T0") +} + +// TestTierTactic_WithinTierLoadSharing verifies that when multiple services +// share a tier, they still share load even when the health filter would +// remove some of them. +func TestTierTactic_WithinTierLoadSharing(t *testing.T) { + lb, hm := newTierTestLB(t) + + a := tierService("share-a", "m1", 0) + b := tierService("share-b", "m1", 0) + backup := tierService("share-backup", "m1", 1) + rule := &typ.Rule{ + UUID: uuid.New().String(), + RequestModel: "test-model", + LBTactic: typ.Tactic{ + Type: loadbalance.TacticTier, + Params: typ.DefaultTierParams(), + }, + Services: []*loadbalance.Service{a, b, backup}, + Active: true, + } + + // Mark service A as unhealthy in the health monitor. Without the fix, + // only B would be visible and the backup would never get picked — but + // crucially, A's breaker is still closed, so the tier tactic should still + // pick it some of the time. + hm.ReportRateLimit(a.ServiceID()) + + counts := map[string]int{} + for i := 0; i < 200; i++ { + svc, err := lb.SelectService(rule) + require.NoError(t, err) + require.NotNil(t, svc) + counts[svc.Provider]++ + } + + assert.Greater(t, counts[a.Provider], 0, + "service A should still receive traffic despite health filter marking it unhealthy") + assert.Greater(t, counts[b.Provider], 0, + "service B should receive traffic") + assert.Equal(t, 0, counts[backup.Provider], + "T1 backup should not be picked when T0 breakers are all closed") +} + +// TestTierTactic_RateLimitDoesNotStickFor5Min is the highest-level +// reproduction of the original bug: a single 429 on T0 should not pin +// traffic to T1 for the full health-monitor window. +func TestTierTactic_RateLimitDoesNotStickFor5Min(t *testing.T) { + lb, hm := newTierTestLB(t) + + primary := tierService("ratelim-p0", "m1", 0) + fallback := tierService("ratelim-p1", "m1", 1) + rule := &typ.Rule{ + UUID: uuid.New().String(), + RequestModel: "test-model", + LBTactic: typ.Tactic{ + Type: loadbalance.TacticTier, + Params: typ.DefaultTierParams(), + }, + Services: []*loadbalance.Service{primary, fallback}, + Active: true, + } + + // Simulate a 429 on the primary — HealthMonitor marks it unhealthy. + hm.ReportRateLimit(primary.ServiceID()) + + // Immediately after the 429, the tier tactic should still see T0 + // (breaker is closed) and route there. + for i := 0; i < 10; i++ { + svc, err := lb.SelectService(rule) + require.NoError(t, err) + assert.Equal(t, primary.Provider, svc.Provider, + fmt.Sprintf("attempt %d: T0 breaker closed, should still pick T0", i)) + } +} + +// TestTierTactic_AllServicesHealthMonitorUnhealthy_AllBreakersOpen tests the +// extreme case: every service is HealthMonitor-unhealthy AND breaker-open. +// The tier tactic should still return a T0 service so the caller gets the +// real upstream error. +func TestTierTactic_AllServicesHealthMonitorUnhealthy_AllBreakersOpen(t *testing.T) { + lb, hm := newTierTestLB(t) + + t0 := tierService("alldown-p0", "m1", 0) + t1 := tierService("alldown-p1", "m1", 1) + rule := &typ.Rule{ + UUID: uuid.New().String(), + RequestModel: "test-model", + LBTactic: typ.Tactic{ + Type: loadbalance.TacticTier, + Params: typ.DefaultTierParams(), + }, + Services: []*loadbalance.Service{t0, t1}, + Active: true, + } + + // Mark all as unhealthy + breakers open. + store := loadbalance.DefaultBreakerStore() + for _, svc := range rule.Services { + hm.ReportRateLimit(svc.ServiceID()) + for i := 0; i < loadbalance.DefaultBreakerFailureThreshold; i++ { + store.RecordFailure(svc.ServiceID()) + } + } + defer func() { + store.RecordSuccess(t0.ServiceID()) + store.RecordSuccess(t1.ServiceID()) + }() + + svc, err := lb.SelectService(rule) + require.NoError(t, err) + require.NotNil(t, svc, "should still return a service for the upstream-error path") + assert.Equal(t, 0, svc.Tier, + "all-open fallback should pick T0 to surface the real upstream error") +} + +// TestTierTactic_AuthErrorStillFiltered verifies that auth errors (401/403) +// are still filtered out for tier rules. Auth errors are permanent — a +// revoked API key never self-heals — so the tier tactic should not keep +// probing the broken service every 30 seconds via the breaker half-open cycle. +func TestTierTactic_AuthErrorStillFiltered(t *testing.T) { + lb, hm := newTierTestLB(t) + + broken := tierService("auth-broken", "m1", 0) + fallback := tierService("auth-fallback", "m1", 1) + rule := &typ.Rule{ + UUID: uuid.New().String(), + RequestModel: "test-model", + LBTactic: typ.Tactic{ + Type: loadbalance.TacticTier, + Params: typ.DefaultTierParams(), + }, + Services: []*loadbalance.Service{broken, fallback}, + Active: true, + } + + // T0 has a revoked API key → auth error. + hm.ReportAuthError(broken.ServiceID(), 401) + + // The tier tactic should NOT pick T0 despite its breaker being closed. + for i := 0; i < 10; i++ { + svc, err := lb.SelectService(rule) + require.NoError(t, err) + require.NotNil(t, svc) + assert.Equal(t, fallback.Provider, svc.Provider, + fmt.Sprintf("attempt %d: auth-error service should be filtered", i)) + } +} + +// TestTierTactic_AuthErrorOnlyFiltersAuthNotRateLimit confirms the filter is +// surgical: a T0 with a rate limit is kept (breaker handles it), while a T0 +// with an auth error is removed. +func TestTierTactic_AuthErrorOnlyFiltersAuthNotRateLimit(t *testing.T) { + lb, hm := newTierTestLB(t) + + rateLimited := tierService("auth-rl-p0", "m1", 0) + authBroken := tierService("auth-rl-p1", "m1", 0) + backup := tierService("auth-rl-p2", "m1", 1) + rule := &typ.Rule{ + UUID: uuid.New().String(), + RequestModel: "test-model", + LBTactic: typ.Tactic{ + Type: loadbalance.TacticTier, + Params: typ.DefaultTierParams(), + }, + Services: []*loadbalance.Service{rateLimited, authBroken, backup}, + Active: true, + } + + // One T0 is rate-limited (transient), the other has an auth error (permanent). + hm.ReportRateLimit(rateLimited.ServiceID()) + hm.ReportAuthError(authBroken.ServiceID(), 403) + + counts := map[string]int{} + for i := 0; i < 50; i++ { + svc, err := lb.SelectService(rule) + require.NoError(t, err) + require.NotNil(t, svc) + counts[svc.Provider]++ + } + + assert.Greater(t, counts[rateLimited.Provider], 0, + "rate-limited T0 should still be reachable (breaker handles transient failures)") + assert.Equal(t, 0, counts[authBroken.Provider], + "auth-error T0 should be filtered out") + assert.Equal(t, 0, counts[backup.Provider], + "T1 should not be picked while a healthy T0 exists") +} + +// TestTierTactic_AllT0AuthError_FallsToT1 ensures that when every T0 service +// has an auth error, the tier tactic falls to T1 via filtering, not via +// breaker cycling. +func TestTierTactic_AllT0AuthError_FallsToT1(t *testing.T) { + lb, hm := newTierTestLB(t) + + t0a := tierService("allauth-a", "m1", 0) + t0b := tierService("allauth-b", "m1", 0) + t1 := tierService("allauth-c", "m1", 1) + rule := &typ.Rule{ + UUID: uuid.New().String(), + RequestModel: "test-model", + LBTactic: typ.Tactic{ + Type: loadbalance.TacticTier, + Params: typ.DefaultTierParams(), + }, + Services: []*loadbalance.Service{t0a, t0b, t1}, + Active: true, + } + + hm.ReportAuthError(t0a.ServiceID(), 401) + hm.ReportAuthError(t0b.ServiceID(), 403) + + for i := 0; i < 10; i++ { + svc, err := lb.SelectService(rule) + require.NoError(t, err) + assert.Equal(t, t1.Provider, svc.Provider, + "all T0 auth errors should route to T1 immediately") + } +} diff --git a/internal/typ/health_filter.go b/internal/typ/health_filter.go index 0bbfd0598..186d823b5 100644 --- a/internal/typ/health_filter.go +++ b/internal/typ/health_filter.go @@ -47,6 +47,33 @@ func (hf *HealthFilter) FilterWithFallback(services []*loadbalance.Service) []*l return healthy } +// FilterAuthErrors removes only services with auth errors (401/403). Transient +// states like rate limits and consecutive errors are kept so that callers with +// their own recovery mechanisms (e.g. tier tactic's circuit breaker) can still +// see those services. +func (hf *HealthFilter) FilterAuthErrors(services []*loadbalance.Service) []*loadbalance.Service { + if hf.monitor == nil { + return services + } + var out []*loadbalance.Service + for i, svc := range services { + if svc == nil || hf.monitor.HasAuthError(svc.ServiceID()) { + if out == nil { + out = make([]*loadbalance.Service, 0, len(services)) + out = append(out, services[:i]...) + } + continue + } + if out != nil { + out = append(out, svc) + } + } + if out != nil { + return out + } + return services +} + // IsHealthy checks if a specific service is healthy func (hf *HealthFilter) IsHealthy(serviceID string) bool { if hf.monitor == nil {