From 41e322c53d7b4b796eb377d0df9c29ecd10ba431 Mon Sep 17 00:00:00 2001 From: Chia Date: Thu, 6 Aug 2026 09:29:41 +1200 Subject: feat: complete commercial control plane, billing, auth, and model catalog - add PostgreSQL control-plane persistence with Redis-degraded hot reload - implement prepaid balance, usage ledger, Stripe top-up and reconciliation - add registration, email verification, password reset, invitations and RBAC - support TOTP, Passkey MFA, device sessions, quotas and rate limits - add tenant billing profiles, audit logs and operational readiness checks - build authenticated admin console, Quickstart, Playground and usage analytics - add public model catalog with pricing, filtering and cost estimation - support OpenAI Responses providers and provider health failover - validate real upstream usage reporting and balance settlement --- internal/providerhealth/tracker.go | 200 ++++++++++++++++++++++++++++++++ internal/providerhealth/tracker_test.go | 50 ++++++++ 2 files changed, 250 insertions(+) create mode 100644 internal/providerhealth/tracker.go create mode 100644 internal/providerhealth/tracker_test.go (limited to 'internal/providerhealth') diff --git a/internal/providerhealth/tracker.go b/internal/providerhealth/tracker.go new file mode 100644 index 0000000..a5af2b2 --- /dev/null +++ b/internal/providerhealth/tracker.go @@ -0,0 +1,200 @@ +package providerhealth + +import ( + "sort" + "sync" + "time" +) + +const recentWindow = 100 + +type RouteKey struct { + ModelID string + ProviderID string + WireAPI string +} + +type Observation struct { + StatusCode int + Latency time.Duration + Failed bool + ObservedAt time.Time +} + +type Status struct { + ModelID string `json:"model_id"` + ProviderID string `json:"provider_id"` + WireAPI string `json:"wire_api"` + State string `json:"state"` + Attempts uint64 `json:"attempts"` + RecentSamples int `json:"recent_samples"` + AvailabilityPercent float64 `json:"availability_percent"` + HeaderLatencyEWMA int64 `json:"header_latency_ewma_ms"` + ConsecutiveFailures uint64 `json:"consecutive_failures"` + LastStatusCode int `json:"last_status_code,omitempty"` + LastObservedAt *time.Time `json:"last_observed_at,omitempty"` + LastHealthyAt *time.Time `json:"last_healthy_at,omitempty"` + CircuitOpenUntil *time.Time `json:"circuit_open_until,omitempty"` +} + +type Options struct { + FailureThreshold uint64 + OpenDuration time.Duration + Now func() time.Time +} + +type Tracker struct { + states sync.Map + failureThreshold uint64 + openDuration time.Duration + now func() time.Time +} + +type routeState struct { + mu sync.RWMutex + attempts uint64 + consecutiveFailures uint64 + lastStatusCode int + lastObservedAt time.Time + lastHealthyAt time.Time + openUntil time.Time + headerLatencyEWMA float64 + recent [recentWindow]bool + recentCount int + recentPosition int + recentHealthy int +} + +func New(options Options) *Tracker { + if options.FailureThreshold == 0 { + options.FailureThreshold = 3 + } + if options.OpenDuration <= 0 { + options.OpenDuration = 30 * time.Second + } + if options.Now == nil { + options.Now = time.Now + } + return &Tracker{failureThreshold: options.FailureThreshold, openDuration: options.OpenDuration, now: options.Now} +} + +func (t *Tracker) Observe(key RouteKey, observation Observation) { + if t == nil || key.ProviderID == "" { + return + } + if observation.ObservedAt.IsZero() { + observation.ObservedAt = t.now() + } + value, _ := t.states.LoadOrStore(key, &routeState{}) + state := value.(*routeState) + state.mu.Lock() + defer state.mu.Unlock() + + state.attempts++ + state.lastStatusCode = observation.StatusCode + state.lastObservedAt = observation.ObservedAt + if observation.Latency > 0 { + latency := float64(observation.Latency.Milliseconds()) + if latency < 1 { + latency = 1 + } + if state.headerLatencyEWMA == 0 { + state.headerLatencyEWMA = latency + } else { + state.headerLatencyEWMA = state.headerLatencyEWMA*0.8 + latency*0.2 + } + } + state.addRecent(!observation.Failed) + if observation.Failed { + state.consecutiveFailures++ + if state.consecutiveFailures >= t.failureThreshold { + state.openUntil = observation.ObservedAt.Add(t.openDuration) + } + return + } + state.consecutiveFailures = 0 + state.openUntil = time.Time{} + state.lastHealthyAt = observation.ObservedAt +} + +func (s *routeState) addRecent(healthy bool) { + if s.recentCount == recentWindow { + if s.recent[s.recentPosition] { + s.recentHealthy-- + } + } else { + s.recentCount++ + } + s.recent[s.recentPosition] = healthy + if healthy { + s.recentHealthy++ + } + s.recentPosition = (s.recentPosition + 1) % recentWindow +} + +func (t *Tracker) CircuitOpen(key RouteKey) bool { + if t == nil { + return false + } + value, ok := t.states.Load(key) + if !ok { + return false + } + state := value.(*routeState) + state.mu.RLock() + defer state.mu.RUnlock() + return state.openUntil.After(t.now()) +} + +func (t *Tracker) Snapshot() []Status { + if t == nil { + return []Status{} + } + now := t.now() + result := make([]Status, 0) + t.states.Range(func(rawKey, rawState any) bool { + key := rawKey.(RouteKey) + state := rawState.(*routeState) + state.mu.RLock() + item := statusFromState(key, state, now) + state.mu.RUnlock() + result = append(result, item) + return true + }) + sort.Slice(result, func(i, j int) bool { + if result[i].ModelID != result[j].ModelID { + return result[i].ModelID < result[j].ModelID + } + return result[i].ProviderID < result[j].ProviderID + }) + return result +} + +func statusFromState(key RouteKey, state *routeState, now time.Time) Status { + item := Status{ModelID: key.ModelID, ProviderID: key.ProviderID, WireAPI: key.WireAPI, + Attempts: state.attempts, RecentSamples: state.recentCount, HeaderLatencyEWMA: int64(state.headerLatencyEWMA + 0.5), + ConsecutiveFailures: state.consecutiveFailures, LastStatusCode: state.lastStatusCode} + if state.recentCount > 0 { + item.AvailabilityPercent = float64(state.recentHealthy) / float64(state.recentCount) * 100 + } + if !state.lastObservedAt.IsZero() { + value := state.lastObservedAt + item.LastObservedAt = &value + } + if !state.lastHealthyAt.IsZero() { + value := state.lastHealthyAt + item.LastHealthyAt = &value + } + if state.openUntil.After(now) { + value := state.openUntil + item.CircuitOpenUntil = &value + item.State = "open" + } else if state.attempts == 0 { + item.State = "unknown" + } else if state.consecutiveFailures > 0 || (state.recentCount >= 5 && item.AvailabilityPercent < 95) { + item.State = "degraded" + } else { + item.State = "healthy" + } + return item +} diff --git a/internal/providerhealth/tracker_test.go b/internal/providerhealth/tracker_test.go new file mode 100644 index 0000000..73bd6d5 --- /dev/null +++ b/internal/providerhealth/tracker_test.go @@ -0,0 +1,50 @@ +package providerhealth + +import ( + "sync" + "testing" + "time" +) + +func TestTrackerOpensAndRecoversCircuit(t *testing.T) { + now := time.Date(2026, time.August, 6, 0, 0, 0, 0, time.UTC) + tracker := New(Options{FailureThreshold: 3, OpenDuration: 30 * time.Second, Now: func() time.Time { return now }}) + key := RouteKey{ModelID: "openai/test", ProviderID: "primary", WireAPI: "responses"} + for _, status := range []int{503, 429, 504} { + tracker.Observe(key, Observation{StatusCode: status, Latency: 100 * time.Millisecond, Failed: true}) + } + if !tracker.CircuitOpen(key) { + t.Fatal("circuit did not open after consecutive retryable failures") + } + snapshot := tracker.Snapshot() + if len(snapshot) != 1 || snapshot[0].State != "open" || snapshot[0].AvailabilityPercent != 0 || snapshot[0].CircuitOpenUntil == nil { + t.Fatalf("unexpected open snapshot: %+v", snapshot) + } + now = now.Add(31 * time.Second) + if tracker.CircuitOpen(key) { + t.Fatal("circuit did not permit a recovery attempt after cooldown") + } + tracker.Observe(key, Observation{StatusCode: 200, Latency: 50 * time.Millisecond}) + snapshot = tracker.Snapshot() + if snapshot[0].State != "healthy" || snapshot[0].ConsecutiveFailures != 0 || snapshot[0].LastHealthyAt == nil || snapshot[0].HeaderLatencyEWMA < 50 { + t.Fatalf("unexpected recovered snapshot: %+v", snapshot[0]) + } +} + +func TestTrackerConcurrentObservations(t *testing.T) { + tracker := New(Options{FailureThreshold: 1000}) + key := RouteKey{ModelID: "model", ProviderID: "provider", WireAPI: "chat_completions"} + var group sync.WaitGroup + for index := range 200 { + group.Add(1) + go func(failed bool) { + defer group.Done() + tracker.Observe(key, Observation{StatusCode: 200, Latency: time.Millisecond, Failed: failed}) + }(index%2 == 0) + } + group.Wait() + snapshot := tracker.Snapshot() + if len(snapshot) != 1 || snapshot[0].Attempts != 200 || snapshot[0].RecentSamples != recentWindow || snapshot[0].AvailabilityPercent < 0 || snapshot[0].AvailabilityPercent > 100 { + t.Fatalf("unexpected concurrent snapshot: %+v", snapshot) + } +} -- cgit v1.2.3