Fix failover and health-monitor recovery for 429/5xx outages

.design/tier-routing.md

@@ -142,6 +142,8 @@ Each request outcome feeds **two channels**:
 
 So 429 and 401/403 exclude a service on the *first* hit (health channel), well before the breaker trips. The `harness lb` simulator models both faithfully (reusing `reportHealthStatus` + the breaker recorder), driving them off one shared clock so a single simulated advance recovers both.
 
+Health recovery runs an optional lightweight probe (`Server.SetProbeFunc`, OPTIONS + `/models`) once the rate-limit window has elapsed; a failed probe pushes the window forward. The probe parses the serviceID with `loadbalance.ParseServiceID` ("provider/model") — it originally split on `":"`, which never matches, so every probe failed and a single 429 excluded the service *forever* (traffic could never return to the primary). Pinned by `TestFailoverTimeline_PrimaryDownThenRecover/429-*` in `internal/protocoltest/failover_timeline_test.go`.
+
 There is also a **third** time-based input: the **affinity TTL** (`AffinityEntry.ExpiresAt`). A lock expires *strictly* at `LockedAt + SessionAffinity` — an in-window request is honored but does **not** slide the expiry; once past it the pin is dropped and the session is re-selected and re-locked. The simulator puts this on the same clock seam (`loadbalance.SetClock` + `routing.SetClock`, both fed one fake clock), so a single `advance` moves breaker recovery, health recovery, *and* affinity expiry together — matching production, where all three read the wall clock. This lets the scenario suite assert the strict-TTL contract and cross-model failover (a failover hop must carry the fallback service's own model, not reuse the primary's) deterministically.
 
 ### End-to-end flow: how the tactic switch actually takes effect
@@ -201,6 +203,8 @@ The "user moves a service card to a different tier" event has to cross five laye
 ┌─ Per-request feedback into the breaker ──────────────────────────────┐
 │  dispatchWithPriorityFailover owns breaker accounting per attempt:    │
 │    gate committed        → RecordServiceSuccess(ruleUUID, serviceID)  │
+│    terminal 2xx (buffered, e.g. non-streaming c.JSON)                 │
+│                          → RecordServiceSuccess(ruleUUID, serviceID)  │
 │    retryable failure     → RecordServiceFailure(ruleUUID, serviceID)  │
 │  Same DefaultBreakerStore the selection logic consulted, so the next  │
 │  request's TierTactic sees the updated state. State is keyed per rule │
@@ -333,9 +337,9 @@ v1 only handled cross-request failover: a request that failed returned the error
 2. **`firstChunkGate`** is a passive, protocol-agnostic byte buffer wrapping `c.Writer`. It makes no decisions in its write path: writes land in memory until an explicit signal commits them. Crucially, **single-service requests skip the gate entirely** (`len(GetActiveServices()) <= 1`), so the common case never touches the buffer — zero blast radius.
 3. **Orchestrator** (`dispatchWithPriorityFailover`) owns the retry decision. After each attempt it reads the gate's state:
    - `gate.Committed()` → the stream's first real chunk already reached the wire; retry is impossible, return.
-   - else `gate.Status()` retryable (429, 500, 502, 503, 504) → `gate.Discard()`, pick the next tier via `selectFallbackService`, try again.
+   - else `gate.Status()` retryable (429 or **any 5xx**, 500–599) → `gate.Discard()`, pick the next tier via `selectFallbackService`, try again. The full 5xx range matters because error forwarding propagates the upstream's status verbatim — enumerating individual codes silently dropped provider-specific statuses like Anthropic's 529 `overloaded_error` (no failover on exactly the outage it signals).
    - else (200, other 4xx, status 0) → terminal; the deferred `gate.CommitIfBuffered()` flushes the captured error to the client.
-4. **Commit seam.** Streaming producers raise `CommitFirstChunk` on their first real chunk (centralised in `ProcessStream`/`StreamLoop`, plus the explicit `message_start` senders), which flushes captured headers + body and switches the gate to pass-through — preserving incremental delivery.
+4. **Commit seam.** Streaming producers raise `CommitFirstChunk` on their first real chunk (centralised in `ProcessStream`/`StreamLoop`, the explicit `message_start` senders, plus the MCP generic stream interceptor's `sendEvent` — the A→A v1 path runs through the interceptor unconditionally, and before it committed, multi-service rules on that path buffered the whole stream and never fed the breaker a success), which flushes captured headers + body and switches the gate to pass-through — preserving incremental delivery.
 5. Budget caps at the number of active services so we never loop unbounded. The same service is never tried twice in one request (in-request `tried` map, complementary to the cross-request breaker). The recorder's bound provider is re-set before each attempt (`SetActiveService`), so a second-attempt failure trips the *second* service's breaker — not the first's.
 
 ### Streaming: priming the first event

internal/loadbalance/health_monitor.go

@@ -73,8 +73,17 @@ type HealthMonitor struct {
 	probeFunc              HealthProbeFunc // Optional probe function for recovery checking
 }
 
-// NewHealthMonitor creates a new health monitor with the given configuration
+// NewHealthMonitor creates a new health monitor with the given configuration.
+// Zero values fall back to defaults (mirroring NewBreaker): the server builds
+// the monitor straight from the persisted config, and an absent health_monitor
+// section used to yield a ZERO recovery window — a 429-marked service
+// "auto-recovered" on the very next request, so the documented rate-limit
+// window never held. ProbeEnabled is left as-is (false is a valid choice and
+// a bool cannot express "unset").
 func NewHealthMonitor(config HealthMonitorConfig) *HealthMonitor {
+	if config.RecoveryTimeoutSeconds <= 0 {
+		config.RecoveryTimeoutSeconds = DefaultHealthMonitorConfig().RecoveryTimeoutSeconds
+	}
 	return &HealthMonitor{
 		services:               make(map[string]*ServiceHealth),
 		config:                 config,
@@ -303,11 +312,15 @@ func (hm *HealthMonitor) RemoveHealth(serviceID string) {
 	delete(hm.services, serviceID)
 }
 
-// UpdateConfig updates the health monitor configuration
+// UpdateConfig updates the health monitor configuration. Zero values fall
+// back to defaults, matching NewHealthMonitor.
 func (hm *HealthMonitor) UpdateConfig(config HealthMonitorConfig) {
 	hm.mutex.Lock()
 	defer hm.mutex.Unlock()
 
+	if config.RecoveryTimeoutSeconds <= 0 {
+		config.RecoveryTimeoutSeconds = DefaultHealthMonitorConfig().RecoveryTimeoutSeconds
+	}
 	hm.config = config
 	hm.defaultRecoveryTimeout = time.Duration(config.RecoveryTimeoutSeconds) * time.Second
 }

internal/loadbalance/service_id.go

@@ -1,6 +1,9 @@
 package loadbalance
 
-import "fmt"
+import (
+	"fmt"
+	"strings"
+)
 
 // ServiceID uniquely identifies a provider+model combination in load balancing.
 type ServiceID struct {
@@ -27,6 +30,18 @@ func FormatServiceID(providerUUID, model string) string {
 	return fmt.Sprintf("%s/%s", providerUUID, model)
 }
 
+// ParseServiceID splits a canonical "provider/model" service ID back into its
+// parts (the inverse of FormatServiceID). Provider UUIDs are slash-free, so
+// splitting on the first "/" is unambiguous even when the model name itself
+// contains slashes. model is "" when serviceID carries no separator.
+func ParseServiceID(serviceID string) (providerUUID, model string) {
+	parts := strings.SplitN(serviceID, "/", 2)
+	if len(parts) < 2 {
+		return parts[0], ""
+	}
+	return parts[0], parts[1]
+}
+
 // FormatBreakerKey formats a (ruleUUID, serviceID) pair into the canonical
 // "ruleUUID/serviceID" string used as the breaker-store key. The breaker is
 // rule-scoped: each rule owns independent breaker state per service so a busy

internal/protocoltest/failover_timeline.go

@@ -0,0 +1,142 @@
+package protocoltest
+
+// Timeline failover fixtures: switchable vmodel upstreams whose availability
+// can be flipped at runtime, so a test can script a wall-clock scenario like
+// "00:00 all up → 00:05 vm1 down (429/500/529) → vm1 up again → traffic must
+// return to vm1 after breaker recovery". Complements failover.go, whose error
+// mocks are statically always-failing.
+
+import (
+	"fmt"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/tingly-dev/tingly-box/internal/constant"
+	"github.com/tingly-dev/tingly-box/internal/loadbalance"
+	"github.com/tingly-dev/tingly-box/internal/protocol"
+	"github.com/tingly-dev/tingly-box/internal/typ"
+	anthropicvm "github.com/tingly-dev/tingly-box/vmodel/anthropic"
+	openaivm "github.com/tingly-dev/tingly-box/vmodel/openai"
+	"github.com/tingly-dev/tingly-box/vmodel/virtualserver"
+)
+
+// SwitchableUpstream is a vmodel-backed httptest provider that can be flipped
+// between healthy (serves the registered virtual models) and down (every
+// request — including probes — gets the configured HTTP status with an error
+// envelope) at any point during a test.
+type SwitchableUpstream struct {
+	Server *httptest.Server
+
+	hits       atomic.Int64
+	downStatus atomic.Int64 // 0 = up; otherwise the HTTP status to fail with
+}
+
+// SetDown makes every subsequent request fail with the given HTTP status.
+func (u *SwitchableUpstream) SetDown(status int) { u.downStatus.Store(int64(status)) }
+
+// SetUp restores normal vmodel serving.
+func (u *SwitchableUpstream) SetUp() { u.downStatus.Store(0) }
+
+// Hits reports how many HTTP requests reached this upstream (successful or
+// failed, chat or probe endpoints alike).
+func (u *SwitchableUpstream) Hits() int64 { return u.hits.Load() }
+
+// newSwitchableUpstream builds the upstream: a virtualserver with both
+// protocol registries (so it serves either API style), wrapped by a counting
+// middleware that injects the outage when SetDown is active.
+func newSwitchableUpstream(t *testing.T) *SwitchableUpstream {
+	t.Helper()
+	gin.SetMode(gin.TestMode)
+
+	u := &SwitchableUpstream{}
+
+	svc := virtualserver.NewService()
+	openaivm.RegisterDefaults(svc.GetOpenAIRegistry())
+	anthropicvm.RegisterDefaults(svc.GetAnthropicRegistry())
+
+	engine := gin.New()
+	engine.Use(func(c *gin.Context) {
+		u.hits.Add(1)
+		if status := int(u.downStatus.Load()); status != 0 {
+			// Shape follows the Anthropic error envelope; both SDKs surface
+			// the HTTP status either way, which is all failover keys off.
+			c.AbortWithStatusJSON(status, gin.H{
+				"type": "error",
+				"error": gin.H{
+					"type":    "api_error",
+					"message": fmt.Sprintf("simulated outage: upstream returned %d", status),
+				},
+			})
+			return
+		}
+		c.Next()
+	})
+	svc.SetupRoutes(engine.Group("/v1"))
+
+	srv := httptest.NewServer(engine)
+	t.Cleanup(srv.Close)
+	u.Server = srv
+	return u
+}
+
+// TimelineFailoverRoute is the handle returned by SetupTimelineFailoverRoute.
+type TimelineFailoverRoute struct {
+	ModelName  string                // gateway-facing request model
+	RuleUUID   string                // for breaker-store introspection
+	VMs        []*SwitchableUpstream // index == tier (VMs[0] is T0)
+	ServiceIDs []string              // loadbalance service IDs, index == tier
+}
+
+// timelineSuccessModel is the vmodel every tier serves while up.
+const timelineSuccessModel = "echo-model"
+
+// SetupTimelineFailoverRoute wires an N-tier rule (T0..T(n-1)) whose tiers are
+// independent switchable upstreams, all serving echo-model while up. All
+// providers use the source protocol's API style (homogeneous failover — the
+// cross-style path has its own suite). label must be unique per test case; it
+// namespaces the provider/rule UUIDs so parallel or sequential cases cannot
+// collide in the shared breaker store.
+func (env *TestEnv) SetupTimelineFailoverRoute(t *testing.T, source protocol.APIType, tiers int, label string) TimelineFailoverRoute {
+	t.Helper()
+
+	apiStyle := targetToAPIStyle(source)
+	requestModel := fmt.Sprintf("fo-timeline-%s-%s", source, label)
+
+	route := TimelineFailoverRoute{
+		ModelName: requestModel,
+		RuleUUID:  requestModel,
+	}
+
+	services := make([]*loadbalance.Service, 0, tiers)
+	for i := 0; i < tiers; i++ {
+		vm := newSwitchableUpstream(t)
+		route.VMs = append(route.VMs, vm)
+
+		apiBase := vm.Server.URL
+		if apiStyle == protocol.APIStyleOpenAI {
+			apiBase = vm.Server.URL + "/v1"
+		}
+
+		providerUUID := fmt.Sprintf("timeline-%s-vm%d", label, i+1)
+		if err := env.appConfig.AddProvider(&typ.Provider{
+			UUID: providerUUID, Name: providerUUID, APIBase: apiBase, APIStyle: apiStyle,
+			Token: "timeline-token", Enabled: true, Timeout: int64(constant.DefaultRequestTimeout),
+		}); err != nil {
+			t.Fatalf("add timeline provider vm%d: %v", i+1, err)
+		}
+
+		services = append(services, tieredService(providerUUID, timelineSuccessModel, i))
+		route.ServiceIDs = append(route.ServiceIDs, loadbalance.FormatServiceID(providerUUID, timelineSuccessModel))
+	}
+
+	rule := newHarnessRule(requestModel, sourceToRuleScenario(source), requestModel, timelineSuccessModel, services...)
+	rule.LBTactic = tierFailoverTactic()
+	if err := env.appConfig.GetGlobalConfig().AddRequestConfig(rule); err != nil {
+		t.Fatalf("add timeline rule: %v", err)
+	}
+
+	return route
+}