haeaeae/silo-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(search): keep meilisearch active during pending sync

Browse Source
This commit is contained in:
Quick
2026-06-26 10:46:42 -04:00
parent dd2784c0ee
commit a181586139
7 changed files with 144 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/handlers/catalog.go
+12 -9
View File
@@ -65,12 +65,14 @@ type catalogResponse struct {
// /api/v1/catalog only when a relevance-sorted search actually ran through a
// CatalogSearchProvider. mode/semantic_used reflect POST-downgrade reality
// (a hybrid request that fell back to keyword reports mode="keyword",
// semantic_used=false). fallback_reason is omitted when empty.
// semantic_used=false). fallback_reason and index_pending_updates are omitted
// when empty.
type searchDiagnostics struct {
Provider string `json:"provider"`
Mode string `json:"mode"`
SemanticUsed bool `json:"semantic_used"`
FallbackReason string `json:"fallback_reason,omitempty"`
Provider string `json:"provider"`
Mode string `json:"mode"`
SemanticUsed bool `json:"semantic_used"`
FallbackReason string `json:"fallback_reason,omitempty"`
IndexPendingUpdates int `json:"index_pending_updates,omitempty"`
}
type catalogFiltersResponse struct {
@@ -234,10 +236,11 @@ func (h *CatalogHandler) writeCatalogResponse(w http.ResponseWriter, result *cat
var diag *searchDiagnostics
if result.Provider != "" {
diag = &searchDiagnostics{
Provider: result.Provider,
Mode: result.Mode,
SemanticUsed: result.SemanticUsed,
FallbackReason: result.FallbackReason,
Provider: result.Provider,
Mode: result.Mode,
SemanticUsed: result.SemanticUsed,
FallbackReason: result.FallbackReason,
IndexPendingUpdates: result.IndexPendingEvents,
}
}
internal/api/handlers/catalog_diagnostics_test.go
+7 -3
View File
@@ -69,9 +69,10 @@ func TestWriteCatalogResponse_DiagnosticsKeywordFallback(t *testing.T) {
func TestWriteCatalogResponse_DiagnosticsHybridOmitsFallbackReason(t *testing.T) {
body := decodeCatalogResponse(t, &catalog.CatalogResult{
Provider: catalog.SearchProviderMeilisearch,
Mode: "hybrid",
SemanticUsed: true,
Provider: catalog.SearchProviderMeilisearch,
Mode: "hybrid",
SemanticUsed: true,
IndexPendingEvents: 7,
}, false)
diagRaw, ok := body["search_diagnostics"]
@@ -88,6 +89,9 @@ func TestWriteCatalogResponse_DiagnosticsHybridOmitsFallbackReason(t *testing.T)
if _, ok := diag["fallback_reason"]; ok {
t.Fatalf("fallback_reason should be omitted when empty: %v", diag)
}
if diag["index_pending_updates"].(float64) != 7 {
t.Fatalf("index_pending_updates = %v, want 7", diag["index_pending_updates"])
}
}
func TestWriteCatalogResponse_NoProviderOmitsDiagnostics(t *testing.T) {
internal/catalog/catalog_resolver.go
+19 -16
View File
@@ -27,14 +27,16 @@ type CatalogResult struct {
HasMore bool
TotalExact bool
SnapshotAt time.Time // pagination fence timestamp
// Provider, Mode, SemanticUsed and FallbackReason are per-query search
// diagnostics. They are only populated on the direct-search path (where a
// CatalogSearchProvider actually ran); browse / preview / grouped paths
// leave them zero-valued so the handler omits search_diagnostics.
Provider string
Mode string
SemanticUsed bool
FallbackReason string
// Provider, Mode, SemanticUsed, FallbackReason and IndexPendingEvents are
// per-query search diagnostics. They are only populated on the direct-search
// path (where a CatalogSearchProvider actually ran); browse / preview /
// grouped paths leave them zero-valued so the handler omits
// search_diagnostics.
Provider string
Mode string
SemanticUsed bool
FallbackReason string
IndexPendingEvents int
}
type CatalogFiltersResult struct {
@@ -297,14 +299,15 @@ func (r *CatalogResolver) resolveDirectSearchSource(ctx context.Context, req Cat
}
return &CatalogResult{
Items: result.Items,
Total: result.Total,
HasMore: result.HasMore,
TotalExact: result.TotalExact,
Provider: result.Provider,
Mode: result.Mode,
SemanticUsed: result.SemanticUsed,
FallbackReason: result.FallbackReason,
Items: result.Items,
Total: result.Total,
HasMore: result.HasMore,
TotalExact: result.TotalExact,
Provider: result.Provider,
Mode: result.Mode,
SemanticUsed: result.SemanticUsed,
FallbackReason: result.FallbackReason,
IndexPendingEvents: result.IndexPendingEvents,
}, nil
}
internal/catalog/catalog_resolver_test.go
+8 -4
View File
@@ -403,10 +403,11 @@ func TestResolveDirectSearchSource_PlumbsDiagnostics(t *testing.T) {
{
name: "hybrid survived",
result: &CatalogSearchResult{
Items: []*models.MediaItem{},
Provider: SearchProviderMeilisearch,
Mode: "hybrid",
SemanticUsed: true,
Items: []*models.MediaItem{},
Provider: SearchProviderMeilisearch,
Mode: "hybrid",
SemanticUsed: true,
IndexPendingEvents: 7,
},
},
}
@@ -437,6 +438,9 @@ func TestResolveDirectSearchSource_PlumbsDiagnostics(t *testing.T) {
if got.FallbackReason != tc.result.FallbackReason {
t.Fatalf("FallbackReason = %q, want %q", got.FallbackReason, tc.result.FallbackReason)
}
if got.IndexPendingEvents != tc.result.IndexPendingEvents {
t.Fatalf("IndexPendingEvents = %d, want %d", got.IndexPendingEvents, tc.result.IndexPendingEvents)
}
})
}
}
internal/catalog/search_meilisearch_provider.go
+15 -9
View File
@@ -38,6 +38,11 @@ var meilisearchTitleSearchAttributes = []string{
"title_variants",
}
type meilisearchIndexStateStore interface {
GetState(ctx context.Context, provider string) (SearchIndexState, error)
PendingCount(ctx context.Context, provider string) (int, error)
}
type MeilisearchProviderConfig struct {
URL string
APIKey string
@@ -58,7 +63,7 @@ type MeilisearchProviderConfig struct {
type MeilisearchSearchProvider struct {
itemRepo *ItemRepository
stateRepo *SearchIndexEventRepository
stateRepo meilisearchIndexStateStore
fallback *PostgresSearchProvider
client *meilisearchClient
config MeilisearchProviderConfig
@@ -127,9 +132,13 @@ func NewMeilisearchSearchProvider(
if err != nil {
return nil, err
}
var stateStore meilisearchIndexStateStore
if stateRepo != nil {
stateStore = stateRepo
}
return &MeilisearchSearchProvider{
itemRepo: itemRepo,
stateRepo: stateRepo,
stateRepo: stateStore,
fallback: fallback,
client: client,
config: config,
@@ -164,13 +173,9 @@ func (p *MeilisearchSearchProvider) Search(ctx context.Context, req CatalogSearc
if state.SchemaVersion != catalogSearchMeilisearchSchemaVersion(p.config.Embedder, p.config.IndexTypes, p.config.SemanticEnabled) {
return p.fallbackSearch(ctx, req, "meilisearch index schema mismatch")
}
pending, err := p.stateRepo.PendingCount(ctx, SearchProviderMeilisearch)
if err != nil {
p.markFallback("index pending state unavailable")
return p.fallback.Search(ctx, req)
}
if pending > 0 {
return p.fallbackSearch(ctx, req, "meilisearch index has pending updates")
pending := 0
if count, err := p.stateRepo.PendingCount(ctx, SearchProviderMeilisearch); err == nil && count > 0 {
pending = count
}
result, err := p.searchMeilisearch(ctx, req, state.ActiveIndexUID)
@@ -185,6 +190,7 @@ func (p *MeilisearchSearchProvider) Search(ctx context.Context, req CatalogSearc
} else {
p.clearFallback()
}
result.IndexPendingEvents = pending
return result, nil
}
internal/catalog/search_provider.go
+7 -6
View File
@@ -42,7 +42,7 @@ const (
DefaultMeilisearchRebuildBatchSize = 5000
DefaultMeilisearchRebuildQueueDepth = 4
DefaultMeilisearchSemanticEnabled = false
DefaultMeilisearchSemanticRatio = 0.30
DefaultMeilisearchSemanticRatio = 0.50
DefaultMeilisearchEmbedder = "silo_recommendations"
MaxMeilisearchSyncBatchSize = 10000
@@ -62,11 +62,12 @@ type CatalogSearchRequest struct {
}
type CatalogSearchResult struct {
Items []*models.MediaItem
Total int
HasMore bool
TotalExact bool
Provider string
Items []*models.MediaItem
Total int
HasMore bool
TotalExact bool
Provider string
IndexPendingEvents int
// Mode reports which retrieval path actually served this result —
// "keyword" or "hybrid". For Meilisearch it reflects POST-downgrade
// reality (a hybrid request that fell back to keyword reports "keyword").
internal/catalog/search_provider_test.go
+76
View File
@@ -4,9 +4,12 @@ import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"testing"
"time"
)
func TestCatalogSearchSettingsFromMapParsesMeilisearchTuning(t *testing.T) {
@@ -442,6 +445,79 @@ func TestMeilisearchCircuitTripsOnServerAndDecodeErrors(t *testing.T) {
}
}
type fakeMeilisearchIndexStateStore struct {
state SearchIndexState
pending int
}
func (f fakeMeilisearchIndexStateStore) GetState(context.Context, string) (SearchIndexState, error) {
return f.state, nil
}
func (f fakeMeilisearchIndexStateStore) PendingCount(context.Context, string) (int, error) {
return f.pending, nil
}
func TestMeilisearchProviderUsesActiveIndexWhenPendingUpdatesExist(t *testing.T) {
requests := 0
var gotMethod, gotPath string
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requests++
gotMethod = r.Method
gotPath = r.URL.Path
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"hits":[],"estimatedTotalHits":0}`))
}))
defer server.Close()
client, err := newMeilisearchClient(server.URL, "", time.Second)
if err != nil {
t.Fatalf("newMeilisearchClient: %v", err)
}
provider := &MeilisearchSearchProvider{
stateRepo: fakeMeilisearchIndexStateStore{
state: SearchIndexState{
ActiveIndexUID: "search-index",
SchemaVersion: catalogSearchMeilisearchSchemaVersion(DefaultMeilisearchEmbedder, nil, false),
},
pending: 7,
},
fallback: &PostgresSearchProvider{},
client: client,
config: MeilisearchProviderConfig{
BatchSize: meilisearchDefaultBatchSize,
CandidateScanCap: meilisearchDefaultCandidateScanCap,
DeepOffsetLimit: meilisearchDefaultDeepOffsetLimit,
MatchingStrategy: DefaultMeilisearchMatchingStrategy,
Embedder: DefaultMeilisearchEmbedder,
},
}
result, err := provider.Search(context.Background(), CatalogSearchRequest{
Query: "sponge in the sea",
Limit: 10,
})
if err != nil {
t.Fatalf("Search returned error: %v", err)
}
if requests != 1 {
t.Fatalf("Meilisearch requests = %d, want 1", requests)
}
if gotMethod != http.MethodPost || gotPath != "/indexes/search-index/search" {
t.Fatalf("unexpected Meilisearch request %s %s", gotMethod, gotPath)
}
if result.Provider != SearchProviderMeilisearch {
t.Fatalf("provider = %q, want %q", result.Provider, SearchProviderMeilisearch)
}
if result.FallbackReason != "" {
t.Fatalf("fallback reason = %q, want empty", result.FallbackReason)
}
if result.IndexPendingEvents != 7 {
t.Fatalf("IndexPendingEvents = %d, want 7", result.IndexPendingEvents)
}
}
func TestMeilisearchProviderFallsBackWhenScopedIndexCannotSatisfyRequest(t *testing.T) {
provider := &MeilisearchSearchProvider{
config: MeilisearchProviderConfig{