nodestorage: IndexSpace releases the storage

nodehead/nodehead.go

@@ -156,11 +156,7 @@ func (n *nodeHead) GetHead(spaceId string) (hash string, err error) {
 }
 
 func (n *nodeHead) ReloadHeadFromStore(ctx context.Context, spaceId string) error {
-	ss, err := n.spaceStore.IndexSpace(ctx, spaceId, true)
-	if err != nil {
-		return err
-	}
-	return ss.Close(ctx)
+	return n.spaceStore.IndexSpace(ctx, spaceId, true)
 }
 
 func (n *nodeHead) registerMetrics(m metric.Metric) {

nodespace/migrator/migrator.go

@@ -120,20 +120,8 @@ func (m *migrator) Run(ctx context.Context) (err error) {
 			return err
 		}
 		log.Info("migrated space", zap.String("spaceId", id), zap.String("total", fmt.Sprintf("%d/%d", idx, len(allIds))), zap.String("time", time.Since(tm).String()))
-		st, err := m.newStorage.WaitSpaceStorage(ctx, id)
-		if err != nil {
-			return fmt.Errorf("migration: failed to get new space storage: %w", err)
-		}
-		state, err := st.StateStorage().GetState(ctx)
-		if err != nil {
-			return fmt.Errorf("migration: failed to get state: %w", err)
-		}
-		err = m.newStorage.IndexStorage().UpdateHash(ctx, nodestorage.SpaceUpdate{
-			SpaceId: id,
-			NewHash: state.NewHash,
-		})
-		if err != nil {
-			return fmt.Errorf("migration: failed to update hash: %w", err)
+		if err = m.newStorage.IndexSpace(ctx, id, false); err != nil {
+			return fmt.Errorf("migration: failed to index space: %w", err)
 		}
 		select {
 		case <-ctx.Done():

nodestorage/storageservice.go

@@ -65,7 +65,7 @@ type (
 type NodeStorage interface {
 	spacestorage.SpaceStorageProvider
 	IndexStorage() IndexStorage
-	IndexSpace(ctx context.Context, spaceId string, setHead bool) (spacestorage.SpaceStorage, error)
+	IndexSpace(ctx context.Context, spaceId string, setHead bool) error
 	SpaceStorage(ctx context.Context, spaceId string) (spacestorage.SpaceStorage, error)
 	TryLockAndDo(ctx context.Context, spaceId string, do DoFunc) (err error)
 	TryLockAndOpenDb(ctx context.Context, spaceId string, do DoAfterOpenFunc) (err error)
@@ -194,7 +194,7 @@ func (s *storageService) Run(ctx context.Context) (err error) {
 		return err
 	}
 	for _, id := range toUpdate {
-		_, err := s.IndexSpace(ctx, id, false)
+		err := s.IndexSpace(ctx, id, false)
 		if err != nil {
 			log.Error("failed to index space", zap.String("spaceId", id), zap.Error(err))
 			continue
@@ -394,11 +394,17 @@ func (s *storageService) SpaceExists(id string) bool {
 	return true
 }
 
-func (s *storageService) IndexSpace(ctx context.Context, spaceId string, setHead bool) (ss spacestorage.SpaceStorage, err error) {
-	ss, err = s.SpaceStorage(ctx, spaceId)
+func (s *storageService) IndexSpace(ctx context.Context, spaceId string, setHead bool) (err error) {
+	ss, err := s.SpaceStorage(ctx, spaceId)
 	if err != nil {
 		return
 	}
+	// an unreleased storage is never evicted from the cache
+	defer func() {
+		if cErr := ss.Close(ctx); cErr != nil {
+			log.Warn("can't close indexed storage", zap.String("spaceId", spaceId), zap.Error(cErr))
+		}
+	}()
 	state, err := ss.StateStorage().GetState(ctx)
 	if err != nil {
 		return

nodestorage/storageservice_test.go

@@ -16,6 +16,7 @@ import (
 
 	anystore "github.com/anyproto/any-store"
 	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/commonspace/headsync/statestorage"
 	"github.com/anyproto/any-sync/commonspace/spacestorage"
 	"github.com/anyproto/any-sync/testutil/anymock"
 	"github.com/stretchr/testify/assert"
@@ -383,6 +384,91 @@ func TestStorageService_TryLockAndOpenDb(t *testing.T) {
 	})
 }
 
+func TestStorageService_IndexSpace(t *testing.T) {
+	newSpace := func(t *testing.T, ss *storageService) (spaceId, hash string) {
+		store, err := ss.CreateSpaceStorage(ctx, NewStorageCreatePayload(t))
+		require.NoError(t, err)
+		spaceId = store.Id()
+		hash = "hash-" + spaceId
+		require.NoError(t, ss.ForceRemove(spaceId))
+		// the hash is set on the raw db: the storage observer would index it asynchronously
+		require.NoError(t, ss.TryLockAndOpenDb(ctx, spaceId, func(db anystore.DB) error {
+			state, err := statestorage.New(ctx, spaceId, db)
+			if err != nil {
+				return err
+			}
+			return state.SetHash(ctx, hash)
+		}))
+		require.NoError(t, ss.ForceRemove(spaceId))
+		// the index entry is left for IndexSpace to write
+		require.NoError(t, ss.IndexStorage().DeleteSpaceEntry(ctx, spaceId))
+		return spaceId, hash
+	}
+	// a released storage is evictable; a pinned one never is
+	requireReleased := func(t *testing.T, ss *storageService, spaceId string) {
+		removed, err := ss.cache.TryRemove(spaceId)
+		require.NoError(t, err)
+		require.True(t, removed)
+	}
+
+	t.Run("indexes and releases", func(t *testing.T) {
+		ss := newStorageService(t)
+		defer ss.Close(ctx)
+		spaceId, hash := newSpace(t, ss)
+		var headSet bool
+		ss.OnWriteHash(func(_ context.Context, _, _ string) { headSet = true })
+
+		require.NoError(t, ss.IndexSpace(ctx, spaceId, false))
+
+		requireReleased(t, ss, spaceId)
+		assert.False(t, headSet)
+		entry, err := ss.IndexStorage().SpaceStatusEntry(ctx, spaceId)
+		require.NoError(t, err)
+		assert.Equal(t, hash, entry.NewHash)
+	})
+	t.Run("set head", func(t *testing.T) {
+		ss := newStorageService(t)
+		defer ss.Close(ctx)
+		spaceId, hash := newSpace(t, ss)
+		var gotId, gotHash string
+		ss.OnWriteHash(func(_ context.Context, id, h string) { gotId, gotHash = id, h })
+
+		require.NoError(t, ss.IndexSpace(ctx, spaceId, true))
+
+		assert.Equal(t, spaceId, gotId)
+		assert.Equal(t, hash, gotHash)
+	})
+	t.Run("held storage stays open", func(t *testing.T) {
+		ss := newStorageService(t)
+		defer ss.Close(ctx)
+		spaceId, _ := newSpace(t, ss)
+		held, err := ss.SpaceStorage(ctx, spaceId)
+		require.NoError(t, err)
+
+		require.NoError(t, ss.IndexSpace(ctx, spaceId, false))
+
+		_, err = held.StateStorage().GetState(ctx)
+		require.NoError(t, err)
+
+		require.NoError(t, held.Close(ctx))
+		requireReleased(t, ss, spaceId)
+	})
+	t.Run("released on error", func(t *testing.T) {
+		ss := newStorageService(t)
+		defer ss.Close(ctx)
+		spaceId, _ := newSpace(t, ss)
+		// a cached storage skips the index status check, so the failure is UpdateHash
+		held, err := ss.SpaceStorage(ctx, spaceId)
+		require.NoError(t, err)
+		require.NoError(t, ss.indexStorage.Close())
+
+		require.Error(t, ss.IndexSpace(ctx, spaceId, false))
+
+		require.NoError(t, held.Close(ctx))
+		requireReleased(t, ss, spaceId)
+	})
+}
+
 func TestSpaceStorage_GetSpaceStats_CalcMedian(t *testing.T) {
 	l1 := []int{1, 3, 2}
 	l2 := []int{1}

repairer/repairer.go

@@ -173,7 +173,7 @@ func (r *repairer) repairSpace(spaceId string) (err error) {
 	if r.storage.SpaceExists(spaceId) {
 		// the error may have been transient (e.g. a failed archive upload):
 		// if the db opens and indexes fine, the space is healthy in place
-		if _, ipErr := r.storage.IndexSpace(ctx, spaceId, true); ipErr == nil {
+		if ipErr := r.storage.IndexSpace(ctx, spaceId, true); ipErr == nil {
 			log.Info("space repaired in place", zap.String("spaceId", spaceId))
 			r.stat.repairedInPlace.Add(1)
 			return nil
@@ -222,7 +222,7 @@ func (r *repairer) repairSpace(spaceId string) (err error) {
 		return errNoValidCopy
 	}
 	// validate and register the pulled copy (index hashes + nodehead)
-	if _, err = r.storage.IndexSpace(ctx, spaceId, true); err != nil {
+	if err = r.storage.IndexSpace(ctx, spaceId, true); err != nil {
 		return err
 	}
 	log.Info("space repaired from a neighbor", zap.String("spaceId", spaceId))

repairer/repairer_test.go

@@ -31,7 +31,7 @@ func TestRepairer_RepairSpace(t *testing.T) {
 		fx.indexStorage.EXPECT().SpaceStatusEntry(gomock.Any(), spaceId).Return(nodestorage.SpaceStatusEntry{Status: nodestorage.SpaceStatusError, NewHash: "someHash"}, nil)
 		fx.indexStorage.EXPECT().SetSpaceStatus(gomock.Any(), spaceId, nodestorage.SpaceStatusOk, "").Return(nil)
 		fx.storage.EXPECT().SpaceExists(spaceId).Return(true)
-		fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(nil, nil)
+		fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(nil)
 
 		require.NoError(t, fx.repairSpace(spaceId))
 		assert.Equal(t, uint32(1), fx.stat.repairedInPlace.Load())
@@ -42,13 +42,13 @@ func TestRepairer_RepairSpace(t *testing.T) {
 		fx.indexStorage.EXPECT().SetSpaceStatus(gomock.Any(), spaceId, nodestorage.SpaceStatusOk, "").Return(nil)
 		fx.storage.EXPECT().SpaceExists(spaceId).Return(true)
 		// in-place validation fails: db is corrupted
-		fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(nil, errors.New("malformed database"))
+		fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(errors.New("malformed database"))
 		fx.storage.EXPECT().QuarantineSpace(gomock.Any(), spaceId).Return("/quarantine/err.space-1", nil)
 		fx.nodeConf.EXPECT().NodeIds(spaceId).Return([]string{"p1", "p2"})
 		// first peer fails, second delivers
 		fx.coldSync.EXPECT().Sync(gomock.Any(), spaceId, "p1").Return(errors.New("unreachable"))
 		fx.coldSync.EXPECT().Sync(gomock.Any(), spaceId, "p2").Return(nil)
-		fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(nil, nil)
+		fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(nil)
 
 		require.NoError(t, fx.repairSpace(spaceId))
 		assert.Equal(t, uint32(1), fx.stat.quarantined.Load())
@@ -61,7 +61,7 @@ func TestRepairer_RepairSpace(t *testing.T) {
 		fx.storage.EXPECT().SpaceExists(spaceId).Return(false)
 		fx.nodeConf.EXPECT().NodeIds(spaceId).Return([]string{"p1"})
 		fx.coldSync.EXPECT().Sync(gomock.Any(), spaceId, "p1").Return(nil)
-		fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(nil, nil)
+		fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(nil)
 
 		require.NoError(t, fx.repairSpace(spaceId))
 		assert.Equal(t, uint32(1), fx.stat.repaired.Load())
@@ -111,7 +111,7 @@ func TestRepairer_RepairSpace(t *testing.T) {
 		fx.storage.EXPECT().SpaceExists(spaceId).Return(false)
 		fx.nodeConf.EXPECT().NodeIds(spaceId).Return([]string{"p1"})
 		fx.coldSync.EXPECT().Sync(gomock.Any(), spaceId, "p1").Return(nil)
-		fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(nil, errors.New("malformed database"))
+		fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(errors.New("malformed database"))
 		fx.indexStorage.EXPECT().SetSpaceStatus(gomock.Any(), spaceId, nodestorage.SpaceStatusError, "").Return(nil)
 
 		assert.Error(t, fx.repairSpace(spaceId))
@@ -133,7 +133,7 @@ func TestRepairer_RepairCycle(t *testing.T) {
 	fx.indexStorage.EXPECT().SpaceStatusEntry(gomock.Any(), spaceId).Return(nodestorage.SpaceStatusEntry{Status: nodestorage.SpaceStatusError, NewHash: "someHash"}, nil)
 	fx.indexStorage.EXPECT().SetSpaceStatus(gomock.Any(), spaceId, nodestorage.SpaceStatusOk, "").Return(nil)
 	fx.storage.EXPECT().SpaceExists(spaceId).Return(true)
-	fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(nil, nil)
+	fx.storage.EXPECT().IndexSpace(gomock.Any(), spaceId, true).Return(nil)
 
 	fx.repairCycle()
 	assert.Equal(t, uint32(0), fx.stat.errored.Load())

resharder/resharder.go

@@ -252,7 +252,7 @@ func (r *resharder) reconcileUnindexed() {
 			continue
 		}
 		ctx, cancel := context.WithTimeout(r.runCtx, time.Minute)
-		if _, iErr := r.storage.IndexSpace(ctx, id, true); iErr != nil {
+		if iErr := r.storage.IndexSpace(ctx, id, true); iErr != nil {
 			log.Warn("drain cycle: can't index unindexed space dir", zap.String("spaceId", id), zap.Error(iErr))
 		} else {
 			log.Info("drain cycle: indexed a space dir without an index entry", zap.String("spaceId", id))