From b380faeb3e5f4fb6fc60d1c75259da8905bd8ab5 Mon Sep 17 00:00:00 2001 From: Marketen Date: Mon, 19 Oct 2026 18:09:19 +0200 Subject: [PATCH] feat: stop sending to providers that keep failing, retry quick failures once A provider that is down used to cost every request a full attempt: on 18-19 Oct our Qwen server answered 502 to every request for about a day. A circuit breaker per provider and upstream model now opens after five provider failures in a row and keeps requests away for 30 s (doubling up to 5 min after failed probes). While it is open, requests go straight to the model's fallback provider, or fail at once with 503 and Retry-After. One request probes the provider to close it again. Without a fallback, a connection error or a quick 502/503 before any output is retried once; a request reaches at most two provider attempts. Rejected requests, rate limits and canceled clients never count as failures. New metrics: nexus_gateway_circuit_open and nexus_gateway_circuit_transitions_total; retries are counted in nexus_gateway_provider_retries_total{reason="transient_retry"}. Co-Authored-By: Claude Opus 5.5 --- README.md | 13 + .../observability/metrics/prometheus.go | 18 + .../internal/application/services/breaker.go | 320 ++++++++++++++++++ .../application/services/breaker_test.go | 216 ++++++++++++ .../application/services/generate_service.go | 10 + .../internal/application/services/proxy.go | 32 +- .../application/services/systemone_test.go | 3 + apps/gateway/main.go | 5 + apps/gateway/selfhealing.go | 44 +++ pkg/domain/error_class.go | 2 + 10 files changed, 656 insertions(+), 7 deletions(-) create mode 100644 apps/gateway/internal/application/services/breaker.go create mode 100644 apps/gateway/internal/application/services/breaker_test.go create mode 100644 apps/gateway/selfhealing.go diff --git a/README.md b/README.md index 74b54d2..c1238c0 100644 --- a/README.md +++ b/README.md @@ -91,6 +91,19 @@ keep a valid `Retry-After` header. Errors include `origin` (`gateway` or details are not exposed. A configured fallback provider is tried only after a rate limit or a server-side failure, never after a rejected request. +The gateway protects callers from a provider that keeps failing. After five +provider failures in a row (outages, timeouts, server errors or a broken +provider account; never a rejected request, a rate limit or a canceled +client), it stops sending to that provider for 30 seconds: requests go +straight to the model's fallback provider, or, without one, fail at once with +503 `provider_unavailable` and a `Retry-After` header. One request then +probes the provider; a failed probe doubles the pause, up to five minutes. +Without a fallback, a connection error or a 502/503 that the provider returned +within five seconds and before any output is retried once. A request reaches +at most two provider attempts. `PROVIDER_BREAKER_FAILURES` (0 turns the +breaker off), `PROVIDER_BREAKER_OPEN_SECONDS`, `PROVIDER_BREAKER_MAX_OPEN_SECONDS` +and `PROVIDER_RETRY_FAST_FAILURES` change these defaults. + Responses have no limit on total duration. Two stall guards apply instead: `PROVIDER_SILENCE_TIMEOUT_SECONDS` (default 900) ends a request when the provider sends nothing, neither headers nor bytes, for that long, and diff --git a/apps/gateway/internal/adapters/observability/metrics/prometheus.go b/apps/gateway/internal/adapters/observability/metrics/prometheus.go index 370a4f8..812710e 100644 --- a/apps/gateway/internal/adapters/observability/metrics/prometheus.go +++ b/apps/gateway/internal/adapters/observability/metrics/prometheus.go @@ -209,6 +209,22 @@ var ( []string{"endpoint", "model", "provider", "fallback_provider", "fault", "error_code"}, ) + CircuitOpen = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "nexus_gateway_circuit_open", + Help: "1 while the gateway has stopped sending to a provider target that kept failing, by provider and upstream model.", + }, + []string{"provider", "provider_model"}, + ) + + CircuitTransitions = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "nexus_gateway_circuit_transitions_total", + Help: "Total provider circuit state changes, by provider, upstream model and new state (open|closed).", + }, + []string{"provider", "provider_model", "state"}, + ) + BuildInfo = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Name: "nexus_gateway_build_info", @@ -275,6 +291,8 @@ func init() { ProviderRetries, ErrorsTotal, FallbacksTotal, + CircuitOpen, + CircuitTransitions, BuildInfo, ) BuildInfo.WithLabelValues(buildinfo.Version, buildinfo.Commit).Set(1) diff --git a/apps/gateway/internal/application/services/breaker.go b/apps/gateway/internal/application/services/breaker.go new file mode 100644 index 0000000..028fd88 --- /dev/null +++ b/apps/gateway/internal/application/services/breaker.go @@ -0,0 +1,320 @@ +package services + +import ( + "context" + "errors" + "math" + "strconv" + "sync" + "time" + + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/observability/metrics" + "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" +) + +// SelfHealing configures how the gateway reacts to a provider that keeps +// failing: a circuit breaker per provider and upstream model, and one quick +// retry for a failure that happened before any output. +type SelfHealing struct { + // BreakerFailures is how many provider failures in a row open the + // circuit; 0 turns the breaker off. + BreakerFailures int + // BreakerOpenFor is how long a newly opened circuit stays open. A probe + // that fails doubles it, up to BreakerMaxOpen. + BreakerOpenFor time.Duration + BreakerMaxOpen time.Duration + // RetryFastFailures retries once a connection error or a 502/503 that the + // provider returned within RetryWithin, before any output. + RetryFastFailures bool + RetryWithin time.Duration + RetryDelay time.Duration +} + +// DefaultSelfHealing is what the gateway runs with unless configured. +var DefaultSelfHealing = SelfHealing{ + BreakerFailures: 5, + BreakerOpenFor: 30 * time.Second, + BreakerMaxOpen: 5 * time.Minute, + RetryFastFailures: true, + RetryWithin: 5 * time.Second, + RetryDelay: 250 * time.Millisecond, +} + +// Circuit states, for logs and metrics. +const ( + circuitClosed = "closed" + circuitOpen = "open" + circuitHalfOpen = "half_open" +) + +// breaker tracks provider failures per target. While a target's circuit is +// open, requests skip it: they go to the model's fallback, or fail at once +// with Retry-After instead of each paying for a provider that is down. When +// the open period ends, one request probes the target; its result closes or +// reopens the circuit. +type breaker struct { + mu sync.Mutex + cfg SelfHealing + now func() time.Time + targets map[string]*circuit +} + +type circuit struct { + failures int + openUntil time.Time + openFor time.Duration + probing bool +} + +// transition reports a circuit changing state, for logs and metrics. +type transition struct { + to string + failures int + openFor time.Duration +} + +func newBreaker(cfg SelfHealing) *breaker { + return &breaker{cfg: cfg, now: time.Now, targets: map[string]*circuit{}} +} + +func (b *breaker) enabled() bool { return b != nil && b.cfg.BreakerFailures > 0 } + +// allow reports whether a request may go to the target, and if not, how long +// until it may try again. After the open period one request is let through +// as a probe; others keep waiting until the probe answers. +func (b *breaker) allow(key string) (bool, time.Duration) { + if !b.enabled() { + return true, 0 + } + b.mu.Lock() + defer b.mu.Unlock() + c := b.targets[key] + if c == nil || c.openUntil.IsZero() { + return true, 0 + } + now := b.now() + if now.Before(c.openUntil) { + return false, c.openUntil.Sub(now) + } + if c.probing { + return false, time.Second + } + c.probing = true + return true, 0 +} + +// success closes the target's circuit. It returns a transition when the +// circuit was open. +func (b *breaker) success(key string) *transition { + if !b.enabled() { + return nil + } + b.mu.Lock() + defer b.mu.Unlock() + c := b.targets[key] + if c == nil { + return nil + } + wasOpen := !c.openUntil.IsZero() + delete(b.targets, key) + if wasOpen { + return &transition{to: circuitClosed} + } + return nil +} + +// failure counts a provider failure. It returns a transition when the +// circuit opens, or reopens after a failed probe. +func (b *breaker) failure(key string) *transition { + if !b.enabled() { + return nil + } + b.mu.Lock() + defer b.mu.Unlock() + c := b.targets[key] + if c == nil { + c = &circuit{} + b.targets[key] = c + } + c.failures++ + switch { + case c.probing: + c.probing = false + c.openFor = minDuration(c.openFor*2, b.cfg.BreakerMaxOpen) + case c.openUntil.IsZero() && c.failures >= b.cfg.BreakerFailures: + c.openFor = b.cfg.BreakerOpenFor + default: + return nil + } + c.openUntil = b.now().Add(c.openFor) + return &transition{to: circuitOpen, failures: c.failures, openFor: c.openFor} +} + +// release frees a probe slot when the probe ended without telling us anything +// about the provider (the client left, or the request was rejected). +func (b *breaker) release(key string) { + if !b.enabled() { + return + } + b.mu.Lock() + defer b.mu.Unlock() + if c := b.targets[key]; c != nil { + c.probing = false + } +} + +// state names the target's circuit state, for metrics. +func (b *breaker) state(key string) string { + if !b.enabled() { + return circuitClosed + } + b.mu.Lock() + defer b.mu.Unlock() + c := b.targets[key] + switch { + case c == nil || c.openUntil.IsZero(): + return circuitClosed + case b.now().Before(c.openUntil): + return circuitOpen + default: + return circuitHalfOpen + } +} + +func minDuration(a, b time.Duration) time.Duration { + if a < b { + return a + } + return b +} + +func breakerKey(target domain.PublicModel) string { + return target.ProviderConfig.ProviderName + "/" + target.UpstreamModelName +} + +// providerOutage reports whether an error says the provider itself is failing +// (down, overloaded, timing out, or refusing our account), as opposed to the +// caller's mistake, a rate limit, a rejected request or a client that left. +func providerOutage(ctx context.Context, err error) bool { + if err == nil || ctx.Err() != nil { + return false + } + var gwErr *domain.GatewayError + if !errors.As(err, &gwErr) { + return true // A transport error with no classification. + } + switch gwErr.Code { + case domain.ErrCodeProviderConfiguration: + return true // Our key or account at the provider: every request fails until it is fixed. + case domain.ErrCodeProviderUnavailable, domain.ErrCodeProviderTimeout, domain.ErrCodeProviderError: + if status := domain.UpstreamStatus(err); status >= 400 && status < 500 { + return false // The provider answered about this request (rejected it, rate limited it). + } + return gwErr.Metadata["cause"] != domain.CauseCircuitOpen && gwErr.Metadata["cause"] != domain.CauseProviderNotRegistered + } + return false +} + +// fastTransient reports whether a failure is worth one quick retry: the +// connection failed, or the provider answered 502 or 503, quickly and before +// any output. A slow failure is not retried; it would only double the wait. +func fastTransient(ctx context.Context, err error, took, within time.Duration) bool { + if !providerOutage(ctx, err) || took > within { + return false + } + var gwErr *domain.GatewayError + if !errors.As(err, &gwErr) { + return true + } + if gwErr.Code == domain.ErrCodeProviderConfiguration { + return false // Our key or account at the provider: a retry fails the same way. + } + switch domain.UpstreamStatus(err) { + case 0, 502, 503: + return true + } + return false +} + +// circuitOpenError is what a request gets when its provider's circuit is +// open and the model has no fallback. +func circuitOpenError(provider string, wait time.Duration) *domain.GatewayError { + err := domain.ErrProviderUnavailable(provider).WithMeta("cause", domain.CauseCircuitOpen) + err.RetryAfter = strconv.Itoa(int(math.Ceil(wait.Seconds()))) + return err +} + +// tryTarget sends a request to one provider target through its circuit +// breaker. When mayRetry is set (the model has no fallback to go to), a quick +// transient failure before any output is retried once. +func (s *GenerateService) tryTarget(ctx context.Context, requestID string, target domain.PublicModel, + attempt func(domain.PublicModel) (*ProxyCall, error), mayRetry bool) (*ProxyCall, error) { + key := breakerKey(target) + provider := target.ProviderConfig.ProviderName + if ok, wait := s.breaker.allow(key); !ok { + return nil, circuitOpenError(provider, wait) + } + for try := 1; ; try++ { + began := time.Now() + call, err := attempt(target) + if err == nil { + s.noteCircuit(requestID, target, s.breaker.success(key)) + return call, nil + } + if !providerOutage(ctx, err) { + s.breaker.release(key) + return nil, err + } + if t := s.breaker.failure(key); t != nil { + s.noteCircuit(requestID, target, t) + return nil, err // The circuit just opened: stop sending to this provider. + } + if try >= 2 || !mayRetry || !s.heal.RetryFastFailures || !fastTransient(ctx, err, time.Since(began), s.heal.RetryWithin) { + return nil, err + } + metrics.ProviderRetries.WithLabelValues(provider, "transient_retry").Inc() + if s.logger != nil { + s.logger.Info("retrying provider after a quick failure", append([]any{ + "request_id", requestID, "model", target.PublicModelID, "provider", provider, + "provider_model", target.UpstreamModelName}, domain.DiagnosticFields(err)...)...) + } + if !sleepContext(ctx, s.heal.RetryDelay) { + return nil, err + } + if ok, _ := s.breaker.allow(key); !ok { + return nil, err + } + } +} + +// noteCircuit records a circuit state change in the metrics and the log. +func (s *GenerateService) noteCircuit(requestID string, target domain.PublicModel, t *transition) { + if t == nil { + return + } + provider, upstream := target.ProviderConfig.ProviderName, target.UpstreamModelName + metrics.CircuitTransitions.WithLabelValues(provider, upstream, t.to).Inc() + fields := []any{"request_id", requestID, "model", target.PublicModelID, "provider", provider, "provider_model", upstream} + if t.to == circuitOpen { + metrics.CircuitOpen.WithLabelValues(provider, upstream).Set(1) + if s.logger != nil { + s.logger.Warn("provider circuit opened", append(fields, "failures", t.failures, "open_seconds", int(t.openFor.Seconds()))...) + } + return + } + metrics.CircuitOpen.WithLabelValues(provider, upstream).Set(0) + if s.logger != nil { + s.logger.Info("provider circuit closed", fields...) + } +} + +func sleepContext(ctx context.Context, d time.Duration) bool { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} diff --git a/apps/gateway/internal/application/services/breaker_test.go b/apps/gateway/internal/application/services/breaker_test.go new file mode 100644 index 0000000..d3c25c2 --- /dev/null +++ b/apps/gateway/internal/application/services/breaker_test.go @@ -0,0 +1,216 @@ +package services + +import ( + "context" + "io" + "strings" + "testing" + "time" + + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/adapters/observability/metrics" + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/ports" + "github.com/dappnode/dappnode-nexus-gateway/pkg/domain" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +func upstream502() error { + return domain.ErrProviderError(502, "provider server error").WithMeta("upstream_status", 502) +} + +// scriptedProvider answers each call with the next error in its script, and +// succeeds once the script runs out. +type scriptedProvider struct { + errs []error + calls int +} + +func (p *scriptedProvider) next() error { + p.calls++ + if len(p.errs) == 0 { + return nil + } + err := p.errs[0] + p.errs = p.errs[1:] + return err +} + +func (p *scriptedProvider) Complete(_ context.Context, _ []byte, _ domain.PublicModel) ([]byte, *domain.TinfoilTransportProof, error) { + if err := p.next(); err != nil { + return nil, nil, err + } + return []byte(stubAnswer), nil, nil +} + +func (p *scriptedProvider) Stream(_ context.Context, _ []byte, _ domain.PublicModel) (ports.ProviderStream, error) { + if err := p.next(); err != nil { + return ports.ProviderStream{}, err + } + return ports.ProviderStream{Body: io.NopCloser(strings.NewReader("data: [DONE]\n\n"))}, nil +} + +func failing(n int, err func() error) []error { + out := make([]error, n) + for i := range out { + out[i] = err() + } + return out +} + +func fastHealing() SelfHealing { + cfg := DefaultSelfHealing + cfg.RetryDelay = time.Millisecond + return cfg +} + +func TestBreakerOpensProbesAndCloses(t *testing.T) { + now := time.Unix(1_000, 0) + b := newBreaker(SelfHealing{BreakerFailures: 3, BreakerOpenFor: 30 * time.Second, BreakerMaxOpen: 50 * time.Second}) + b.now = func() time.Time { return now } + + for i := 0; i < 2; i++ { + if tr := b.failure("p/m"); tr != nil { + t.Fatalf("opened after %d failures", i+1) + } + } + tr := b.failure("p/m") + if tr == nil || tr.to != circuitOpen || tr.openFor != 30*time.Second { + t.Fatalf("third failure: %+v, want open for 30s", tr) + } + if ok, wait := b.allow("p/m"); ok || wait != 30*time.Second { + t.Fatalf("allow while open = %v, %v", ok, wait) + } + if ok, _ := b.allow("other/m"); !ok { + t.Fatal("another target was blocked") + } + + now = now.Add(31 * time.Second) // Half-open: one probe, the rest wait. + if ok, _ := b.allow("p/m"); !ok { + t.Fatal("no probe after the open period") + } + if ok, _ := b.allow("p/m"); ok { + t.Fatal("a second request went through while the probe was out") + } + if tr := b.failure("p/m"); tr == nil || tr.openFor != 50*time.Second { + t.Fatalf("failed probe: %+v, want reopen for 50s (doubled, capped)", tr) + } + + now = now.Add(51 * time.Second) + if ok, _ := b.allow("p/m"); !ok { + t.Fatal("no second probe") + } + if tr := b.success("p/m"); tr == nil || tr.to != circuitClosed { + t.Fatalf("successful probe: %+v, want closed", tr) + } + if ok, _ := b.allow("p/m"); !ok || b.state("p/m") != circuitClosed { + t.Fatal("circuit not closed after a successful probe") + } +} + +func TestBreakerReleaseFreesTheProbe(t *testing.T) { + now := time.Unix(1_000, 0) + b := newBreaker(SelfHealing{BreakerFailures: 1, BreakerOpenFor: time.Second, BreakerMaxOpen: time.Second}) + b.now = func() time.Time { return now } + b.failure("p/m") + now = now.Add(2 * time.Second) + b.allow("p/m") + b.release("p/m") // The probe's client left: we learned nothing about the provider. + if ok, _ := b.allow("p/m"); !ok { + t.Fatal("probe slot not released") + } +} + +func TestBreakerOff(t *testing.T) { + b := newBreaker(SelfHealing{}) + for i := 0; i < 10; i++ { + b.failure("p/m") + } + if ok, _ := b.allow("p/m"); !ok { + t.Fatal("a disabled breaker blocked a request") + } +} + +// A quick 502 with no fallback is retried once, and the caller gets the answer. +func TestProxy_RetriesAQuickFailureOnce(t *testing.T) { + provider := &scriptedProvider{errs: []error{upstream502()}} + svc := proxyService(&resultMeter{}, "", map[string]ports.Provider{"primary": provider}, false) + svc.SetSelfHealing(fastHealing()) + retries := metrics.ProviderRetries.WithLabelValues("primary", "transient_retry") + before := testutil.ToFloat64(retries) + + if _, err := svc.Proxy(context.Background(), domain.EndpointChatCompletions, []byte(`{}`), domain.GenerateRequest{PublicModelID: "zai-org/glm-5.3-flash"}, "key"); err != nil { + t.Fatalf("Proxy: %v", err) + } + if provider.calls != 2 { + t.Fatalf("provider calls = %d, want 2", provider.calls) + } + if got := testutil.ToFloat64(retries) - before; got != 1 { + t.Fatalf("retries delta = %v, want 1", got) + } +} + +// Requests the provider rejects, rate limits and canceled clients are not +// outages: no retry, and the circuit never opens. +func TestProxy_DoesNotRetryOrTripOnNonOutages(t *testing.T) { + for name, errFn := range map[string]func() error{ + "rejected": func() error { return domain.ErrProviderError(400, "bad").WithMeta("upstream_status", 400) }, + "rate limited": func() error { return domain.ErrProviderError(429, "slow down").WithMeta("upstream_status", 429) }, + "canceled": func() error { return domain.ErrClientCanceled() }, + } { + t.Run(name, func(t *testing.T) { + provider := &scriptedProvider{errs: failing(20, errFn)} + svc := proxyService(&resultMeter{}, "", map[string]ports.Provider{"primary": provider}, false) + svc.SetSelfHealing(fastHealing()) + for i := 0; i < 10; i++ { + svc.Proxy(context.Background(), domain.EndpointChatCompletions, []byte(`{}`), domain.GenerateRequest{PublicModelID: "zai-org/glm-5.3-flash"}, "key") + } + if provider.calls != 10 { + t.Fatalf("provider calls = %d, want 10 (one per request)", provider.calls) + } + }) + } +} + +// A provider that keeps failing gets its circuit opened: later requests fail +// at once with 503 and Retry-After, without reaching it. +func TestProxy_OpensTheCircuitOnAProviderThatKeepsFailing(t *testing.T) { + provider := &scriptedProvider{errs: failing(100, upstream502)} + svc := proxyService(&resultMeter{}, "", map[string]ports.Provider{"primary": provider}, false) + svc.SetSelfHealing(fastHealing()) + run := func() error { + _, err := svc.Proxy(context.Background(), domain.EndpointChatCompletions, []byte(`{}`), domain.GenerateRequest{PublicModelID: "zai-org/glm-5.3-flash"}, "key") + return err + } + for i := 0; i < 3; i++ { // Two attempts each (with the retry): the fifth failure opens it. + run() + } + calls := provider.calls + err := run() + gwErr, ok := err.(*domain.GatewayError) + if !ok || gwErr.HTTPStatus != 503 || gwErr.Metadata["cause"] != domain.CauseCircuitOpen || gwErr.RetryAfter == "" { + t.Fatalf("error = %#v, want 503 circuit_open with Retry-After", err) + } + if provider.calls != calls { + t.Fatalf("provider called while its circuit was open (%d -> %d)", calls, provider.calls) + } + if v := testutil.ToFloat64(metrics.CircuitOpen.WithLabelValues("primary", "glm-5-3-flash")); v != 1 { + t.Fatalf("circuit_open gauge = %v, want 1", v) + } +} + +// With a fallback, an open circuit sends requests straight to the fallback. +func TestProxy_SkipsAnOpenCircuitForTheFallback(t *testing.T) { + primary := &scriptedProvider{errs: failing(100, upstream502)} + fallback := &scriptedProvider{} + svc := proxyService(&resultMeter{}, "", map[string]ports.Provider{"primary": primary, "fallback": fallback}, true) + svc.SetSelfHealing(fastHealing()) + for i := 0; i < 8; i++ { + if _, err := svc.Proxy(context.Background(), domain.EndpointChatCompletions, []byte(`{}`), domain.GenerateRequest{PublicModelID: "zai-org/glm-5.3-flash"}, "key"); err != nil { + t.Fatalf("request %d: %v", i, err) + } + } + // One attempt per request until the fifth failure opens the circuit; then + // the primary is skipped and every request goes to the fallback. + if primary.calls != 5 || fallback.calls != 8 { + t.Fatalf("calls = primary %d, fallback %d; want 5 and 8", primary.calls, fallback.calls) + } +} diff --git a/apps/gateway/internal/application/services/generate_service.go b/apps/gateway/internal/application/services/generate_service.go index feeac0f..618bc7d 100644 --- a/apps/gateway/internal/application/services/generate_service.go +++ b/apps/gateway/internal/application/services/generate_service.go @@ -26,6 +26,8 @@ type GenerateService struct { piiFailOpen bool tinfoilProofs ports.TinfoilTransportProofRepository logger ports.Logger + heal SelfHealing + breaker *breaker } func NewGenerateService( @@ -46,9 +48,17 @@ func NewGenerateService( pii: pii, piiLang: "en", logger: logger, + heal: DefaultSelfHealing, + breaker: newBreaker(DefaultSelfHealing), } } +// SetSelfHealing configures the provider circuit breaker and quick retries. +func (s *GenerateService) SetSelfHealing(cfg SelfHealing) { + s.heal = cfg + s.breaker = newBreaker(cfg) +} + // SetPIIOptions configures the PII filter language and failure mode. // failOpen=true allows requests through when the filter is unreachable. func (s *GenerateService) SetPIIOptions(language string, failOpen bool) { diff --git a/apps/gateway/internal/application/services/proxy.go b/apps/gateway/internal/application/services/proxy.go index 5dde257..6076ca0 100644 --- a/apps/gateway/internal/application/services/proxy.go +++ b/apps/gateway/internal/application/services/proxy.go @@ -162,14 +162,32 @@ func (s *GenerateService) Proxy(ctx context.Context, endpoint string, raw []byte return &ProxyCall{Model: target, Stream: stream}, nil } - call, err := attempt(model) + // The model's provider first, then its fallback if the first one failed + // on the provider's side. A provider whose circuit is open is skipped at + // once, so a dead provider costs callers nothing when a fallback exists. + targets := []domain.PublicModel{model} + if model.Fallback != nil { + if fallback := withProviderTarget(model, *model.Fallback); fallback.SupportsEndpoint(endpoint) { + targets = append(targets, fallback) + } + } + var call *ProxyCall executed := model - if err != nil && shouldTryFallback(ctx, err, model.Fallback) && - withProviderTarget(model, *model.Fallback).SupportsEndpoint(endpoint) { - executed = withProviderTarget(model, *model.Fallback) - s.logFallback(requestID, model, executed, sanitizeErrorWithPIIMapping(err, mapping)) - metrics.RecordFallback(endpoint, model.PublicModelID, model.ProviderConfig.ProviderName, executed.ProviderConfig.ProviderName, err) - call, err = attempt(executed) + for i, target := range targets { + if i > 0 { + if !shouldTryFallback(ctx, err, model.Fallback) { + break + } + s.logFallback(requestID, executed, target, sanitizeErrorWithPIIMapping(err, mapping)) + metrics.RecordFallback(endpoint, model.PublicModelID, executed.ProviderConfig.ProviderName, target.ProviderConfig.ProviderName, err) + } + executed = target + // At most two provider attempts per request: the provider and its + // fallback, or, with no fallback, the provider and one quick retry. + call, err = s.tryTarget(ctx, requestID, target, attempt, len(targets) == 1) + if err == nil { + break + } } if err != nil { defer stopLease() diff --git a/apps/gateway/internal/application/services/systemone_test.go b/apps/gateway/internal/application/services/systemone_test.go index c0a1879..b5f9f31 100644 --- a/apps/gateway/internal/application/services/systemone_test.go +++ b/apps/gateway/internal/application/services/systemone_test.go @@ -44,6 +44,9 @@ func TestSystemOne(t *testing.T) { filter := &fakePIIFilter{enabled: true, byText: map[string][]domain.PIIEntity{"jane@example.com": {entity("EMAIL", "jane@example.com", "jane@example.com")}}} catalog := &stubModelCatalog{model: model} svc := NewGenerateService(auth, catalog, nil, &stubProviderRegistry{providers: map[string]ports.Provider{"typesafe": typesafe.NewAdapter(time.Second)}}, meter, filter, stubLogger{}) + // These checks repeat failures on purpose; the circuit breaker would answer + // in the provider's place after a few, so it is off here. + svc.SetSelfHealing(SelfHealing{}) req, err := mapper.SystemOneRequestToDomain([]byte(evaluationRequest)) if err != nil { t.Fatal(err) diff --git a/apps/gateway/main.go b/apps/gateway/main.go index 9f4c1ff..43e6fe4 100644 --- a/apps/gateway/main.go +++ b/apps/gateway/main.go @@ -92,6 +92,11 @@ func main() { listModelsSvc := services.NewListModelsServiceWithRateProvider(modelCatalog, zapLogger, fxProvider, eurToUSDFallbackRate) generateSvc := services.NewGenerateService(authService, modelCatalog, routerClient, providerRegistry, meteringClient, piiFilter, zapLogger) generateSvc.SetPIIOptions(piiLang, piiFailOpen) + selfHealing, err := loadSelfHealing() + if err != nil { + log.Fatal(err) + } + generateSvc.SetSelfHealing(selfHealing) generateSvc.SetTinfoilProofRepository(tinfoilProofRepo) chatCompletionsSvc := services.NewChatCompletionsService(generateSvc, zapLogger) diff --git a/apps/gateway/selfhealing.go b/apps/gateway/selfhealing.go new file mode 100644 index 0000000..064473f --- /dev/null +++ b/apps/gateway/selfhealing.go @@ -0,0 +1,44 @@ +package main + +import ( + "fmt" + "os" + "strconv" + + "github.com/dappnode/dappnode-nexus-gateway/apps/gateway/internal/application/services" +) + +// loadSelfHealing reads the provider circuit breaker and retry settings: +// +// PROVIDER_BREAKER_FAILURES failures in a row that open a circuit (default 5; 0 turns it off) +// PROVIDER_BREAKER_OPEN_SECONDS first open period (default 30) +// PROVIDER_BREAKER_MAX_OPEN_SECONDS longest open period after failed probes (default 300) +// PROVIDER_RETRY_FAST_FAILURES retry a quick failure once before any output (default true) +func loadSelfHealing() (services.SelfHealing, error) { + cfg := services.DefaultSelfHealing + if v := os.Getenv("PROVIDER_BREAKER_FAILURES"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n < 0 { + return cfg, fmt.Errorf("PROVIDER_BREAKER_FAILURES must be a whole number, 0 or more") + } + cfg.BreakerFailures = n + } + var err error + if cfg.BreakerOpenFor, err = timeoutSeconds("PROVIDER_BREAKER_OPEN_SECONDS", cfg.BreakerOpenFor); err != nil { + return cfg, err + } + if cfg.BreakerMaxOpen, err = timeoutSeconds("PROVIDER_BREAKER_MAX_OPEN_SECONDS", cfg.BreakerMaxOpen); err != nil { + return cfg, err + } + if cfg.BreakerMaxOpen < cfg.BreakerOpenFor { + return cfg, fmt.Errorf("PROVIDER_BREAKER_MAX_OPEN_SECONDS must not be shorter than PROVIDER_BREAKER_OPEN_SECONDS") + } + if v := os.Getenv("PROVIDER_RETRY_FAST_FAILURES"); v != "" { + b, err := strconv.ParseBool(v) + if err != nil { + return cfg, fmt.Errorf("PROVIDER_RETRY_FAST_FAILURES must be true or false") + } + cfg.RetryFastFailures = b + } + return cfg, nil +} diff --git a/pkg/domain/error_class.go b/pkg/domain/error_class.go index e7dab64..9bad81e 100644 --- a/pkg/domain/error_class.go +++ b/pkg/domain/error_class.go @@ -33,6 +33,8 @@ const ( CauseRouterBadResponse = "router_bad_response" CauseReplayStoreFull = "replay_store_full" CauseUntypedError = "untyped_error" + // The provider kept failing, so the gateway stopped sending to it for a while. + CauseCircuitOpen = "circuit_open" ) var faultByCode = map[string]string{