test(failover): cover retry per endpoint family and remote fail-back

Assisted-by: Claude:claude-opus-5-5

tests/e2e/cloud_proxy_helpers_test.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"slices"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -46,6 +47,9 @@ type fakeOpenAIUpstreamServer struct {
 
 	mu     sync.Mutex
 	script func(req []byte) (status int, body string, contentType string)
+	// models is what GET /v1/models lists: failover liveness probes check
+	// that the upstream still serves the target's model.
+	models []string
 }
 
 func newFakeOpenAIUpstream() *fakeOpenAIUpstreamServer {
@@ -59,6 +63,20 @@ func newFakeOpenAIUpstream() *fakeOpenAIUpstreamServer {
 }
 
 func (f *fakeOpenAIUpstreamServer) serve(w http.ResponseWriter, r *http.Request) {
+	// Answered before recording: periodic probes must not overwrite the
+	// request a spec is about to assert on.
+	if r.Method == http.MethodGet && r.URL.Path == "/v1/models" {
+		f.mu.Lock()
+		ids := slices.Clone(f.models)
+		f.mu.Unlock()
+		data := make([]map[string]string, 0, len(ids))
+		for _, id := range ids {
+			data = append(data, map[string]string{"id": id})
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{"data": data})
+		return
+	}
 	atomic.AddInt32(&f.recorder.RequestHits, 1)
 	body, _ := io.ReadAll(r.Body)
 	f.recorder.mu.Lock()
@@ -80,6 +98,12 @@ func (f *fakeOpenAIUpstreamServer) serve(w http.ResponseWriter, r *http.Request)
 func (f *fakeOpenAIUpstreamServer) URL() string { return f.srv.URL }
 func (f *fakeOpenAIUpstreamServer) Close()      { f.srv.Close() }
 
+func (f *fakeOpenAIUpstreamServer) SetModels(ids ...string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.models = ids
+}
+
 func (f *fakeOpenAIUpstreamServer) SetScript(script func(req []byte) (status int, body string, contentType string)) {
 	f.mu.Lock()
 	defer f.mu.Unlock()

tests/e2e/e2e_failover_test.go

@@ -0,0 +1,172 @@
+package e2e_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"mime/multipart"
+	"net/http"
+	"os"
+	"path/filepath"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"gopkg.in/yaml.v3"
+)
+
+var _ = Describe("Failover chains", Label("failover"), func() {
+	postJSON := func(path string, body map[string]any) *http.Response {
+		b, err := json.Marshal(body)
+		Expect(err).ToNot(HaveOccurred())
+		resp, err := http.Post(apiURL+path, "application/json", bytes.NewReader(b))
+		Expect(err).ToNot(HaveOccurred())
+		return resp
+	}
+	expectServedByFallback := func(resp *http.Response) {
+		defer func() { _ = resp.Body.Close() }()
+		body, _ := io.ReadAll(resp.Body)
+		Expect(resp.StatusCode).To(BeNumerically("<", 300), "%s\nheaders: %v", body, resp.Header)
+		Expect(resp.Header.Get("X-LocalAI-Served-Model")).To(Equal("failover-fallback"))
+		Expect(resp.Header.Get("X-LocalAI-Failover")).To(Equal("fallback"))
+	}
+
+	// The entry name is the chain suffix: chain-<name> is written by the suite.
+	DescribeTable("retries every endpoint family on the next target",
+		func(path string, body func(model string) map[string]any) {
+			expectServedByFallback(postJSON(path, body("chain-"+CurrentSpecReport().LeafNodeText)))
+		},
+		Entry("chat", "/chat/completions", func(m string) map[string]any {
+			return map[string]any{"model": m, "messages": []map[string]string{{"role": "user", "content": "hi"}}}
+		}),
+		Entry("completion", "/completions", func(m string) map[string]any {
+			return map[string]any{"model": m, "prompt": "hi"}
+		}),
+		Entry("embeddings", "/embeddings", func(m string) map[string]any {
+			return map[string]any{"model": m, "input": "hi"}
+		}),
+		Entry("tts", "/audio/speech", func(m string) map[string]any {
+			return map[string]any{"model": m, "input": "hi", "voice": "default"}
+		}),
+		Entry("image", "/images/generations", func(m string) map[string]any {
+			return map[string]any{"model": m, "prompt": "a cat", "size": "256x256"}
+		}),
+		Entry("rerank", "/rerank", func(m string) map[string]any {
+			return map[string]any{"model": m, "query": "q", "documents": []string{"a", "b"}}
+		}),
+		Entry("vad", "/vad", func(m string) map[string]any {
+			return map[string]any{"model": m, "audio": []float32{0, 0, 0, 0}}
+		}),
+	)
+
+	It("retries transcription with the multipart body", func() {
+		var body bytes.Buffer
+		mw := multipart.NewWriter(&body)
+		Expect(mw.WriteField("model", "chain-transcription")).To(Succeed())
+		fw, err := mw.CreateFormFile("file", "a.wav")
+		Expect(err).ToNot(HaveOccurred())
+		// 200 ms of 16 kHz mono silence.
+		_, err = fw.Write(wavFromPCM(make([]byte, 6400), 16000))
+		Expect(err).ToNot(HaveOccurred())
+		Expect(mw.Close()).To(Succeed())
+		resp, err := http.Post(apiURL+"/audio/transcriptions", mw.FormDataContentType(), &body)
+		Expect(err).ToNot(HaveOccurred())
+		expectServedByFallback(resp)
+	})
+
+	Describe("remote targets", Ordered, func() {
+		var up1, up2 *fakeOpenAIUpstreamServer
+
+		chatReply := func([]byte) (int, string, string) {
+			return 200, `{"id":"x","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}`, "application/json"
+		}
+
+		BeforeAll(func() {
+			if cloudProxyPath == "" {
+				Skip("cloud-proxy backend binary not built (make build-cloud-proxy-backend)")
+			}
+			up1, up2 = newFakeOpenAIUpstream(), newFakeOpenAIUpstream()
+			DeferCleanup(up1.Close)
+			DeferCleanup(up2.Close)
+			up1.SetModels("up-1")
+			up2.SetModels("up-2")
+			registerFailoverRemoteModels(up1.URL(), up2.URL())
+		})
+
+		It("fails over when the primary upstream errors and fails back when it recovers", func() {
+			up1.SetScript(func([]byte) (int, string, string) {
+				return 503, `{"error":"no healthy nodes"}`, "application/json"
+			})
+			up2.SetScript(chatReply)
+
+			resp := postJSON("/chat/completions", map[string]any{"model": "chain-remote", "messages": []map[string]string{{"role": "user", "content": "hi"}}})
+			defer func() { _ = resp.Body.Close() }()
+			body, _ := io.ReadAll(resp.Body)
+			Expect(resp.StatusCode).To(Equal(200), string(body))
+			Expect(resp.Header.Get("X-LocalAI-Served-Model")).To(Equal("up-2"))
+			Expect(resp.Header.Get("X-LocalAI-Failover")).To(Equal("fallback"))
+			Expect(chainActive("chain-remote")).To(Equal("up-2"))
+
+			up1.SetScript(chatReply)
+			Eventually(func() string { return chainActive("chain-remote") }, 30*time.Second, 500*time.Millisecond).
+				Should(Equal("up-1"))
+
+			resp2 := postJSON("/chat/completions", map[string]any{"model": "chain-remote", "messages": []map[string]string{{"role": "user", "content": "hi"}}})
+			defer func() { _ = resp2.Body.Close() }()
+			Expect(resp2.StatusCode).To(Equal(200))
+			Expect(resp2.Header.Get("X-LocalAI-Served-Model")).To(Equal("up-1"))
+			Expect(resp2.Header.Get("X-LocalAI-Failover")).To(BeEmpty())
+		})
+	})
+})
+
+// chainActive returns the active target of a chain as the REST status reports
+// it, or "" when the status cannot be read.
+func chainActive(chain string) string {
+	r, err := http.Get(anthropicBaseURL + "/api/failover/" + chain)
+	if err != nil {
+		return ""
+	}
+	defer func() { _ = r.Body.Close() }()
+	var st struct {
+		Active string `json:"active"`
+	}
+	_ = json.NewDecoder(r.Body).Decode(&st)
+	return st.Active
+}
+
+// registerFailoverRemoteModels registers two cloud-proxy passthrough models
+// (up-1, up-2) and a chain over them. The upstream URLs exist only at runtime,
+// so the YAMLs are written after startup and the loader re-reads the models
+// directory; the failover manager picks the chain up on its next tick.
+func registerFailoverRemoteModels(url1, url2 string) {
+	proxyModel := func(name, upstream string) map[string]any {
+		return map[string]any{
+			"name":       name,
+			"backend":    "cloud-proxy",
+			"parameters": map[string]any{"model": name + ".bin"},
+			"proxy": map[string]any{
+				"mode":         "passthrough",
+				"provider":     "openai",
+				"upstream_url": upstream + "/v1/chat/completions",
+				"api_key_env":  "CLOUD_PROXY_E2E_OPENAI_KEY",
+			},
+		}
+	}
+	chain := map[string]any{
+		"name": "chain-remote",
+		"failover": map[string]any{
+			"targets":  []map[string]any{{"model": "up-1"}, {"model": "up-2"}},
+			"probe":    map[string]any{"interval": "1s"},
+			"recovery": map[string]any{"probes": 2, "min_dwell": "2s"},
+		},
+	}
+	for _, cfg := range []map[string]any{proxyModel("up-1", url1), proxyModel("up-2", url2), chain} {
+		data, err := yaml.Marshal(cfg)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(os.WriteFile(filepath.Join(modelsPath, cfg["name"].(string)+".yaml"), data, 0644)).To(Succeed())
+	}
+	Expect(localAIApp.ModelConfigLoader().LoadModelConfigsFromPath(modelsPath)).To(Succeed())
+	Eventually(func() string { return chainActive("chain-remote") }, 10*time.Second, 200*time.Millisecond).
+		Should(Equal("up-1"))
+}

tests/e2e/e2e_suite_test.go

@@ -118,6 +118,39 @@ var _ = BeforeSuite(func() {
 	Expect(err).ToNot(HaveOccurred())
 	Expect(os.WriteFile(configPath, configYAML, 0644)).To(Succeed())
 
+	// Failover chains, one per endpoint family: target 0 is a mock model whose
+	// load always fails (the mock rejects models named fail-load*), so every
+	// request exercises the retry onto failover-fallback. The fallback's model
+	// file must exist: the failover manager's liveness probe marks a local
+	// target with a missing file down, and mock-model.bin is never created.
+	Expect(os.WriteFile(filepath.Join(modelsPath, "failover-fallback.bin"), nil, 0644)).To(Succeed())
+	fallbackData, err := yaml.Marshal(map[string]any{
+		"name":       "failover-fallback",
+		"backend":    "mock-backend",
+		"parameters": map[string]any{"model": "failover-fallback.bin"},
+	})
+	Expect(err).ToNot(HaveOccurred())
+	Expect(os.WriteFile(filepath.Join(modelsPath, "failover-fallback.yaml"), fallbackData, 0644)).To(Succeed())
+	for _, family := range []string{"chat", "completion", "embeddings", "transcription", "tts", "image", "rerank", "vad"} {
+		for _, cfg := range []map[string]any{
+			{
+				"name":       "fail-" + family,
+				"backend":    "mock-backend",
+				"parameters": map[string]any{"model": "fail-load-" + family},
+			},
+			{
+				"name": "chain-" + family,
+				"failover": map[string]any{
+					"targets": []map[string]any{{"model": "fail-" + family}, {"model": "failover-fallback"}},
+				},
+			},
+		} {
+			data, err := yaml.Marshal(cfg)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(os.WriteFile(filepath.Join(modelsPath, cfg["name"].(string)+".yaml"), data, 0644)).To(Succeed())
+		}
+	}
+
 	// Create model config for autoparser tests (NoGrammar so tool calls
 	// are driven entirely by the backend's ChatDeltas, not grammar enforcement)
 	autoparserConfig := map[string]any{

tests/e2e/mock-backend/main.go

@@ -93,6 +93,10 @@ func (m *MockBackend) LoadModel(ctx context.Context, in *pb.ModelOptions) (*pb.R
 		"draft_model", in.DraftModel,
 		"mmproj", in.MMProj)
 	recordLoadParams(in)
+	// Lets e2e specs build a failover target whose backend cannot load.
+	if strings.HasPrefix(in.Model, "fail-load") {
+		return &pb.Result{Message: "mock: load failure", Success: false}, nil
+	}
 	return &pb.Result{
 		Message: "Model loaded successfully (mocked)",
 		Success: true,