fix(search): bound memory of the bleve descendant lookup

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

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(`+=&|><!(){}[]^\"~*?:\/`) + `\-\s])`)
 
 func getFieldValue[T any](m map[string]any, key string) (out T) {
 	val, ok := m[key]
@@ -85,6 +83,3 @@ func matchToResource(match *bleveSearch.DocumentMatch) *search.Resource {
 	return mapping.Deserialize[search.Resource](match.Fields)
 }
 
-func escapeQuery(s string) string {
-	return queryEscape.ReplaceAllString(s, "\\$1")
-}

services/search/pkg/bleve/descendants_bench_test.go

@@ -0,0 +1,89 @@
+package bleve
+
+import (
+	"fmt"
+	"runtime"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+// BenchmarkSearchResourcesByPath measures the descendant lookup that backs
+// folder Delete/Move/Restore/Purge. Cumulative allocations are similar for any
+// implementation that visits every descendant; what OOMs servers is the peak
+// LIVE heap while the lookup runs, so that is reported as peak-MB.
+func BenchmarkSearchResourcesByPath(b *testing.B) {
+	for _, n := range []int{20_000, 100_000} {
+		b.Run(fmt.Sprintf("docs=%d", n), func(b *testing.B) {
+			idx, _, err := NewIndex(b.TempDir(), log.NopLogger())
+			if err != nil {
+				b.Fatal(err)
+			}
+			defer idx.Close()
+
+			const rootID = "storage$space!root"
+			batch := idx.NewBatch()
+			for i := 0; i < n; i++ {
+				id := fmt.Sprintf("storage$space!f%06d", i)
+				err := batch.Index(id, search.Resource{
+					ID:       id,
+					RootID:   rootID,
+					ParentID: "storage$space!big",
+					Path:     fmt.Sprintf("./big/dir%02d/file-%06d-%032x.txt", i%50, i, uint64(i)*2654435761),
+					Type:     1,
+				})
+				if err != nil {
+					b.Fatal(err)
+				}
+				if batch.Size() >= 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")
+		})
+	}
+}

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:<folder>/* 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)
+	}
+}

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:<lookupPath>/*") 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