Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0c4e3a2227 |
+41
-59
@@ -4,7 +4,6 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
|
||||||
|
|
||||||
"github.com/gorilla/mux"
|
"github.com/gorilla/mux"
|
||||||
)
|
)
|
||||||
@@ -39,6 +38,36 @@ type MetadataSearchRequest struct {
|
|||||||
Property string `json:"property,omitempty"`
|
Property string `json:"property,omitempty"`
|
||||||
PropertyValue string `json:"propertyValue,omitempty"`
|
PropertyValue string `json:"propertyValue,omitempty"`
|
||||||
Limit int `json:"limit,omitempty"`
|
Limit int `json:"limit,omitempty"`
|
||||||
|
|
||||||
|
// Filters is the structured form: AND across entries, OR within an
|
||||||
|
// entry's Values. Evaluated in memory against the search index alongside
|
||||||
|
// the free-text Query, so an id-anchored lookup costs a map probe per
|
||||||
|
// record rather than a Redis round-trip per record.
|
||||||
|
//
|
||||||
|
// Property/PropertyValue is the single-filter legacy shape and folds into
|
||||||
|
// this list (see normalizedFilters) with Exact=false, preserving its
|
||||||
|
// substring semantics byte-for-byte.
|
||||||
|
Filters []MetadataFilter `json:"filters,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// MetadataFilter is one structured constraint on a record.
|
||||||
|
//
|
||||||
|
// A record satisfies it when ANY of Values matches, and a search result must
|
||||||
|
// satisfy EVERY filter (OR within, AND across). An entry with no usable value
|
||||||
|
// constrains nothing — same rule the gateway's `genres_filter_matches` uses, so
|
||||||
|
// a caller that lowers an empty filter doesn't accidentally empty the result.
|
||||||
|
type MetadataFilter struct {
|
||||||
|
// Property is the field name, WITHOUT a key-set suffix: `genres`, not
|
||||||
|
// `genres/Action`. Both storage shapes are probed — see recordMatchesFilter.
|
||||||
|
Property string `json:"property"`
|
||||||
|
// Values are OR'd. Compared case-insensitively.
|
||||||
|
Values []string `json:"values,omitempty"`
|
||||||
|
// Exact selects CSV-exact comparison (the semantics meta-gateway's
|
||||||
|
// `any_csv_value_matches` applies at its own Layer-B re-filter) over
|
||||||
|
// substring containment. Callers doing identity lookups want this: with a
|
||||||
|
// result limit, a substring match on `tmdbid=1405` also admits `21405` and
|
||||||
|
// can crowd the real matches out of the page.
|
||||||
|
Exact bool `json:"exact,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// MetadataSearchResult is a single search result
|
// MetadataSearchResult is a single search result
|
||||||
@@ -415,65 +444,18 @@ func (s *Server) handleSearchMetadata(w http.ResponseWriter, r *http.Request) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Free-text query: served from the in-memory search index (search_index.go)
|
// Everything else — free text, structured filters, both, or neither
|
||||||
// so we match in memory instead of doing a SCAN+MGET Redis round-trip per
|
// (which returns the first `limit`) — is served from the in-memory search
|
||||||
// record (that was 40s+ over ~72k records). Full metadata is fetched only
|
// index (search_index.go). Matching and the returned metadata both come
|
||||||
// for the <=limit matches.
|
// from memory, so a query costs zero Redis round-trips.
|
||||||
if req.Query != "" && req.Property == "" {
|
//
|
||||||
|
// There is deliberately no second path. The property branch used to fall
|
||||||
|
// through to a `GetAllHashIDs` + per-record `GetMetadataFlat` scan, which
|
||||||
|
// is what made an id-anchored lookup 3.63s over ~20k records (vs 10ms for
|
||||||
|
// the free-text branch on the same corpus) and kept meta-core out of the
|
||||||
|
// gateway's anchored search path entirely. The index already holds every
|
||||||
|
// field of every record, so that scan was re-reading data it had in hand.
|
||||||
s.serveIndexedSearch(w, req)
|
s.serveIndexedSearch(w, req)
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// Property filter (rare) or empty query (returns the first `limit`): the
|
|
||||||
// legacy scan. Empty query only fetches `limit` records; property search is
|
|
||||||
// low-volume and not on the hot search path.
|
|
||||||
allHashIDs, err := s.storage.GetAllHashIDs()
|
|
||||||
if err != nil {
|
|
||||||
writeError(w, http.StatusInternalServerError, err.Error())
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
var results []MetadataSearchResult
|
|
||||||
total := len(allHashIDs)
|
|
||||||
|
|
||||||
for _, hashID := range allHashIDs {
|
|
||||||
if len(results) >= req.Limit {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
|
|
||||||
metadata, err := s.storage.GetMetadataFlat(hashID)
|
|
||||||
if err != nil || metadata == nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
matches := false
|
|
||||||
|
|
||||||
// Property search
|
|
||||||
if req.Property != "" && req.PropertyValue != "" {
|
|
||||||
value, ok := metadata[req.Property]
|
|
||||||
if ok && strings.Contains(strings.ToLower(value), strings.ToLower(req.PropertyValue)) {
|
|
||||||
matches = true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// If no query or property filter, return all
|
|
||||||
if req.Query == "" && req.Property == "" {
|
|
||||||
matches = true
|
|
||||||
}
|
|
||||||
|
|
||||||
if matches {
|
|
||||||
results = append(results, MetadataSearchResult{
|
|
||||||
HashID: hashID,
|
|
||||||
Metadata: metadata,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
writeJSON(w, http.StatusOK, MetadataSearchResponse{
|
|
||||||
Results: results,
|
|
||||||
Count: len(results),
|
|
||||||
Total: total,
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// handleBatchUpdate handles POST /api/metadata/batch
|
// handleBatchUpdate handles POST /api/metadata/batch
|
||||||
|
|||||||
@@ -0,0 +1,211 @@
|
|||||||
|
package api
|
||||||
|
|
||||||
|
import "testing"
|
||||||
|
|
||||||
|
// Tests for the structured-filter matcher that lets meta-core answer an
|
||||||
|
// id-anchored query from the in-memory search index (search_index.go).
|
||||||
|
//
|
||||||
|
// Why this matters beyond "filters work": meta-core is a *coarse pre-filter*
|
||||||
|
// for meta-gateway, which re-evaluates every record it emits through its own
|
||||||
|
// `query_eval::record_matches`. So the property that has to hold is that this
|
||||||
|
// matcher stays a SUPERSET of that one — it may admit records the gateway will
|
||||||
|
// drop, but it must never drop a record the gateway would have kept. Every
|
||||||
|
// case below is written from that angle:
|
||||||
|
//
|
||||||
|
// - key-set fields (`genres/Action`) must match a `genres:action` filter, or
|
||||||
|
// card-tier records vanish from a genre-scoped query;
|
||||||
|
// - a missing field fails CLOSED, matching record_matches;
|
||||||
|
// - an empty/unusable filter constrains NOTHING, so a caller that lowers a
|
||||||
|
// blank filter gets its results rather than an empty page.
|
||||||
|
|
||||||
|
func f(property string, exact bool, values ...string) MetadataFilter {
|
||||||
|
return MetadataFilter{Property: property, Values: values, Exact: exact}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRecordMatchesFilter(t *testing.T) {
|
||||||
|
fields := map[string]string{
|
||||||
|
"fileType": "video",
|
||||||
|
"videoType": "tv,anime",
|
||||||
|
"tmdbid": "1405",
|
||||||
|
"genres/Action": "true",
|
||||||
|
"genres/Science Fiction": "true",
|
||||||
|
"source/gateway:nyaa.si": "true",
|
||||||
|
"categories/XXX": "true",
|
||||||
|
"retracted": "false",
|
||||||
|
}
|
||||||
|
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
filter MetadataFilter
|
||||||
|
want bool
|
||||||
|
}{
|
||||||
|
// Flat scalar.
|
||||||
|
{"flat exact hit", f("fileType", true, "video"), true},
|
||||||
|
{"flat exact miss", f("fileType", true, "audio"), false},
|
||||||
|
{"flat exact is case-insensitive", f("fileType", true, "VIDEO"), true},
|
||||||
|
|
||||||
|
// csv-set field: each comma piece is compared independently.
|
||||||
|
{"csv member hit", f("videoType", true, "anime"), true},
|
||||||
|
{"csv member miss", f("videoType", true, "movie"), false},
|
||||||
|
|
||||||
|
// Exact vs contains — the reason Exact exists. A substring match on an
|
||||||
|
// identity field admits neighbours and, under a result limit, crowds
|
||||||
|
// the real matches off the page.
|
||||||
|
{"contains admits a superstring", f("tmdbid", false, "140"), true},
|
||||||
|
{"exact rejects a superstring", f("tmdbid", true, "140"), false},
|
||||||
|
{"exact accepts the whole value", f("tmdbid", true, "1405"), true},
|
||||||
|
|
||||||
|
// Key-set shape. The stored member is verbatim (`genres/Action`) while
|
||||||
|
// query values arrive lowercased, so the case-insensitive sweep is what
|
||||||
|
// makes a genre-scoped row work at all.
|
||||||
|
{"key-set exact casing", f("genres", true, "Action"), true},
|
||||||
|
{"key-set lowercased query", f("genres", true, "action"), true},
|
||||||
|
{"key-set with a space", f("genres", true, "science fiction"), true},
|
||||||
|
{"key-set miss", f("genres", true, "horror"), false},
|
||||||
|
{"key-set label containing a colon", f("source", true, "gateway:nyaa.si"), true},
|
||||||
|
|
||||||
|
// OR within one filter's values.
|
||||||
|
{"or hits on the second value", f("fileType", true, "audio", "video"), true},
|
||||||
|
{"or misses on all values", f("fileType", true, "audio", "image"), false},
|
||||||
|
|
||||||
|
// Fail-closed on an absent field — mirrors record_matches, and is what
|
||||||
|
// keeps a filtered query from returning unrelated records.
|
||||||
|
{"absent field fails closed", f("imdbid", true, "tt0903747"), false},
|
||||||
|
|
||||||
|
// A filter that constrains nothing must not empty the result.
|
||||||
|
{"no values constrains nothing", f("fileType", true), true},
|
||||||
|
{"blank value constrains nothing", f("fileType", true, " "), true},
|
||||||
|
{"no property constrains nothing", f("", true, "video"), true},
|
||||||
|
|
||||||
|
// A FLAT field whose value happens to be "false" is still a value, and
|
||||||
|
// an exact filter for it matches. Only the key-set arm treats "false"
|
||||||
|
// as non-membership (pinned in the test below).
|
||||||
|
{"flat field valued false still matches", f("retracted", true, "false"), true},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
if got := recordMatchesFilter(fields, &tc.filter); got != tc.want {
|
||||||
|
t.Fatalf("recordMatchesFilter(%+v) = %v, want %v", tc.filter, got, tc.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// `retracted = "false"` is a flat field, not a key-set member, so the flat arm
|
||||||
|
// legitimately matches the literal string "false". The key-set arm is what must
|
||||||
|
// treat it as non-membership — pinned separately so the two arms don't get
|
||||||
|
// conflated if either is refactored.
|
||||||
|
func TestFalseyKeySetMemberIsNotMembership(t *testing.T) {
|
||||||
|
fields := map[string]string{"genres/Action": "false", "genres/Drama": "0"}
|
||||||
|
for _, v := range []string{"Action", "Drama"} {
|
||||||
|
filter := f("genres", true, v)
|
||||||
|
if recordMatchesFilter(fields, &filter) {
|
||||||
|
t.Fatalf("genres/%s with a falsey value must not assert membership", v)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRecordMatchesFiltersIsAndAcrossEntries(t *testing.T) {
|
||||||
|
fields := map[string]string{"fileType": "video", "season": "2"}
|
||||||
|
|
||||||
|
all := []MetadataFilter{f("fileType", true, "video"), f("season", true, "2")}
|
||||||
|
if !recordMatchesFilters(fields, all) {
|
||||||
|
t.Fatal("a record satisfying every filter must match")
|
||||||
|
}
|
||||||
|
|
||||||
|
one := []MetadataFilter{f("fileType", true, "video"), f("season", true, "3")}
|
||||||
|
if recordMatchesFilters(fields, one) {
|
||||||
|
t.Fatal("filters are AND across entries — one miss must reject the record")
|
||||||
|
}
|
||||||
|
|
||||||
|
if !recordMatchesFilters(fields, nil) {
|
||||||
|
t.Fatal("no filters must match everything")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The legacy Property/PropertyValue pair keeps its substring semantics: it is
|
||||||
|
// the shape existing callers already use, and tightening it silently would
|
||||||
|
// change results for them. New callers pass Filters and opt into Exact.
|
||||||
|
func TestNormalizedFiltersFoldsTheLegacyPair(t *testing.T) {
|
||||||
|
got := normalizedFilters(MetadataSearchRequest{Property: "tmdbid", PropertyValue: "1405"})
|
||||||
|
if len(got) != 1 {
|
||||||
|
t.Fatalf("legacy pair must fold into exactly one filter, got %d", len(got))
|
||||||
|
}
|
||||||
|
if got[0].Property != "tmdbid" || len(got[0].Values) != 1 || got[0].Values[0] != "1405" {
|
||||||
|
t.Fatalf("legacy pair folded incorrectly: %+v", got[0])
|
||||||
|
}
|
||||||
|
if got[0].Exact {
|
||||||
|
t.Fatal("the legacy pair must stay substring (Exact=false) for compatibility")
|
||||||
|
}
|
||||||
|
|
||||||
|
// A half-specified pair is not a filter — it must not become one that
|
||||||
|
// matches nothing and empties the response.
|
||||||
|
if n := len(normalizedFilters(MetadataSearchRequest{Property: "tmdbid"})); n != 0 {
|
||||||
|
t.Fatalf("a property with no value must not fold into a filter, got %d", n)
|
||||||
|
}
|
||||||
|
if n := len(normalizedFilters(MetadataSearchRequest{PropertyValue: "1405"})); n != 0 {
|
||||||
|
t.Fatalf("a value with no property must not fold into a filter, got %d", n)
|
||||||
|
}
|
||||||
|
|
||||||
|
both := normalizedFilters(MetadataSearchRequest{
|
||||||
|
Property: "tmdbid",
|
||||||
|
PropertyValue: "1405",
|
||||||
|
Filters: []MetadataFilter{f("fileType", true, "video")},
|
||||||
|
})
|
||||||
|
if len(both) != 2 {
|
||||||
|
t.Fatalf("the legacy pair must be appended to explicit filters, got %d", len(both))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The acceptance bar for routing the anchored path through the index: a
|
||||||
|
// full-miss filtered sweep over a realistic corpus must stay in the same class
|
||||||
|
// as the free-text branch (measured 10 ms over 19 622 live records), not the
|
||||||
|
// legacy per-record Redis scan it replaces (measured 3.63 s over the same
|
||||||
|
// corpus). Full-miss is the worst case — the loop cannot short-circuit on the
|
||||||
|
// result limit, so every record is evaluated.
|
||||||
|
//
|
||||||
|
// Run: go test ./internal/api/ -bench BenchmarkFilterSweep -benchtime 10x
|
||||||
|
func BenchmarkFilterSweep(b *testing.B) {
|
||||||
|
const records = 20000
|
||||||
|
corpus := make([]map[string]string, records)
|
||||||
|
for i := range corpus {
|
||||||
|
corpus[i] = map[string]string{
|
||||||
|
"title": "Some Release",
|
||||||
|
"fileType": "video",
|
||||||
|
"videoType": "tv",
|
||||||
|
"tmdbid": "1405",
|
||||||
|
"season": "2",
|
||||||
|
"genres/Action": "true",
|
||||||
|
"genres/Drama": "true",
|
||||||
|
"source/gateway:nyaa.si": "true",
|
||||||
|
"categories/TV/Anime": "true",
|
||||||
|
"indexer": "nyaa.si",
|
||||||
|
"fileName": "some.release.s02e05.mkv",
|
||||||
|
"description/eng": "text",
|
||||||
|
"poster": "bafy",
|
||||||
|
"sizeByte": "123456789",
|
||||||
|
"contentKind": "episode",
|
||||||
|
"anchoredBy/tmdb:1405": "true",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// A miss on the identity field: the shape an id-anchored gateway query
|
||||||
|
// takes, and the one that must not degrade.
|
||||||
|
filters := []MetadataFilter{
|
||||||
|
f("tmdbid", true, "999999"),
|
||||||
|
f("fileType", true, "video"),
|
||||||
|
f("genres", true, "action"),
|
||||||
|
}
|
||||||
|
b.ResetTimer()
|
||||||
|
for n := 0; n < b.N; n++ {
|
||||||
|
hits := 0
|
||||||
|
for _, fields := range corpus {
|
||||||
|
if recordMatchesFilters(fields, filters) {
|
||||||
|
hits++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if hits != 0 {
|
||||||
|
b.Fatalf("expected a full miss, got %d hits", hits)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+246
-14
@@ -3,6 +3,7 @@ package api
|
|||||||
import (
|
import (
|
||||||
"log"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
@@ -17,19 +18,46 @@ import (
|
|||||||
// batched bulk read (storage.GetSearchIndexTuples), off the request path.
|
// batched bulk read (storage.GetSearchIndexTuples), off the request path.
|
||||||
const searchIndexRefreshInterval = 60 * time.Second
|
const searchIndexRefreshInterval = 60 * time.Second
|
||||||
|
|
||||||
|
// searchIndexReconcileEvery is how many refresh ticks pass between full
|
||||||
|
// rebuilds. Every other tick is incremental. 30 ticks × 60 s = a full
|
||||||
|
// reconcile every 30 minutes — see runRefreshLoop for why one is kept at all.
|
||||||
|
const searchIndexReconcileEvery = 30
|
||||||
|
|
||||||
// searchIndex is an in-memory snapshot of every record's search haystack
|
// searchIndex is an in-memory snapshot of every record's search haystack
|
||||||
// (hashID + lowercased searchable fields), rebuilt periodically from Redis. It
|
// (hashID + lowercased searchable fields), kept current from Redis. It exists
|
||||||
// exists so POST /api/metadata/search matches in memory (string ops over a
|
// so POST /api/metadata/search matches in memory (string ops over a slice)
|
||||||
// slice) instead of doing a SCAN+MGET Redis round-trip per record — the latter
|
// instead of doing a SCAN+MGET Redis round-trip per record — the latter took
|
||||||
// took 40s+ over ~72k records and made a live parallel search path unusable.
|
// 40s+ over ~72k records and made a live parallel search path unusable.
|
||||||
|
//
|
||||||
|
// # Two update paths, and why
|
||||||
|
//
|
||||||
|
// `refresh` rebuilds everything: a whole-keyspace SCAN, batched MGETs, and a
|
||||||
|
// complete fresh copy. That is O(corpus) regardless of how much changed, and it
|
||||||
|
// used to run every 60 s forever. Measured on a 19 622-record staging corpus
|
||||||
|
// (2026-08-30): the loop accounted for ~45 % of ALL Redis CPU, and the Go side
|
||||||
|
// held 783 MB for data Redis stores in 135 MB — largely the double copy.
|
||||||
|
//
|
||||||
|
// `refreshIncremental` re-reads only the records that changed, tracked by
|
||||||
|
// storage's dirty set (search_dirty.go). Cost is O(change). It is the steady
|
||||||
|
// state; the full rebuild runs once at startup (there is nothing to be
|
||||||
|
// incremental against) and then only as an explicit repair.
|
||||||
|
//
|
||||||
|
// The distinction matters because meta-core is the platform's permanent,
|
||||||
|
// additive store — every search hit on every gateway is retained — so a rebuild
|
||||||
|
// priced against the whole corpus is a cost that only ever grows.
|
||||||
type searchIndex struct {
|
type searchIndex struct {
|
||||||
mu sync.RWMutex
|
mu sync.RWMutex
|
||||||
|
// byID is the authoritative content; tuples is a stable-order view of it
|
||||||
|
// rebuilt on each update. Map iteration order in Go is randomised, so
|
||||||
|
// serving a limited query straight off the map would return a different
|
||||||
|
// subset each call — the slice is what keeps results stable.
|
||||||
|
byID map[string]storage.SearchTuple
|
||||||
tuples []storage.SearchTuple
|
tuples []storage.SearchTuple
|
||||||
ready atomic.Bool
|
ready atomic.Bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// snapshot returns the current tuples slice and whether the index has built at
|
// snapshot returns the current tuples slice and whether the index has built at
|
||||||
// least once. The refresh swaps in a fresh slice under the write lock (never
|
// least once. Every update swaps in a fresh slice under the write lock (never
|
||||||
// mutates in place), so the returned slice is safe to read lock-free.
|
// mutates in place), so the returned slice is safe to read lock-free.
|
||||||
func (idx *searchIndex) snapshot() ([]storage.SearchTuple, bool) {
|
func (idx *searchIndex) snapshot() ([]storage.SearchTuple, bool) {
|
||||||
idx.mu.RLock()
|
idx.mu.RLock()
|
||||||
@@ -37,7 +65,21 @@ func (idx *searchIndex) snapshot() ([]storage.SearchTuple, bool) {
|
|||||||
return idx.tuples, idx.ready.Load()
|
return idx.tuples, idx.ready.Load()
|
||||||
}
|
}
|
||||||
|
|
||||||
// refresh rebuilds the index from Redis once (a batched bulk read).
|
// rebuildViewLocked regenerates the stable-order slice from byID. Caller holds
|
||||||
|
// the write lock.
|
||||||
|
func (idx *searchIndex) rebuildViewLocked() {
|
||||||
|
view := make([]storage.SearchTuple, 0, len(idx.byID))
|
||||||
|
for _, t := range idx.byID {
|
||||||
|
view = append(view, t)
|
||||||
|
}
|
||||||
|
// Stable across rebuilds so a limited query returns a consistent page
|
||||||
|
// rather than a fresh random subset every interval.
|
||||||
|
sort.Slice(view, func(i, j int) bool { return view[i].HashID < view[j].HashID })
|
||||||
|
idx.tuples = view
|
||||||
|
}
|
||||||
|
|
||||||
|
// refresh rebuilds the whole index from Redis (a batched bulk read). Used for
|
||||||
|
// the initial build and as an explicit repair — not the steady state.
|
||||||
func (idx *searchIndex) refresh(stor *storage.Client) {
|
func (idx *searchIndex) refresh(stor *storage.Client) {
|
||||||
if stor == nil || !stor.IsConnected() {
|
if stor == nil || !stor.IsConnected() {
|
||||||
return
|
return
|
||||||
@@ -47,27 +89,101 @@ func (idx *searchIndex) refresh(stor *storage.Client) {
|
|||||||
log.Printf("[API] search index refresh failed: %v", err)
|
log.Printf("[API] search index refresh failed: %v", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
byID := make(map[string]storage.SearchTuple, len(tuples))
|
||||||
|
for _, t := range tuples {
|
||||||
|
byID[t.HashID] = t
|
||||||
|
}
|
||||||
idx.mu.Lock()
|
idx.mu.Lock()
|
||||||
idx.tuples = tuples
|
idx.byID = byID
|
||||||
|
idx.rebuildViewLocked()
|
||||||
idx.mu.Unlock()
|
idx.mu.Unlock()
|
||||||
idx.ready.Store(true)
|
idx.ready.Store(true)
|
||||||
}
|
}
|
||||||
|
|
||||||
// runRefreshLoop builds the index immediately, then rebuilds on a ticker for the
|
// refreshIncremental drains the dirty set and patches only those records.
|
||||||
// process lifetime.
|
// Returns whether more dirty entries remain, so a burst is caught up over
|
||||||
|
// consecutive passes rather than one per interval.
|
||||||
|
//
|
||||||
|
// A record in the dirty set but absent from storage was deleted and is evicted
|
||||||
|
// here — otherwise it would keep answering searches until the next full
|
||||||
|
// rebuild. A batch that fails after the (destructive) drain is re-marked by
|
||||||
|
// storage so it is retried rather than silently lost.
|
||||||
|
func (idx *searchIndex) refreshIncremental(stor *storage.Client) (more bool) {
|
||||||
|
if stor == nil || !stor.IsConnected() {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
ids, more, err := stor.DrainDirty()
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("[API] search index drain failed: %v", err)
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if len(ids) == 0 {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
tuples, gone, err := stor.GetSearchIndexTuplesFor(ids)
|
||||||
|
if err != nil {
|
||||||
|
stor.MarkDirty(ids...)
|
||||||
|
log.Printf("[API] incremental search index refresh failed: %v", err)
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
idx.mu.Lock()
|
||||||
|
if idx.byID == nil {
|
||||||
|
idx.byID = make(map[string]storage.SearchTuple, len(tuples))
|
||||||
|
}
|
||||||
|
for _, t := range tuples {
|
||||||
|
idx.byID[t.HashID] = t
|
||||||
|
}
|
||||||
|
for _, id := range gone {
|
||||||
|
delete(idx.byID, id)
|
||||||
|
}
|
||||||
|
idx.rebuildViewLocked()
|
||||||
|
idx.mu.Unlock()
|
||||||
|
|
||||||
|
log.Printf("[API] search index: %d updated, %d evicted (%d records)", len(tuples), len(gone), len(idx.tuples))
|
||||||
|
return more
|
||||||
|
}
|
||||||
|
|
||||||
|
// runRefreshLoop builds the index once in full, then keeps it current from the
|
||||||
|
// dirty set, with a periodic full reconcile as a safety net.
|
||||||
|
//
|
||||||
|
// The reconcile exists because a *missed* dirty mark is silent: no error, just
|
||||||
|
// a record whose indexed copy quietly diverges from storage until something
|
||||||
|
// forces a rebuild. Marking is spread across eight write paths plus anything
|
||||||
|
// added later, and the cost of getting it wrong is invisible — so the index
|
||||||
|
// should not depend on that coverage being perfect forever. Every
|
||||||
|
// searchIndexReconcileEvery-th tick pays the old O(corpus) pass and repairs
|
||||||
|
// whatever drifted.
|
||||||
|
//
|
||||||
|
// That still drops the full rebuild from every 60 s to every 30 min — a ~30×
|
||||||
|
// reduction in the steady-state cost that measured at ~45 % of all Redis CPU —
|
||||||
|
// while keeping the self-healing property that made the naive loop safe.
|
||||||
func (idx *searchIndex) runRefreshLoop(stor *storage.Client) {
|
func (idx *searchIndex) runRefreshLoop(stor *storage.Client) {
|
||||||
idx.refresh(stor)
|
idx.refresh(stor)
|
||||||
ticker := time.NewTicker(searchIndexRefreshInterval)
|
ticker := time.NewTicker(searchIndexRefreshInterval)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
ticks := 0
|
||||||
for range ticker.C {
|
for range ticker.C {
|
||||||
|
ticks++
|
||||||
|
if ticks%searchIndexReconcileEvery == 0 {
|
||||||
idx.refresh(stor)
|
idx.refresh(stor)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// Drain until caught up, so a burst larger than one batch does not take
|
||||||
|
// one interval per batch to land.
|
||||||
|
for idx.refreshIncremental(stor) {
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// serveIndexedSearch answers a free-text (non-property) query from the in-memory
|
// serveIndexedSearch answers a search from the in-memory index: a token-AND
|
||||||
// index: token-AND match against each record's haystack (or a whole-query hashID
|
// match against each record's haystack (or a whole-query hashID substring,
|
||||||
// substring, preserving the legacy behaviour), then fetch full metadata for only
|
// preserving the legacy behaviour) AND every structured filter, then returns
|
||||||
// the <=limit matches. No Redis round-trip per scanned record.
|
// the full metadata for the <=limit survivors. No Redis round-trip per scanned
|
||||||
|
// record — matching and results both come from memory.
|
||||||
|
//
|
||||||
|
// This is the ONLY search path. Free text, filters, both, or neither all land
|
||||||
|
// here; only the `hashId` exact-match branch in handleSearchMetadata bypasses it.
|
||||||
func (s *Server) serveIndexedSearch(w http.ResponseWriter, req MetadataSearchRequest) {
|
func (s *Server) serveIndexedSearch(w http.ResponseWriter, req MetadataSearchRequest) {
|
||||||
tuples, ready := s.search.snapshot()
|
tuples, ready := s.search.snapshot()
|
||||||
if !ready {
|
if !ready {
|
||||||
@@ -78,15 +194,20 @@ func (s *Server) serveIndexedSearch(w http.ResponseWriter, req MetadataSearchReq
|
|||||||
}
|
}
|
||||||
tokens := strings.Fields(strings.ToLower(req.Query))
|
tokens := strings.Fields(strings.ToLower(req.Query))
|
||||||
queryLower := strings.ToLower(req.Query)
|
queryLower := strings.ToLower(req.Query)
|
||||||
|
filters := normalizedFilters(req)
|
||||||
results := make([]MetadataSearchResult, 0, req.Limit)
|
results := make([]MetadataSearchResult, 0, req.Limit)
|
||||||
for _, t := range tuples {
|
for _, t := range tuples {
|
||||||
if len(results) >= req.Limit {
|
if len(results) >= req.Limit {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
if !matchAllTokens(t.Haystack, tokens) &&
|
if req.Query != "" &&
|
||||||
|
!matchAllTokens(t.Haystack, tokens) &&
|
||||||
!strings.Contains(strings.ToLower(t.HashID), queryLower) {
|
!strings.Contains(strings.ToLower(t.HashID), queryLower) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
if !recordMatchesFilters(t.Fields, filters) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
// Full metadata is cached in the index — return it straight from memory,
|
// Full metadata is cached in the index — return it straight from memory,
|
||||||
// no Redis round-trip per result.
|
// no Redis round-trip per result.
|
||||||
results = append(results, MetadataSearchResult{HashID: t.HashID, Metadata: t.Fields})
|
results = append(results, MetadataSearchResult{HashID: t.HashID, Metadata: t.Fields})
|
||||||
@@ -108,3 +229,114 @@ func matchAllTokens(haystack string, tokens []string) bool {
|
|||||||
}
|
}
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// normalizedFilters returns the request's structured filters with the legacy
|
||||||
|
// Property/PropertyValue pair folded in as one more entry.
|
||||||
|
//
|
||||||
|
// The legacy pair keeps Exact=false so its long-standing substring semantics
|
||||||
|
// are unchanged for existing callers; new callers pass Filters and choose.
|
||||||
|
func normalizedFilters(req MetadataSearchRequest) []MetadataFilter {
|
||||||
|
if req.Property == "" || req.PropertyValue == "" {
|
||||||
|
return req.Filters
|
||||||
|
}
|
||||||
|
out := make([]MetadataFilter, 0, len(req.Filters)+1)
|
||||||
|
out = append(out, req.Filters...)
|
||||||
|
return append(out, MetadataFilter{
|
||||||
|
Property: req.Property,
|
||||||
|
Values: []string{req.PropertyValue},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// recordMatchesFilters reports whether fields satisfies every filter (AND).
|
||||||
|
func recordMatchesFilters(fields map[string]string, filters []MetadataFilter) bool {
|
||||||
|
for i := range filters {
|
||||||
|
if !recordMatchesFilter(fields, &filters[i]) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
// recordMatchesFilter reports whether fields satisfies one filter (OR within
|
||||||
|
// its values), probing BOTH storage shapes a field can take:
|
||||||
|
//
|
||||||
|
// - a flat scalar/csv-set field — `fileType = "video"`, `videoType = "tv,anime"`;
|
||||||
|
// - a key-set member — `genres/Action = "true"`, `source/gateway:nyaa.si = "true"`
|
||||||
|
// (METADATA_KEYS "How the store is shaped").
|
||||||
|
//
|
||||||
|
// Probing both is what lets one filter shape serve the whole vocabulary. The
|
||||||
|
// gateway needs a `genres:action` filter to find a record whose genre lives in
|
||||||
|
// the key `genres/Action`, and a `fileType:video` filter to find one whose value
|
||||||
|
// lives in the field `fileType` — without meta-core knowing which fields are
|
||||||
|
// key-sets. It also means this stays a SUPERSET of the gateway's own
|
||||||
|
// `record_matches`, which is the contract: meta-core pre-filters, the gateway
|
||||||
|
// re-filters authoritatively.
|
||||||
|
//
|
||||||
|
// A filter with no usable value constrains nothing (matches). A filter whose
|
||||||
|
// field is absent fails closed, matching `record_matches`.
|
||||||
|
func recordMatchesFilter(fields map[string]string, f *MetadataFilter) bool {
|
||||||
|
values := make([]string, 0, len(f.Values))
|
||||||
|
for _, v := range f.Values {
|
||||||
|
if v = strings.TrimSpace(v); v != "" {
|
||||||
|
values = append(values, v)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if f.Property == "" || len(values) == 0 {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
// Key-set shape: `<property>/<value>` present and truthy. Try the exact
|
||||||
|
// key first — that hits whenever the caller's casing matches what was
|
||||||
|
// written, which is the common case and costs one map probe.
|
||||||
|
for _, v := range values {
|
||||||
|
if isTruthyMember(fields[f.Property+"/"+v]) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Members are written verbatim (`genres/Action`) while query values arrive
|
||||||
|
// lowercased, so fall back to one case-insensitive sweep. Single pass over
|
||||||
|
// the record's fields for ALL values, not one pass per value.
|
||||||
|
wanted := make(map[string]struct{}, len(values))
|
||||||
|
for _, v := range values {
|
||||||
|
wanted[strings.ToLower(v)] = struct{}{}
|
||||||
|
}
|
||||||
|
prefix := strings.ToLower(f.Property) + "/"
|
||||||
|
for k, val := range fields {
|
||||||
|
if len(k) <= len(prefix) || !strings.EqualFold(k[:len(prefix)], prefix) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, ok := wanted[strings.ToLower(k[len(prefix):])]; ok && isTruthyMember(val) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Flat shape: the field's own value, split on `,` for csv-set fields.
|
||||||
|
fieldValue, ok := fields[f.Property]
|
||||||
|
if !ok {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
for _, piece := range strings.Split(fieldValue, ",") {
|
||||||
|
piece = strings.TrimSpace(piece)
|
||||||
|
if piece == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
for _, v := range values {
|
||||||
|
if f.Exact {
|
||||||
|
if strings.EqualFold(piece, v) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
} else if strings.Contains(strings.ToLower(piece), strings.ToLower(v)) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
// isTruthyMember reports whether a key-set member's value asserts membership.
|
||||||
|
// Mirrors the gateway's `is_truthy_member`: presence with a non-empty,
|
||||||
|
// non-"false"/"0" value.
|
||||||
|
func isTruthyMember(v string) bool {
|
||||||
|
v = strings.TrimSpace(strings.ToLower(v))
|
||||||
|
return v != "" && v != "false" && v != "0"
|
||||||
|
}
|
||||||
|
|||||||
@@ -309,6 +309,10 @@ func (c *Client) DeleteRoot(uuid string) error {
|
|||||||
if err := c.client.SRem(ctx, c.buildIndexKey(), uuid).Err(); err != nil {
|
if err := c.client.SRem(ctx, c.buildIndexKey(), uuid).Err(); err != nil {
|
||||||
return fmt.Errorf("srem index: %w", err)
|
return fmt.Errorf("srem index: %w", err)
|
||||||
}
|
}
|
||||||
|
// Marked so the incremental search-index refresh reads it, finds it gone,
|
||||||
|
// and evicts it — otherwise a deleted root keeps answering searches until
|
||||||
|
// the next full rebuild.
|
||||||
|
c.markDirtyLocked(ctx, uuid)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
|
"os"
|
||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -173,6 +174,7 @@ func (c *Client) SetMetadataFlat(hashID string, metadata map[string]string) erro
|
|||||||
if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
||||||
return fmt.Errorf("sadd index failed: %w", err)
|
return fmt.Errorf("sadd index failed: %w", err)
|
||||||
}
|
}
|
||||||
|
c.markDirtyLocked(ctx, hashID)
|
||||||
|
|
||||||
// Auto-register reverse-index aliases for any cids/<cid> key-set members in
|
// Auto-register reverse-index aliases for any cids/<cid> key-set members in
|
||||||
// the payload. That is the ONLY shape recognised — the legacy flat cid_* /
|
// the payload. That is the ONLY shape recognised — the legacy flat cid_* /
|
||||||
@@ -257,6 +259,7 @@ func (c *Client) SetProperty(hashID, property, value string) error {
|
|||||||
if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
||||||
return fmt.Errorf("sadd index failed: %w", err)
|
return fmt.Errorf("sadd index failed: %w", err)
|
||||||
}
|
}
|
||||||
|
c.markDirtyLocked(ctx, hashID)
|
||||||
|
|
||||||
// Auto-register reverse-index alias if this is a CID field.
|
// Auto-register reverse-index alias if this is a CID field.
|
||||||
_ = c.maybeAddAliasFromFieldLocked(ctx, hashID, property, value)
|
_ = c.maybeAddAliasFromFieldLocked(ctx, hashID, property, value)
|
||||||
@@ -325,6 +328,10 @@ func (c *Client) DeleteMetadata(hashID string) (int64, error) {
|
|||||||
if err := c.client.SRem(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
if err := c.client.SRem(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
||||||
return deletedCount, fmt.Errorf("srem index failed: %w", err)
|
return deletedCount, fmt.Errorf("srem index failed: %w", err)
|
||||||
}
|
}
|
||||||
|
// Marked, not unmarked: the incremental refresh reads the record, finds it
|
||||||
|
// gone, and evicts it from the search index. Skipping this would leave a
|
||||||
|
// deleted record answering searches until the next full rebuild.
|
||||||
|
c.markDirtyLocked(ctx, hashID)
|
||||||
|
|
||||||
return deletedCount, nil
|
return deletedCount, nil
|
||||||
}
|
}
|
||||||
@@ -441,6 +448,7 @@ func (c *Client) MergeMetadataFlat(hashID string, metadata map[string]string) (i
|
|||||||
if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
||||||
return len(metadata), fmt.Errorf("sadd index failed: %w", err)
|
return len(metadata), fmt.Errorf("sadd index failed: %w", err)
|
||||||
}
|
}
|
||||||
|
c.markDirtyLocked(ctx, hashID)
|
||||||
|
|
||||||
// Auto-register reverse-index aliases for any cids/<cid> key-set members in
|
// Auto-register reverse-index aliases for any cids/<cid> key-set members in
|
||||||
// the payload. That is the ONLY shape recognised — the legacy flat cid_* /
|
// the payload. That is the ONLY shape recognised — the legacy flat cid_* /
|
||||||
@@ -468,6 +476,7 @@ func (c *Client) DeleteProperty(hashID, property string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
c.removeFieldLocked(ctx, hashID, property)
|
c.removeFieldLocked(ctx, hashID, property)
|
||||||
|
c.markDirtyLocked(ctx, hashID)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -521,6 +530,7 @@ func (c *Client) AddToSet(hashID, property, value string) (bool, error) {
|
|||||||
if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
if err := c.client.SAdd(ctx, c.buildIndexKey(), hashID).Err(); err != nil {
|
||||||
return true, fmt.Errorf("sadd index failed: %w", err)
|
return true, fmt.Errorf("sadd index failed: %w", err)
|
||||||
}
|
}
|
||||||
|
c.markDirtyLocked(ctx, hashID)
|
||||||
|
|
||||||
return true, nil
|
return true, nil
|
||||||
}
|
}
|
||||||
@@ -574,11 +584,16 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) {
|
|||||||
}
|
}
|
||||||
// The key is gone, so the field name must leave the index too.
|
// The key is gone, so the field name must leave the index too.
|
||||||
c.removeFieldLocked(ctx, hashID, property)
|
c.removeFieldLocked(ctx, hashID, property)
|
||||||
|
c.markDirtyLocked(ctx, hashID)
|
||||||
return true, nil
|
return true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
newValue := strings.Join(newValues, "|")
|
newValue := strings.Join(newValues, "|")
|
||||||
return true, c.client.Set(ctx, key, newValue, 0).Err()
|
err = c.client.Set(ctx, key, newValue, 0).Err()
|
||||||
|
// The field survives with fewer members, so no field-index change — but the
|
||||||
|
// VALUE changed, which is exactly what the search index caches.
|
||||||
|
c.markDirtyLocked(ctx, hashID)
|
||||||
|
return true, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// FileTuple holds the (uuid, path, size, mtime, midhash) record for one
|
// FileTuple holds the (uuid, path, size, mtime, midhash) record for one
|
||||||
@@ -691,6 +706,44 @@ type SearchTuple struct {
|
|||||||
// haystack — must stay in sync with the fields the search handler documents.
|
// haystack — must stay in sync with the fields the search handler documents.
|
||||||
var searchIndexFields = []string{"title", "fileName", "originaltitle", "showtitle", "filePath"}
|
var searchIndexFields = []string{"title", "fileName", "originaltitle", "showtitle", "filePath"}
|
||||||
|
|
||||||
|
// defaultIndexExcludePrefixes are field-name prefixes kept OUT of the in-memory
|
||||||
|
// search index. Redis still stores them and every per-record read still returns
|
||||||
|
// them — this narrows the *index*, never the store.
|
||||||
|
//
|
||||||
|
// `categories/newznab/<code>` is the raw Newznab numeric category code, which
|
||||||
|
// METADATA_KEYS documents as a mirror of the canonical `categories/<label>`
|
||||||
|
// vocabulary that rides alongside it. Nothing in meta-gateway, meta-search,
|
||||||
|
// meta-watch or meta-core reads it, and it is the single largest contributor to
|
||||||
|
// the index: measured on a live 19.6k-record corpus, `categories/*` was 40% of
|
||||||
|
// the whole keyspace and the `newznab` sub-vocabulary was 59% of that — ~24% of
|
||||||
|
// all keys, for zero readers.
|
||||||
|
//
|
||||||
|
// ⚠ Excluding the WHOLE `categories/` family instead would be a security bug,
|
||||||
|
// not an optimisation. `categories/XXX` (and `categories/XXX/<leaf>`) is the
|
||||||
|
// server-side adult-content enforcement point — meta-search's
|
||||||
|
// `query_eval::fields_are_adult` and meta-gateway's `cards::fields_are_adult`
|
||||||
|
// both match on it, and the `exclude_adult` preference defaults to ON. A record
|
||||||
|
// served from the index without its categories would read as non-adult and the
|
||||||
|
// filter would fail OPEN. Only prefixes with no semantic readers belong here.
|
||||||
|
var defaultIndexExcludePrefixes = []string{"categories/newznab/"}
|
||||||
|
|
||||||
|
// indexExcludePrefixes resolves the exclusion list, allowing an operator to
|
||||||
|
// override it with a comma-separated META_CORE_INDEX_EXCLUDE_PREFIXES. An
|
||||||
|
// explicitly empty value disables exclusion entirely (index everything).
|
||||||
|
func indexExcludePrefixes() []string {
|
||||||
|
raw, set := os.LookupEnv("META_CORE_INDEX_EXCLUDE_PREFIXES")
|
||||||
|
if !set {
|
||||||
|
return defaultIndexExcludePrefixes
|
||||||
|
}
|
||||||
|
out := make([]string, 0, 4)
|
||||||
|
for _, p := range strings.Split(raw, ",") {
|
||||||
|
if p = strings.TrimSpace(p); p != "" {
|
||||||
|
out = append(out, p)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
// GetSearchIndexTuples reads EVERY record's full metadata in a single keyspace
|
// GetSearchIndexTuples reads EVERY record's full metadata in a single keyspace
|
||||||
// SCAN pass over the flat `file:{hashID}/{field}` keys plus batched MGETs,
|
// SCAN pass over the flat `file:{hashID}/{field}` keys plus batched MGETs,
|
||||||
// grouped by record — one bulk pass instead of a SCAN+MGET per record (which is
|
// grouped by record — one bulk pass instead of a SCAN+MGET per record (which is
|
||||||
@@ -712,6 +765,7 @@ func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) {
|
|||||||
indexKey := c.buildIndexKey() // {filePrefix}__index__ (a SET, skip it)
|
indexKey := c.buildIndexKey() // {filePrefix}__index__ (a SET, skip it)
|
||||||
pattern := filePrefix + "*"
|
pattern := filePrefix + "*"
|
||||||
|
|
||||||
|
excluded := indexExcludePrefixes()
|
||||||
records := make(map[string]map[string]string)
|
records := make(map[string]map[string]string)
|
||||||
keyBatch := make([]string, 0, 1000)
|
keyBatch := make([]string, 0, 1000)
|
||||||
|
|
||||||
@@ -762,6 +816,11 @@ func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) {
|
|||||||
if strings.HasSuffix(k, "/"+FieldsField) {
|
if strings.HasSuffix(k, "/"+FieldsField) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
// Index-excluded fields are skipped here rather than after the
|
||||||
|
// MGET, so an excluded key costs neither the fetch nor the memory.
|
||||||
|
if _, field, ok := c.splitRecordKey(k); ok && hasAnyPrefix(field, excluded) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
keyBatch = append(keyBatch, k)
|
keyBatch = append(keyBatch, k)
|
||||||
if len(keyBatch) >= 1000 {
|
if len(keyBatch) >= 1000 {
|
||||||
if err := flush(); err != nil {
|
if err := flush(); err != nil {
|
||||||
@@ -780,18 +839,23 @@ func (c *Client) GetSearchIndexTuples() ([]SearchTuple, error) {
|
|||||||
|
|
||||||
out := make([]SearchTuple, 0, len(records))
|
out := make([]SearchTuple, 0, len(records))
|
||||||
for hashID, fields := range records {
|
for hashID, fields := range records {
|
||||||
var sb strings.Builder
|
// Shared with the incremental path (search_dirty.go) so the two can
|
||||||
for _, f := range searchIndexFields {
|
// never disagree about what is searchable.
|
||||||
if v, ok := fields[f]; ok && v != "" {
|
out = append(out, SearchTuple{HashID: hashID, Haystack: buildHaystack(fields), Fields: fields})
|
||||||
sb.WriteString(strings.ToLower(v))
|
|
||||||
sb.WriteByte('\n')
|
|
||||||
}
|
|
||||||
}
|
|
||||||
out = append(out, SearchTuple{HashID: hashID, Haystack: sb.String(), Fields: fields})
|
|
||||||
}
|
}
|
||||||
return out, nil
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// hasAnyPrefix reports whether s starts with any of the given prefixes.
|
||||||
|
func hasAnyPrefix(s string, prefixes []string) bool {
|
||||||
|
for _, p := range prefixes {
|
||||||
|
if strings.HasPrefix(s, p) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
// GetMemoryInfo returns Redis memory usage information
|
// GetMemoryInfo returns Redis memory usage information
|
||||||
func (c *Client) GetMemoryInfo() (string, error) {
|
func (c *Client) GetMemoryInfo() (string, error) {
|
||||||
c.mu.RLock()
|
c.mu.RLock()
|
||||||
|
|||||||
@@ -0,0 +1,214 @@
|
|||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"log"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Incremental search-index maintenance.
|
||||||
|
//
|
||||||
|
// # Why this exists
|
||||||
|
//
|
||||||
|
// The API's in-memory search index used to be rebuilt in full every 60 s:
|
||||||
|
// GetSearchIndexTuples does a whole-keyspace SCAN plus batched MGETs, builds an
|
||||||
|
// intermediate map of every record, then allocates a fresh []SearchTuple — all
|
||||||
|
// while the previous index is still live serving reads. That cost is O(corpus),
|
||||||
|
// not O(change): adding 200 records in a minute re-read every one of the other
|
||||||
|
// ~620 000 field keys to notice them.
|
||||||
|
//
|
||||||
|
// Measured on a 19 622-record staging corpus (2026-08-30): SCAN was 57.8 % and
|
||||||
|
// MGET 19.1 % of ALL Redis CPU, and ~90 % of the MGET calls plus ~48 % of the
|
||||||
|
// SCAN calls were this loop — roughly 45 % of Redis's total work, spent
|
||||||
|
// continuously whether or not anything changed. The Go side held 783 MB for
|
||||||
|
// data Redis stores in 135 MB, largely because of the double copy.
|
||||||
|
//
|
||||||
|
// That was tolerable while the corpus was small and mostly static. It stops
|
||||||
|
// being tolerable now that meta-core is the platform's permanent, additive
|
||||||
|
// store: every search hit on every gateway is retained forever, so the rebuild
|
||||||
|
// cost is a ratchet.
|
||||||
|
//
|
||||||
|
// # The shape
|
||||||
|
//
|
||||||
|
// Every write path marks its record in a dirty set; the refresh drains that set
|
||||||
|
// and re-reads only those records. Cost becomes O(change).
|
||||||
|
//
|
||||||
|
// Marking rides along with the field-index maintenance every write path already
|
||||||
|
// performs, so there is no new bookkeeping discipline to remember — if you are
|
||||||
|
// adding a write path, the same line that keeps `__fields__` correct keeps this
|
||||||
|
// correct.
|
||||||
|
//
|
||||||
|
// # Correctness
|
||||||
|
//
|
||||||
|
// The drain is a destructive SPOP, so a write that lands *during* a refresh is
|
||||||
|
// either already in the drained batch or still in the set for the next tick —
|
||||||
|
// never dropped. The cost of the race is one tick of staleness, which is well
|
||||||
|
// inside the interval the index was already stale by.
|
||||||
|
//
|
||||||
|
// A refresh that fails after draining WOULD lose its batch, so drainDirty's
|
||||||
|
// caller re-marks on error (see MarkDirty / the API's refreshIncremental).
|
||||||
|
//
|
||||||
|
// ⚠ **The destructive drain assumes ONE index consumer per Redis.** Two
|
||||||
|
// meta-core processes sharing a keyspace would steal each other's marks, and
|
||||||
|
// each would end up with an index missing whatever the other drained. That is
|
||||||
|
// fine today — multi-instance leader election was deprecated and the dev/prod
|
||||||
|
// stacks run a single meta-core — but it is the assumption to revisit first if
|
||||||
|
// a second instance is ever pointed at the same Redis. A per-consumer dirty key
|
||||||
|
// (`file:__dirty__:<consumer>`) is the shape that would fix it.
|
||||||
|
//
|
||||||
|
// The API's periodic full reconcile (searchIndexReconcileEvery) is the backstop
|
||||||
|
// for both this and for a write path that forgets to mark: a missed mark costs
|
||||||
|
// staleness until the next reconcile, never permanent divergence.
|
||||||
|
|
||||||
|
// dirtyField is the reserved key holding the set of record ids whose metadata
|
||||||
|
// changed since the last search-index refresh. It sits under the same `file:`
|
||||||
|
// namespace as `__index__` and is likewise not a record.
|
||||||
|
const dirtyField = "file:__dirty__"
|
||||||
|
|
||||||
|
// dirtyDrainCap bounds one incremental refresh. A burst larger than this is
|
||||||
|
// drained over several ticks rather than in one long-held pass — the point of
|
||||||
|
// the change is to keep any single refresh cheap. Sized well above a normal
|
||||||
|
// search-persist flush (which writes at most a few hundred records).
|
||||||
|
const dirtyDrainCap = 5000
|
||||||
|
|
||||||
|
// buildDirtyKey constructs the key for the dirty-record set.
|
||||||
|
func (c *Client) buildDirtyKey() string {
|
||||||
|
return c.buildKey(dirtyField)
|
||||||
|
}
|
||||||
|
|
||||||
|
// markDirtyLocked records that hashID's metadata changed. Caller must hold c.mu.
|
||||||
|
//
|
||||||
|
// Best-effort by design: a failure here costs one record its incremental
|
||||||
|
// refresh, and the periodic full rebuild still repairs it. Failing the caller's
|
||||||
|
// write instead would trade a stale index entry for a lost write, which is
|
||||||
|
// strictly worse.
|
||||||
|
func (c *Client) markDirtyLocked(ctx context.Context, hashID string) {
|
||||||
|
if hashID == "" {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := c.client.SAdd(ctx, c.buildDirtyKey(), hashID).Err(); err != nil {
|
||||||
|
log.Printf("[STORAGE] dirty-mark failed for %s: %v (index repairs on the next full rebuild)", hashID, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// MarkDirty records that hashID needs re-indexing. Exported for the API layer,
|
||||||
|
// which re-marks a batch whose refresh failed after the drain.
|
||||||
|
func (c *Client) MarkDirty(hashIDs ...string) {
|
||||||
|
c.mu.RLock()
|
||||||
|
defer c.mu.RUnlock()
|
||||||
|
if c.client == nil || len(hashIDs) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
members := make([]interface{}, 0, len(hashIDs))
|
||||||
|
for _, h := range hashIDs {
|
||||||
|
if h != "" {
|
||||||
|
members = append(members, h)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(members) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := c.client.SAdd(ctx, c.buildDirtyKey(), members...).Err(); err != nil {
|
||||||
|
log.Printf("[STORAGE] dirty re-mark failed for %d records: %v", len(members), err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// DrainDirty removes and returns up to dirtyDrainCap record ids from the dirty
|
||||||
|
// set. `more` reports whether entries remain, so the caller can schedule
|
||||||
|
// another pass instead of waiting a full interval to catch up on a burst.
|
||||||
|
//
|
||||||
|
// Destructive on purpose: SPOP makes concurrent refreshes disjoint and stops a
|
||||||
|
// record being re-read every tick forever. The caller owns the drained ids
|
||||||
|
// until it has indexed them — on failure it must MarkDirty them again.
|
||||||
|
func (c *Client) DrainDirty() (ids []string, more bool, err error) {
|
||||||
|
c.mu.RLock()
|
||||||
|
defer c.mu.RUnlock()
|
||||||
|
|
||||||
|
if c.client == nil {
|
||||||
|
return nil, false, fmt.Errorf("not connected")
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
key := c.buildDirtyKey()
|
||||||
|
ids, err = c.client.SPopN(ctx, key, dirtyDrainCap).Result()
|
||||||
|
if err != nil {
|
||||||
|
return nil, false, fmt.Errorf("spop dirty: %w", err)
|
||||||
|
}
|
||||||
|
if len(ids) < dirtyDrainCap {
|
||||||
|
return ids, false, nil
|
||||||
|
}
|
||||||
|
remaining, err := c.client.SCard(ctx, key).Result()
|
||||||
|
if err != nil {
|
||||||
|
// We have the batch; not knowing the remainder just means the next
|
||||||
|
// tick picks it up on schedule.
|
||||||
|
return ids, false, nil
|
||||||
|
}
|
||||||
|
return ids, remaining > 0, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetSearchIndexTuplesFor reads the search tuples for a specific set of records
|
||||||
|
// — the incremental counterpart to GetSearchIndexTuples.
|
||||||
|
//
|
||||||
|
// Reads go through the per-record field index (GetMetadataFlat), so this is two
|
||||||
|
// round-trips per record rather than a keyspace scan. That trade is what makes
|
||||||
|
// it O(change): it beats the full pass whenever the changed set is a small
|
||||||
|
// fraction of the corpus, and the caller bounds the batch (dirtyDrainCap).
|
||||||
|
//
|
||||||
|
// A record present in the dirty set but absent from storage was deleted; it is
|
||||||
|
// reported in `gone` so the caller can evict it from the index rather than
|
||||||
|
// leaving a tombstone that still answers searches.
|
||||||
|
func (c *Client) GetSearchIndexTuplesFor(hashIDs []string) (tuples []SearchTuple, gone []string, err error) {
|
||||||
|
if len(hashIDs) == 0 {
|
||||||
|
return nil, nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
excluded := indexExcludePrefixes()
|
||||||
|
tuples = make([]SearchTuple, 0, len(hashIDs))
|
||||||
|
|
||||||
|
for _, hashID := range hashIDs {
|
||||||
|
fields, ferr := c.GetMetadataFlat(hashID)
|
||||||
|
if ferr != nil {
|
||||||
|
// Re-mark so the record is retried rather than silently missing
|
||||||
|
// from the index until the next full rebuild.
|
||||||
|
c.MarkDirty(hashID)
|
||||||
|
log.Printf("[STORAGE] incremental index read failed for %s: %v", hashID, ferr)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if len(fields) == 0 {
|
||||||
|
gone = append(gone, hashID)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
for field := range fields {
|
||||||
|
if hasAnyPrefix(field, excluded) {
|
||||||
|
delete(fields, field)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
tuples = append(tuples, SearchTuple{
|
||||||
|
HashID: hashID,
|
||||||
|
Haystack: buildHaystack(fields),
|
||||||
|
Fields: fields,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return tuples, gone, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// buildHaystack folds a record's searchable text fields into one lowercased
|
||||||
|
// blob. Shared by the full and incremental index reads so the two can never
|
||||||
|
// disagree about what is searchable — a drift there would make a record's
|
||||||
|
// matchability depend on which path happened to index it.
|
||||||
|
func buildHaystack(fields map[string]string) string {
|
||||||
|
var sb strings.Builder
|
||||||
|
for _, f := range searchIndexFields {
|
||||||
|
if v, ok := fields[f]; ok && v != "" {
|
||||||
|
sb.WriteString(strings.ToLower(v))
|
||||||
|
sb.WriteByte('\n')
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return sb.String()
|
||||||
|
}
|
||||||
@@ -0,0 +1,211 @@
|
|||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"sort"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Tests for incremental search-index maintenance (search_dirty.go).
|
||||||
|
//
|
||||||
|
// The dirty set is derived state on the hot write path, which makes its failure
|
||||||
|
// mode the dangerous kind: a MISSING mark doesn't error, it just leaves the
|
||||||
|
// search index quietly serving a stale copy of that record until someone
|
||||||
|
// triggers a full rebuild. So the property worth pinning is coverage — every
|
||||||
|
// path that changes what a search would match must mark — and that is what the
|
||||||
|
// bulk of these tests assert, one write path at a time.
|
||||||
|
//
|
||||||
|
// The second property is that a delete is a mark too, not an unmark: the
|
||||||
|
// refresh has to LEARN the record is gone in order to evict it. Marking on
|
||||||
|
// delete is counter-intuitive enough to be worth a test of its own.
|
||||||
|
|
||||||
|
func dirtyMembers(t *testing.T, c *Client) []string {
|
||||||
|
t.Helper()
|
||||||
|
got, err := c.client.SMembers(context.Background(), c.buildDirtyKey()).Result()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("smembers dirty: %v", err)
|
||||||
|
}
|
||||||
|
sort.Strings(got)
|
||||||
|
return got
|
||||||
|
}
|
||||||
|
|
||||||
|
func clearDirty(t *testing.T, c *Client) {
|
||||||
|
t.Helper()
|
||||||
|
if err := c.client.Del(context.Background(), c.buildDirtyKey()).Err(); err != nil {
|
||||||
|
t.Fatalf("del dirty: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Every write path must mark. A path that doesn't leaves the index serving a
|
||||||
|
// stale record with no error anywhere.
|
||||||
|
func TestDirty_EveryWritePathMarks(t *testing.T) {
|
||||||
|
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||||
|
cases := []struct {
|
||||||
|
name string
|
||||||
|
id string
|
||||||
|
write func(id string) error
|
||||||
|
}{
|
||||||
|
{"SetMetadataFlat", "r-set", func(id string) error {
|
||||||
|
return c.SetMetadataFlat(id, map[string]string{"title": "A"})
|
||||||
|
}},
|
||||||
|
{"SetProperty", "r-prop", func(id string) error {
|
||||||
|
return c.SetProperty(id, "title", "B")
|
||||||
|
}},
|
||||||
|
{"MergeMetadataFlat", "r-merge", func(id string) error {
|
||||||
|
_, err := c.MergeMetadataFlat(id, map[string]string{"title": "C"})
|
||||||
|
return err
|
||||||
|
}},
|
||||||
|
{"AddToSet", "r-add", func(id string) error {
|
||||||
|
_, err := c.AddToSet(id, "tags", "x")
|
||||||
|
return err
|
||||||
|
}},
|
||||||
|
}
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
clearDirty(t, c)
|
||||||
|
if err := tc.write(tc.id); err != nil {
|
||||||
|
t.Fatalf("%s: %v", tc.name, err)
|
||||||
|
}
|
||||||
|
got := dirtyMembers(t, c)
|
||||||
|
if len(got) != 1 || got[0] != tc.id {
|
||||||
|
t.Fatalf("%s must mark %q dirty, got %v", tc.name, tc.id, got)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// RemoveFromSet's non-empty branch changes a field's VALUE without changing the
|
||||||
|
// field-name index, so it is the one path that a mark riding on field-index
|
||||||
|
// maintenance would have missed. Pinned explicitly for that reason.
|
||||||
|
func TestDirty_RemoveFromSetMarksEvenWhenTheFieldSurvives(t *testing.T) {
|
||||||
|
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||||
|
const id = "r-rem"
|
||||||
|
for _, v := range []string{"x", "y"} {
|
||||||
|
if _, err := c.AddToSet(id, "tags", v); err != nil {
|
||||||
|
t.Fatalf("AddToSet: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
clearDirty(t, c)
|
||||||
|
|
||||||
|
if _, err := c.RemoveFromSet(id, "tags", "x"); err != nil {
|
||||||
|
t.Fatalf("RemoveFromSet: %v", err)
|
||||||
|
}
|
||||||
|
// `tags` still exists (y remains) — no field-index change, but the
|
||||||
|
// searchable value changed.
|
||||||
|
if got := dirtyMembers(t, c); len(got) != 1 || got[0] != id {
|
||||||
|
t.Fatalf("a value-only change must still mark dirty, got %v", got)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// A delete MARKS rather than unmarks: the refresh must read the record, find it
|
||||||
|
// gone, and evict it. Unmarking would leave a deleted record answering searches
|
||||||
|
// until the next full rebuild.
|
||||||
|
func TestDirty_DeletesMarkSoTheRefreshCanEvict(t *testing.T) {
|
||||||
|
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||||
|
const id = "r-del"
|
||||||
|
if err := c.SetMetadataFlat(id, map[string]string{"title": "Doomed"}); err != nil {
|
||||||
|
t.Fatalf("SetMetadataFlat: %v", err)
|
||||||
|
}
|
||||||
|
clearDirty(t, c)
|
||||||
|
|
||||||
|
if _, err := c.DeleteMetadata(id); err != nil {
|
||||||
|
t.Fatalf("DeleteMetadata: %v", err)
|
||||||
|
}
|
||||||
|
if got := dirtyMembers(t, c); len(got) != 1 || got[0] != id {
|
||||||
|
t.Fatalf("a delete must mark dirty so the index evicts it, got %v", got)
|
||||||
|
}
|
||||||
|
|
||||||
|
// …and the incremental read reports it as gone rather than as an error
|
||||||
|
// or an empty-but-present tuple.
|
||||||
|
tuples, gone, err := c.GetSearchIndexTuplesFor([]string{id})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetSearchIndexTuplesFor: %v", err)
|
||||||
|
}
|
||||||
|
if len(tuples) != 0 {
|
||||||
|
t.Fatalf("a deleted record must produce no tuple, got %d", len(tuples))
|
||||||
|
}
|
||||||
|
if len(gone) != 1 || gone[0] != id {
|
||||||
|
t.Fatalf("a deleted record must be reported gone, got %v", gone)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// The drain is destructive so concurrent refreshes stay disjoint and a record
|
||||||
|
// isn't re-read every tick forever. The caller owns what it drained.
|
||||||
|
func TestDirty_DrainIsDestructive(t *testing.T) {
|
||||||
|
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||||
|
clearDirty(t, c)
|
||||||
|
c.MarkDirty("a", "b", "c")
|
||||||
|
|
||||||
|
ids, more, err := c.DrainDirty()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("DrainDirty: %v", err)
|
||||||
|
}
|
||||||
|
sort.Strings(ids)
|
||||||
|
if len(ids) != 3 {
|
||||||
|
t.Fatalf("expected 3 drained ids, got %v", ids)
|
||||||
|
}
|
||||||
|
if more {
|
||||||
|
t.Fatal("a fully-drained set must not report more")
|
||||||
|
}
|
||||||
|
if got := dirtyMembers(t, c); len(got) != 0 {
|
||||||
|
t.Fatalf("drain must empty the set, got %v", got)
|
||||||
|
}
|
||||||
|
|
||||||
|
// A second drain is empty, not an error — the steady state when nothing
|
||||||
|
// has been written since the last tick.
|
||||||
|
ids, _, err = c.DrainDirty()
|
||||||
|
if err != nil || len(ids) != 0 {
|
||||||
|
t.Fatalf("draining an empty set must be a clean no-op, got %v / %v", ids, err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// The incremental read must produce tuples indistinguishable from the full
|
||||||
|
// rebuild's — same haystack, same fields, same exclusions. A drift here would
|
||||||
|
// make a record's matchability depend on which path happened to index it, which
|
||||||
|
// is invisible until someone searches for the one term that differs.
|
||||||
|
func TestDirty_IncrementalTuplesMatchTheFullRebuild(t *testing.T) {
|
||||||
|
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||||
|
const id = "r-parity"
|
||||||
|
if err := c.SetMetadataFlat(id, map[string]string{
|
||||||
|
"title": "Parity Release",
|
||||||
|
"fileName": "parity.s01e01.mkv",
|
||||||
|
"fileType": "video",
|
||||||
|
"categories/XXX": "true",
|
||||||
|
"categories/newznab/5070": "true",
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("SetMetadataFlat: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
full := tuplesByHashID(t, c)[id]
|
||||||
|
incr, gone, err := c.GetSearchIndexTuplesFor([]string{id})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetSearchIndexTuplesFor: %v", err)
|
||||||
|
}
|
||||||
|
if len(gone) != 0 || len(incr) != 1 {
|
||||||
|
t.Fatalf("expected exactly one live tuple, got %d live / %d gone", len(incr), len(gone))
|
||||||
|
}
|
||||||
|
|
||||||
|
if incr[0].Haystack != full.Haystack {
|
||||||
|
t.Errorf("haystack drift:\n incremental %q\n full %q", incr[0].Haystack, full.Haystack)
|
||||||
|
}
|
||||||
|
if len(incr[0].Fields) != len(full.Fields) {
|
||||||
|
t.Errorf("field-count drift: incremental %d, full %d", len(incr[0].Fields), len(full.Fields))
|
||||||
|
}
|
||||||
|
for k, v := range full.Fields {
|
||||||
|
if incr[0].Fields[k] != v {
|
||||||
|
t.Errorf("field %q drift: incremental %q, full %q", k, incr[0].Fields[k], v)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// The index projection must apply on both paths.
|
||||||
|
if _, ok := incr[0].Fields["categories/newznab/5070"]; ok {
|
||||||
|
t.Error("the incremental path must honour the index exclusion list too")
|
||||||
|
}
|
||||||
|
if _, ok := incr[0].Fields["categories/XXX"]; !ok {
|
||||||
|
t.Error("the incremental path must keep the adult-filter category")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
@@ -0,0 +1,116 @@
|
|||||||
|
package storage
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Tests for the search-index field projection (GetSearchIndexTuples +
|
||||||
|
// indexExcludePrefixes).
|
||||||
|
//
|
||||||
|
// The projection narrows the in-memory index only — Redis keeps every field and
|
||||||
|
// every per-record read still returns it. What has to be pinned is the boundary
|
||||||
|
// between the two, because getting it wrong in either direction is silent:
|
||||||
|
// exclude too little and the index keeps paying for fields nothing reads;
|
||||||
|
// exclude too much and a record served *from the index* comes back missing a
|
||||||
|
// field a downstream consumer depends on.
|
||||||
|
//
|
||||||
|
// The specific hazard the default guards is adult-content filtering:
|
||||||
|
// `categories/XXX` is the server-side enforcement point (meta-search's
|
||||||
|
// `query_eval::fields_are_adult`, meta-gateway's `cards::fields_are_adult`),
|
||||||
|
// and `exclude_adult` defaults to ON. Excluding the whole `categories/` family
|
||||||
|
// would make every index-served record read as non-adult and the filter would
|
||||||
|
// fail OPEN — so the default must keep `categories/XXX` while dropping the
|
||||||
|
// reader-less `categories/newznab/` mirror.
|
||||||
|
|
||||||
|
func tuplesByHashID(t *testing.T, c *Client) map[string]SearchTuple {
|
||||||
|
t.Helper()
|
||||||
|
got, err := c.GetSearchIndexTuples()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetSearchIndexTuples: %v", err)
|
||||||
|
}
|
||||||
|
out := make(map[string]SearchTuple, len(got))
|
||||||
|
for _, tu := range got {
|
||||||
|
out[tu.HashID] = tu
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSearchIndex_ExcludesNewznabCategoriesButKeepsTheRest(t *testing.T) {
|
||||||
|
forEachPrefix(t, func(t *testing.T, c *Client) {
|
||||||
|
const id = "root-1"
|
||||||
|
if err := c.SetMetadataFlat(id, map[string]string{
|
||||||
|
"title": "Some Release",
|
||||||
|
"fileType": "video",
|
||||||
|
"categories/TV/Anime": "true",
|
||||||
|
"categories/XXX": "true",
|
||||||
|
"categories/newznab/5070": "true",
|
||||||
|
"categories/newznab/2040": "true",
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("SetMetadataFlat: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
fields := tuplesByHashID(t, c)[id].Fields
|
||||||
|
if fields == nil {
|
||||||
|
t.Fatal("record missing from the search index")
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, excluded := range []string{"categories/newznab/5070", "categories/newznab/2040"} {
|
||||||
|
if _, ok := fields[excluded]; ok {
|
||||||
|
t.Errorf("%s must not be indexed — it is a reader-less mirror vocabulary", excluded)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// The load-bearing ones must survive. XXX especially: without it the
|
||||||
|
// adult filter fails open on every index-served record.
|
||||||
|
for _, kept := range []string{"categories/XXX", "categories/TV/Anime", "title", "fileType"} {
|
||||||
|
if _, ok := fields[kept]; !ok {
|
||||||
|
t.Errorf("%s must stay in the index", kept)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The store itself is untouched — the projection is index-only.
|
||||||
|
stored, err := c.GetMetadataFlat(id)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetMetadataFlat: %v", err)
|
||||||
|
}
|
||||||
|
if _, ok := stored["categories/newznab/5070"]; !ok {
|
||||||
|
t.Error("an index-excluded field must still be stored and readable per-record")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestIndexExcludePrefixes_EnvOverride(t *testing.T) {
|
||||||
|
if got := indexExcludePrefixes(); len(got) != 1 || got[0] != "categories/newznab/" {
|
||||||
|
t.Fatalf("unset env must yield the default, got %v", got)
|
||||||
|
}
|
||||||
|
|
||||||
|
t.Setenv("META_CORE_INDEX_EXCLUDE_PREFIXES", "foo/, bar/ ,")
|
||||||
|
got := indexExcludePrefixes()
|
||||||
|
if len(got) != 2 || got[0] != "foo/" || got[1] != "bar/" {
|
||||||
|
t.Fatalf("override must trim and drop blanks, got %v", got)
|
||||||
|
}
|
||||||
|
|
||||||
|
// An explicitly empty value is the escape hatch: index everything.
|
||||||
|
t.Setenv("META_CORE_INDEX_EXCLUDE_PREFIXES", "")
|
||||||
|
if got := indexExcludePrefixes(); len(got) != 0 {
|
||||||
|
t.Fatalf("an empty override must disable exclusion, got %v", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHasAnyPrefix(t *testing.T) {
|
||||||
|
prefixes := []string{"categories/newznab/", "tmp/"}
|
||||||
|
cases := map[string]bool{
|
||||||
|
"categories/newznab/5070": true,
|
||||||
|
"tmp/scratch": true,
|
||||||
|
"categories/XXX": false,
|
||||||
|
"categories/newznab": false, // the family root is not the mirror
|
||||||
|
"title": false,
|
||||||
|
}
|
||||||
|
for field, want := range cases {
|
||||||
|
if got := hasAnyPrefix(field, prefixes); got != want {
|
||||||
|
t.Errorf("hasAnyPrefix(%q) = %v, want %v", field, got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if hasAnyPrefix("anything", nil) {
|
||||||
|
t.Error("an empty prefix list must exclude nothing")
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user