diff --git a/nodehead/nodehead.go b/nodehead/nodehead.go index 9b771a9d..4064e471 100644 --- a/nodehead/nodehead.go +++ b/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) { diff --git a/nodespace/migrator/migrator.go b/nodespace/migrator/migrator.go index bb7e98c9..0cd7ea48 100644 --- a/nodespace/migrator/migrator.go +++ b/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(): diff --git a/nodestorage/mock_nodestorage/mock_nodestorage.go b/nodestorage/mock_nodestorage/mock_nodestorage.go index b592ad82..cdc37d2c 100644 --- a/nodestorage/mock_nodestorage/mock_nodestorage.go +++ b/nodestorage/mock_nodestorage/mock_nodestorage.go @@ -146,12 +146,11 @@ func (mr *MockNodeStorageMockRecorder) GetStats(ctx, id, treeTop any) *gomock.Ca } // IndexSpace mocks base method. -func (m *MockNodeStorage) IndexSpace(ctx context.Context, spaceId string, setHead bool) (spacestorage.SpaceStorage, error) { +func (m *MockNodeStorage) IndexSpace(ctx context.Context, spaceId string, setHead bool) error { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "IndexSpace", ctx, spaceId, setHead) - ret0, _ := ret[0].(spacestorage.SpaceStorage) - ret1, _ := ret[1].(error) - return ret0, ret1 + ret0, _ := ret[0].(error) + return ret0 } // IndexSpace indicates an expected call of IndexSpace. diff --git a/nodestorage/storageservice.go b/nodestorage/storageservice.go index e427e8c7..91ca58e1 100644 --- a/nodestorage/storageservice.go +++ b/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 diff --git a/nodestorage/storageservice_test.go b/nodestorage/storageservice_test.go index d67d9af6..e76e32a3 100644 --- a/nodestorage/storageservice_test.go +++ b/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} diff --git a/repairer/repairer.go b/repairer/repairer.go index 489241d6..e057499c 100644 --- a/repairer/repairer.go +++ b/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)) diff --git a/repairer/repairer_test.go b/repairer/repairer_test.go index 1c1acec3..9f7e4d6e 100644 --- a/repairer/repairer_test.go +++ b/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()) diff --git a/resharder/resharder.go b/resharder/resharder.go index 6279a79a..81e2c529 100644 --- a/resharder/resharder.go +++ b/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))