From f40611409d112bd6815c235f758bdce9be9f24d4 Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Thu, 25 Jun 2026 23:06:51 -0400 Subject: [PATCH] feat(search): semantic coverage tracker with hysteresis gate Add the in-memory coverage gate the search hot path will consult to decide whether semantic results may be served for a set of item types. The tracker reads an atomic.Pointer snapshot with no DB and no lock on the read path, refreshes single-flight under a mutex on a 2-minute ticker, and applies per-type hysteresis (enable 0.90 / disable 0.80, holding the previous latch inside the band) to avoid flapping. It collapses to not-ready when the active embedding model changes so a stale latch cannot leak across models, and fails safe: a not-yet- computed snapshot reports not-ready (never panics), and a model-lookup or count-query error retains the last-good snapshot instead of publishing zeros. Refresh counts through the existing catalogSemanticCoverageByType via an injectable fetch seam, so the behavior is unit-testable with canned counts and a fake model provider (no pgx.Rows faking). Reuses the Task-1 coverageQuerier and extends catalogTypeCoverage with Ratio/Ready, filled by the pure computeCoverageSnapshot helper. Adds the SemanticCoverageGate interface beside the other provider interfaces; Task 4 wires the gate and starts the refresher. Co-Authored-By: Claude Opus 4.8 --- internal/catalog/search_provider.go | 10 + internal/catalog/semantic_coverage.go | 253 +++++++++++ internal/catalog/semantic_coverage_test.go | 463 +++++++++++++++++++++ 3 files changed, 726 insertions(+) create mode 100644 internal/catalog/semantic_coverage_test.go diff --git a/internal/catalog/search_provider.go b/internal/catalog/search_provider.go index 65625cad..75a954d1 100644 --- a/internal/catalog/search_provider.go +++ b/internal/catalog/search_provider.go @@ -85,6 +85,16 @@ type CatalogSemanticModelProvider interface { ActiveEmbeddingModel(ctx context.Context) (string, error) } +// SemanticCoverageGate answers, on the search hot path, whether the active +// embedding model covers enough of the requested item types to serve semantic +// results. Implementations read an in-memory snapshot with no database access or +// locking, and fail safe (a not-yet-computed gate reports not-ready rather than +// panicking). The boolean reports readiness; the string is a human-readable +// reason when not ready (empty when ready). +type SemanticCoverageGate interface { + CoverageReady(itemTypes []string) (ready bool, reason string) +} + type CatalogSearchCandidateRetriever interface { CandidateIDs(ctx context.Context, vector []float32, itemTypes []string, limit int) ([]string, error) } diff --git a/internal/catalog/semantic_coverage.go b/internal/catalog/semantic_coverage.go index b2cbec1b..4b32d5eb 100644 --- a/internal/catalog/semantic_coverage.go +++ b/internal/catalog/semantic_coverage.go @@ -3,10 +3,27 @@ package catalog import ( "context" "fmt" + "log/slog" + "sync" + "sync/atomic" + "time" "github.com/jackc/pgx/v5" ) +const ( + // semanticCoverageEnableRatio is the per-type vector-coverage fraction at or + // above which semantic search becomes eligible for that type. + semanticCoverageEnableRatio = 0.90 + // semanticCoverageDisableRatio is the per-type fraction below which semantic + // search is disabled for that type. The gap to the enable ratio is the + // hysteresis band that prevents flapping near the threshold. + semanticCoverageDisableRatio = 0.80 + // semanticCoverageRefreshInterval is how often the background tracker + // recomputes the coverage snapshot. + semanticCoverageRefreshInterval = 2 * time.Minute +) + // coverageQuerier is the read surface the semantic-coverage counts need from a // database handle. *pgxpool.Pool satisfies it, so the existing callers continue // to pass their pool unchanged. A later task layers a cached snapshot tracker on @@ -20,10 +37,18 @@ type coverageQuerier interface { // embed-eligible items exist (Eligible) and how many of those carry a // current-model embedding (Vectorized). By construction Vectorized <= Eligible // for every row. +// +// Ratio and Ready are populated by computeCoverageSnapshot, not by the raw +// catalogSemanticCoverageByType count query (which leaves them zero): Ratio is +// Vectorized/Eligible, and Ready is the hysteresis-latched readiness decision +// for the type. Carrying them on the same struct lets a published snapshot store +// counts and the derived gate state together. type catalogTypeCoverage struct { Type string Eligible int Vectorized int + Ratio float64 + Ready bool } // semanticCoverageEligibleByTypeSQL counts embed-eligible items per media type @@ -126,3 +151,231 @@ func catalogSemanticCoverageByType(ctx context.Context, q coverageQuerier, itemT } return out, nil } + +// semanticCoverageSnapshot is an immutable, point-in-time view of per-type +// vector coverage for a single embedding model. Every field, including the +// PerType map and its values, is built fresh by computeCoverageSnapshot and is +// never mutated after the snapshot is published via atomic.Pointer.Store, so +// concurrent readers on the search hot path observe it without locking. +type semanticCoverageSnapshot struct { + PerType map[string]catalogTypeCoverage + Overall float64 + Model string + UpdatedAt time.Time +} + +// computeCoverageSnapshot derives a fresh snapshot from raw per-type counts. It +// applies hysteresis per type: at or above the enable ratio a type is ready; +// below the disable ratio it is not; inside the band it holds the previous +// snapshot's latch for that type (defaulting to not-ready when no prior latch +// exists). A type with no eligible items is never ready. Overall is the global +// vectorized/eligible fraction across the supplied types. +// +// prev supplies only the previous per-type Ready latches for band entries; +// callers pass nil to start fresh (e.g. immediately after a model collapse) so a +// stale latch can never carry into a new model. +func computeCoverageSnapshot(types []catalogTypeCoverage, model string, prev *semanticCoverageSnapshot, now time.Time) *semanticCoverageSnapshot { + per := make(map[string]catalogTypeCoverage, len(types)) + sumEligible, sumVectorized := 0, 0 + for _, c := range types { + ratio := 0.0 + if c.Eligible > 0 { + ratio = float64(c.Vectorized) / float64(c.Eligible) + } + ready := false + switch { + case c.Eligible == 0: + ready = false // no data to gate on + case ratio >= semanticCoverageEnableRatio: + ready = true + case ratio < semanticCoverageDisableRatio: + ready = false + default: + // Hysteresis band [disable, enable): hold the previous latch. + if prev != nil { + if pc, ok := prev.PerType[c.Type]; ok { + ready = pc.Ready + } + } + } + per[c.Type] = catalogTypeCoverage{ + Type: c.Type, + Eligible: c.Eligible, + Vectorized: c.Vectorized, + Ratio: ratio, + Ready: ready, + } + sumEligible += c.Eligible + sumVectorized += c.Vectorized + } + overall := 0.0 + if sumEligible > 0 { + overall = float64(sumVectorized) / float64(sumEligible) + } + return &semanticCoverageSnapshot{ + PerType: per, + Overall: overall, + Model: model, + UpdatedAt: now, + } +} + +// semanticCoverageTracker maintains the in-memory coverage snapshot consulted by +// the search hot path. Reads (CoverageReady/Snapshot) are lock-free via an +// atomic.Pointer; refreshes are single-flighted under mu and publish a freshly +// built, never-mutated snapshot. The fetch seam decouples Refresh from any +// concrete database handle so it is unit-testable without a real pool. +type semanticCoverageTracker struct { + fetch func(ctx context.Context, model string) ([]catalogTypeCoverage, error) + models CatalogSemanticModelProvider + + mu sync.Mutex + snap atomic.Pointer[semanticCoverageSnapshot] + + clock func() time.Time +} + +// Compile-time assertion that the tracker satisfies the hot-path gate contract. +var _ SemanticCoverageGate = (*semanticCoverageTracker)(nil) + +// newSemanticCoverageTracker wires the tracker to a real query surface. The +// fetch seam closes over q so production code counts coverage through +// catalogSemanticCoverageByType, while tests inject a canned fetch directly. +func newSemanticCoverageTracker(q coverageQuerier, indexTypes []string, models CatalogSemanticModelProvider) *semanticCoverageTracker { + return &semanticCoverageTracker{ + fetch: func(ctx context.Context, model string) ([]catalogTypeCoverage, error) { + return catalogSemanticCoverageByType(ctx, q, indexTypes, model) + }, + models: models, + clock: time.Now, + } +} + +// now reads the configured clock, defaulting to time.Now when unset (e.g. a +// struct literal that omits clock). +func (t *semanticCoverageTracker) now() time.Time { + if t.clock != nil { + return t.clock() + } + return time.Now() +} + +// Refresh recomputes and publishes the coverage snapshot. It is single-flighted +// under mu and fails safe: if the active model cannot be resolved or the count +// query errors, the last-good snapshot is retained (no zeroed snapshot is +// published). When the active model changes, the prior snapshot is collapsed to +// not-ready before recompute so stale per-type latches cannot leak across +// models. +func (t *semanticCoverageTracker) Refresh(ctx context.Context) error { + t.mu.Lock() + defer t.mu.Unlock() + + model := "" + if t.models != nil { + m, err := t.models.ActiveEmbeddingModel(ctx) + if err != nil { + // Retain last-good: do not publish on a provider error. + slog.Warn("catalog semantic coverage: active model lookup failed; retaining last snapshot", "err", err) + return err + } + model = m + } + + if model == "" { + // No active embedding model (no lock or no provider): publish an empty + // not-ready snapshot so the gate reports not-ready deterministically. + t.snap.Store(&semanticCoverageSnapshot{ + PerType: map[string]catalogTypeCoverage{}, + Model: "", + UpdatedAt: t.now(), + }) + return nil + } + + prev := t.snap.Load() + if prev != nil && prev.Model != model { + // Model changed: collapse immediately and drop stale latches so the + // recompute below cannot inherit readiness from the previous model. + t.snap.Store(&semanticCoverageSnapshot{ + PerType: map[string]catalogTypeCoverage{}, + Model: model, + UpdatedAt: t.now(), + }) + prev = nil + } + + types, err := t.fetch(ctx, model) + if err != nil { + // Retain last-good: do not overwrite a healthy snapshot with zeros. + slog.Warn("catalog semantic coverage: count query failed; retaining last snapshot", "err", err) + return err + } + + t.snap.Store(computeCoverageSnapshot(types, model, prev, t.now())) + return nil +} + +// CoverageReady reports whether semantic search may serve the requested item +// types. It is lock-free and fail-safe: a not-yet-computed (nil/empty) snapshot +// reports not-ready. An explicit scope is an AND over its types; the first +// not-ready type's reason is returned. An empty scope requires every snapshot +// type to be ready. Requested types absent from the snapshot (no eligible items) +// are not gated; a scope consisting only of such types reports not-ready. +func (t *semanticCoverageTracker) CoverageReady(itemTypes []string) (bool, string) { + s := t.snap.Load() + if s == nil || len(s.PerType) == 0 { + return false, "coverage not yet computed" + } + + requested := normalizeCatalogSearchItemTypes(itemTypes) + if len(requested) == 0 { + requested = make([]string, 0, len(s.PerType)) + for k := range s.PerType { + requested = append(requested, k) + } + } + + anyPresent := false + for _, ty := range requested { + c, ok := s.PerType[ty] + if !ok { + // No eligible items of this type: nothing to gate. + continue + } + anyPresent = true + if !c.Ready { + return false, fmt.Sprintf("type %q coverage %.0f%% below threshold", ty, c.Ratio*100) + } + } + if !anyPresent { + return false, "no embeddable items in requested scope" + } + return true, "" +} + +// Snapshot returns the current published snapshot, which may be nil before the +// first successful Refresh. Callers must nil-check. +func (t *semanticCoverageTracker) Snapshot() *semanticCoverageSnapshot { + return t.snap.Load() +} + +// Run refreshes once immediately, then on a fixed ticker until ctx is done. +// Refresh errors are logged and swallowed so a transient failure does not stop +// the loop (the last-good snapshot is retained by Refresh). +func (t *semanticCoverageTracker) Run(ctx context.Context) { + if err := t.Refresh(ctx); err != nil { + slog.Warn("catalog semantic coverage: initial refresh failed", "err", err) + } + ticker := time.NewTicker(semanticCoverageRefreshInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := t.Refresh(ctx); err != nil { + slog.Warn("catalog semantic coverage: refresh failed", "err", err) + } + } + } +} diff --git a/internal/catalog/semantic_coverage_test.go b/internal/catalog/semantic_coverage_test.go new file mode 100644 index 00000000..aaa37006 --- /dev/null +++ b/internal/catalog/semantic_coverage_test.go @@ -0,0 +1,463 @@ +package catalog + +import ( + "context" + "errors" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// fakeCoverageModels is an in-memory CatalogSemanticModelProvider. It returns a +// queued sequence of (model, err) results so a test can flip the active model +// between Refresh calls without touching a real recommendations engine. +type fakeCoverageModels struct { + mu sync.Mutex + results []struct { + model string + err error + } + idx int +} + +func (f *fakeCoverageModels) push(model string, err error) { + f.mu.Lock() + defer f.mu.Unlock() + f.results = append(f.results, struct { + model string + err error + }{model, err}) +} + +func (f *fakeCoverageModels) ActiveEmbeddingModel(_ context.Context) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.results) == 0 { + return "", nil + } + if f.idx >= len(f.results) { + // Repeat the last queued result for steady-state callers (e.g. the + // -race loops) so they keep observing a stable model. + last := f.results[len(f.results)-1] + return last.model, last.err + } + r := f.results[f.idx] + f.idx++ + return r.model, r.err +} + +// constCoverageModels always returns the same model with no error. +type constCoverageModels struct{ model string } + +func (c constCoverageModels) ActiveEmbeddingModel(_ context.Context) (string, error) { + return c.model, nil +} + +// fatalFetch is a fetch seam that fails the test if it is ever invoked. It +// proves the read path never reaches the database. +func fatalFetch(t *testing.T) func(context.Context, string) ([]catalogTypeCoverage, error) { + t.Helper() + return func(context.Context, string) ([]catalogTypeCoverage, error) { + t.Fatal("fetch must not be called") + return nil, nil + } +} + +// TestCoverageReadyNilSnapshotIsSafe verifies the fail-safe contract: a +// freshly-constructed tracker that has never refreshed reports not-ready, never +// panics, and never touches the fetch seam (read path is lock/DB free). +func TestCoverageReadyNilSnapshotIsSafe(t *testing.T) { + tr := &semanticCoverageTracker{fetch: fatalFetch(t), clock: time.Now} + + ready, reason := tr.CoverageReady(nil) + if ready { + t.Fatalf("nil-snapshot tracker reported ready") + } + if reason != "coverage not yet computed" { + t.Fatalf("reason = %q, want %q", reason, "coverage not yet computed") + } + if tr.Snapshot() != nil { + t.Fatalf("Snapshot() = %+v, want nil before any refresh", tr.Snapshot()) + } +} + +// TestCoverageRefreshEmptyModelDoesNotFetch verifies that when no embedding +// model is active (no lock / no provider value) Refresh publishes an empty +// not-ready snapshot without ever consulting the fetch seam. +func TestCoverageRefreshEmptyModelDoesNotFetch(t *testing.T) { + models := &fakeCoverageModels{} + models.push("", nil) + tr := &semanticCoverageTracker{fetch: fatalFetch(t), models: models, clock: time.Now} + + if err := tr.Refresh(context.Background()); err != nil { + t.Fatalf("Refresh returned error: %v", err) + } + ready, reason := tr.CoverageReady(nil) + if ready { + t.Fatalf("empty-model tracker reported ready (reason %q)", reason) + } + snap := tr.Snapshot() + if snap == nil { + t.Fatalf("Snapshot() = nil after Refresh, want empty snapshot") + } + if snap.Model != "" { + t.Fatalf("snapshot model = %q, want empty", snap.Model) + } +} + +// TestCoverageRefreshModelErrorRetainsLastGood verifies that a provider error +// is fail-safe: the previous good snapshot is retained, not overwritten with a +// not-ready one. +func TestCoverageRefreshModelErrorRetainsLastGood(t *testing.T) { + models := &fakeCoverageModels{} + models.push("model-a", nil) + models.push("", errors.New("lock unavailable")) + + calls := 0 + tr := &semanticCoverageTracker{ + fetch: func(_ context.Context, model string) ([]catalogTypeCoverage, error) { + calls++ + return []catalogTypeCoverage{{Type: "movie", Eligible: 10, Vectorized: 10}}, nil + }, + models: models, + clock: time.Now, + } + + if err := tr.Refresh(context.Background()); err != nil { + t.Fatalf("first Refresh returned error: %v", err) + } + good := tr.Snapshot() + if good == nil || good.Model != "model-a" { + t.Fatalf("first snapshot = %+v, want model-a", good) + } + + if err := tr.Refresh(context.Background()); err == nil { + t.Fatalf("second Refresh should surface provider error") + } + after := tr.Snapshot() + if after != good { + t.Fatalf("provider error replaced snapshot: before=%p after=%p", good, after) + } + if calls != 1 { + t.Fatalf("fetch called %d times, want 1 (provider error short-circuits)", calls) + } +} + +// TestComputeCoverageSnapshotHysteresis exercises the pure hysteresis helper +// across the enable/disable band: 0.85 (nil prev) -> not ready, 0.92 -> ready, +// 0.85 (prev ready) -> stays ready (latched), 0.79 -> not ready. +func TestComputeCoverageSnapshotHysteresis(t *testing.T) { + now := time.Unix(0, 0) + step := func(prev *semanticCoverageSnapshot, vectorized int) *semanticCoverageSnapshot { + return computeCoverageSnapshot( + []catalogTypeCoverage{{Type: "movie", Eligible: 100, Vectorized: vectorized}}, + "m", prev, now, + ) + } + + s1 := step(nil, 85) // 0.85, band, nil prev -> not ready + if s1.PerType["movie"].Ready { + t.Fatalf("0.85 with nil prev should not be ready") + } + + s2 := step(s1, 92) // 0.92 >= enable -> ready + if !s2.PerType["movie"].Ready { + t.Fatalf("0.92 should be ready") + } + + s3 := step(s2, 85) // 0.85, band, prev ready -> stays ready + if !s3.PerType["movie"].Ready { + t.Fatalf("0.85 with ready prev should latch ready") + } + + s4 := step(s3, 79) // 0.79 < disable -> not ready + if s4.PerType["movie"].Ready { + t.Fatalf("0.79 should drop ready below disable threshold") + } + + // Sanity check the recorded ratio on a band entry. + if got := s3.PerType["movie"].Ratio; got < 0.849 || got > 0.851 { + t.Fatalf("ratio = %v, want ~0.85", got) + } +} + +// TestComputeCoverageSnapshotZeroEligibleNotReady verifies a type with no +// eligible items is never marked ready and contributes nothing to overall. +func TestComputeCoverageSnapshotZeroEligibleNotReady(t *testing.T) { + s := computeCoverageSnapshot( + []catalogTypeCoverage{{Type: "movie", Eligible: 0, Vectorized: 0}}, + "m", nil, time.Unix(0, 0), + ) + c := s.PerType["movie"] + if c.Ready { + t.Fatalf("zero-eligible type should not be ready") + } + if c.Ratio != 0 { + t.Fatalf("zero-eligible ratio = %v, want 0", c.Ratio) + } + if s.Overall != 0 { + t.Fatalf("overall = %v, want 0 when nothing eligible", s.Overall) + } +} + +// TestCoverageReadyScopeAND verifies scope semantics: an explicit scope is an +// AND over requested types, the reason names the first failing type, and an +// empty scope requires every snapshot type to be ready. +func TestCoverageReadyScopeAND(t *testing.T) { + now := time.Unix(0, 0) + snap := computeCoverageSnapshot( + []catalogTypeCoverage{ + {Type: "movie", Eligible: 100, Vectorized: 95}, // ready + {Type: "series", Eligible: 100, Vectorized: 50}, // not ready + }, + "m", nil, now, + ) + tr := &semanticCoverageTracker{fetch: fatalFetch(t), clock: time.Now} + tr.snap.Store(snap) + + if ready, reason := tr.CoverageReady([]string{"movie", "series"}); ready { + t.Fatalf("scope with not-ready series reported ready") + } else if !strings.Contains(reason, "series") { + t.Fatalf("reason = %q, should name the failing type series", reason) + } + + if ready, _ := tr.CoverageReady([]string{"movie"}); !ready { + t.Fatalf("movie-only scope should be ready") + } + + // Empty scope = all snapshot types; series is not ready so overall fails. + if ready, reason := tr.CoverageReady(nil); ready { + t.Fatalf("empty scope should require all types ready, reason=%q", reason) + } +} + +// TestCoverageReadyScopeUnknownTypeNotGated verifies that a requested type with +// no snapshot row (no eligible items) is skipped, and a scope consisting only +// of such types reports the no-embeddable-items reason rather than panicking. +func TestCoverageReadyScopeUnknownTypeNotGated(t *testing.T) { + snap := computeCoverageSnapshot( + []catalogTypeCoverage{{Type: "movie", Eligible: 100, Vectorized: 95}}, + "m", nil, time.Unix(0, 0), + ) + tr := &semanticCoverageTracker{fetch: fatalFetch(t), clock: time.Now} + tr.snap.Store(snap) + + // movie present+ready, "music" absent -> gated only on the present type. + if ready, _ := tr.CoverageReady([]string{"movie", "music"}); !ready { + t.Fatalf("absent type should not gate a ready present type") + } + + // Scope of only-absent types -> nothing to gate -> not ready with reason. + if ready, reason := tr.CoverageReady([]string{"music"}); ready { + t.Fatalf("scope with no embeddable items reported ready") + } else if !strings.Contains(reason, "no embeddable items") { + t.Fatalf("reason = %q, want no-embeddable-items message", reason) + } +} + +// TestCoverageRefreshModelCollapse verifies that when the active embedding model +// changes the next Refresh immediately drops the prior latches: the published +// snapshot carries the new model and is not-ready before recompute uses stale +// hysteresis state. +func TestCoverageRefreshModelCollapse(t *testing.T) { + models := &fakeCoverageModels{} + models.push("model-a", nil) + models.push("model-b", nil) + + // fetch always reports full coverage; only the model identity changes. + tr := &semanticCoverageTracker{ + fetch: func(_ context.Context, model string) ([]catalogTypeCoverage, error) { + return []catalogTypeCoverage{{Type: "movie", Eligible: 100, Vectorized: 100}}, nil + }, + models: models, + clock: time.Now, + } + + if err := tr.Refresh(context.Background()); err != nil { + t.Fatalf("Refresh(A) error: %v", err) + } + a := tr.Snapshot() + if a.Model != "model-a" || !a.PerType["movie"].Ready { + t.Fatalf("after Refresh(A): model=%q ready=%v, want model-a ready", a.Model, a.PerType["movie"].Ready) + } + + // Collapse: the latch must drop the instant the model changes. We assert on + // the recomputed snapshot which now reports model-b. The prev passed into + // compute is the collapsed (empty) snapshot, so the band latch can never + // carry model-a readiness into model-b. + if err := tr.Refresh(context.Background()); err != nil { + t.Fatalf("Refresh(B) error: %v", err) + } + b := tr.Snapshot() + if b.Model != "model-b" { + t.Fatalf("after Refresh(B): model=%q, want model-b", b.Model) + } +} + +// TestCoverageRefreshBandCollapseDropsLatch is the sharper collapse assertion: +// it drives a band ratio (0.85) under the new model so readiness can only come +// from a latch. Because the collapse zeroes prev, the new-model snapshot must be +// not-ready even though the same type was ready under the old model. +func TestCoverageRefreshBandCollapseDropsLatch(t *testing.T) { + models := &fakeCoverageModels{} + models.push("model-a", nil) // full coverage -> ready latch under A + models.push("model-b", nil) // band coverage under B + + var vectorized atomic.Int64 + vectorized.Store(100) + tr := &semanticCoverageTracker{ + fetch: func(_ context.Context, model string) ([]catalogTypeCoverage, error) { + return []catalogTypeCoverage{{ + Type: "movie", Eligible: 100, Vectorized: int(vectorized.Load()), + }}, nil + }, + models: models, + clock: time.Now, + } + + if err := tr.Refresh(context.Background()); err != nil { + t.Fatalf("Refresh(A) error: %v", err) + } + if !tr.Snapshot().PerType["movie"].Ready { + t.Fatalf("movie should be ready under model-a at full coverage") + } + + vectorized.Store(85) // band [0.80,0.90) under model-b + if err := tr.Refresh(context.Background()); err != nil { + t.Fatalf("Refresh(B) error: %v", err) + } + b := tr.Snapshot() + if b.Model != "model-b" { + t.Fatalf("snapshot model = %q, want model-b", b.Model) + } + if b.PerType["movie"].Ready { + t.Fatalf("band coverage under a new model must not inherit the old-model latch") + } +} + +// TestCoverageConcurrentRefreshAndRead is the -race gate: many goroutines call +// Refresh while many call CoverageReady. UpdatedAt must advance monotonically +// (single-flight + immutable publish) and the race detector must stay silent. +func TestCoverageConcurrentRefreshAndRead(t *testing.T) { + // Monotonic clock so successive successful publishes carry strictly + // increasing UpdatedAt regardless of wall-clock resolution. + var ticks atomic.Int64 + clock := func() time.Time { return time.Unix(0, ticks.Add(1)) } + + tr := &semanticCoverageTracker{ + fetch: func(_ context.Context, model string) ([]catalogTypeCoverage, error) { + return []catalogTypeCoverage{ + {Type: "movie", Eligible: 100, Vectorized: 95}, + {Type: "series", Eligible: 100, Vectorized: 85}, + }, nil + }, + models: constCoverageModels{model: "m"}, + clock: clock, + } + + ctx := context.Background() + var wg sync.WaitGroup + + // Writers. + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 200; j++ { + if err := tr.Refresh(ctx); err != nil { + t.Errorf("Refresh error: %v", err) + return + } + } + }() + } + + // Readers. + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + var last time.Time + for j := 0; j < 500; j++ { + _, _ = tr.CoverageReady([]string{"movie"}) + if s := tr.Snapshot(); s != nil { + if s.UpdatedAt.Before(last) { + t.Errorf("UpdatedAt went backwards: %v < %v", s.UpdatedAt, last) + return + } + last = s.UpdatedAt + // Touch the published map to prove it is never mutated + // after Store (would trip the race detector otherwise). + _ = s.PerType["movie"].Ready + } + } + }() + } + + wg.Wait() +} + +// TestCoverageRunStopsOnContextCancel verifies Run performs an immediate refresh +// and returns promptly when the context is canceled (ticker lifecycle). +func TestCoverageRunStopsOnContextCancel(t *testing.T) { + var refreshes atomic.Int64 + tr := &semanticCoverageTracker{ + fetch: func(_ context.Context, model string) ([]catalogTypeCoverage, error) { + refreshes.Add(1) + return []catalogTypeCoverage{{Type: "movie", Eligible: 1, Vectorized: 1}}, nil + }, + models: constCoverageModels{model: "m"}, + clock: time.Now, + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + tr.Run(ctx) + close(done) + }() + + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not return after context cancel") + } + if refreshes.Load() == 0 { + t.Fatalf("Run should perform an immediate refresh before ticking") + } +} + +// TestNewSemanticCoverageTrackerCoverageDBSmoke is the single DB-backed test. It +// proves the default fetch wiring reaches catalogSemanticCoverageByType through +// a real pool. Skipped unless SILO_TEST_DATABASE_URL is set. +func TestNewSemanticCoverageTrackerCoverageDBSmoke(t *testing.T) { + pool := newSemanticCoverageTestPool(t) + const prefix = "covtrack-smoke-" + cleanupSemanticCoverageItems(t, pool, prefix) + + seedSemanticCoverageMediaItem(t, pool, prefix+"m1", "movie", "matched") + seedSemanticCoverageMediaItem(t, pool, prefix+"m2", "movie", "matched") + seedSemanticCoverageEmbedding(t, pool, prefix+"m1", "smoke-model") + + tr := newSemanticCoverageTracker(pool, []string{"movie"}, constCoverageModels{model: "smoke-model"}) + if err := tr.Refresh(context.Background()); err != nil { + t.Fatalf("Refresh against real pool: %v", err) + } + snap := tr.Snapshot() + if snap == nil { + t.Fatalf("Snapshot() = nil after DB-backed refresh") + } + cov := snap.PerType["movie"] + if cov.Eligible < 2 { + t.Fatalf("movie eligible = %d, want >= 2 seeded rows", cov.Eligible) + } + if cov.Vectorized < 1 { + t.Fatalf("movie vectorized = %d, want >= 1 seeded embedding", cov.Vectorized) + } + if cov.Vectorized > cov.Eligible { + t.Fatalf("vectorized %d exceeds eligible %d", cov.Vectorized, cov.Eligible) + } +}