diff --git a/README.md b/README.md index a58f1e5..c878b42 100644 --- a/README.md +++ b/README.md @@ -260,9 +260,15 @@ timer and cached in memory. serving that model and picks one by the configured **routing strategy** (`round_robin` | `lowest_cost`, set from the dashboard Routing page). - **Admin overrides** — per-model `enabled`/`blocked`, parameter overrides, - pricing, context window, and rate limits live alongside the discovered data - and survive syncs. Clearing an override in the dashboard falls back to the - provider-reported value. + pricing, context window, rate limits, and a backup model live alongside the + discovered data and survive syncs. Clearing an override in the dashboard + falls back to the provider-reported value. +- **Backup models** — any model can name another model as its backup. When + the model is offline, rate limited, or its upstream fails before the client + has received anything, the gateway sends the same request to the backup and + returns the answer under the requested model's name. The client sees no + difference. The request log records each failed attempt, the address that + served the request, and the backup's cost. - **Curated aliases** — the schema has `models` / `model_targets` tables for friendly names with priority/weight failover. No code uses them yet. diff --git a/backend/dashboard/src/lib/types.ts b/backend/dashboard/src/lib/types.ts index c0021c7..55701e6 100644 --- a/backend/dashboard/src/lib/types.ts +++ b/backend/dashboard/src/lib/types.ts @@ -45,6 +45,9 @@ export interface AdminModel { price_cached_per_mtok: number | null rpm_limit: number | null tpm_limit: number | null + /** Backup model served when this one is down or fails. */ + fallback_model_id: string | null + fallback_address: string | null } export interface ChatMessage { role: string; content: unknown } @@ -55,6 +58,8 @@ export interface RequestLog { RequestID: string KeyPrefix: string ModelName: string + /** Set when a backup model served the request instead of ModelName. */ + ServedModel: string ProviderName: string StatusCode: number Streamed: boolean @@ -146,6 +151,7 @@ export interface ModelOverride { rpm_limit?: number | null tpm_limit?: number | null context_length?: number | null + fallback_model_id?: string | null } export type Settings = { diff --git a/backend/dashboard/src/pages/Logs.tsx b/backend/dashboard/src/pages/Logs.tsx index 360e5cb..8f3776b 100644 --- a/backend/dashboard/src/pages/Logs.tsx +++ b/backend/dashboard/src/pages/Logs.tsx @@ -43,6 +43,7 @@ function splitThink(s: string): { reasoning: string; answer: string } { function Detail({ log, content, loading }: { log: RequestLog; content?: LogContent; loading: boolean }) { const meta = [ ['Request ID', log.RequestID || '—'], + ['Served by', log.ServedModel ? `${log.ServedModel} (backup)` : log.ModelName], ['Streamed', log.Streamed ? 'yes' : 'no'], ['Prompt tokens', log.PromptTokens.toLocaleString()], ['Cached tokens', log.CachedTokens.toLocaleString()], @@ -204,7 +205,10 @@ export default function Logs() { onKeyDown={e => { if (e.key === 'Enter' || e.key === ' ') { e.preventDefault(); toggle(l.ID) } }} className="row row-hover cursor-pointer" style={{ gridTemplateColumns: COLS, borderBottom: 'none' }}> {fmtTime(l.CreatedAt)} - {l.ModelName} + + {l.ModelName} + {l.ServedModel && ↪ backup} + {l.ProviderName} {l.KeyPrefix || '—'} {fmtTokens(l.TotalTokens)} diff --git a/backend/dashboard/src/pages/Models.tsx b/backend/dashboard/src/pages/Models.tsx index 2e42f32..13046e9 100644 --- a/backend/dashboard/src/pages/Models.tsx +++ b/backend/dashboard/src/pages/Models.tsx @@ -8,7 +8,7 @@ import { Badge, Button, Card, Checkbox, CodeLine, CopyIconButton, Empty, ErrorNo type Filter = 'all' | 'live' | 'offline' | 'blocked' | 'overridden' function hasOverride(m: AdminModel) { - return m.price_input_per_mtok != null || m.price_output_per_mtok != null || m.price_cached_per_mtok != null || m.rpm_limit != null || m.tpm_limit != null || m.context_length_override != null + return m.price_input_per_mtok != null || m.price_output_per_mtok != null || m.price_cached_per_mtok != null || m.rpm_limit != null || m.tpm_limit != null || m.context_length_override != null || m.fallback_model_id != null } function price(v: number | null | undefined) { @@ -24,13 +24,14 @@ function parseField(raw: string, int = false): number | null | 'invalid' { return n } -function OverrideModal({ model, onClose, onSaved }: { model: AdminModel; onClose: () => void; onSaved: () => void }) { +function OverrideModal({ model, models, onClose, onSaved }: { model: AdminModel; models: AdminModel[]; onClose: () => void; onSaved: () => void }) { const [inPrice, setInPrice] = useState(price(model.price_input_per_mtok)) const [outPrice, setOutPrice] = useState(price(model.price_output_per_mtok)) const [cachedPrice, setCachedPrice] = useState(price(model.price_cached_per_mtok)) const [rpm, setRpm] = useState(price(model.rpm_limit)) const [tpm, setTpm] = useState(price(model.tpm_limit)) const [ctx, setCtx] = useState(price(model.context_length_override)) + const [fallback, setFallback] = useState(model.fallback_model_id ?? '') const [enabled, setEnabled] = useState(model.enabled) const [blocked, setBlocked] = useState(model.blocked) const [loading, setLoading] = useState(false) @@ -57,7 +58,7 @@ function OverrideModal({ model, onClose, onSaved }: { model: AdminModel; onClose } setLoading(true) try { - await api.models.update(model.id, { enabled, blocked, ...(fields as Partial) }) + await api.models.update(model.id, { enabled, blocked, fallback_model_id: fallback || null, ...(fields as Partial) }) onSaved() onClose() } catch (e) { @@ -68,6 +69,7 @@ function OverrideModal({ model, onClose, onSaved }: { model: AdminModel; onClose } const discIn = model.disc_input_per_mtok, discOut = model.disc_output_per_mtok + const backups = models.filter(m => m.id !== model.id && !m.blocked && m.enabled) return ( @@ -89,6 +91,12 @@ function OverrideModal({ model, onClose, onSaved }: { model: AdminModel; onClose setCtx(e.target.value)} placeholder={model.context_length != null ? String(model.context_length) : 'unknown'} /> + + +
@@ -213,7 +221,10 @@ export default function Models() { {address}
- {m.display_name || —} + + {m.display_name || —} + {m.fallback_address && ↪ backup} + {(m.context_length_override ?? m.context_length) ? fmtContext(m.context_length_override ?? m.context_length!).replace(' ctx', '') : '—'}{m.context_length_override != null && *} @@ -232,7 +243,7 @@ export default function Models() { - {selected && setSelected(null)} onSaved={() => { toast('Model updated', 'good'); load() }} />} + {selected && setSelected(null)} onSaved={() => { toast('Model updated', 'good'); load() }} />} ) } diff --git a/backend/internal/api/admin.go b/backend/internal/api/admin.go index 7b9553e..a123edc 100644 --- a/backend/internal/api/admin.go +++ b/backend/internal/api/admin.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "log/slog" "net/http" @@ -15,6 +16,7 @@ import ( "github.com/go-chi/chi/v5" "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgconn" "git.govekar.net/LovroG05/localmodelroutex/internal/catalog" "git.govekar.net/LovroG05/localmodelroutex/internal/providers" @@ -436,6 +438,14 @@ func (a *API) adminUpdateModel(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "context_length must be 1 or more") return } + if o.FallbackModelID, err = patchField[uuid.UUID](fields, "fallback_model_id"); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + if o.FallbackModelID.Value != nil && *o.FallbackModelID.Value == id { + writeError(w, http.StatusBadRequest, "a model cannot be its own backup") + return + } for _, p := range []store.Patch[float64]{o.InputPrice, o.OutputPrice, o.CachedPrice} { if p.Value != nil && *p.Value < 0 { writeError(w, http.StatusBadRequest, "prices must be 0 or more") @@ -450,6 +460,11 @@ func (a *API) adminUpdateModel(w http.ResponseWriter, r *http.Request) { } if err := a.Store.UpdateModelOverride(r.Context(), id, o); err != nil { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == "23503" /* foreign_key_violation */ { + writeError(w, http.StatusBadRequest, "backup model not found") + return + } writeError(w, http.StatusInternalServerError, err.Error()) return } diff --git a/backend/internal/api/chat.go b/backend/internal/api/chat.go index af23cc8..fc3cb1b 100644 --- a/backend/internal/api/chat.go +++ b/backend/internal/api/chat.go @@ -43,101 +43,123 @@ func (a *API) chatCompletions(w http.ResponseWriter, r *http.Request) { key := keyFrom(r.Context()) address := req.Model - entry, err := a.Cache.Resolve(address) + route, err := a.Cache.Route(address) if err != nil { writeError(w, http.StatusNotFound, err.Error()) return } - if !allowed(key, entry.Address) { + if !allowed(key, route.Primary) { writeError(w, http.StatusForbidden, fmt.Sprintf("model %q is not allowed for this key", address)) return } - if !a.Limiter.AllowRequest(entry.Address, ptrInt(entry.RPMLimit)) { - writeError(w, http.StatusTooManyRequests, "rate limit exceeded (RPM)") - return - } - if !a.Limiter.AllowTokens(entry.Address, ptrInt(entry.TPMLimit)) { - writeError(w, http.StatusTooManyRequests, "rate limit exceeded (TPM)") - return - } - prov, ok := a.Manager.Get(entry.ProviderName) - if !ok { - writeError(w, http.StatusServiceUnavailable, "provider unavailable") - return - } - cc, ok := prov.(providers.ChatCompleter) - if !ok { - writeError(w, http.StatusNotImplemented, "provider does not support chat completions") - return - } - - // Snapshot the messages before we overwrite req.Model below. messages := marshalMessages(req.Payload["messages"]) - req.Model = entry.Slug - applyOverrides(req.Payload, entry.ParamOverrides) - - start := time.Now() - if req.Stream { - a.streamChat(w, r, cc, req, entry, key, address, messages, start) - return + var last *failure + for i, entry := range route.Targets { + if i > 0 && r.Context().Err() != nil { + break // the client is gone; a backup cannot help + } + prov, fail := a.admit(entry, 1) + if fail != nil { + last = fail + continue + } + cc, ok := prov.(providers.ChatCompleter) + if !ok { + last = &failure{status: http.StatusNotImplemented, msg: "provider does not support chat completions"} + continue + } + attempt := &providers.ChatRequest{Model: entry.Slug, Stream: req.Stream, Payload: targetPayload(req.Payload, entry)} + next := nextTarget(route, i) + start := time.Now() + if req.Stream { + fail = a.streamChat(w, r, cc, attempt, entry, route, next, key, messages, start) + } else { + fail = a.completeChat(w, r, cc, attempt, entry, route, next, key, messages, start) + } + if fail == nil { + return + } + last = fail } + last.write(w) +} +// completeChat serves a non-streaming completion from one target. It returns +// nil once the client has an answer, or the failure when the request should +// move on to next. An upstream error with no backup behind it is relayed. +func (a *API) completeChat(w http.ResponseWriter, r *http.Request, cc providers.ChatCompleter, req *providers.ChatRequest, entry *catalog.Entry, route *catalog.Route, next *catalog.Entry, key *store.VirtualKey, messages json.RawMessage, start time.Time) *failure { resp, err := cc.ChatCompletion(r.Context(), req) if err != nil { - writeError(w, http.StatusBadGateway, err.Error()) - a.logRequest(r, key, entry, address, http.StatusBadGateway, false, providers.Usage{}, err.Error(), start, nil, messages, "") - return + a.logRequest(r, key, entry, route, http.StatusBadGateway, false, providers.Usage{}, err.Error()+switchNote(next), start, nil, messages, "") + return &failure{status: http.StatusBadGateway, msg: err.Error()} } a.Limiter.RecordTokens(entry.Address, resp.Usage.TotalTokens) - respBody := resp.Body - errMsg := "" if resp.StatusCode >= 400 { - errMsg = upstreamErrorMessage(resp.Body) - } else { - respBody = injectCreated(resp.Body) + errMsg := upstreamErrorMessage(resp.Body) + if next != nil && retryable(resp.StatusCode) { + a.logRequest(r, key, entry, route, resp.StatusCode, false, resp.Usage, errMsg+switchNote(next), start, nil, messages, "") + return &failure{status: resp.StatusCode, body: resp.Body} + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(resp.StatusCode) + _, _ = w.Write(resp.Body) + a.logRequest(r, key, entry, route, resp.StatusCode, false, resp.Usage, errMsg, start, nil, messages, "") + return nil } w.Header().Set("Content-Type", "application/json") w.WriteHeader(resp.StatusCode) - _, _ = w.Write(respBody) + _, _ = w.Write(clientBody(resp.Body, time.Now().Unix(), reportedModel(route, entry))) - completion := extractCompletion(resp.Body) - a.logRequest(r, key, entry, address, resp.StatusCode, false, resp.Usage, errMsg, start, nil, messages, completion) + a.logRequest(r, key, entry, route, resp.StatusCode, false, resp.Usage, "", start, nil, messages, extractCompletion(resp.Body)) + return nil } -func (a *API) streamChat(w http.ResponseWriter, r *http.Request, cc providers.ChatCompleter, req *providers.ChatRequest, entry *catalog.Entry, key *store.VirtualKey, address string, messages json.RawMessage, start time.Time) { +// streamChat serves a streaming completion from one target. The first chunk is +// read before the status line goes out, so an upstream that fails at once can +// still hand the request to a backup. After the first byte reaches the +// client, a failure can only be recorded in the log. +func (a *API) streamChat(w http.ResponseWriter, r *http.Request, cc providers.ChatCompleter, req *providers.ChatRequest, entry *catalog.Entry, route *catalog.Route, next *catalog.Entry, key *store.VirtualKey, messages json.RawMessage, start time.Time) *failure { stream, err := cc.ChatCompletionStream(r.Context(), req) if err != nil { var apiErr *providers.APIError if errors.As(err, &apiErr) { + errMsg := upstreamErrorMessage(apiErr.Body) + if next != nil && retryable(apiErr.StatusCode) { + a.logRequest(r, key, entry, route, apiErr.StatusCode, true, providers.Usage{}, errMsg+switchNote(next), start, nil, messages, "") + return &failure{status: apiErr.StatusCode, body: apiErr.Body} + } w.Header().Set("Content-Type", "application/json") w.WriteHeader(apiErr.StatusCode) _, _ = w.Write(apiErr.Body) - a.logRequest(r, key, entry, address, apiErr.StatusCode, true, providers.Usage{}, upstreamErrorMessage(apiErr.Body), start, nil, messages, "") - return + a.logRequest(r, key, entry, route, apiErr.StatusCode, true, providers.Usage{}, errMsg, start, nil, messages, "") + return nil } - writeError(w, http.StatusBadGateway, err.Error()) - a.logRequest(r, key, entry, address, http.StatusBadGateway, true, providers.Usage{}, err.Error(), start, nil, messages, "") - return + a.logRequest(r, key, entry, route, http.StatusBadGateway, true, providers.Usage{}, err.Error()+switchNote(next), start, nil, messages, "") + return &failure{status: http.StatusBadGateway, msg: err.Error()} } defer stream.Close() + ev, err := stream.Recv() + if err != nil && !errors.Is(err, io.EOF) { + errMsg := "stream interrupted: " + err.Error() + a.logRequest(r, key, entry, route, http.StatusBadGateway, true, providers.Usage{}, errMsg+switchNote(next), start, nil, messages, "") + return &failure{status: http.StatusBadGateway, msg: errMsg} + } + flusher, _ := w.(http.Flusher) w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") w.WriteHeader(http.StatusOK) + model := reportedModel(route, entry) var ttft *int var reasoning, content strings.Builder created := time.Now().Unix() errMsg := "" - for { - ev, err := stream.Recv() - if errors.Is(err, io.EOF) { - break - } + for !errors.Is(err, io.EOF) { if err != nil { // The status line is already sent, so the error can only be // recorded in the log row and signalled by closing the stream. @@ -154,13 +176,14 @@ func (a *API) streamChat(w http.ResponseWriter, r *http.Request, cc providers.Ch } // Accumulate reasoning and content deltas separately so the recorded // completion wraps reasoning in think tags, matching extractCompletion. - r, c := extractDelta(ev.Data) - reasoning.WriteString(r) + rs, c := extractDelta(ev.Data) + reasoning.WriteString(rs) content.WriteString(c) - _, _ = fmt.Fprintf(w, "data: %s\n\n", injectCreatedChunk(ev.Data, created)) + _, _ = fmt.Fprintf(w, "data: %s\n\n", clientBody(ev.Data, created, model)) if flusher != nil { flusher.Flush() } + ev, err = stream.Recv() } _, _ = fmt.Fprint(w, "data: [DONE]\n\n") if flusher != nil { @@ -173,21 +196,26 @@ func (a *API) streamChat(w http.ResponseWriter, r *http.Request, cc providers.Ch } usage := stream.Usage() a.Limiter.RecordTokens(entry.Address, usage.TotalTokens) - a.logRequest(r, key, entry, address, http.StatusOK, true, usage, errMsg, start, ttft, messages, completion) + a.logRequest(r, key, entry, route, http.StatusOK, true, usage, errMsg, start, ttft, messages, completion) + return nil } -func applyOverrides(payload, overrides map[string]any) { - maps.Copy(payload, overrides) -} - -func (a *API) logRequest(r *http.Request, key *store.VirtualKey, entry *catalog.Entry, address string, status int, streamed bool, usage providers.Usage, errMsg string, start time.Time, ttft *int, messages json.RawMessage, completion string) { +// logRequest records one attempt. ModelName is the address the client sent; +// when a backup served the request, ServedModel names it and Cost uses its +// prices, so the switch is visible in the log but not to the client. +func (a *API) logRequest(r *http.Request, key *store.VirtualKey, entry *catalog.Entry, route *catalog.Route, status int, streamed bool, usage providers.Usage, errMsg string, start time.Time, ttft *int, messages json.RawMessage, completion string) { entryID := entry.ID providerID := entry.ProviderID + served := "" + if route.IsFallback(entry) { + served = entry.Address + } rec := store.RequestLog{ ProviderModelID: &entryID, ProviderID: &providerID, ProviderName: entry.ProviderName, - ModelName: address, + ModelName: route.Address, + ServedModel: served, StatusCode: status, Streamed: streamed, Error: errMsg, @@ -381,34 +409,3 @@ func clientIP(r *http.Request) string { } return host } - -// injectCreated ensures a non-streaming response has a "created" timestamp. -// If the upstream already provided one, it is preserved. -func injectCreated(body []byte) []byte { - var obj map[string]any - if json.Unmarshal(body, &obj) == nil { - if _, ok := obj["created"]; ok { - return body // upstream already set it - } - obj["created"] = time.Now().Unix() - if out, err := json.Marshal(obj); err == nil { - return out - } - } - return body // fall back to raw body on any error -} - -// injectCreatedChunk ensures a streaming SSE chunk has a "created" timestamp. -func injectCreatedChunk(data []byte, created int64) []byte { - var obj map[string]any - if json.Unmarshal(data, &obj) == nil { - if _, ok := obj["created"]; ok { - return data // upstream already set it - } - obj["created"] = created - if out, err := json.Marshal(obj); err == nil { - return out - } - } - return data // fall back to raw chunk on any error -} diff --git a/backend/internal/api/embeddings.go b/backend/internal/api/embeddings.go index 030f74a..cf177ec 100644 --- a/backend/internal/api/embeddings.go +++ b/backend/internal/api/embeddings.go @@ -28,57 +28,62 @@ func (a *API) embeddings(w http.ResponseWriter, r *http.Request) { key := keyFrom(r.Context()) address := req.Model - entry, err := a.Cache.Resolve(address) + route, err := a.Cache.Route(address) if err != nil { writeError(w, http.StatusNotFound, err.Error()) return } - if !allowed(key, entry.Address) { + if !allowed(key, route.Primary) { writeError(w, http.StatusForbidden, fmt.Sprintf("model %q is not allowed for this key", address)) return } - if !a.Limiter.AllowRequest(entry.Address, ptrInt(entry.RPMLimit)) { - writeError(w, http.StatusTooManyRequests, "rate limit exceeded (RPM)") + + var last *failure + for i, entry := range route.Targets { + if i > 0 && r.Context().Err() != nil { + break + } + prov, fail := a.admit(entry, 1) + if fail != nil { + last = fail + continue + } + emb, ok := prov.(providers.Embedder) + if !ok { + last = &failure{status: http.StatusNotImplemented, msg: "provider does not support embeddings"} + continue + } + attempt := &providers.EmbeddingRequest{Model: entry.Slug, Payload: targetPayload(req.Payload, entry)} + next := nextTarget(route, i) + start := time.Now() + resp, err := emb.Embeddings(r.Context(), attempt) + if err != nil { + a.logRequest(r, key, entry, route, http.StatusBadGateway, false, providers.Usage{}, err.Error()+switchNote(next), start, nil, nil, "") + last = &failure{status: http.StatusBadGateway, msg: err.Error()} + continue + } + a.Limiter.RecordTokens(entry.Address, resp.Usage.TotalTokens) + + errMsg := "" + body := resp.Body + if resp.StatusCode >= 400 { + errMsg = upstreamErrorMessage(resp.Body) + if next != nil && retryable(resp.StatusCode) { + a.logRequest(r, key, entry, route, resp.StatusCode, false, resp.Usage, errMsg+switchNote(next), start, nil, nil, "") + last = &failure{status: resp.StatusCode, body: resp.Body} + continue + } + } else { + body = withModel(resp.Body, reportedModel(route, entry)) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(resp.StatusCode) + _, _ = w.Write(body) + + a.logRequest(r, key, entry, route, resp.StatusCode, false, resp.Usage, errMsg, start, nil, nil, "") return } - if !a.Limiter.AllowTokens(entry.Address, ptrInt(entry.TPMLimit)) { - writeError(w, http.StatusTooManyRequests, "rate limit exceeded (TPM)") - return - } - - prov, ok := a.Manager.Get(entry.ProviderName) - if !ok { - writeError(w, http.StatusServiceUnavailable, "provider unavailable") - return - } - emb, ok := prov.(providers.Embedder) - if !ok { - writeError(w, http.StatusNotImplemented, "provider does not support embeddings") - return - } - - req.Model = entry.Slug - applyOverrides(req.Payload, entry.ParamOverrides) - - start := time.Now() - resp, err := emb.Embeddings(r.Context(), req) - if err != nil { - writeError(w, http.StatusBadGateway, err.Error()) - a.logRequest(r, key, entry, address, http.StatusBadGateway, false, providers.Usage{}, err.Error(), start, nil, nil, "") - return - } - - a.Limiter.RecordTokens(entry.Address, resp.Usage.TotalTokens) - - errMsg := "" - if resp.StatusCode >= 400 { - errMsg = upstreamErrorMessage(resp.Body) - } - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(resp.StatusCode) - _, _ = w.Write(resp.Body) - - a.logRequest(r, key, entry, address, resp.StatusCode, false, resp.Usage, errMsg, start, nil, nil, "") + last.write(w) } // ptrInt dereferences an *int, returning 0 (no limit) when nil. diff --git a/backend/internal/api/images.go b/backend/internal/api/images.go index ff67fc3..4d09fb4 100644 --- a/backend/internal/api/images.go +++ b/backend/internal/api/images.go @@ -47,69 +47,73 @@ func (a *API) imageGenerations(w http.ResponseWriter, r *http.Request) { key := keyFrom(r.Context()) address := req.Model - entry, err := a.Cache.Resolve(address) + route, err := a.Cache.Route(address) if err != nil { writeError(w, http.StatusNotFound, err.Error()) return } - if !allowed(key, entry.Address) { + if !allowed(key, route.Primary) { writeError(w, http.StatusForbidden, fmt.Sprintf("model %q is not allowed for this key", address)) return } - if !slices.Contains(entry.Capabilities, string(providers.CapabilityImages)) { - writeError(w, http.StatusBadRequest, fmt.Sprintf("model %q does not generate images", address)) - return - } - // Each requested image is one upstream call, so charge all of them - // against the RPM limit up front. - if !a.Limiter.AllowRequests(entry.Address, ptrInt(entry.RPMLimit), max(req.N, 1)) { - writeError(w, http.StatusTooManyRequests, "rate limit exceeded (RPM)") - return - } - if !a.Limiter.AllowTokens(entry.Address, ptrInt(entry.TPMLimit)) { - writeError(w, http.StatusTooManyRequests, "rate limit exceeded (TPM)") - return - } - - prov, ok := a.Manager.Get(entry.ProviderName) - if !ok { - writeError(w, http.StatusServiceUnavailable, "provider unavailable") - return - } - gen, ok := prov.(providers.ImageGenerator) - if !ok { - writeError(w, http.StatusNotImplemented, "provider does not support image generation") - return - } - // Log the prompt as a one-message conversation so it shows in the request // log like a chat turn. messages, _ := json.Marshal([]map[string]any{{"role": "user", "content": req.Prompt}}) - req.Model = entry.Slug - applyOverrides(req.Payload, entry.ParamOverrides) + var last *failure + for i, entry := range route.Targets { + if i > 0 && r.Context().Err() != nil { + break + } + if !slices.Contains(entry.Capabilities, string(providers.CapabilityImages)) { + last = &failure{status: http.StatusBadRequest, msg: fmt.Sprintf("model %q does not generate images", address)} + continue + } + // Each requested image is one upstream call, so charge all of them + // against the RPM limit up front. + prov, fail := a.admit(entry, max(req.N, 1)) + if fail != nil { + last = fail + continue + } + gen, ok := prov.(providers.ImageGenerator) + if !ok { + last = &failure{status: http.StatusNotImplemented, msg: "provider does not support image generation"} + continue + } + attempt := *req + attempt.Model = entry.Slug + attempt.Payload = targetPayload(req.Payload, entry) + next := nextTarget(route, i) + start := time.Now() + resp, err := gen.ImageGeneration(r.Context(), &attempt) + if err != nil { + a.logRequest(r, key, entry, route, http.StatusBadGateway, false, providers.Usage{}, err.Error()+switchNote(next), start, nil, messages, "") + last = &failure{status: http.StatusBadGateway, msg: err.Error()} + continue + } + a.Limiter.RecordTokens(entry.Address, resp.Usage.TotalTokens) - start := time.Now() - resp, err := gen.ImageGeneration(r.Context(), req) - if err != nil { - writeError(w, http.StatusBadGateway, err.Error()) - a.logRequest(r, key, entry, address, http.StatusBadGateway, false, providers.Usage{}, err.Error(), start, nil, messages, "") + errMsg := "" + completion := "" + if resp.StatusCode >= 400 { + errMsg = upstreamErrorMessage(resp.Body) + if next != nil && retryable(resp.StatusCode) { + a.logRequest(r, key, entry, route, resp.StatusCode, false, resp.Usage, errMsg+switchNote(next), start, nil, messages, "") + last = &failure{status: resp.StatusCode, body: resp.Body} + continue + } + } else { + completion = imageCompletionSummary(resp.Body) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(resp.StatusCode) + _, _ = w.Write(resp.Body) + + a.logRequest(r, key, entry, route, resp.StatusCode, false, resp.Usage, errMsg, start, nil, messages, completion) return } - a.Limiter.RecordTokens(entry.Address, resp.Usage.TotalTokens) - - errMsg := "" - completion := "" - if resp.StatusCode >= 400 { - errMsg = upstreamErrorMessage(resp.Body) - } else { - completion = imageCompletionSummary(resp.Body) - } - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(resp.StatusCode) - _, _ = w.Write(resp.Body) - - a.logRequest(r, key, entry, address, resp.StatusCode, false, resp.Usage, errMsg, start, nil, messages, completion) + last.write(w) } // imageCompletionSummary describes an images response for the content log diff --git a/backend/internal/api/proxy_test.go b/backend/internal/api/proxy_test.go index 8e57d9e..bc8e159 100644 --- a/backend/internal/api/proxy_test.go +++ b/backend/internal/api/proxy_test.go @@ -76,17 +76,27 @@ func (f *fakeLogs) TouchVirtualKey(_ context.Context, id uuid.UUID) error { return nil } -// wait blocks until one request has been fully logged. -func (f *fakeLogs) wait(t *testing.T) store.RequestLog { +// waitN blocks until n requests have been logged and returns the rows in the +// order they were written. +func (f *fakeLogs) waitN(t *testing.T, n int) []store.RequestLog { t.Helper() - select { - case <-f.done: - case <-time.After(3 * time.Second): - t.Fatal("request log was not written") + for range n { + select { + case <-f.done: + case <-time.After(3 * time.Second): + t.Fatal("request log was not written") + } } f.mu.Lock() defer f.mu.Unlock() - return f.rows[len(f.rows)-1] + return append([]store.RequestLog(nil), f.rows...) +} + +// wait blocks until one request has been fully logged. +func (f *fakeLogs) wait(t *testing.T) store.RequestLog { + t.Helper() + rows := f.waitN(t, 1) + return rows[len(rows)-1] } // upstream is a fake OpenRouter that records the last body it received. @@ -473,3 +483,187 @@ func TestAllowedPatterns(t *testing.T) { t.Error("allow-all key must pass") } } + +// failPrimary answers gpt-4o with status and lets every other model through +// to chatReply. +func failPrimary(status int) func(http.ResponseWriter, *http.Request, map[string]any) { + return func(w http.ResponseWriter, r *http.Request, body map[string]any) { + if body["model"] == "openai/gpt-4o" { + w.WriteHeader(status) + _, _ = io.WriteString(w, `{"error":{"message":"primary down"}}`) + return + } + chatReply(w, r, body) + } +} + +// withBackup makes the image model the backup of gpt-4o, at a different price. +func withBackup(rows []store.CatalogRow) { + id := rows[0].ID + rows[1].FallbackModelID = &id + rows[0].InputPricePerMTok = 1 + rows[0].OutputPricePerMTok = 2 +} + +// rowsByStatus indexes log rows by status code. +func rowsByStatus(rows []store.RequestLog) map[int]store.RequestLog { + out := map[int]store.RequestLog{} + for _, r := range rows { + out[r.StatusCode] = r + } + return out +} + +func TestChatCompletionsFallsBackOnUpstreamError(t *testing.T) { + env := newTestEnv(t, failPrimary(http.StatusServiceUnavailable), withBackup) + rec := call(env.api.chatCompletions, allowAllKey(), http.MethodPost, + `{"model":"or/openai/gpt-4o","messages":[{"role":"user","content":"hi"}]}`) + if rec.Code != http.StatusOK { + t.Fatalf("status=%d body=%s", rec.Code, rec.Body) + } + // The client sees the model it asked for. + var out map[string]any + _ = json.Unmarshal(rec.Body.Bytes(), &out) + if out["model"] != "openai/gpt-4o" { + t.Errorf("reported model = %v, want the requested one", out["model"]) + } + if calls, body := env.upstream.last(); calls != 2 || body["model"] != "google/gemini-2.5-flash-image" { + t.Errorf("upstream calls=%d last model=%v", calls, body["model"]) + } + + // The log sees both attempts and the backup's cost. + rows := rowsByStatus(env.logs.waitN(t, 2)) + failed, ok := rows[http.StatusServiceUnavailable] + if !ok || !strings.Contains(failed.Error, "primary down") || !strings.Contains(failed.Error, "switched to backup or/google/gemini-2.5-flash-image") { + t.Errorf("failed attempt row = %+v", failed) + } + if failed.ServedModel != "" || failed.ModelName != "or/openai/gpt-4o" { + t.Errorf("failed attempt must be logged against the requested model: %+v", failed) + } + served := rows[http.StatusOK] + if served.ModelName != "or/openai/gpt-4o" || served.ServedModel != "or/google/gemini-2.5-flash-image" { + t.Errorf("served row = %+v", served) + } + want := 100.0/1e6*1 + 50.0/1e6*2 + if served.Cost < want-1e-12 || served.Cost > want+1e-12 { + t.Errorf("cost = %v, want the backup's price %v", served.Cost, want) + } +} + +func TestChatCompletionsStreamFallsBack(t *testing.T) { + env := newTestEnv(t, failPrimary(http.StatusBadGateway), withBackup) + rec := call(env.api.chatCompletions, allowAllKey(), http.MethodPost, + `{"model":"or/openai/gpt-4o","messages":[],"stream":true}`) + if rec.Code != http.StatusOK || rec.Header().Get("Content-Type") != "text/event-stream" { + t.Fatalf("status=%d type=%q body=%s", rec.Code, rec.Header().Get("Content-Type"), rec.Body) + } + body := rec.Body.String() + if strings.Count(body, "data: ") != 4 || strings.Count(body, `"model":"openai/gpt-4o"`) != 3 { + t.Errorf("stream body = %q", body) + } + rows := rowsByStatus(env.logs.waitN(t, 2)) + if rows[http.StatusOK].ServedModel != "or/google/gemini-2.5-flash-image" || !rows[http.StatusOK].Streamed { + t.Errorf("served row = %+v", rows[http.StatusOK]) + } +} + +func TestChatCompletionsUnavailableModelServedByBackup(t *testing.T) { + env := newTestEnv(t, chatReply, func(rows []store.CatalogRow) { + withBackup(rows) + rows[1].Available = false + }) + rec := call(env.api.chatCompletions, allowAllKey(), http.MethodPost, + `{"model":"or/openai/gpt-4o","messages":[]}`) + if rec.Code != http.StatusOK { + t.Fatalf("status=%d body=%s", rec.Code, rec.Body) + } + var out map[string]any + _ = json.Unmarshal(rec.Body.Bytes(), &out) + if out["model"] != "openai/gpt-4o" { + t.Errorf("reported model = %v", out["model"]) + } + if calls, _ := env.upstream.last(); calls != 1 { + t.Errorf("upstream calls = %d, want 1", calls) + } + row := env.logs.wait(t) + if row.ModelName != "or/openai/gpt-4o" || row.ServedModel != "or/google/gemini-2.5-flash-image" { + t.Errorf("log row = %+v", row) + } + + // The down model stays listed, because clients can still call it. + list := call(env.api.listModels, allowAllKey(), http.MethodGet, "") + if !strings.Contains(list.Body.String(), `"or/openai/gpt-4o"`) { + t.Errorf("models list = %s", list.Body) + } +} + +func TestChatCompletionsRateLimitOverflowsToBackup(t *testing.T) { + env := newTestEnv(t, chatReply, func(rows []store.CatalogRow) { + withBackup(rows) + one := 1 + rows[1].RPMLimit = &one + }) + body := `{"model":"or/openai/gpt-4o","messages":[]}` + if rec := call(env.api.chatCompletions, allowAllKey(), http.MethodPost, body); rec.Code != http.StatusOK { + t.Fatalf("first call status=%d", rec.Code) + } + if row := env.logs.wait(t); row.ServedModel != "" { + t.Errorf("first call must use the requested model: %+v", row) + } + if rec := call(env.api.chatCompletions, allowAllKey(), http.MethodPost, body); rec.Code != http.StatusOK { + t.Fatalf("second call status=%d body=%s", rec.Code, rec.Body) + } + if row := env.logs.wait(t); row.ServedModel != "or/google/gemini-2.5-flash-image" { + t.Errorf("second call must overflow to the backup: %+v", row) + } +} + +func TestChatCompletionsClientErrorIsNotRetried(t *testing.T) { + env := newTestEnv(t, failPrimary(http.StatusBadRequest), withBackup) + rec := call(env.api.chatCompletions, allowAllKey(), http.MethodPost, + `{"model":"or/openai/gpt-4o","messages":[]}`) + if rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "primary down") { + t.Errorf("status=%d body=%s", rec.Code, rec.Body) + } + if calls, _ := env.upstream.last(); calls != 1 { + t.Errorf("upstream calls = %d, want 1", calls) + } + if row := env.logs.wait(t); strings.Contains(row.Error, "switched") { + t.Errorf("log row = %+v", row) + } +} + +func TestChatCompletionsLastFailureIsRelayed(t *testing.T) { + // Both the model and its backup are down: the client gets the last error. + env := newTestEnv(t, func(w http.ResponseWriter, _ *http.Request, _ map[string]any) { + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = io.WriteString(w, `{"error":{"message":"all down"}}`) + }, withBackup) + rec := call(env.api.chatCompletions, allowAllKey(), http.MethodPost, + `{"model":"or/openai/gpt-4o","messages":[]}`) + if rec.Code != http.StatusServiceUnavailable || !strings.Contains(rec.Body.String(), "all down") { + t.Errorf("status=%d body=%s", rec.Code, rec.Body) + } + if calls, _ := env.upstream.last(); calls != 2 { + t.Errorf("upstream calls = %d, want 2", calls) + } + env.logs.waitN(t, 2) +} + +func TestEmbeddingsFallBack(t *testing.T) { + env := newTestEnv(t, func(w http.ResponseWriter, _ *http.Request, body map[string]any) { + if body["model"] == "openai/gpt-4o" { + w.WriteHeader(http.StatusInternalServerError) + return + } + _, _ = io.WriteString(w, `{"object":"list","model":"backup","data":[{"embedding":[0.1]}],"usage":{"prompt_tokens":4,"total_tokens":4}}`) + }, withBackup) + rec := call(env.api.embeddings, allowAllKey(), http.MethodPost, `{"model":"or/openai/gpt-4o","input":"x"}`) + if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), `"model":"openai/gpt-4o"`) { + t.Errorf("status=%d body=%s", rec.Code, rec.Body) + } + rows := rowsByStatus(env.logs.waitN(t, 2)) + if rows[http.StatusOK].ServedModel != "or/google/gemini-2.5-flash-image" { + t.Errorf("served row = %+v", rows[http.StatusOK]) + } +} diff --git a/backend/internal/api/route.go b/backend/internal/api/route.go new file mode 100644 index 0000000..5928048 --- /dev/null +++ b/backend/internal/api/route.go @@ -0,0 +1,132 @@ +package api + +import ( + "encoding/json" + "maps" + "net/http" + + "git.govekar.net/LovroG05/localmodelroutex/internal/catalog" + "git.govekar.net/LovroG05/localmodelroutex/internal/providers" +) + +// The proxy handlers walk a route's targets in order. The requested model is +// tried first. When it is rate limited, has no provider, or its upstream +// fails before the client has received anything, the next backup is tried +// with the same request. A backup answers under the requested model's name, +// so the client cannot tell the difference. The request log can: each failed +// attempt gets its own row, and the row for the backup carries both the +// requested address and the address that served it, with the backup's cost. + +// failure is one target's refusal. When no target answers, the last failure +// is sent to the client. +type failure struct { + status int + body []byte // upstream error payload to relay; nil writes msg as a gateway error + msg string +} + +func (f *failure) write(w http.ResponseWriter) { + if f.body != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(f.status) + _, _ = w.Write(f.body) + return + } + writeError(w, f.status, f.msg) +} + +// retryable reports whether an upstream status means the request should move +// on to a backup. Client errors are not retried: a backup would get the same +// request and most likely give the same answer. +func retryable(status int) bool { + return status == http.StatusRequestTimeout || status == http.StatusTooManyRequests || status >= http.StatusInternalServerError +} + +// admit charges requests against a target's rate limits and returns its +// provider. The failure says why the target cannot serve. +func (a *API) admit(e *catalog.Entry, requests int) (providers.Provider, *failure) { + if !a.Limiter.AllowRequests(e.Address, ptrInt(e.RPMLimit), requests) { + return nil, &failure{status: http.StatusTooManyRequests, msg: "rate limit exceeded (RPM)"} + } + if !a.Limiter.AllowTokens(e.Address, ptrInt(e.TPMLimit)) { + return nil, &failure{status: http.StatusTooManyRequests, msg: "rate limit exceeded (TPM)"} + } + prov, ok := a.Manager.Get(e.ProviderName) + if !ok { + return nil, &failure{status: http.StatusServiceUnavailable, msg: "provider unavailable"} + } + return prov, nil +} + +// targetPayload copies the client payload for one target and applies the +// target's parameter overrides. Each target gets its own copy, so the +// overrides of a failed target do not leak into the request to its backup. +func targetPayload(payload map[string]any, e *catalog.Entry) map[string]any { + out := maps.Clone(payload) + maps.Copy(out, e.ParamOverrides) + return out +} + +// switchNote is appended to the log row of a failed attempt when the request +// moves on to a backup. Empty when there is no backup after the attempt. +func switchNote(next *catalog.Entry) string { + if next == nil { + return "" + } + return "; switched to backup " + next.Address +} + +// nextTarget returns the target after index i, or nil. +func nextTarget(route *catalog.Route, i int) *catalog.Entry { + if i+1 < len(route.Targets) { + return route.Targets[i+1] + } + return nil +} + +// reportedModel is the model id a response from e must carry. A backup +// reports the requested model's id; the requested model keeps what its +// upstream sent. +func reportedModel(route *catalog.Route, e *catalog.Entry) string { + if route.IsFallback(e) { + return route.Slug + } + return "" +} + +// editJSON decodes an object body, applies edit, and encodes it again. A body +// that is not a JSON object is returned unchanged. +func editJSON(body []byte, edit func(map[string]any)) []byte { + var obj map[string]any + if json.Unmarshal(body, &obj) != nil || obj == nil { + return body + } + edit(obj) + out, err := json.Marshal(obj) + if err != nil { + return body + } + return out +} + +// clientBody prepares an upstream chat body or chunk for the client. A missing +// "created" timestamp is added, and when model is set the reported model is +// replaced with it. +func clientBody(body []byte, created int64, model string) []byte { + return editJSON(body, func(obj map[string]any) { + if _, ok := obj["created"]; !ok { + obj["created"] = created + } + if model != "" { + obj["model"] = model + } + }) +} + +// withModel replaces the reported model when model is set. +func withModel(body []byte, model string) []byte { + if model == "" { + return body + } + return editJSON(body, func(obj map[string]any) { obj["model"] = model }) +} diff --git a/backend/internal/catalog/cache.go b/backend/internal/catalog/cache.go index 9d4095b..4ce004b 100644 --- a/backend/internal/catalog/cache.go +++ b/backend/internal/catalog/cache.go @@ -8,6 +8,8 @@ import ( "sync" "sync/atomic" + "github.com/google/uuid" + "git.govekar.net/LovroG05/localmodelroutex/internal/store" ) @@ -23,17 +25,26 @@ type Cache struct { mu sync.RWMutex byAddr map[string]*Entry + byID map[uuid.UUID]*Entry entries []*Entry strategy Strategy + // standby holds enabled, unblocked models the provider does not serve + // right now. They are not servable, but a request for one can still be + // routed to its backup. + standby map[string]*Entry rr sync.Map // map[string]*atomic.Uint64, round-robin counters per query } // NewCache returns an empty cache; call Reload to populate it. func NewCache(s CatalogStore) *Cache { - return &Cache{store: s, byAddr: map[string]*Entry{}, strategy: StrategyRoundRobin} + return &Cache{store: s, byAddr: map[string]*Entry{}, byID: map[uuid.UUID]*Entry{}, standby: map[string]*Entry{}, strategy: StrategyRoundRobin} } +// maxFallbackHops bounds a backup chain. A chain may legitimately loop (two +// models that back each other up), so the walk also tracks visited ids. +const maxFallbackHops = 8 + // Reload rebuilds the cache (servable entries only) and routing strategy from // the store. func (c *Cache) Reload(ctx context.Context) error { @@ -43,9 +54,11 @@ func (c *Cache) Reload(ctx context.Context) error { } byAddr := make(map[string]*Entry, len(rows)) + byID := make(map[uuid.UUID]*Entry, len(rows)) + standby := map[string]*Entry{} entries := make([]*Entry, 0, len(rows)) for _, r := range rows { - if !r.Enabled || r.Blocked || !r.Available { + if !r.Enabled || r.Blocked { continue } var overrides map[string]any @@ -69,8 +82,14 @@ func (c *Cache) Reload(ctx context.Context) error { CachedPricePerMTok: r.CachedPricePerMTok, RPMLimit: r.RPMLimit, TPMLimit: r.TPMLimit, + FallbackID: r.FallbackModelID, + } + if !r.Available { + standby[e.Address] = e + continue } byAddr[e.Address] = e + byID[e.ID] = e entries = append(entries, e) } @@ -82,15 +101,29 @@ func (c *Cache) Reload(ctx context.Context) error { c.mu.Lock() c.byAddr = byAddr + c.byID = byID + c.standby = standby c.entries = entries c.strategy = strategy c.mu.Unlock() return nil } -// Resolve maps a model address to a servable entry. "any/" selects among -// matching providers using the configured strategy. +// Resolve maps a model address to the servable entry a request would reach +// first: the model itself, or its first live backup when the model is down. func (c *Cache) Resolve(address string) (*Entry, error) { + r, err := c.Route(address) + if err != nil { + return nil, err + } + return r.Targets[0], nil +} + +// Route maps a model address to the entries that can serve it, in order. +// "any/" selects among matching providers using the configured strategy; +// the selected entry's backups follow it. A model that is enabled but not +// available routes straight to its backups. +func (c *Cache) Route(address string) (*Route, error) { provider, slug, ok := SplitAddress(address) if !ok { return nil, fmt.Errorf("invalid model address %q (want /)", address) @@ -104,14 +137,51 @@ func (c *Cache) Resolve(address string) (*Entry, error) { if len(cands) == 0 { return nil, fmt.Errorf("no provider serves model %q", slug) } - return c.selectLocked(slug, cands), nil + e := c.selectLocked(slug, cands) + return &Route{Address: address, Primary: e.Address, Slug: e.Slug, Targets: c.chainLocked(e)}, nil } - e, ok := c.byAddr[address] - if !ok { - return nil, fmt.Errorf("model %q not found", address) + if e, ok := c.byAddr[address]; ok { + return &Route{Address: address, Primary: address, Slug: e.Slug, Targets: c.chainLocked(e)}, nil } - return e, nil + if e, ok := c.standby[address]; ok { + if targets := c.chainLocked(e); len(targets) > 0 { + return &Route{Address: address, Primary: address, Slug: e.Slug, Targets: targets}, nil + } + } + return nil, fmt.Errorf("model %q not found", address) +} + +// chainLocked returns e, when it is servable, followed by its live backups. +// Backups that are not servable are skipped and the walk continues through +// them, so a down backup does not hide the ones behind it. +func (c *Cache) chainLocked(e *Entry) []*Entry { + var out []*Entry + seen := map[uuid.UUID]bool{} + for hops := 0; e != nil && hops <= maxFallbackHops && !seen[e.ID]; hops++ { + seen[e.ID] = true + if c.byID[e.ID] == e { + out = append(out, e) + } + e = c.lookupLocked(e.FallbackID) + } + return out +} + +// lookupLocked finds an enabled, unblocked entry by id, servable or standby. +func (c *Cache) lookupLocked(id *uuid.UUID) *Entry { + if id == nil { + return nil + } + if e, ok := c.byID[*id]; ok { + return e + } + for _, e := range c.standby { + if e.ID == *id { + return e + } + } + return nil } // candidatesLocked returns servable entries whose slug equals slug or ends with @@ -149,12 +219,18 @@ func (c *Cache) selectLocked(key string, cands []*Entry) *Entry { func cost(e *Entry) float64 { return e.InputPricePerMTok + e.OutputPricePerMTok } -// List returns a snapshot of servable entries. +// List returns a snapshot of the entries clients can call: the servable ones, +// plus unavailable models that have a live backup. func (c *Cache) List() []*Entry { c.mu.RLock() defer c.mu.RUnlock() - out := make([]*Entry, len(c.entries)) + out := make([]*Entry, len(c.entries), len(c.entries)+len(c.standby)) copy(out, c.entries) + for _, e := range c.standby { + if len(c.chainLocked(e)) > 0 { + out = append(out, e) + } + } return out } diff --git a/backend/internal/catalog/cache_test.go b/backend/internal/catalog/cache_test.go index 161c048..5007fd3 100644 --- a/backend/internal/catalog/cache_test.go +++ b/backend/internal/catalog/cache_test.go @@ -1,19 +1,139 @@ package catalog -import "testing" +import ( + "testing" + + "github.com/google/uuid" +) func newTestCache(strategy Strategy, entries ...*Entry) *Cache { - c := &Cache{byAddr: map[string]*Entry{}, strategy: strategy} + c := &Cache{byAddr: map[string]*Entry{}, byID: map[uuid.UUID]*Entry{}, standby: map[string]*Entry{}, strategy: strategy} for _, e := range entries { if e.Address == "" { e.Address = e.ProviderName + "/" + e.Slug } + if e.ID == uuid.Nil { + e.ID = uuid.New() + } c.byAddr[e.Address] = e + c.byID[e.ID] = e c.entries = append(c.entries, e) } return c } +// addStandby registers an enabled model the provider does not serve right now. +func addStandby(c *Cache, e *Entry) *Entry { + if e.Address == "" { + e.Address = e.ProviderName + "/" + e.Slug + } + if e.ID == uuid.Nil { + e.ID = uuid.New() + } + c.standby[e.Address] = e + return e +} + +func backup(e *Entry) *uuid.UUID { id := e.ID; return &id } + +func addresses(entries []*Entry) []string { + out := make([]string, len(entries)) + for i, e := range entries { + out[i] = e.Address + } + return out +} + +func TestRouteFollowsBackupChainAndSkipsDownBackups(t *testing.T) { + a := &Entry{ProviderName: "local", Slug: "a"} + b := &Entry{ProviderName: "local", Slug: "b"} + cEntry := &Entry{ProviderName: "cloud", Slug: "c"} + c := newTestCache(StrategyRoundRobin, a, cEntry) + addStandby(c, b) + a.FallbackID = backup(b) + b.FallbackID = backup(cEntry) + + r, err := c.Route("local/a") + if err != nil { + t.Fatal(err) + } + if got := addresses(r.Targets); len(got) != 2 || got[0] != "local/a" || got[1] != "cloud/c" { + t.Fatalf("targets = %v, want [local/a cloud/c]", got) + } + if r.Primary != "local/a" || r.Slug != "a" || r.IsFallback(a) || !r.IsFallback(cEntry) { + t.Errorf("route = %+v", r) + } +} + +func TestRouteUnavailableModelGoesToBackup(t *testing.T) { + b := &Entry{ProviderName: "cloud", Slug: "org/a"} + c := newTestCache(StrategyRoundRobin, b) + a := addStandby(c, &Entry{ProviderName: "local", Slug: "a"}) + a.FallbackID = backup(b) + + r, err := c.Route("local/a") + if err != nil { + t.Fatal(err) + } + if got := addresses(r.Targets); len(got) != 1 || got[0] != "cloud/org/a" { + t.Fatalf("targets = %v", got) + } + // The client asked for local/a, so the allowlist and the reported model + // use it, not the backup. + if r.Address != "local/a" || r.Primary != "local/a" || r.Slug != "a" || !r.IsFallback(b) { + t.Errorf("route = %+v", r) + } + if e, err := c.Resolve("local/a"); err != nil || e != b { + t.Errorf("Resolve = %v, %v", e, err) + } + if got := addresses(c.List()); len(got) != 2 { + t.Errorf("List must include the down model with a live backup: %v", got) + } +} + +func TestRouteUnavailableModelWithoutBackupIsNotFound(t *testing.T) { + c := newTestCache(StrategyRoundRobin) + addStandby(c, &Entry{ProviderName: "local", Slug: "a"}) + if _, err := c.Route("local/a"); err == nil { + t.Fatal("expected not found") + } + if got := c.List(); len(got) != 0 { + t.Errorf("List = %v", addresses(got)) + } +} + +func TestRouteMutualBackupsDoNotLoop(t *testing.T) { + a := &Entry{ProviderName: "x", Slug: "a"} + b := &Entry{ProviderName: "x", Slug: "b"} + c := newTestCache(StrategyRoundRobin, a, b) + a.FallbackID = backup(b) + b.FallbackID = backup(a) + r, err := c.Route("x/b") + if err != nil { + t.Fatal(err) + } + if got := addresses(r.Targets); len(got) != 2 || got[0] != "x/b" || got[1] != "x/a" { + t.Fatalf("targets = %v", got) + } +} + +func TestRouteAnyCarriesBackups(t *testing.T) { + a := &Entry{ProviderName: "openai", Slug: "gpt-4o"} + b := &Entry{ProviderName: "cloud", Slug: "other"} + c := newTestCache(StrategyRoundRobin, a, b) + a.FallbackID = backup(b) + r, err := c.Route("any/gpt-4o") + if err != nil { + t.Fatal(err) + } + if got := addresses(r.Targets); len(got) != 2 || got[1] != "cloud/other" { + t.Fatalf("targets = %v", got) + } + if r.Address != "any/gpt-4o" || r.Primary != "openai/gpt-4o" { + t.Errorf("route = %+v", r) + } +} + func TestSplitAddress(t *testing.T) { cases := []struct { in string diff --git a/backend/internal/catalog/entry.go b/backend/internal/catalog/entry.go index 6702503..ec6cd4a 100644 --- a/backend/internal/catalog/entry.go +++ b/backend/internal/catalog/entry.go @@ -41,4 +41,27 @@ type Entry struct { CachedPricePerMTok float64 RPMLimit *int TPMLimit *int + // FallbackID is the admin-configured backup model, when set. The cache + // resolves it to servable entries in Route. + FallbackID *uuid.UUID } + +// Route is the ordered list of servable entries for one requested address: +// the model the client asked for first, then its backups. A backup is used +// when the model before it is unavailable or its upstream fails, and the +// client sees no difference in the response. +type Route struct { + // Address is the address the client sent, for the request log. + Address string + // Primary is the address of the model the client asked for, after any + // "any/" selection. Key allowlists are checked against it. + Primary string + // Slug is the primary model's upstream id. Responses served by a backup + // report this id, so the switch stays invisible to the client. + Slug string + // Targets holds the servable entries to try, in order. It is never empty. + Targets []*Entry +} + +// IsFallback reports whether e is a backup rather than the requested model. +func (r *Route) IsFallback(e *Entry) bool { return e.Address != r.Primary } diff --git a/backend/internal/influx/influx.go b/backend/internal/influx/influx.go index 44952f0..e739826 100644 --- a/backend/internal/influx/influx.go +++ b/backend/internal/influx/influx.go @@ -147,6 +147,12 @@ func (e *Exporter) requestPoint(rec store.RequestLog) Point { "key_prefix": keyPrefix, "status_code": strconv.Itoa(rec.StatusCode), "streamed": strconv.FormatBool(rec.Streamed), + // fallback is true when a backup model served the request; then + // served_model names it and cost is the backup's cost. + "fallback": strconv.FormatBool(rec.ServedModel != ""), + } + if rec.ServedModel != "" { + tags["served_model"] = rec.ServedModel } if rec.UserID != nil { tags["user_id"] = rec.UserID.String() diff --git a/backend/internal/influx/influx_test.go b/backend/internal/influx/influx_test.go index fbc84d8..12f8ac9 100644 --- a/backend/internal/influx/influx_test.go +++ b/backend/internal/influx/influx_test.go @@ -165,7 +165,7 @@ func TestRecordWritesRequestPoint(t *testing.T) { if srv.auth != "Bearer secret" { t.Fatalf("auth = %q", srv.auth) } - want := `lmr_request,key_prefix=lmr_abc,model=openai/gpt-4o,provider=openai,status_code=200,streamed=true,user_id=11111111-1111-1111-1111-111111111111 ` + + want := `lmr_request,fallback=false,key_prefix=lmr_abc,model=openai/gpt-4o,provider=openai,status_code=200,streamed=true,user_id=11111111-1111-1111-1111-111111111111 ` + `cached_tokens=2i,completion_tokens=5i,cost=0.00125,error="",errors=0i,finish_reason="",latency_ms=321i,prompt_tokens=10i,request_id="req-1",requests=1i,total_tokens=15i,ttft_ms=42i 100000000000` + "\n" if len(srv.bodies) != 1 || srv.bodies[0] != want { t.Fatalf("bodies = %q\nwant %q", srv.bodies, want) @@ -206,6 +206,21 @@ func TestRequestPointWritesAllFields(t *testing.T) { if p.Tags["key_prefix"] != "(no key)" { t.Errorf("key_prefix = %q", p.Tags["key_prefix"]) } + if p.Tags["fallback"] != "false" { + t.Errorf("fallback = %q", p.Tags["fallback"]) + } + if _, ok := p.Tags["served_model"]; ok { + t.Errorf("served_model must be absent when the requested model served") + } +} + +func TestRequestPointTagsBackupModel(t *testing.T) { + srv := newCaptureServer(t) + e := newTestExporter(t, srv, Config{}) + p := e.requestPoint(store.RequestLog{StatusCode: 200, ModelName: "local/a", ServedModel: "cloud/b"}) + if p.Tags["fallback"] != "true" || p.Tags["served_model"] != "cloud/b" || p.Tags["model"] != "local/a" { + t.Errorf("tags = %v", p.Tags) + } } func TestBatchAndFlushOnClose(t *testing.T) { diff --git a/backend/internal/store/provider_models.go b/backend/internal/store/provider_models.go index c10ed39..f8162d3 100644 --- a/backend/internal/store/provider_models.go +++ b/backend/internal/store/provider_models.go @@ -95,6 +95,8 @@ type CatalogRow struct { CachedPricePerMTok float64 RPMLimit *int TPMLimit *int + // FallbackModelID is the admin-configured backup model, when set. + FallbackModelID *uuid.UUID } // ListCatalog returns every catalog entry belonging to an enabled provider. @@ -109,7 +111,7 @@ func (s *Store) ListCatalog(ctx context.Context) ([]CatalogRow, error) { COALESCE(pm.price_input_per_mtok, pm.disc_input_price_per_mtok, 0)::double precision, COALESCE(pm.price_output_per_mtok, pm.disc_output_price_per_mtok, 0)::double precision, COALESCE(pm.price_cached_per_mtok, 0)::double precision, - pm.rpm_limit, pm.tpm_limit + pm.rpm_limit, pm.tpm_limit, pm.fallback_model_id FROM provider_models pm JOIN providers p ON p.id = pm.provider_id WHERE p.enabled = true`) @@ -126,7 +128,7 @@ func (s *Store) ListCatalog(ctx context.Context) ([]CatalogRow, error) { &r.DisplayName, &r.Capabilities, &r.ContextLength, &r.Enabled, &r.Blocked, &r.Available, &r.ParamOverrides, &r.InputPricePerMTok, &r.OutputPricePerMTok, &r.CachedPricePerMTok, - &r.RPMLimit, &r.TPMLimit, + &r.RPMLimit, &r.TPMLimit, &r.FallbackModelID, ); err != nil { return nil, err } diff --git a/backend/internal/store/provider_models_admin.go b/backend/internal/store/provider_models_admin.go index f3a4c45..059a325 100644 --- a/backend/internal/store/provider_models_admin.go +++ b/backend/internal/store/provider_models_admin.go @@ -28,6 +28,9 @@ type AdminModelRow struct { PriceCachedPerMTok *float64 `json:"price_cached_per_mtok"` RPMLimit *int `json:"rpm_limit"` TPMLimit *int `json:"tpm_limit"` + FallbackModelID *uuid.UUID `json:"fallback_model_id"` + // FallbackAddress is the backup model's "/" for display. + FallbackAddress *string `json:"fallback_address"` } // ListModelsAdmin lists every catalog entry (no servability filter) for admin. @@ -37,9 +40,12 @@ func (s *Store) ListModelsAdmin(ctx context.Context) ([]AdminModelRow, error) { pm.available, pm.enabled, pm.blocked, pm.context_length, pm.context_length_override, pm.last_seen_at, pm.disc_input_price_per_mtok, pm.disc_output_price_per_mtok, pm.price_input_per_mtok, pm.price_output_per_mtok, pm.price_cached_per_mtok, - pm.rpm_limit, pm.tpm_limit + pm.rpm_limit, pm.tpm_limit, + pm.fallback_model_id, fp.name || '/' || fb.slug FROM provider_models pm JOIN providers p ON p.id = pm.provider_id + LEFT JOIN provider_models fb ON fb.id = pm.fallback_model_id + LEFT JOIN providers fp ON fp.id = fb.provider_id ORDER BY p.name, pm.slug`) if err != nil { return nil, err @@ -52,7 +58,7 @@ func (s *Store) ListModelsAdmin(ctx context.Context) ([]AdminModelRow, error) { &m.Available, &m.Enabled, &m.Blocked, &m.ContextLength, &m.ContextLengthOvr, &m.LastSeenAt, &m.DiscInputPerMTok, &m.DiscOutputPerMTok, &m.PriceInputPerMTok, &m.PriceOutputPerMTok, &m.PriceCachedPerMTok, - &m.RPMLimit, &m.TPMLimit); err != nil { + &m.RPMLimit, &m.TPMLimit, &m.FallbackModelID, &m.FallbackAddress); err != nil { return nil, err } out = append(out, m) @@ -79,6 +85,8 @@ type ModelOverride struct { TPMLimit Patch[int] // ContextLength overrides the provider-reported context window. ContextLength Patch[int] + // FallbackModelID is the backup model served when this one fails. + FallbackModelID Patch[uuid.UUID] } func (s *Store) UpdateModelOverride(ctx context.Context, id uuid.UUID, o ModelOverride) error { @@ -96,7 +104,8 @@ func (s *Store) UpdateModelOverride(ctx context.Context, id uuid.UUID, o ModelOv price_cached_per_mtok = CASE WHEN $9::boolean THEN $10 ELSE price_cached_per_mtok END, rpm_limit = CASE WHEN $11::boolean THEN $12 ELSE rpm_limit END, tpm_limit = CASE WHEN $13::boolean THEN $14 ELSE tpm_limit END, - context_length_override = CASE WHEN $15::boolean THEN $16 ELSE context_length_override END + context_length_override = CASE WHEN $15::boolean THEN $16 ELSE context_length_override END, + fallback_model_id = CASE WHEN $17::boolean THEN $18 ELSE fallback_model_id END WHERE id = $1`, id, o.Enabled, o.Blocked, params, o.InputPrice.Set, o.InputPrice.Value, @@ -105,6 +114,7 @@ func (s *Store) UpdateModelOverride(ctx context.Context, id uuid.UUID, o ModelOv o.RPMLimit.Set, o.RPMLimit.Value, o.TPMLimit.Set, o.TPMLimit.Value, o.ContextLength.Set, o.ContextLength.Value, + o.FallbackModelID.Set, o.FallbackModelID.Value, ) return err } diff --git a/backend/internal/store/request_logs.go b/backend/internal/store/request_logs.go index 2231b3c..db7df31 100644 --- a/backend/internal/store/request_logs.go +++ b/backend/internal/store/request_logs.go @@ -19,6 +19,7 @@ type RequestLog struct { ProviderID *uuid.UUID KeyPrefix string ModelName string // the address the client requested + ServedModel string // the address that served it, when a backup model did ProviderName string StatusCode int Streamed bool @@ -43,14 +44,15 @@ func (s *Store) InsertRequestLog(ctx context.Context, e RequestLog) (uuid.UUID, key_prefix, model_name, provider_name, status_code, streamed, finish_reason, error, prompt_tokens, completion_tokens, cached_tokens, total_tokens, - cost, latency_ms, ttft_ms, request_id, client_ip - ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20) + cost, latency_ms, ttft_ms, request_id, client_ip, served_model + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21) RETURNING id`, e.VirtualKeyID, e.UserID, e.ProviderModelID, e.ProviderID, nullString(e.KeyPrefix), nullString(e.ModelName), nullString(e.ProviderName), e.StatusCode, e.Streamed, nullString(e.FinishReason), nullString(e.Error), e.PromptTokens, e.CompletionTokens, e.CachedTokens, e.TotalTokens, e.Cost, e.LatencyMS, e.TTFTMs, nullString(e.RequestID), nullInet(e.ClientIP), + nullString(e.ServedModel), ).Scan(&id) return id, err } @@ -76,6 +78,7 @@ type RequestLogListRow struct { RequestID string KeyPrefix string ModelName string + ServedModel string // set when a backup model served the request ProviderName string StatusCode int Streamed bool @@ -132,7 +135,8 @@ func (s *Store) ListRequestLogs(ctx context.Context, p ListLogsParams) ([]Reques } rows, err := s.pool.Query(ctx, ` SELECT rl.id::text, rl.created_at, COALESCE(rl.request_id,''), - COALESCE(rl.key_prefix,''), COALESCE(rl.model_name,''), COALESCE(rl.provider_name,''), + COALESCE(rl.key_prefix,''), COALESCE(rl.model_name,''), COALESCE(rl.served_model,''), + COALESCE(rl.provider_name,''), rl.status_code, rl.streamed, COALESCE(rl.error,''), rl.prompt_tokens, rl.completion_tokens, rl.cached_tokens, rl.total_tokens, rl.cost, rl.latency_ms, rl.ttft_ms @@ -145,6 +149,7 @@ func (s *Store) ListRequestLogs(ctx context.Context, p ListLogsParams) ([]Reques AND ($6::timestamptz IS NULL OR rl.created_at >= $6) AND ($7::timestamptz IS NULL OR (rl.created_at, rl.id) < ($7, $8)) AND ($9 = '' OR rl.model_name ILIKE '%' || $9 || '%' + OR rl.served_model ILIKE '%' || $9 || '%' OR rl.provider_name ILIKE '%' || $9 || '%' OR rl.key_prefix ILIKE '%' || $9 || '%' OR rl.request_id ILIKE '%' || $9 || '%') @@ -161,7 +166,7 @@ func (s *Store) ListRequestLogs(ctx context.Context, p ListLogsParams) ([]Reques var r RequestLogListRow if err := rows.Scan( &r.ID, &r.CreatedAt, &r.RequestID, - &r.KeyPrefix, &r.ModelName, &r.ProviderName, + &r.KeyPrefix, &r.ModelName, &r.ServedModel, &r.ProviderName, &r.StatusCode, &r.Streamed, &r.Error, &r.PromptTokens, &r.CompletionTokens, &r.CachedTokens, &r.TotalTokens, &r.Cost, &r.LatencyMS, &r.TTFTMs, diff --git a/backend/internal/store/store_integration_test.go b/backend/internal/store/store_integration_test.go index 7706fa1..ca79826 100644 --- a/backend/internal/store/store_integration_test.go +++ b/backend/internal/store/store_integration_test.go @@ -119,14 +119,17 @@ func TestStoreQueries(t *testing.T) { if err != nil { t.Fatalf("ListModelsAdmin: %v", err) } - var m1 uuid.UUID + var m1, m2 uuid.UUID for _, r := range rows { if r.ProviderName == "local" && r.Slug == "m1" { m1 = r.ID } + if r.ProviderName == "local" && r.Slug == "m2" { + m2 = r.ID + } } - if m1 == uuid.Nil { - t.Fatalf("local/m1 missing from %+v", rows) + if m1 == uuid.Nil || m2 == uuid.Nil { + t.Fatalf("local/m1 or local/m2 missing from %+v", rows) } price := 2.5 rpm := 30 @@ -157,10 +160,55 @@ func TestStoreQueries(t *testing.T) { } } + // Backup model: set, read back through the admin list and the catalog, + // reject an unknown target, then clear. + if err := st.UpdateModelOverride(ctx, m1, store.ModelOverride{FallbackModelID: store.Patch[uuid.UUID]{Set: true, Value: &m2}}); err != nil { + t.Fatalf("UpdateModelOverride fallback: %v", err) + } + rows, _ = st.ListModelsAdmin(ctx) + for _, r := range rows { + if r.ID == m1 && (r.FallbackModelID == nil || *r.FallbackModelID != m2 || r.FallbackAddress == nil || *r.FallbackAddress != "local/m2") { + t.Fatalf("fallback not applied: %+v", r) + } + } + on := true + if err := st.UpdateProvider(ctx, provID, store.UpdateProviderParams{Enabled: &on}); err != nil { + t.Fatalf("UpdateProvider enable: %v", err) + } + cat, err := st.ListCatalog(ctx) + if err != nil { + t.Fatalf("ListCatalog: %v", err) + } + var sawM1 bool + for _, r := range cat { + if r.ID == m1 { + sawM1 = true + if r.FallbackModelID == nil || *r.FallbackModelID != m2 { + t.Fatalf("catalog row lacks fallback: %+v", r) + } + } + } + if !sawM1 { + t.Fatalf("local/m1 missing from catalog %+v", cat) + } + unknown := uuid.New() + if err := st.UpdateModelOverride(ctx, m1, store.ModelOverride{FallbackModelID: store.Patch[uuid.UUID]{Set: true, Value: &unknown}}); err == nil { + t.Fatal("UpdateModelOverride must reject an unknown backup model") + } + if err := st.UpdateModelOverride(ctx, m1, store.ModelOverride{FallbackModelID: store.Patch[uuid.UUID]{Set: true}}); err != nil { + t.Fatalf("UpdateModelOverride clear fallback: %v", err) + } + rows, _ = st.ListModelsAdmin(ctx) + for _, r := range rows { + if r.ID == m1 && (r.FallbackModelID != nil || r.FallbackAddress != nil) { + t.Fatalf("fallback not cleared: %+v", r) + } + } + // Logs: insert, then list and summarize. logID, err := st.InsertRequestLog(ctx, store.RequestLog{ - VirtualKeyID: &keyID, UserID: &user.ID, ProviderID: &provID, ProviderModelID: &m1, - KeyPrefix: "sk-00000001", ModelName: "local/m1", ProviderName: "local", + VirtualKeyID: &keyID, UserID: &user.ID, ProviderID: &provID, ProviderModelID: &m2, + KeyPrefix: "sk-00000001", ModelName: "local/m1", ServedModel: "local/m2", ProviderName: "local", StatusCode: 502, Error: "upstream down", LatencyMS: 12, ClientIP: "127.0.0.1", }) if err != nil { @@ -170,9 +218,13 @@ func TestStoreQueries(t *testing.T) { t.Fatalf("InsertLogContent: %v", err) } list, err := st.ListRequestLogs(ctx, store.ListLogsParams{UserID: &user.ID, Query: "local/m1", StatusGte: 400}) - if err != nil || len(list) != 1 || list[0].Error != "upstream down" { + if err != nil || len(list) != 1 || list[0].Error != "upstream down" || list[0].ServedModel != "local/m2" { t.Fatalf("ListRequestLogs: %v %+v", err, list) } + // The search also matches the address that served the request. + if list, err := st.ListRequestLogs(ctx, store.ListLogsParams{UserID: &user.ID, Query: "local/m2"}); err != nil || len(list) != 1 { + t.Fatalf("ListRequestLogs by served model: %v %+v", err, list) + } sum, err = st.GetLogsSummary(ctx, time.Now().Add(-time.Hour), &user.ID) if err != nil || sum.Requests != 1 || sum.Errors != 1 { t.Fatalf("summary = %+v err=%v", sum, err) diff --git a/backend/migrations/000006_fallback_model.down.sql b/backend/migrations/000006_fallback_model.down.sql new file mode 100644 index 0000000..30a7906 --- /dev/null +++ b/backend/migrations/000006_fallback_model.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE request_logs DROP COLUMN IF EXISTS served_model; +ALTER TABLE provider_models DROP COLUMN IF EXISTS fallback_model_id; diff --git a/backend/migrations/000006_fallback_model.up.sql b/backend/migrations/000006_fallback_model.up.sql new file mode 100644 index 0000000..472af39 --- /dev/null +++ b/backend/migrations/000006_fallback_model.up.sql @@ -0,0 +1,9 @@ +-- Admin-configured backup model. When the model is unavailable or its upstream +-- fails, the gateway serves the request with the backup instead. NULL means +-- there is no backup. Deleting the backup clears the reference. +ALTER TABLE provider_models + ADD COLUMN fallback_model_id uuid REFERENCES provider_models (id) ON DELETE SET NULL; + +-- The address that served a request when it differs from model_name, which is +-- the address the client requested. NULL when the requested model served it. +ALTER TABLE request_logs ADD COLUMN served_model text; diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 9808f58..96ca3dc 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -109,13 +109,37 @@ router, and `/upstream//props` for models llama-swap has loaded. It never asks llama-swap about a stopped model, because that would start it. Admin overrides live in separate columns on the same row and survive syncs: -enabled, blocked, parameter overrides, prices, context window, and rate -limits. A model that disappears from the upstream list is marked unavailable, -not deleted. +enabled, blocked, parameter overrides, prices, context window, rate limits, +and a backup model. A model that disappears from the upstream list is marked +unavailable, not deleted. The cache holds every servable row in memory, keyed by address. It is rebuilt after each sync, so the request path never queries the database for routing. +### Backup models + +A model's `fallback_model_id` names another catalog row. The cache turns an +address into a route: the requested model first, then its backups in chain +order, with models that are not servable skipped. Two models may back each +other up; the walk tracks visited ids and stops after a fixed number of hops. +A model that is enabled but unavailable still routes, straight to its backups, +and still appears in `/v1/models`. + +The proxy handlers walk the route. A target is skipped when the rate limiter +refuses it, its provider is missing, the transport fails, or the upstream +answers 408, 429, or 5xx. Client errors (4xx) are relayed, because the backup +would most likely answer the same. For a stream, the first chunk is read +before the status line goes to the client, so a stream that dies at once can +still move to a backup. After the first byte is sent, a failure can only be +logged. + +The switch is invisible to the client and visible in the log. A backup's +response reports the requested model's id. The key allowlist is checked +against the requested address. Every failed attempt gets its own log row, with +the error and a note that names the backup. The row for the backup keeps the +requested address in `model_name`, puts the backup in `served_model`, and +uses the backup's prices for the cost. + ## Authentication Two systems that do not mix: