From 3331a1af578019d91c5ce4998a9a47aa4d7f6660 Mon Sep 17 00:00:00 2001 From: Damilola Edwards Date: Thu, 23 Jul 2026 16:33:05 +0100 Subject: [PATCH] Recover fully from a partial reindex failure When one search index rebuilt successfully but another failed in the same reindex pass, only the failed index was left parked. The one that succeeded stayed live under the new embedding model even though builtModel was never advanced to record that, so the runtime's own bookkeeping no longer matched what was actually live. Recovery also depended on the proxy's served model changing again on a later tick; if it reverted to the model builtModel already recorded, the model-change guard saw no difference and never retried, leaving the failed index parked indefinitely with no way back short of a restart. On a partial failure, park every index the reindex pass touched, not just the one that failed, so nothing is ever left live in a space the runtime isn't tracking. Track that a reindex is incomplete and have the background refresher retry on every tick while it is, regardless of whether the served model still looks different from what's recorded, so a retry against the same model can still succeed and fully restore service. --- pkg/searchruntime/reindex_recovery_test.go | 206 +++++++++++++++++++++ pkg/searchruntime/runtime.go | 39 +++- 2 files changed, 240 insertions(+), 5 deletions(-) create mode 100644 pkg/searchruntime/reindex_recovery_test.go diff --git a/pkg/searchruntime/reindex_recovery_test.go b/pkg/searchruntime/reindex_recovery_test.go new file mode 100644 index 00000000..2269e21e --- /dev/null +++ b/pkg/searchruntime/reindex_recovery_test.go @@ -0,0 +1,206 @@ +package searchruntime + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "sync" + "testing" + + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/require" + + "github.com/ethpandaops/panda/pkg/embedding" + "github.com/ethpandaops/panda/pkg/module" + "github.com/ethpandaops/panda/pkg/resource" + "github.com/ethpandaops/panda/pkg/types" + "github.com/ethpandaops/panda/runbooks" +) + +// fakeEmbeddingBackend serves the proxy's v2 embedding routes well enough for +// embedding.RemoteEmbedder to talk to it, with a switch to make every real +// embed call fail on demand — letting a test control exactly which of +// reindex's index builds succeeds. +type fakeEmbeddingBackend struct { + mu sync.Mutex + model string + dims int + fail bool +} + +func (b *fakeEmbeddingBackend) configure(model string, dims int, fail bool) { + b.mu.Lock() + defer b.mu.Unlock() + + b.model, b.dims, b.fail = model, dims, fail +} + +func (b *fakeEmbeddingBackend) snapshot() (string, int, bool) { + b.mu.Lock() + defer b.mu.Unlock() + + return b.model, b.dims, b.fail +} + +func (b *fakeEmbeddingBackend) handler() http.Handler { + mux := http.NewServeMux() + + mux.HandleFunc("/v2/embedding/check", func(w http.ResponseWriter, _ *http.Request) { + model, dims, _ := b.snapshot() + + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "model": model, "dimensions": dims, "cached": []any{}, + }) + }) + + mux.HandleFunc("/v2/embedding", func(w http.ResponseWriter, r *http.Request) { + model, dims, fail := b.snapshot() + + if fail { + http.Error(w, "embedding backend unavailable", http.StatusInternalServerError) + + return + } + + var req struct { + Items []struct { + Hash string `json:"hash"` + Text string `json:"text"` + } `json:"items"` + } + _ = json.NewDecoder(r.Body).Decode(&req) + + results := make([]map[string]any, len(req.Items)) + for i, item := range req.Items { + vec := make([]float32, dims) + if dims > 0 { + vec[0] = 1 + } + + results[i] = map[string]any{"hash": item.Hash, "vector": vec} + } + + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "results": results, "model": model, "dimensions": dims, + }) + }) + + return mux +} + +// stubEmbeddingProxy is a minimal proxy.Service whose URL points at a fake +// embedding backend under test control. +type stubEmbeddingProxy struct{ baseURL string } + +func (s *stubEmbeddingProxy) Start(context.Context) error { return nil } +func (s *stubEmbeddingProxy) Stop(context.Context) error { return nil } +func (s *stubEmbeddingProxy) URL() string { return s.baseURL } +func (s *stubEmbeddingProxy) Ready() bool { return true } +func (s *stubEmbeddingProxy) RegisterToken() string { return "" } +func (s *stubEmbeddingProxy) Invalidate() {} +func (s *stubEmbeddingProxy) RevokeToken() {} + +func (s *stubEmbeddingProxy) ClickHouseDatasources() []string { return nil } +func (s *stubEmbeddingProxy) ClickHouseDatasourceInfo() []types.DatasourceInfo { return nil } + +func (s *stubEmbeddingProxy) ClickHouseQuery(context.Context, string, string, url.Values) ([]byte, error) { + return nil, nil +} + +func (s *stubEmbeddingProxy) PrometheusDatasourceInfo() []types.DatasourceInfo { return nil } +func (s *stubEmbeddingProxy) LokiDatasourceInfo() []types.DatasourceInfo { return nil } +func (s *stubEmbeddingProxy) BenchmarkoorDatasourceInfo() []types.DatasourceInfo { return nil } +func (s *stubEmbeddingProxy) ComputeDatasourceInfo() []types.DatasourceInfo { return nil } +func (s *stubEmbeddingProxy) EthNodeAvailable() bool { return false } +func (s *stubEmbeddingProxy) EthNodeDatasourceInfo() []types.DatasourceInfo { return nil } +func (s *stubEmbeddingProxy) EmbeddingAvailable() bool { return false } +func (s *stubEmbeddingProxy) EmbeddingModel() string { return "" } + +// TestReindexPartialFailureParksEverythingAndRecovers drives the real reindex +// through a full lifecycle: a successful baseline build, a partial failure +// where one index succeeds and another fails, and a recovery once the same +// model is retried. It asserts every index — including the one that +// individually succeeded during the partial failure — ends up parked +// not-ready rather than left live in a space builtModel no longer records, +// and that a retry against the unchanged model fully restores service. +func TestReindexPartialFailureParksEverythingAndRecovers(t *testing.T) { + log := logrus.New() + log.SetLevel(logrus.PanicLevel) + + backend := &fakeEmbeddingBackend{} + srv := httptest.NewServer(backend.handler()) + t.Cleanup(srv.Close) + + runbookReg, err := runbooks.NewRegistry(log) + require.NoError(t, err) + + if runbookReg.Count() == 0 { + t.Skip("no embedded runbooks available to exercise a real second index build") + } + + r := &Runtime{ + stop: make(chan struct{}), + log: log, + // Zero modules registered means GetQueryExamples returns no examples, so + // ExampleIndex's build never touches the network and always succeeds — + // isolating the induced failure to RunbookIndex. EIP/Specs are left nil + // so reindex skips them. + moduleRegistry: module.NewRegistry(log), + proxyService: &stubEmbeddingProxy{baseURL: srv.URL}, + ExampleIndex: resource.NewRefreshableExampleIndex(nil), + RunbookRegistry: runbookReg, + RunbookIndex: resource.NewRefreshableRunbookIndex(nil), + } + + const ( + modelA = "model-a" + dims = 4 + ) + + // Baseline: a fully successful reindex at modelA. + backend.configure(modelA, dims, false) + r.reindex(modelA, dims, embedding.ProtocolV2) + + require.Equal(t, modelA, r.builtModel) + require.False(t, r.reindexIncomplete) + + _, err = r.RunbookIndex.Search("finality delay", 3) + require.NoError(t, err, "RunbookIndex should be live after a fully successful reindex") + + // Partial failure: same model, but the backend now rejects embed calls. + // ExampleIndex has nothing to embed and "succeeds" trivially; RunbookIndex + // has real content to embed and fails. + backend.configure(modelA, dims, true) + r.reindex(modelA, dims, embedding.ProtocolV2) + + require.True(t, r.reindexIncomplete, "a partial failure must mark the runtime as needing a retry") + + backend.configure(modelA, dims, false) // backend recovers; only the parked state matters now + + _, exampleErr := r.ExampleIndex.Search("anything", 3) + require.Error(t, exampleErr, "ExampleIndex must be re-parked even though it individually succeeded this round") + + _, runbookErr := r.RunbookIndex.Search("finality delay", 3) + require.Error(t, runbookErr, "RunbookIndex must stay parked after its own build failed") + + // The background refresher's guard: the served model is unchanged from + // builtModel, so embeddingSpaceChanged alone would not retry — recovery + // depends on reindexIncomplete also being checked. + require.False(t, embeddingSpaceChanged(r.builtModel, r.builtDims, r.builtProtocol, modelA, dims, embedding.ProtocolV2)) + + // Retry against the same model, as the refresher now does whenever + // reindexIncomplete is set, regardless of embeddingSpaceChanged. + r.reindex(modelA, dims, embedding.ProtocolV2) + + require.False(t, r.reindexIncomplete, "a fully successful retry must clear the incomplete flag") + + _, exampleErr = r.ExampleIndex.Search("anything", 3) + require.NoError(t, exampleErr, "ExampleIndex should be live again after the retry succeeds") + + _, runbookErr = r.RunbookIndex.Search("finality delay", 3) + require.NoError(t, runbookErr, "RunbookIndex should be live again after the retry succeeds") +} diff --git a/pkg/searchruntime/runtime.go b/pkg/searchruntime/runtime.go index 5828c560..77249f07 100644 --- a/pkg/searchruntime/runtime.go +++ b/pkg/searchruntime/runtime.go @@ -47,6 +47,15 @@ type Runtime struct { builtDims int builtProtocol embedding.Protocol + // reindexIncomplete is set when the last reindex attempt left one or more + // indices parked not-ready because their rebuild failed. While set, the + // background refresher retries reindex on every tick regardless of + // whether the served model still differs from builtModel — otherwise a + // served model that reverts to builtModel before a retry succeeds would + // satisfy the model-change guard trivially and the parked indices would + // never be retried. + reindexIncomplete bool + // activated reports whether search has been brought online (indices built // and the background refresher started). activating is a single-flight guard // so concurrent discovery events don't trigger overlapping activation builds. @@ -342,7 +351,9 @@ func (r *Runtime) startRefresh(initialSig uint64) { return case <-ticker.C: model, dims, protocol := resolveModel(context.Background(), r.proxyService) - if model != "" && embeddingSpaceChanged(r.builtModel, r.builtDims, r.builtProtocol, model, dims, protocol) { + spaceChanged := embeddingSpaceChanged(r.builtModel, r.builtDims, r.builtProtocol, model, dims, protocol) + + if model != "" && (spaceChanged || r.reindexIncomplete) { r.reindex(model, dims, protocol) lastSig = exampleSignature(resource.GetQueryExamples(r.moduleRegistry)) @@ -461,11 +472,28 @@ func (r *Runtime) reindex(model string, dims int, protocol embedding.Protocol) { } if !ok { - // Don't advance builtModel — the next tick will re-detect the model - // change and retry. Any index that failed stays not-ready (never mixing - // model spaces) until a retry rebuilds it. + // Don't advance builtModel — the next tick retries. Re-park every index, + // including the ones that rebuilt fine above: builtModel isn't advancing, + // so leaving them live would mean a live index in a space builtModel no + // longer records, exactly the desync a partial failure must not cause. + r.ExampleIndex.Swap(nil) + + if r.RunbookIndex != nil { + r.RunbookIndex.Swap(nil) + } + + if r.EIPIndex != nil { + r.EIPIndex.Swap(nil) + } + + if r.SpecsIndex != nil { + r.SpecsIndex.Swap(nil) + } + + r.reindexIncomplete = true + r.log.WithField("model", model). - Warn("Re-index incomplete; some indices failed to rebuild — will retry on the next tick") + Warn("Re-index incomplete; some indices failed to rebuild — will retry every tick until it fully succeeds") return } @@ -474,6 +502,7 @@ func (r *Runtime) reindex(model string, dims int, protocol embedding.Protocol) { r.builtModel = model r.builtDims = dims r.builtProtocol = protocol + r.reindexIncomplete = false r.log.WithFields(logrus.Fields{"model": model, "dims": dims, "protocol": protocol}).Info("Re-index complete") }