GO-4202 make the ft reindex logic atomic

core/indexer/fulltext.go

@@ -216,15 +216,16 @@ func (i *indexer) ftInit() error {
 			return err
 		}
 		if docCount == 0 {
+			// query objects that are existing in the store
+			// if they are not existing in the object store, they will be indexed and added via reindexOutdatedObjects or on receiving via any-sync
 			ids, err := i.store.ListIdsCrossSpace()
 			if err != nil {
 				return err
 			}
-			for _, id := range ids {
-				if err := i.store.AddToIndexQueue(i.runCtx, id); err != nil {
-					return err
-				}
+			if err := i.store.AddToIndexQueue(i.runCtx, ids...); err != nil {
+				return err
 			}
+
 		}
 	}
 	return nil

pkg/lib/localstore/objectstore/indexer_store.go

@@ -8,17 +8,26 @@ import (
 	"github.com/anyproto/anytype-heart/pkg/lib/pb/model"
 )
 
-func (s *dsObjectStore) AddToIndexQueue(ctx context.Context, id string) error {
+func (s *dsObjectStore) AddToIndexQueue(ctx context.Context, ids ...string) error {
+	txn, err := s.fulltextQueue.WriteTx(ctx)
+	if err != nil {
+		return fmt.Errorf("start write tx: %w", err)
+	}
 	arena := s.arenaPool.Get()
 	defer func() {
 		arena.Reset()
 		s.arenaPool.Put(arena)
 	}()
-	obj := arena.NewObject()
-	obj.Set("id", arena.NewString(id))
 
-	_, err := s.fulltextQueue.UpsertOne(ctx, obj)
-	return err
+	obj := arena.NewObject()
+	for _, id := range ids {
+		obj.Set("id", arena.NewString(id))
+		_, err = s.fulltextQueue.UpsertOne(txn.Context(), obj)
+		if err != nil {
+			return fmt.Errorf("upsert: %w", err)
+		}
+	}
+	return txn.Commit()
 }
 
 func (s *dsObjectStore) BatchProcessFullTextQueue(ctx context.Context, limit int, processIds func(ids []string) error) error {

pkg/lib/localstore/objectstore/service.go

@@ -55,7 +55,7 @@ type ObjectStore interface {
 }
 
 type IndexerStore interface {
-	AddToIndexQueue(ctx context.Context, id string) error
+	AddToIndexQueue(ctx context.Context, id ...string) error
 	ListIdsFromFullTextQueue(limit int) ([]string, error)
 	RemoveIdsFromFullTextQueue(ids []string) error
 	GetGlobalChecksums() (checksums *model.ObjectStoreChecksums, err error)

pkg/lib/localstore/objectstore/spaceindex/fixture.go

@@ -50,11 +50,13 @@ func (q *dummyFulltextQueue) RemoveIdsFromFullTextQueue(ids []string) error {
 	return nil
 }
 
-func (q *dummyFulltextQueue) AddToIndexQueue(ctx context.Context, id string) error {
+func (q *dummyFulltextQueue) AddToIndexQueue(ctx context.Context, ids ...string) error {
 	q.lock.Lock()
 	defer q.lock.Unlock()
-	if !lo.Contains(q.ids, id) {
-		q.ids = append(q.ids, id)
+	for _, id := range ids {
+		if !lo.Contains(q.ids, id) {
+			q.ids = append(q.ids, id)
+		}
 	}
 	return nil
 }

pkg/lib/localstore/objectstore/spaceindex/store.go

@@ -93,7 +93,7 @@ type SourceDetailsFromID interface {
 
 type FulltextQueue interface {
 	RemoveIdsFromFullTextQueue(ids []string) error
-	AddToIndexQueue(ctx context.Context, id string) error
+	AddToIndexQueue(ctx context.Context, ids ...string) error
 	ListIdsFromFullTextQueue(limit int) ([]string, error)
 }