feat(search): check the index schema on startup and refuse breaking changes

services/search/pkg/bleve/index.go

@@ -1,11 +1,15 @@
 package bleve
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
 	"math"
 	"path/filepath"
 	"reflect"
+	"slices"
+	"strings"
 
 	"github.com/blevesearch/bleve/v2"
 	"github.com/blevesearch/bleve/v2/analysis/analyzer/custom"
@@ -17,27 +21,170 @@ import (
 	"github.com/blevesearch/bleve/v2/mapping"
 	storageProvider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 
+	"github.com/opencloud-eu/opencloud/pkg/log"
 	searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping"
 	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
 )
 
-func NewIndex(root string) (bleve.Index, error) {
+// bolt_timeout makes a second process on the same datapath fail after 5s
+// instead of blocking forever on the file lock.
+var openRuntimeConfig = map[string]interface{}{"bolt_timeout": "5s"}
+
+// NewIndex opens (or creates) the bleve index at root and reconciles its schema
+// against NewMapping() via searchmapping.Reconcile.
+func NewIndex(root string, logger log.Logger) (bleve.Index, searchmapping.Classification, error) {
 	destination := filepath.Join(root, fmt.Sprintf("bleve-v%d", search.SchemaVersion))
-	index, err := bleve.Open(destination)
-	if errors.Is(bleve.ErrorIndexPathDoesNotExist, err) {
+	index, err := bleve.OpenUsing(destination, openRuntimeConfig)
+	if errors.Is(err, bleve.ErrorIndexPathDoesNotExist) {
 		indexMapping, err := NewMapping()
 		if err != nil {
-			return nil, err
+			return nil, searchmapping.Classification{}, err
 		}
 		index, err = bleve.New(destination, indexMapping)
 		if err != nil {
-			return nil, err
+			return nil, searchmapping.Classification{}, err
 		}
 
-		return index, nil
+		searchmapping.LogNewIndexCreated(logger, destination)
+		return index, searchmapping.Classification{Verdict: searchmapping.VerdictEqual}, nil
 	}
+	if err != nil {
+		return nil, searchmapping.Classification{}, err
+	}
+
+	r := &bleveReconciler{index: index, destination: destination}
+	classification, err := searchmapping.Reconcile(destination, r, logger)
+	if err != nil {
+		if r.index != nil {
+			_ = r.index.Close()
+		}
+		return nil, classification, err
+	}
+
+	return r.index, classification, nil
+}
+
+// bleveReconciler adapts a bleve index to searchmapping.SchemaReconciler.
+type bleveReconciler struct {
+	index       bleve.Index
+	destination string
+	codeB       []byte // marshaled code mapping, produced by Classify, used by ApplyAdditive
+}
 
-	return index, err
+func (r *bleveReconciler) Classify() (searchmapping.Classification, error) {
+	classification, codeB, err := classifyStoredMapping(r.index)
+	r.codeB = codeB
+	return classification, err
+}
+
+// ApplyAdditive persists the code mapping and reopens so the live mapping picks
+// it up. persisted=true once SetInternal succeeds, even if the reopen then
+// fails; on error it closes the index and clears the handle.
+func (r *bleveReconciler) ApplyAdditive() (bool, error) {
+	if err := r.index.SetInternal([]byte("_mapping"), r.codeB); err != nil {
+		_ = r.index.Close()
+		r.index = nil
+		return false, fmt.Errorf("failed to store the updated index mapping: %w", err)
+	}
+	if err := r.index.Close(); err != nil {
+		r.index = nil
+		return true, err
+	}
+	index, err := bleve.OpenUsing(r.destination, openRuntimeConfig)
+	if err != nil {
+		r.index = nil
+		return true, err
+	}
+	r.index = index
+	return true, nil
+}
+
+// classifyStoredMapping diffs the stored mapping against NewMapping() and
+// returns the marshaled code mapping. New-in-code fields that already hold
+// data (previously indexed dynamically) are breaking. The compare is only
+// stable within one bleve version: a changed marshaling default fails towards
+// breaking, normalize the affected key here if that ever fires.
+func classifyStoredMapping(index bleve.Index) (searchmapping.Classification, []byte, error) {
+	storedB, err := index.GetInternal([]byte("_mapping"))
+	if err != nil {
+		return searchmapping.Classification{}, nil, fmt.Errorf("failed to read the stored index mapping: %w", err)
+	}
+	codeMapping, err := NewMapping()
+	if err != nil {
+		return searchmapping.Classification{}, nil, err
+	}
+	codeB, err := json.Marshal(codeMapping)
+	if err != nil {
+		return searchmapping.Classification{}, nil, err
+	}
+
+	var stored, code map[string]any
+	if err := json.Unmarshal(storedB, &stored); err != nil {
+		return searchmapping.Classification{}, nil, fmt.Errorf("failed to parse the stored index mapping: %w", err)
+	}
+	if err := json.Unmarshal(codeB, &code); err != nil {
+		return searchmapping.Classification{}, nil, err
+	}
+
+	fields, err := index.Fields()
+	if err != nil {
+		return searchmapping.Classification{}, nil, fmt.Errorf("failed to list the indexed fields: %w", err)
+	}
+	indexedFields := make(map[string]struct{}, len(fields))
+	for _, f := range fields {
+		if !strings.HasPrefix(f, "_") { // skip bleve-internal fields like _all
+			indexedFields[f] = struct{}{}
+		}
+	}
+
+	storedDM, _ := stored["default_mapping"].(map[string]any)
+	codeDM, _ := code["default_mapping"].(map[string]any)
+	storedProps, _ := storedDM["properties"].(map[string]any)
+	codeProps, _ := codeDM["properties"].(map[string]any)
+
+	classification := searchmapping.Classify(storedProps, codeProps, func(path string) bool {
+		if _, ok := indexedFields[path]; ok {
+			return true
+		}
+		nested := path + "."
+		for f := range indexedFields {
+			if strings.HasPrefix(f, nested) {
+				return true
+			}
+		}
+		return false
+	})
+
+	// everything outside default_mapping.properties (analyzer definitions,
+	// default analyzer, dynamic flags, ...) must match exactly
+	var reasons []string
+	compareKeysExcept(stored, code, "default_mapping", "", &reasons)
+	compareKeysExcept(storedDM, codeDM, "properties", "default_mapping.", &reasons)
+	if len(reasons) > 0 {
+		classification.Verdict = searchmapping.VerdictBreaking
+		classification.Reasons = append(reasons, classification.Reasons...)
+	}
+
+	return classification, codeB, nil
+}
+
+// compareKeysExcept deep-compares all keys present on either side except skip.
+func compareKeysExcept(stored, code map[string]any, skip, prefix string, reasons *[]string) {
+	keys := slices.Collect(maps.Keys(stored))
+	for k := range code {
+		if _, ok := stored[k]; !ok {
+			keys = append(keys, k)
+		}
+	}
+	slices.Sort(keys)
+	for _, k := range keys {
+		if k == skip {
+			continue
+		}
+		if !reflect.DeepEqual(stored[k], code[k]) {
+			*reasons = append(*reasons, fmt.Sprintf("%s%s changed", prefix, k))
+		}
+	}
 }
 
 func NewMapping() (mapping.IndexMapping, error) {

services/search/pkg/bleve/index_test.go

@@ -0,0 +1,202 @@
+package bleve_test
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	bleveSearch "github.com/blevesearch/bleve/v2"
+	"github.com/blevesearch/bleve/v2/analysis/analyzer/custom"
+	"github.com/blevesearch/bleve/v2/analysis/token/lowercase"
+	"github.com/blevesearch/bleve/v2/analysis/tokenizer/unicode"
+	bleveMapping "github.com/blevesearch/bleve/v2/mapping"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/bleve"
+	searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping"
+	"github.com/opencloud-eu/opencloud/services/search/pkg/search"
+)
+
+var _ = Describe("NewIndex", func() {
+	var root string
+
+	BeforeEach(func() {
+		root = GinkgoT().TempDir()
+	})
+
+	codeMapping := func() *bleveMapping.IndexMappingImpl {
+		m, err := bleve.NewMapping()
+		Expect(err).ToNot(HaveOccurred())
+		impl, ok := m.(*bleveMapping.IndexMappingImpl)
+		Expect(ok).To(BeTrue())
+		return impl
+	}
+
+	// buildIndex simulates an index left behind by an older release
+	buildIndex := func(m bleveMapping.IndexMapping, docs map[string]map[string]any) {
+		idx, err := bleveSearch.New(filepath.Join(root, fmt.Sprintf("bleve-v%d", search.SchemaVersion)), m)
+		Expect(err).ToNot(HaveOccurred())
+		for id, doc := range docs {
+			Expect(idx.Index(id, doc)).To(Succeed())
+		}
+		Expect(idx.Close()).To(Succeed())
+	}
+
+	It("creates a fresh index", func() {
+		idx, classification, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).ToNot(HaveOccurred())
+		Expect(classification.Verdict).To(Equal(searchmapping.VerdictEqual))
+		Expect(idx.Close()).To(Succeed())
+	})
+
+	It("opens an index with an identical schema", func() {
+		buildIndex(codeMapping(), nil)
+
+		idx, classification, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).ToNot(HaveOccurred())
+		Expect(classification.Verdict).To(Equal(searchmapping.VerdictEqual))
+		Expect(classification.NewFields).To(BeEmpty())
+		Expect(idx.Close()).To(Succeed())
+	})
+
+	It("treats a genuinely new field as additive", func() {
+		old := codeMapping()
+		Expect(old.DefaultMapping.Properties).To(HaveKey("Title"))
+		delete(old.DefaultMapping.Properties, "Title")
+		buildIndex(old, nil)
+
+		idx, classification, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).ToNot(HaveOccurred())
+		Expect(classification.Verdict).To(Equal(searchmapping.VerdictAdditive))
+		Expect(classification.NewFields).To(ConsistOf("Title"))
+		Expect(idx.Index("1", map[string]any{"Title": "hello"})).To(Succeed())
+		Expect(idx.Close()).To(Succeed())
+	})
+
+	It("treats a new nested field as additive", func() {
+		old := codeMapping()
+		photo := old.DefaultMapping.Properties["photo"]
+		Expect(photo).ToNot(BeNil())
+		Expect(photo.Properties).To(HaveKey("cameraMake"))
+		delete(photo.Properties, "cameraMake")
+		buildIndex(old, nil)
+
+		idx, classification, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).ToNot(HaveOccurred())
+		Expect(classification.Verdict).To(Equal(searchmapping.VerdictAdditive))
+		Expect(classification.NewFields).To(ConsistOf("photo.cameraMake"))
+		Expect(idx.Close()).To(Succeed())
+	})
+
+	It("persists an additive schema change so later startups classify it as equal", func() {
+		old := codeMapping()
+		delete(old.DefaultMapping.Properties, "Title")
+		buildIndex(old, nil)
+
+		idx, classification, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).ToNot(HaveOccurred())
+		Expect(classification.Verdict).To(Equal(searchmapping.VerdictAdditive))
+		Expect(idx.Index("1", map[string]any{"Title": "hello"})).To(Succeed())
+		Expect(idx.Close()).To(Succeed())
+
+		idx, classification, err = bleve.NewIndex(root, log.NopLogger())
+		Expect(err).ToNot(HaveOccurred())
+		Expect(classification.Verdict).To(Equal(searchmapping.VerdictEqual))
+		Expect(idx.Close()).To(Succeed())
+	})
+
+	It("refuses when a new field already has data in the index", func() {
+		old := codeMapping()
+		Expect(old.DefaultMapping.Properties).To(HaveKey("Mtime"))
+		delete(old.DefaultMapping.Properties, "Mtime")
+		buildIndex(old, map[string]map[string]any{"1": {"Mtime": "2026-01-02T03:04:05Z"}})
+
+		idx, _, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).To(MatchError(searchmapping.ErrManualActionRequired))
+		Expect(idx).To(BeNil())
+	})
+
+	It("refuses when a new object field already has nested data in the index", func() {
+		old := codeMapping()
+		Expect(old.DefaultMapping.Properties).To(HaveKey("photo"))
+		delete(old.DefaultMapping.Properties, "photo")
+		buildIndex(old, map[string]map[string]any{"1": {"photo": map[string]any{"cameraMake": "ACME"}}})
+
+		_, _, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).To(MatchError(searchmapping.ErrManualActionRequired))
+	})
+
+	It("refuses on a changed field definition", func() {
+		old := codeMapping()
+		name := old.DefaultMapping.Properties["Name"]
+		Expect(name).ToNot(BeNil())
+		Expect(name.Fields).ToNot(BeEmpty())
+		name.Fields[0].Analyzer = "fulltext"
+		buildIndex(old, nil)
+
+		_, _, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).To(MatchError(searchmapping.ErrManualActionRequired))
+	})
+
+	It("refuses when a stored field was removed from the code schema", func() {
+		old := codeMapping()
+		old.DefaultMapping.AddFieldMappingsAt("Legacy", bleveSearch.NewTextFieldMapping())
+		buildIndex(old, nil)
+
+		_, _, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).To(MatchError(searchmapping.ErrManualActionRequired))
+	})
+
+	It("refuses when a default_mapping attribute changed", func() {
+		old := codeMapping()
+		old.DefaultMapping.Dynamic = false
+		buildIndex(old, nil)
+
+		_, _, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).To(MatchError(searchmapping.ErrManualActionRequired))
+	})
+
+	It("refuses on a changed analyzer definition", func() {
+		old := codeMapping()
+		Expect(old.CustomAnalysis.Analyzers).To(HaveKey("fulltext"))
+		old.CustomAnalysis.Analyzers["fulltext"] = map[string]any{
+			"type":          custom.Name,
+			"tokenizer":     unicode.Name,
+			"token_filters": []string{lowercase.Name},
+		}
+		buildIndex(old, nil)
+
+		_, _, err := bleve.NewIndex(root, log.NopLogger())
+		Expect(err).To(MatchError(searchmapping.ErrManualActionRequired))
+	})
+})
+
+var _ = Describe("NewMapping", func() {
+	It("only references registered analyzers", func() {
+		m, err := bleve.NewMapping()
+		Expect(err).ToNot(HaveOccurred())
+		impl, ok := m.(*bleveMapping.IndexMappingImpl)
+		Expect(ok).To(BeTrue())
+		Expect(impl.Validate()).To(Succeed())
+	})
+
+	// A diff here means existing indexes will classify as breaking (schema or
+	// bleve marshaling changed); update the golden file only deliberately.
+	It("matches the committed golden mapping", func() {
+		m, err := bleve.NewMapping()
+		Expect(err).ToNot(HaveOccurred())
+		b, err := json.Marshal(m)
+		Expect(err).ToNot(HaveOccurred())
+		var got, golden map[string]any
+		Expect(json.Unmarshal(b, &got)).To(Succeed())
+
+		goldenB, err := os.ReadFile("testdata/mapping.golden.json")
+		Expect(err).ToNot(HaveOccurred())
+		Expect(json.Unmarshal(goldenB, &golden)).To(Succeed())
+
+		Expect(got).To(Equal(golden))
+	})
+})