diff --git a/changelog/unreleased/fix-search-descendant-lookup-memory.md b/changelog/unreleased/fix-search-descendant-lookup-memory.md new file mode 100644 index 0000000000..5b61e44c9e --- /dev/null +++ b/changelog/unreleased/fix-search-descendant-lookup-memory.md @@ -0,0 +1,17 @@ +Bugfix: Bound memory of the search descendant lookup + +Deleting, moving, restoring or purging a folder made the search service look +up every descendant of that folder with a Path wildcard query. Path is a +keyword field, so bleve expanded the wildcard into one term searcher per +descendant, all alive at once. Peak live memory scaled with the number of +descendants and the kernel OOM-killed the whole server on folder deletes; on +one production instance a routine delete held 2.24 GB of a 2.29 GB live heap +in this single query. + +The lookup now enumerates the matching path terms from the field dictionary +and fetches the documents in bounded batches of exact term queries, returning +the same result set with O(1) live searcher memory: a 100k-file folder went +from 1194 MB peak to 102 MB, slightly faster than before. + +https://github.com/opencloud-eu/opencloud/issues/1269 +https://github.com/opencloud-eu/opencloud/issues/3469 diff --git a/services/search/pkg/bleve/bleve.go b/services/search/pkg/bleve/bleve.go index 1bbf53857d..d5320e8165 100644 --- a/services/search/pkg/bleve/bleve.go +++ b/services/search/pkg/bleve/bleve.go @@ -1,7 +1,6 @@ package bleve import ( - "regexp" bleveSearch "github.com/blevesearch/bleve/v2/search" storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" @@ -11,7 +10,6 @@ import ( "github.com/opencloud-eu/opencloud/services/search/pkg/search" ) -var queryEscape = regexp.MustCompile(`([` + regexp.QuoteMeta(`+=&|>= 1000 { + if err := idx.Batch(batch); err != nil { + b.Fatal(err) + } + batch.Reset() + } + } + if err := idx.Batch(batch); err != nil { + b.Fatal(err) + } + + runtime.GC() + var base runtime.MemStats + runtime.ReadMemStats(&base) + + var peak atomic.Uint64 + stop := make(chan struct{}) + go func() { + var m runtime.MemStats + for { + select { + case <-stop: + return + default: + runtime.ReadMemStats(&m) + if m.HeapInuse > peak.Load() { + peak.Store(m.HeapInuse) + } + time.Sleep(200 * time.Microsecond) + } + } + }() + + b.ResetTimer() + for b.Loop() { + res, err := searchResourcesByPath(rootID, "./big", idx) + if err != nil { + b.Fatal(err) + } + if len(res) != n { + b.Fatalf("expected %d descendants, got %d", n, len(res)) + } + } + b.StopTimer() + close(stop) + b.ReportMetric(float64(peak.Load()-base.HeapInuse)/1e6, "peak-MB") + }) + } +} diff --git a/services/search/pkg/bleve/descendants_test.go b/services/search/pkg/bleve/descendants_test.go new file mode 100644 index 0000000000..124e15b048 --- /dev/null +++ b/services/search/pkg/bleve/descendants_test.go @@ -0,0 +1,155 @@ +package bleve + +import ( + "fmt" + "runtime" + "sort" + "sync/atomic" + "testing" + "time" + + "github.com/blevesearch/bleve/v2" + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/services/search/pkg/search" +) + +func newTestIndex(t testing.TB) bleve.Index { + t.Helper() + idx, _, err := NewIndex(t.TempDir(), log.NopLogger()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = idx.Close() }) + return idx +} + +func indexResources(t testing.TB, idx bleve.Index, resources ...search.Resource) { + t.Helper() + batch := idx.NewBatch() + for _, r := range resources { + if err := batch.Index(r.ID, r); err != nil { + t.Fatal(err) + } + if batch.Size() >= 1000 { + if err := idx.Batch(batch); err != nil { + t.Fatal(err) + } + batch.Reset() + } + } + if err := idx.Batch(batch); err != nil { + t.Fatal(err) + } +} + +func TestSearchResourcesByPath(t *testing.T) { + idx := newTestIndex(t) + + const rootA, rootB = "s$a!root", "s$b!root" + var docs []search.Resource + add := func(root, id, path string) { + docs = append(docs, search.Resource{ID: id, RootID: root, Path: path, Type: 1}) + } + // 1001 descendants: crosses the 500-term batch boundary twice, once + // mid-batch and once with a single-element tail + var wantIDs []string + for i := 0; i < 1001; i++ { + id := fmt.Sprintf("s$a!f%04d", i) + add(rootA, id, fmt.Sprintf("./big/f%04d.txt", i)) + wantIDs = append(wantIDs, id) + } + add(rootA, "s$a!big", "./big") // the folder itself: not a descendant + add(rootA, "s$a!big2", "./big2/x.txt") // sibling with prefix name: excluded + add(rootB, "s$b!clone", "./big/f0000.txt") // same path, other space: excluded + // special characters the old query-string escaping had to handle + add(rootA, "s$a!odd", `./odd name*[1]/file:with spaces?.txt`) + indexResources(t, idx, docs...) + + got, err := searchResourcesByPath(rootA, "./big", idx) + if err != nil { + t.Fatal(err) + } + gotIDs := make([]string, 0, len(got)) + for _, r := range got { + gotIDs = append(gotIDs, r.ID) + } + sort.Strings(gotIDs) + sort.Strings(wantIDs) + if len(gotIDs) != len(wantIDs) { + t.Fatalf("expected %d descendants, got %d", len(wantIDs), len(gotIDs)) + } + for i := range wantIDs { + if gotIDs[i] != wantIDs[i] { + t.Fatalf("descendant sets differ at %d: want %s, got %s", i, wantIDs[i], gotIDs[i]) + } + } + + odd, err := searchResourcesByPath(rootA, "./odd name*[1]", idx) + if err != nil { + t.Fatal(err) + } + if len(odd) != 1 || odd[0].ID != "s$a!odd" { + t.Fatalf("special-character path: expected [s$a!odd], got %v", odd) + } +} + +// TestSearchResourcesByPathMemoryBounded guards against the descendant lookup +// regressing to an implementation whose live memory scales with the number of +// descendants (e.g. the former Path:/* wildcard, which materialised +// one term searcher per descendant and OOM-killed servers on folder deletes; +// it holds ~200MB here). The batched term-query implementation stays under +// ~20MB regardless of folder size. +func TestSearchResourcesByPathMemoryBounded(t *testing.T) { + if testing.Short() { + t.Skip("indexes 20k documents") + } + idx := newTestIndex(t) + + const n, rootID = 20_000, "s$mem!root" + docs := make([]search.Resource, 0, n) + for i := 0; i < n; i++ { + id := fmt.Sprintf("s$mem!f%05d", i) + docs = append(docs, search.Resource{ + ID: id, RootID: rootID, Type: 1, + Path: fmt.Sprintf("./big/dir%02d/file-%05d-%032x.txt", i%50, i, uint64(i)*2654435761), + }) + } + indexResources(t, idx, docs...) + + runtime.GC() + var base runtime.MemStats + runtime.ReadMemStats(&base) + + var peak atomic.Uint64 + stop := make(chan struct{}) + go func() { + var m runtime.MemStats + for { + select { + case <-stop: + return + default: + runtime.ReadMemStats(&m) + if m.HeapInuse > peak.Load() { + peak.Store(m.HeapInuse) + } + time.Sleep(200 * time.Microsecond) + } + } + }() + + res, err := searchResourcesByPath(rootID, "./big", idx) + close(stop) + if err != nil { + t.Fatal(err) + } + if len(res) != n { + t.Fatalf("expected %d descendants, got %d", n, len(res)) + } + + const limit = 64 << 20 // generous 3x headroom over the fix, far below the wildcard's cost + if delta := peak.Load() - base.HeapInuse; delta > limit { + t.Fatalf("descendant lookup held %dMB live heap for %d docs (limit %dMB): "+ + "searcher memory must stay bounded, not scale with folder size", delta>>20, n, limit>>20) + } +} diff --git a/services/search/pkg/bleve/index.go b/services/search/pkg/bleve/index.go index fd4d50e0d6..9b463ba83c 100644 --- a/services/search/pkg/bleve/index.go +++ b/services/search/pkg/bleve/index.go @@ -16,6 +16,7 @@ import ( "github.com/blevesearch/bleve/v2/analysis/token/lowercase" "github.com/blevesearch/bleve/v2/analysis/tokenizer/unicode" "github.com/blevesearch/bleve/v2/mapping" + "github.com/blevesearch/bleve/v2/search/query" storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" "github.com/opencloud-eu/opencloud/pkg/log" @@ -227,21 +228,58 @@ func searchResourceByID(id string, index bleve.Index) (*search.Resource, error) } func searchResourcesByPath(rootID string, lookupPath string, index bleve.Index) ([]*search.Resource, error) { - q := bleve.NewConjunctionQuery( - bleve.NewQueryStringQuery("RootID:"+rootID), - bleve.NewQueryStringQuery("Path:"+escapeQuery(lookupPath+"/*")), - ) - bleveReq := bleve.NewSearchRequest(q) - bleveReq.Size = math.MaxInt - bleveReq.Fields = []string{"*"} - res, err := index.Search(bleveReq) + // Path is a keyword field: one term per document, the full path. A wildcard + // query ("Path:/*") materialises one term searcher per + // descendant, all alive at once, each holding segment dictionary and FST + // readers -- gigabytes for a big folder, OOM-killing the server on any + // folder delete/move/restore/purge. Enumerate the matching terms from the + // field dictionary instead and fetch the documents in bounded batches of + // exact term queries. + dict, err := index.FieldDictPrefix("Path", []byte(lookupPath+"/")) if err != nil { return nil, err } + var paths []string + for { + entry, err := dict.Next() + if err != nil { + _ = dict.Close() + return nil, err + } + if entry == nil { + break + } + paths = append(paths, entry.Term) + } + if err := dict.Close(); err != nil { + return nil, err + } - resources := make([]*search.Resource, 0, res.Hits.Len()) - for _, match := range res.Hits { - resources = append(resources, matchToResource(match)) + rootQuery := bleve.NewTermQuery(rootID) + rootQuery.SetField("RootID") + + const termBatchSize = 500 // bounds the number of term searchers alive at once + resources := make([]*search.Resource, 0, len(paths)) + for start := 0; start < len(paths); start += termBatchSize { + pathQueries := make([]query.Query, 0, termBatchSize) + for _, p := range paths[start:min(start+termBatchSize, len(paths))] { + pq := bleve.NewTermQuery(p) + pq.SetField("Path") + pathQueries = append(pathQueries, pq) + } + bleveReq := bleve.NewSearchRequest(bleve.NewConjunctionQuery( + rootQuery, + bleve.NewDisjunctionQuery(pathQueries...), + )) + bleveReq.Size = math.MaxInt + bleveReq.Fields = []string{"*"} + res, err := index.Search(bleveReq) + if err != nil { + return nil, err + } + for _, match := range res.Hits { + resources = append(resources, matchToResource(match)) + } } return resources, nil