diff --git a/services/search/pkg/bleve/migrate.go b/services/search/pkg/bleve/migrate.go index 760844f8dd..27aba850f2 100644 --- a/services/search/pkg/bleve/migrate.go +++ b/services/search/pkg/bleve/migrate.go @@ -3,7 +3,6 @@ package bleve import ( "errors" "fmt" - "io/fs" "os" "path/filepath" "strconv" @@ -15,235 +14,66 @@ import ( "github.com/opencloud-eu/opencloud/services/search/pkg/search" ) -const ( - revisionKey = "_schema_revision" - migratingSuffix = ".migrating" - backupSuffix = ".bak" - migrateBatch = 100 - migrateLogInterval = 50000 // how often buildReplacement logs progress, in documents -) +const revisionKey = "_schema_revision" // OpenOrMigrate opens the index and, when autoMigrate is set and the stored -// schema is incompatible, rebuilds it in place first. The migration triggers on -// a schema revision bump, not on a mapping diff, so a breaking change without a -// matching bump still refuses to start. Returns the number of documents migrated -// (0 for none). Opening first means a fresh install and the common up-to-date -// case open the index only once; the migration path is taken only on a mismatch. +// schema is incompatible, resets it first so a re-crawl can repopulate it at the +// current schema. The reset triggers on a schema revision bump, not on a mapping +// diff, so a breaking change without a matching bump still refuses to start. func OpenOrMigrate(root string, autoMigrate bool, logger log.Logger) (bleve.Index, searchmapping.Classification, int, error) { idx, classification, openErr := NewIndex(root) if openErr == nil || !autoMigrate || !errors.Is(openErr, searchmapping.ErrManualActionRequired) { - // opened cleanly, or auto-migrate is off, or the failure is not a - // schema mismatch we could migrate away return idx, classification, 0, openErr } - // the stored schema is incompatible; rebuild only if the revision was bumped, - // otherwise surface the original error (a schema change without a bump) - migrated, err := MigrateIndex(root, logger) + // incompatible schema: MigrateIndex drops a bumped-revision index so the + // reopen below rebuilds it empty; a breaking change without a bump is a no-op + // there and is surfaced by that reopen instead + dropped, err := MigrateIndex(root, logger) if err != nil { return nil, classification, 0, err } - if migrated == 0 { - return nil, classification, 0, openErr - } idx, classification, err = NewIndex(root) - return idx, classification, migrated, err + return idx, classification, dropped, err } -// MigrateIndex rebuilds the index from the documents it already holds when the -// stored schema revision is older than search.SchemaRevision, else a no-op. It -// takes the exclusive bolt lock, so the search service must be stopped. +// MigrateIndex resets the index when its stored schema revision is older than +// search.SchemaRevision: it drops the outdated index so a full re-crawl +// ('opencloud search index --all-spaces --force-rescan') repopulates it at the +// current schema. No-op when the index is missing or already current. It takes +// the exclusive bolt lock, so the search service must be stopped. Returns the +// number of documents that were dropped. func MigrateIndex(root string, logger log.Logger) (int, error) { dest := filepath.Join(root, "bleve") - tmp := dest + migratingSuffix - bak := dest + backupSuffix - if err := recoverMigration(root); err != nil { - return 0, err - } - - old, err := bleve.OpenUsing(dest, openRuntimeConfig) + idx, err := bleve.OpenUsing(dest, openRuntimeConfig) if errors.Is(err, bleve.ErrorIndexPathDoesNotExist) { - return 0, nil // nothing to migrate yet; NewIndex creates a fresh index + return 0, nil // fresh install, nothing to migrate } if err != nil { return 0, fmt.Errorf("open index (is the search service still running?): %w", err) } - - stored, err := readRevision(old) - if err != nil { - _ = old.Close() - return 0, err - } - if stored >= search.SchemaRevision { - _ = old.Close() - return 0, nil + stored, rerr := readRevision(idx) + count, cerr := idx.DocCount() + _ = idx.Close() + switch { + case rerr != nil: + return 0, rerr + case cerr != nil: + return 0, cerr + case stored >= search.SchemaRevision: + return 0, nil // already current } - total, err := old.DocCount() - if err != nil { - _ = old.Close() + if err := os.RemoveAll(dest); err != nil { return 0, err } - logger.Info().Uint64("documents", total).Msg("starting search index migration") - - copied, err := buildReplacement(old, tmp, total, logger) - if err != nil { - _ = old.Close() - _ = os.RemoveAll(tmp) - return 0, err - } - - // verify before the swap, so a failure leaves the original untouched - if err := verifyMigrated(tmp, total); err != nil { - _ = old.Close() - _ = os.RemoveAll(tmp) - return 0, err - } - - if err := swap(old, dest, tmp, bak); err != nil { - return 0, err - } - logger.Info().Int("documents", copied).Msg("search index migration complete") - return copied, nil + logger.Warn().Msgf("the bleve index at %s was reset for schema revision %d (%d documents dropped); run 'opencloud search index --all-spaces --force-rescan' to repopulate", root, search.SchemaRevision, count) + return int(count), nil } -// buildReplacement copies every document from src into a fresh index at tmp via -// the same Deserialize -> PrepareForIndex round-trip the Move/Delete/Restore -// paths use, and stamps the current revision. -func buildReplacement(src bleve.Index, tmp string, total uint64, logger log.Logger) (int, error) { - if err := os.RemoveAll(tmp); err != nil { - return 0, err - } - m, err := NewMapping() - if err != nil { - return 0, err - } - dst, err := bleve.New(tmp, m) - if err != nil { - return 0, err - } - - batch, err := NewBatch(dst, migrateBatch) - if err != nil { - _ = dst.Close() - return 0, err - } - - var copied int - nextLog := migrateLogInterval - var after []string - for { - req := bleve.NewSearchRequest(bleve.NewMatchAllQuery()) - req.Size = migrateBatch - req.Fields = []string{"*"} - req.SortBy([]string{"_id"}) // total order, no skips or dupes across pages - req.SearchAfter = after - res, err := src.Search(req) - if err != nil { - _ = dst.Close() - return 0, err - } - if len(res.Hits) == 0 { - break - } - for _, hit := range res.Hits { - r := legacyHitToResource(hit.Fields) - if err := batch.Upsert(hit.ID, r); err != nil { // hit.ID: the bleve key is authoritative - _ = dst.Close() - return 0, err - } - copied++ - } - if copied >= nextLog { - logger.Info().Msgf("migrating search index: %d of %d documents", copied, total) - nextLog += migrateLogInterval - } - after = []string{res.Hits[len(res.Hits)-1].ID} - } - if err := batch.Push(); err != nil { - _ = dst.Close() - return 0, err - } - if err := writeRevision(dst); err != nil { - _ = dst.Close() - return 0, err - } - return copied, dst.Close() -} - -// legacyHitToResource is the fixup hook before re-indexing, symmetric to the -// opensearch side. bleve Deserialize is fail-soft, so nothing is stripped here. -func legacyHitToResource(fields map[string]any) search.Resource { - return *searchmapping.Deserialize[search.Resource](fields) -} - -// verifyMigrated checks the rebuilt index holds want documents before the swap -// replaces the original. The count catches a pagination drop or batch-push loss; -// the mapping needs no check, it is built deterministically from NewMapping(). -func verifyMigrated(path string, want uint64) error { - idx, err := bleve.OpenUsing(path, openRuntimeConfig) - if err != nil { - return err - } - defer func() { _ = idx.Close() }() - - got, err := idx.DocCount() - if err != nil { - return err - } - if got != want { - return fmt.Errorf("migrated index holds %d documents, expected %d", got, want) - } - return nil -} - -// swap replaces the live index with the rebuilt one. old still holds the lock -// and is closed here right before the rename. -func swap(old bleve.Index, dest, tmp, bak string) error { - if err := old.Close(); err != nil { - return err - } - if err := os.Rename(dest, bak); err != nil { - return err - } - if err := os.Rename(tmp, dest); err != nil { - _ = os.Rename(bak, dest) // best effort roll back - return err - } - return os.RemoveAll(bak) -} - -// recoverMigration cleans up a crashed MigrateIndex and must run before the -// index is opened, else NewIndex would recreate an empty index mid-swap. Rolls -// back, never forward: a truncated tmp is a valid index, completeness is not -// observable. -func recoverMigration(root string) error { - dest := filepath.Join(root, "bleve") - tmp := dest + migratingSuffix - bak := dest + backupSuffix - - if _, err := os.Stat(dest); err == nil { - return errors.Join(os.RemoveAll(tmp), os.RemoveAll(bak)) // index present: tmp/bak are leftovers - } else if !errors.Is(err, fs.ErrNotExist) { - return err - } - - if _, err := os.Stat(bak); err == nil { - // crashed mid-swap: roll back - if err := os.RemoveAll(tmp); err != nil { - return err - } - return os.Rename(bak, dest) - } else if !errors.Is(err, fs.ErrNotExist) { - return err - } - - return os.RemoveAll(tmp) // fresh install, or index removed by hand -} - -// writeRevision stamps the current revision so a fresh or migrated index is not -// seen as outdated. +// writeRevision stamps the current revision so a fresh index is not seen as +// outdated on the next start. func writeRevision(index bleve.Index) error { return index.SetInternal([]byte(revisionKey), []byte(strconv.Itoa(search.SchemaRevision))) } diff --git a/services/search/pkg/bleve/migrate_fields_test.go b/services/search/pkg/bleve/migrate_fields_test.go deleted file mode 100644 index b570fa0424..0000000000 --- a/services/search/pkg/bleve/migrate_fields_test.go +++ /dev/null @@ -1,89 +0,0 @@ -package bleve_test - -import ( - "path/filepath" - "sort" - "testing" - "time" - - bleveSearch "github.com/blevesearch/bleve/v2" - libregraph "github.com/opencloud-eu/libre-graph-api-go" - "github.com/stretchr/testify/require" - - "github.com/opencloud-eu/opencloud/pkg/conversions" - "github.com/opencloud-eu/opencloud/pkg/log" - "github.com/opencloud-eu/opencloud/services/search/pkg/bleve" - "github.com/opencloud-eu/opencloud/services/search/pkg/content" - searchmapping "github.com/opencloud-eu/opencloud/services/search/pkg/mapping" - "github.com/opencloud-eu/opencloud/services/search/pkg/search" -) - -// fullyPopulated exercises every field, including the facets and slices the -// count-only test skipped. -func fullyPopulated() search.Resource { - return search.Resource{ - ID: "1$2!3", - RootID: "1$2!2", - ParentID: "1$2!2", - Path: "./a/b/song.mp3", - Type: 2, - Deleted: true, - Hidden: true, - Document: content.Document{ - Title: "the title", - Name: "song.mp3", - Content: "some extracted content", - Size: 123456, - Mtime: conversions.ToPointer(time.Date(2026, 3, 15, 12, 0, 0, 0, time.UTC)), - MimeType: "audio/mpeg", - Tags: []string{"alpha", "beta", "gamma"}, - Favorites: []string{"user-a", "user-b"}, - Audio: &libregraph.Audio{ - Album: libregraph.PtrString("the album"), - Artist: libregraph.PtrString("the artist"), - Track: libregraph.PtrInt32(7), - Year: libregraph.PtrInt32(1998), - HasDrm: libregraph.PtrBool(false), - }, - Image: &libregraph.Image{Width: libregraph.PtrInt32(1920), Height: libregraph.PtrInt32(1080)}, - Photo: &libregraph.Photo{CameraMake: libregraph.PtrString("Canon"), Iso: libregraph.PtrInt32(400), FNumber: libregraph.PtrFloat64(2.8)}, - Location: &libregraph.GeoCoordinates{ - Longitude: libregraph.PtrFloat64(11.103870357204285), - Latitude: libregraph.PtrFloat64(49.48675890884328), - Altitude: libregraph.PtrFloat64(300.0), - }, - }, - } -} - -func TestMigratePreservesAllFields(t *testing.T) { - root := t.TempDir() - want := fullyPopulated() - - old, err := bleveSearch.New(filepath.Join(root, "bleve"), oldMainMapping(t)) - require.NoError(t, err) - require.NoError(t, old.Index(want.ID, want)) - require.NoError(t, old.Close()) - - _, err = bleve.MigrateIndex(root, log.NopLogger()) - require.NoError(t, err) - - idx, _, err := bleve.NewIndex(root) - require.NoError(t, err) - defer func() { _ = idx.Close() }() - - // read the migrated document back through the production reconstruction path - req := bleveSearch.NewSearchRequest(bleveSearch.NewMatchAllQuery()) - req.Fields = []string{"*"} - res, err := idx.Search(req) - require.NoError(t, err) - require.Len(t, res.Hits, 1) - got := searchmapping.Deserialize[search.Resource](res.Hits[0].Fields) - - require.NotNil(t, got.Mtime, "Mtime") - require.True(t, want.Mtime.Equal(*got.Mtime), "Mtime: want %v got %v", want.Mtime, got.Mtime) - got.Mtime = want.Mtime // normalize time.Time location/monotonic - sort.Strings(got.Tags) // bleve may return multi-value fields reordered - sort.Strings(got.Favorites) // - require.Equal(t, want, *got) -} diff --git a/services/search/pkg/bleve/migrate_test.go b/services/search/pkg/bleve/migrate_test.go index 0c1ab3489a..dbaf9c3697 100644 --- a/services/search/pkg/bleve/migrate_test.go +++ b/services/search/pkg/bleve/migrate_test.go @@ -7,7 +7,6 @@ import ( bleveSearch "github.com/blevesearch/bleve/v2" bleveMapping "github.com/blevesearch/bleve/v2/mapping" - "github.com/blevesearch/bleve/v2/search/query" libregraph "github.com/opencloud-eu/libre-graph-api-go" "github.com/stretchr/testify/require" @@ -129,35 +128,23 @@ func TestMigrateIndex(t *testing.T) { _, _, err = bleve.NewIndex(root) require.ErrorIs(t, err, searchmapping.ErrManualActionRequired) - // 3) migrate + // 3) migrate resets the outdated index, dropping its documents n, err := bleve.MigrateIndex(root, log.NopLogger()) require.NoError(t, err) require.Equal(t, 3, n) - // 4) the migrated index classifies as equal and holds every document + // 4) the reset index classifies as equal and is empty, awaiting a re-crawl migrated, classification, err := bleve.NewIndex(root) require.NoError(t, err) require.Equal(t, searchmapping.VerdictEqual, classification.Verdict) count, err := migrated.DocCount() require.NoError(t, err) - require.Equal(t, uint64(3), count) - - // 4a) the trashed document survived (a rescan would have dropped it) - trashed, err := migrated.Document("1$1!3") - require.NoError(t, err) - require.NotNil(t, trashed) - - // 5) location_geopoint was synthesized during migration (absent in the old index) - near := query.NewGeoDistanceQuery(11.103870357204285, 49.48675890884328, "1km") - near.SetField("location" + searchmapping.GeopointSuffix) - res, err := migrated.Search(bleveSearch.NewSearchRequest(near)) - require.NoError(t, err) - require.Equal(t, uint64(3), res.Total, "all three docs match a geo-distance query on the new sibling field") + require.Equal(t, uint64(0), count, "reset drops the documents; a re-crawl repopulates") require.NoError(t, migrated.Close()) // release the lock before the second run - // 6) idempotent: a second run is a no-op because the revision is now current + // 5) idempotent: a second run is a no-op because the revision is now current n, err = bleve.MigrateIndex(root, log.NopLogger()) require.NoError(t, err) require.Equal(t, 0, n)