Merge pull request #1401 from anyproto/GO-3769-sync-status-updates

GO-3769: Sync status updates

Author: mcrakhman

.golangci.yml

@@ -13,6 +13,7 @@ issues:
     - pb
   exclude-files:
     - '.*_test.go'
+    - 'mock*'
     - 'testMock/*'
     - 'clientlibrary/service/service.pb.go'
 

.mockery.yaml

@@ -192,6 +192,9 @@ packages:
     interfaces:
       PeerStatusChecker:
       SyncDetailsUpdater:
+  github.com/anyproto/anytype-heart/core/syncstatus/nodestatus:
+    interfaces:
+      NodeStatus:
   github.com/anyproto/anytype-heart/core/syncstatus/objectsyncstatus:
     interfaces:
       Updater:
@@ -210,4 +213,6 @@ packages:
   github.com/anyproto/anytype-heart/core/syncstatus/spacesyncstatus:
     interfaces:
       SpaceIdGetter:
+      NodeUsage:
+      NetworkConfig:
       Updater:

core/anytype/bootstrap.go

@@ -81,6 +81,7 @@ import (
 	"github.com/anyproto/anytype-heart/core/syncstatus/detailsupdater"
 	"github.com/anyproto/anytype-heart/core/syncstatus/nodestatus"
 	"github.com/anyproto/anytype-heart/core/syncstatus/spacesyncstatus"
+	"github.com/anyproto/anytype-heart/core/syncstatus/syncsubscriptions"
 	"github.com/anyproto/anytype-heart/core/wallet"
 	"github.com/anyproto/anytype-heart/metrics"
 	"github.com/anyproto/anytype-heart/pkg/lib/core"
@@ -263,7 +264,7 @@ func Bootstrap(a *app.App, components ...app.Component) {
 		Register(treemanager.New()).
 		Register(block.New()).
 		Register(indexer.New()).
-		Register(detailsupdater.NewUpdater()).
+		Register(detailsupdater.New()).
 		Register(session.NewHookRunner()).
 		Register(spacesyncstatus.NewSpaceSyncStatus()).
 		Register(nodestatus.NewNodeStatus()).
@@ -277,6 +278,7 @@ func Bootstrap(a *app.App, components ...app.Component) {
 		Register(debug.New()).
 		Register(collection.New()).
 		Register(subscription.New()).
+		Register(syncsubscriptions.New()).
 		Register(builtinobjects.New()).
 		Register(bookmark.New()).
 		Register(importer.New()).

core/block/object/treesyncer/mock_treesyncer/mock_SyncDetailsUpdater.go

@@ -15,8 +15,6 @@ import (
 	"github.com/anyproto/any-sync/net/streampool"
 	"github.com/anyproto/any-sync/nodeconf"
 	"go.uber.org/zap"
-
-	"github.com/anyproto/anytype-heart/core/domain"
 )
 
 var log = logger.NewNamed(treemanager.CName)
@@ -62,14 +60,9 @@ type SyncedTreeRemover interface {
 	RemoveAllExcept(senderId string, differentRemoteIds []string)
 }
 
-type PeerStatusChecker interface {
-	app.Component
-	IsPeerOffline(peerId string) bool
-}
-
 type SyncDetailsUpdater interface {
 	app.Component
-	UpdateDetails(objectId []string, status domain.ObjectSyncStatus, syncError domain.SyncError, spaceId string)
+	UpdateSpaceDetails(existing, missing []string, spaceId string)
 }
 
 type treeSyncer struct {
@@ -84,7 +77,6 @@ type treeSyncer struct {
 	treeManager        treemanager.TreeManager
 	isRunning          bool
 	isSyncing          bool
-	peerManager        PeerStatusChecker
 	nodeConf           nodeconf.NodeConf
 	syncedTreeRemover  SyncedTreeRemover
 	syncDetailsUpdater SyncDetailsUpdater
@@ -106,7 +98,6 @@ func NewTreeSyncer(spaceId string) treesyncer.TreeSyncer {
 func (t *treeSyncer) Init(a *app.App) (err error) {
 	t.isSyncing = true
 	t.treeManager = app.MustComponent[treemanager.TreeManager](a)
-	t.peerManager = app.MustComponent[PeerStatusChecker](a)
 	t.nodeConf = app.MustComponent[nodeconf.NodeConf](a)
 	t.syncedTreeRemover = app.MustComponent[SyncedTreeRemover](a)
 	t.syncDetailsUpdater = app.MustComponent[SyncDetailsUpdater](a)
@@ -161,13 +152,11 @@ func (t *treeSyncer) ShouldSync(peerId string) bool {
 	return t.isSyncing
 }
 
-func (t *treeSyncer) SyncAll(ctx context.Context, peerId string, existing, missing []string) error {
+func (t *treeSyncer) SyncAll(ctx context.Context, peerId string, existing, missing []string) (err error) {
 	t.Lock()
 	defer t.Unlock()
-	var err error
 	isResponsible := slices.Contains(t.nodeConf.NodeIds(t.spaceId), peerId)
-	defer t.sendResultEvent(err, isResponsible, peerId, existing)
-	t.sendSyncingEvent(peerId, existing, missing, isResponsible)
+	t.sendSyncEvents(existing, missing, isResponsible)
 	reqExec, exists := t.requestPools[peerId]
 	if !exists {
 		reqExec = newExecutor(t.requests, 0)
@@ -206,31 +195,15 @@ func (t *treeSyncer) SyncAll(ctx context.Context, peerId string, existing, missi
 	return nil
 }
 
-func (t *treeSyncer) sendSyncingEvent(peerId string, existing []string, missing []string, nodePeer bool) {
+func (t *treeSyncer) sendSyncEvents(existing, missing []string, nodePeer bool) {
 	if !nodePeer {
 		return
 	}
-	if t.peerManager.IsPeerOffline(peerId) {
-		t.sendDetailsUpdates(existing, domain.ObjectError, domain.NetworkError)
-		return
-	}
-	if len(existing) != 0 || len(missing) != 0 {
-		t.sendDetailsUpdates(existing, domain.ObjectSyncing, domain.Null)
-	}
+	t.sendDetailsUpdates(existing, missing)
 }
 
-func (t *treeSyncer) sendResultEvent(err error, nodePeer bool, peerId string, existing []string) {
-	if nodePeer && !t.peerManager.IsPeerOffline(peerId) {
-		if err != nil {
-			t.sendDetailsUpdates(existing, domain.ObjectError, domain.NetworkError)
-		} else {
-			t.sendDetailsUpdates(existing, domain.ObjectSynced, domain.Null)
-		}
-	}
-}
-
-func (t *treeSyncer) sendDetailsUpdates(existing []string, status domain.ObjectSyncStatus, syncError domain.SyncError) {
-	t.syncDetailsUpdater.UpdateDetails(existing, status, syncError, t.spaceId)
+func (t *treeSyncer) sendDetailsUpdates(existing, missing []string) {
+	t.syncDetailsUpdater.UpdateSpaceDetails(existing, missing, t.spaceId)
 }
 
 func (t *treeSyncer) requestTree(peerId, id string) {
@@ -257,6 +230,7 @@ func (t *treeSyncer) updateTree(peerId, id string) {
 	syncTree, ok := tr.(synctree.SyncTree)
 	if !ok {
 		log.Warn("not a sync tree")
+		return
 	}
 	if err = syncTree.SyncWithPeer(ctx, peerId); err != nil {
 		log.Warn("synctree.SyncWithPeer error", zap.Error(err))

core/block/object/treesyncer/treesyncer.go

@@ -16,7 +16,6 @@ import (
 	"go.uber.org/mock/gomock"
 
 	"github.com/anyproto/anytype-heart/core/block/object/treesyncer/mock_treesyncer"
-	"github.com/anyproto/anytype-heart/core/domain"
 	"github.com/anyproto/anytype-heart/tests/testutil"
 )
 
@@ -26,7 +25,6 @@ type fixture struct {
 	missingMock        *mock_objecttree.MockObjectTree
 	existingMock       *mock_synctree.MockSyncTree
 	treeManager        *mock_treemanager.MockTreeManager
-	checker            *mock_treesyncer.MockPeerStatusChecker
 	nodeConf           *mock_nodeconf.MockService
 	syncStatus         *mock_treesyncer.MockSyncedTreeRemover
 	syncDetailsUpdater *mock_treesyncer.MockSyncDetailsUpdater
@@ -37,16 +35,13 @@ func newFixture(t *testing.T, spaceId string) *fixture {
 	treeManager := mock_treemanager.NewMockTreeManager(ctrl)
 	missingMock := mock_objecttree.NewMockObjectTree(ctrl)
 	existingMock := mock_synctree.NewMockSyncTree(ctrl)
-	checker := mock_treesyncer.NewMockPeerStatusChecker(t)
-	checker.EXPECT().Name().Return("checker").Maybe()
 	nodeConf := mock_nodeconf.NewMockService(ctrl)
 	nodeConf.EXPECT().Name().Return("nodeConf").AnyTimes()
 	syncStatus := mock_treesyncer.NewMockSyncedTreeRemover(t)
 	syncDetailsUpdater := mock_treesyncer.NewMockSyncDetailsUpdater(t)
 
 	a := new(app.App)
 	a.Register(testutil.PrepareMock(context.Background(), a, treeManager)).
-		Register(testutil.PrepareMock(context.Background(), a, checker)).
 		Register(testutil.PrepareMock(context.Background(), a, syncStatus)).
 		Register(testutil.PrepareMock(context.Background(), a, nodeConf)).
 		Register(testutil.PrepareMock(context.Background(), a, syncDetailsUpdater))
@@ -59,7 +54,6 @@ func newFixture(t *testing.T, spaceId string) *fixture {
 		missingMock:        missingMock,
 		existingMock:       existingMock,
 		treeManager:        treeManager,
-		checker:            checker,
 		nodeConf:           nodeConf,
 		syncStatus:         syncStatus,
 		syncDetailsUpdater: syncDetailsUpdater,
@@ -91,6 +85,25 @@ func TestTreeSyncer(t *testing.T) {
 		fx.Close(ctx)
 	})
 
+	t.Run("delayed sync notify sync status", func(t *testing.T) {
+		ctx := context.Background()
+		fx := newFixture(t, spaceId)
+		fx.treeManager.EXPECT().GetTree(gomock.Any(), spaceId, existingId).Return(fx.existingMock, nil)
+		fx.existingMock.EXPECT().SyncWithPeer(gomock.Any(), peerId).Return(nil)
+		fx.treeManager.EXPECT().GetTree(gomock.Any(), spaceId, missingId).Return(fx.missingMock, nil)
+		fx.nodeConf.EXPECT().NodeIds(spaceId).Return([]string{peerId})
+		fx.syncDetailsUpdater.EXPECT().UpdateSpaceDetails([]string{existingId}, []string{missingId}, spaceId)
+		fx.syncStatus.EXPECT().RemoveAllExcept(peerId, []string{existingId}).Return()
+		err := fx.SyncAll(context.Background(), peerId, []string{existingId}, []string{missingId})
+		require.NoError(t, err)
+		require.NotNil(t, fx.requestPools[peerId])
+		require.NotNil(t, fx.headPools[peerId])
+
+		fx.StartSync()
+		time.Sleep(100 * time.Millisecond)
+		fx.Close(ctx)
+	})
+
 	t.Run("sync after run", func(t *testing.T) {
 		ctx := context.Background()
 		fx := newFixture(t, spaceId)
@@ -189,45 +202,5 @@ func TestTreeSyncer(t *testing.T) {
 		require.Equal(t, []string{"before close", "after done"}, events)
 		mutex.Unlock()
 	})
-	t.Run("send offline event", func(t *testing.T) {
-		ctx := context.Background()
-		fx := newFixture(t, spaceId)
-		fx.treeManager.EXPECT().GetTree(gomock.Any(), spaceId, existingId).Return(fx.existingMock, nil)
-		fx.existingMock.EXPECT().SyncWithPeer(gomock.Any(), peerId).Return(nil)
-		fx.treeManager.EXPECT().GetTree(gomock.Any(), spaceId, missingId).Return(fx.missingMock, nil)
-		fx.nodeConf.EXPECT().NodeIds(spaceId).Return([]string{peerId})
-		fx.checker.EXPECT().IsPeerOffline(peerId).Return(true)
-		fx.syncStatus.EXPECT().RemoveAllExcept(peerId, []string{existingId}).Return()
-		fx.syncDetailsUpdater.EXPECT().UpdateDetails([]string{"existing"}, domain.ObjectError, domain.NetworkError, "spaceId").Return()
-
-		fx.StartSync()
-		err := fx.SyncAll(context.Background(), peerId, []string{existingId}, []string{missingId})
-		require.NoError(t, err)
-		require.NotNil(t, fx.requestPools[peerId])
-		require.NotNil(t, fx.headPools[peerId])
-
-		time.Sleep(100 * time.Millisecond)
-		fx.Close(ctx)
-	})
-	t.Run("send syncing and synced event", func(t *testing.T) {
-		ctx := context.Background()
-		fx := newFixture(t, spaceId)
-		fx.treeManager.EXPECT().GetTree(gomock.Any(), spaceId, existingId).Return(fx.existingMock, nil)
-		fx.existingMock.EXPECT().SyncWithPeer(gomock.Any(), peerId).Return(nil)
-		fx.treeManager.EXPECT().GetTree(gomock.Any(), spaceId, missingId).Return(fx.missingMock, nil)
-		fx.nodeConf.EXPECT().NodeIds(spaceId).Return([]string{peerId})
-		fx.checker.EXPECT().IsPeerOffline(peerId).Return(false)
-		fx.syncStatus.EXPECT().RemoveAllExcept(peerId, []string{existingId}).Return()
-		fx.syncDetailsUpdater.EXPECT().UpdateDetails([]string{"existing"}, domain.ObjectSynced, domain.Null, "spaceId").Return()
-		fx.syncDetailsUpdater.EXPECT().UpdateDetails([]string{"existing"}, domain.ObjectSyncing, domain.Null, "spaceId").Return()
-
-		fx.StartSync()
-		err := fx.SyncAll(context.Background(), peerId, []string{existingId}, []string{missingId})
-		require.NoError(t, err)
-		require.NotNil(t, fx.requestPools[peerId])
-		require.NotNil(t, fx.headPools[peerId])
 
-		time.Sleep(100 * time.Millisecond)
-		fx.Close(ctx)
-	})
 }

core/block/object/treesyncer/treesyncer_test.go

@@ -1,53 +1,29 @@
 package domain
 
-type SyncType int32
-
-const (
-	Objects SyncType = 0
-	Files   SyncType = 1
-)
-
 type SpaceSyncStatus int32
 
 const (
-	Synced  SpaceSyncStatus = 0
-	Syncing SpaceSyncStatus = 1
-	Error   SpaceSyncStatus = 2
-	Offline SpaceSyncStatus = 3
-	Unknown SpaceSyncStatus = 4
+	SpaceSyncStatusSynced  SpaceSyncStatus = 0
+	SpaceSyncStatusSyncing SpaceSyncStatus = 1
+	SpaceSyncStatusError   SpaceSyncStatus = 2
+	SpaceSyncStatusOffline SpaceSyncStatus = 3
+	SpaceSyncStatusUnknown SpaceSyncStatus = 4
 )
 
 type ObjectSyncStatus int32
 
 const (
-	ObjectSynced  ObjectSyncStatus = 0
-	ObjectSyncing ObjectSyncStatus = 1
-	ObjectError   ObjectSyncStatus = 2
-	ObjectQueued  ObjectSyncStatus = 3
+	ObjectSyncStatusSynced  ObjectSyncStatus = 0
+	ObjectSyncStatusSyncing ObjectSyncStatus = 1
+	ObjectSyncStatusError   ObjectSyncStatus = 2
+	ObjectSyncStatusQueued  ObjectSyncStatus = 3
 )
 
 type SyncError int32
 
 const (
-	Null                SyncError = 0
-	StorageLimitExceed  SyncError = 1
-	IncompatibleVersion SyncError = 2
-	NetworkError        SyncError = 3
-	Oversized           SyncError = 4
+	SyncErrorNull                SyncError = 0
+	SyncErrorIncompatibleVersion SyncError = 2
+	SyncErrorNetworkError        SyncError = 3
+	SyncErrorOversized           SyncError = 4
 )
-
-type SpaceSync struct {
-	SpaceId   string
-	Status    SpaceSyncStatus
-	SyncError SyncError
-	SyncType  SyncType
-}
-
-func MakeSyncStatus(spaceId string, status SpaceSyncStatus, syncError SyncError, syncType SyncType) *SpaceSync {
-	return &SpaceSync{
-		SpaceId:   spaceId,
-		Status:    status,
-		SyncError: syncError,
-		SyncType:  syncType,
-	}
-}