Change headsync logic

commonspace/headsync/diffsyncer.go

@@ -137,16 +137,19 @@ func (d *diffSyncer) syncWithPeer(ctx context.Context, p peer.Peer) (err error)
 
 	totalLen := len(newIds) + len(changedIds) + len(removedIds)
 	// not syncing ids which were removed through settings document
-	filteredIds := d.deletionState.FilterJoin(newIds, changedIds, removedIds)
+	missingIds := d.deletionState.Filter(newIds)
+	existingIds := append(d.deletionState.Filter(removedIds), d.deletionState.Filter(changedIds)...)
 
-	d.syncStatus.RemoveAllExcept(p.Id(), filteredIds, stateCounter)
-
-	d.syncTrees(ctx, p.Id(), filteredIds)
+	d.syncStatus.RemoveAllExcept(p.Id(), existingIds, stateCounter)
 
+	err = d.cache.SyncTrees(ctx, existingIds, missingIds)
+	if err != nil {
+		return err
+	}
 	d.log.Info("sync done:", zap.Int("newIds", len(newIds)),
 		zap.Int("changedIds", len(changedIds)),
 		zap.Int("removedIds", len(removedIds)),
-		zap.Int("already deleted ids", totalLen-len(filteredIds)),
+		zap.Int("already deleted ids", totalLen-len(existingIds)-len(missingIds)),
 		zap.String("peerId", p.Id()),
 	)
 	return

commonspace/headsync/diffsyncer_test.go

@@ -135,12 +135,10 @@ func TestDiffSyncer_Sync(t *testing.T) {
 		diffMock.EXPECT().
 			Diff(gomock.Any(), gomock.Eq(NewRemoteDiff(spaceId, clientMock))).
 			Return([]string{"new"}, []string{"changed"}, nil, nil)
-		delState.EXPECT().FilterJoin(gomock.Any()).Return([]string{"new", "changed"})
-		for _, arg := range []string{"new", "changed"} {
-			cacheMock.EXPECT().
-				GetTree(gomock.Any(), spaceId, arg).
-				Return(nil, nil)
-		}
+		delState.EXPECT().Filter([]string{"new"}).Return([]string{"new"}).Times(1)
+		delState.EXPECT().Filter([]string{"changed"}).Return([]string{"changed"}).Times(1)
+		delState.EXPECT().Filter(nil).Return(nil).Times(1)
+		cacheMock.EXPECT().SyncTrees(gomock.Any(), []string{"changed"}, []string{"new"}).Return(nil)
 		require.NoError(t, diffSyncer.Sync(ctx))
 	})
 

commonspace/headsync/headsync.go

@@ -29,7 +29,6 @@ type TreeHeads struct {
 
 type HeadSync interface {
 	Init(objectIds []string, deletionState settingsstate.ObjectDeletionState)
-	Run()
 
 	UpdateHeads(id string, heads []string)
 	HandleRangeRequest(ctx context.Context, req *spacesyncproto.HeadSyncRequest) (resp *spacesyncproto.HeadSyncResponse, err error)
@@ -49,7 +48,6 @@ type headSync struct {
 	syncer         DiffSyncer
 	configuration  nodeconf.NodeConf
 	spaceIsDeleted *atomic.Bool
-	isRunning      bool
 
 	syncPeriod int
 }
@@ -95,11 +93,7 @@ func NewHeadSync(
 func (d *headSync) Init(objectIds []string, deletionState settingsstate.ObjectDeletionState) {
 	d.fillDiff(objectIds)
 	d.syncer.Init(deletionState)
-}
-
-func (d *headSync) Run() {
 	d.periodicSync.Run()
-	d.isRunning = true
 }
 
 func (d *headSync) HandleRangeRequest(ctx context.Context, req *spacesyncproto.HeadSyncRequest) (resp *spacesyncproto.HeadSyncResponse, err error) {
@@ -141,9 +135,7 @@ func (d *headSync) RemoveObjects(ids []string) {
 }
 
 func (d *headSync) Close() (err error) {
-	if d.isRunning {
-		d.periodicSync.Close()
-	}
+	d.periodicSync.Close()
 	return
 }
 

commonspace/headsync/headsync_test.go

@@ -51,7 +51,6 @@ func TestDiffService(t *testing.T) {
 		storageMock.EXPECT().WriteSpaceHash(hash)
 		pSyncMock.EXPECT().Run()
 		service.Init([]string{initId}, delState)
-		service.Run()
 	})
 
 	t.Run("update heads", func(t *testing.T) {
@@ -65,9 +64,7 @@ func TestDiffService(t *testing.T) {
 	})
 
 	t.Run("close", func(t *testing.T) {
-		pSyncMock.EXPECT().Run()
 		pSyncMock.EXPECT().Close()
-		service.Run()
 		service.Close()
 	})
 }

commonspace/object/treemanager/mock_treemanager/mock_treemanager.go

@@ -134,3 +134,17 @@ func (mr *MockTreeManagerMockRecorder) Run(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Run", reflect.TypeOf((*MockTreeManager)(nil).Run), arg0)
 }
+
+// SyncTrees mocks base method.
+func (m *MockTreeManager) SyncTrees(arg0 context.Context, arg1, arg2 []string) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SyncTrees", arg0, arg1, arg2)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SyncTrees indicates an expected call of SyncTrees.
+func (mr *MockTreeManagerMockRecorder) SyncTrees(arg0, arg1, arg2 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncTrees", reflect.TypeOf((*MockTreeManager)(nil).SyncTrees), arg0, arg1, arg2)
+}

commonspace/object/treemanager/treemanager.go

@@ -14,4 +14,5 @@ type TreeManager interface {
 	GetTree(ctx context.Context, spaceId, treeId string) (objecttree.ObjectTree, error)
 	MarkTreeDeleted(ctx context.Context, spaceId, treeId string) error
 	DeleteTree(ctx context.Context, spaceId, treeId string) error
+	SyncTrees(ctx context.Context, exiting, removed []string) error
 }

commonspace/settings/settingsstate/deletionstate.go

@@ -16,7 +16,7 @@ type ObjectDeletionState interface {
 	GetQueued() (ids []string)
 	Delete(id string) (err error)
 	Exists(id string) bool
-	FilterJoin(ids ...[]string) (filtered []string)
+	Filter(ids []string) (filtered []string)
 }
 
 type objectDeletionState struct {
@@ -113,19 +113,14 @@ func (st *objectDeletionState) Exists(id string) bool {
 	return st.exists(id)
 }
 
-func (st *objectDeletionState) FilterJoin(ids ...[]string) (filtered []string) {
+func (st *objectDeletionState) Filter(ids []string) (filtered []string) {
 	st.RLock()
 	defer st.RUnlock()
-	filter := func(ids []string) {
-		for _, id := range ids {
-			if !st.exists(id) {
-				filtered = append(filtered, id)
-			}
+	for _, id := range ids {
+		if !st.exists(id) {
+			filtered = append(filtered, id)
 		}
 	}
-	for _, arr := range ids {
-		filter(arr)
-	}
 	return
 }
 

commonspace/settings/settingsstate/deletionstate_test.go

@@ -80,8 +80,8 @@ func TestDeletionState_FilterJoin(t *testing.T) {
 	fx.delState.queued["id1"] = struct{}{}
 	fx.delState.queued["id2"] = struct{}{}
 
-	filtered := fx.delState.FilterJoin([]string{"id1"}, []string{"id3", "id2"}, []string{"id4"})
-	require.Equal(t, []string{"id3", "id4"}, filtered)
+	filtered := fx.delState.Filter([]string{"id3", "id2"})
+	require.Equal(t, []string{"id3"}, filtered)
 }
 
 func TestDeletionState_AddObserver(t *testing.T) {

commonspace/settings/settingsstate/mock_settingsstate/mock_settingsstate.go

@@ -87,22 +87,18 @@ func (mr *MockObjectDeletionStateMockRecorder) Exists(arg0 interface{}) *gomock.
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Exists", reflect.TypeOf((*MockObjectDeletionState)(nil).Exists), arg0)
 }
 
-// FilterJoin mocks base method.
-func (m *MockObjectDeletionState) FilterJoin(arg0 ...[]string) []string {
+// Filter mocks base method.
+func (m *MockObjectDeletionState) Filter(arg0 []string) []string {
 	m.ctrl.T.Helper()
-	varargs := []interface{}{}
-	for _, a := range arg0 {
-		varargs = append(varargs, a)
-	}
-	ret := m.ctrl.Call(m, "FilterJoin", varargs...)
+	ret := m.ctrl.Call(m, "Filter", arg0)
 	ret0, _ := ret[0].([]string)
 	return ret0
 }
 
-// FilterJoin indicates an expected call of FilterJoin.
-func (mr *MockObjectDeletionStateMockRecorder) FilterJoin(arg0 ...interface{}) *gomock.Call {
+// Filter indicates an expected call of Filter.
+func (mr *MockObjectDeletionStateMockRecorder) Filter(arg0 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FilterJoin", reflect.TypeOf((*MockObjectDeletionState)(nil).FilterJoin), arg0...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Filter", reflect.TypeOf((*MockObjectDeletionState)(nil).Filter), arg0)
 }
 
 // GetQueued mocks base method.

commonspace/spaceutils_test.go

@@ -224,6 +224,10 @@ type mockTreeManager struct {
 	markedIds  []string
 }
 
+func (t *mockTreeManager) SyncTrees(ctx context.Context, exiting, removed []string) error {
+	return nil
+}
+
 func (t *mockTreeManager) MarkTreeDeleted(ctx context.Context, spaceId, treeId string) error {
 	t.markedIds = append(t.markedIds, treeId)
 	return nil