Merge pull request #22 from anytypeio/fix-space-gc

Fix space GC

commonspace/object/tree/synctree/synctree.go

@@ -15,7 +15,6 @@ import (
 	"github.com/anytypeio/any-sync/net/peer"
 	"github.com/anytypeio/any-sync/nodeconf"
 	"go.uber.org/zap"
-	"sync/atomic"
 )
 
 var (
@@ -46,7 +45,7 @@ type syncTree struct {
 	syncStatus syncstatus.StatusUpdater
 	notifiable HeadNotifiable
 	listener   updatelistener.UpdateListener
-	treeUsage  *atomic.Int32
+	onClose    func(id string)
 	isClosed   bool
 	isDeleted  bool
 }
@@ -69,7 +68,7 @@ type BuildDeps struct {
 	AclList            list.AclList
 	SpaceStorage       spacestorage.SpaceStorage
 	TreeStorage        treestorage.TreeStorage
-	TreeUsage          *atomic.Int32
+	OnClose            func(id string)
 	SyncStatus         syncstatus.StatusUpdater
 	PeerGetter         ResponsiblePeersGetter
 	WaitTreeRemoteSync bool
@@ -106,7 +105,7 @@ func buildSyncTree(ctx context.Context, isFirstBuild bool, deps BuildDeps) (t Sy
 		ObjectTree: objTree,
 		syncClient: syncClient,
 		notifiable: deps.HeadNotifiable,
-		treeUsage:  deps.TreeUsage,
+		onClose:    deps.OnClose,
 		listener:   deps.Listener,
 		syncStatus: deps.SyncStatus,
 	}
@@ -213,7 +212,7 @@ func (s *syncTree) Close() (err error) {
 	if s.isClosed {
 		return ErrSyncTreeClosed
 	}
-	s.treeUsage.Add(-1)
+	s.onClose(s.Id())
 	s.isClosed = true
 	return
 }
@@ -239,7 +238,6 @@ func (s *syncTree) afterBuild() {
 	if s.listener != nil {
 		s.listener.Rebuild(s)
 	}
-	s.treeUsage.Add(1)
 	if s.notifiable != nil {
 		s.notifiable.UpdateHeads(s.Id(), s.Heads())
 	}

commonspace/space.go

@@ -128,7 +128,7 @@ type space struct {
 	handleQueue multiqueue.MultiQueue[HandleMessage]
 
 	isClosed  *atomic.Bool
-	treesUsed atomic.Int32
+	treesUsed *atomic.Int32
 }
 
 func (s *space) LastUsage() time.Time {
@@ -295,7 +295,7 @@ func (s *space) PutTree(ctx context.Context, payload treestorage.TreeStorageCrea
 		Listener:       listener,
 		AclList:        s.aclList,
 		SpaceStorage:   s.storage,
-		TreeUsage:      &s.treesUsed,
+		OnClose:        func(id string) {},
 		SyncStatus:     s.syncStatus,
 		PeerGetter:     s.peerManager,
 	}
@@ -326,12 +326,16 @@ func (s *space) BuildTree(ctx context.Context, id string, opts BuildTreeOpts) (t
 		Listener:           opts.Listener,
 		AclList:            s.aclList,
 		SpaceStorage:       s.storage,
-		TreeUsage:          &s.treesUsed,
+		OnClose:            s.onObjectClose,
 		SyncStatus:         s.syncStatus,
 		WaitTreeRemoteSync: opts.WaitTreeRemoteSync,
 		PeerGetter:         s.peerManager,
 	}
-	return synctree.BuildSyncTreeOrGetRemote(ctx, id, deps)
+	if t, err = synctree.BuildSyncTreeOrGetRemote(ctx, id, deps); err != nil {
+		return nil, err
+	}
+	s.treesUsed.Add(1)
+	return
 }
 
 func (s *space) BuildHistoryTree(ctx context.Context, id string, opts HistoryTreeOpts) (t objecttree.HistoryTree, err error) {
@@ -388,10 +392,19 @@ func (s *space) handleMessage(msg HandleMessage) {
 	}
 
 	if err := s.objectSync.HandleMessage(ctx, msg.SenderId, msg.Message); err != nil {
+		if msg.Message.ObjectId != "" {
+			// cleanup thread on error
+			_ = s.handleQueue.CloseThread(msg.Message.ObjectId)
+		}
 		log.InfoCtx(ctx, "handleMessage error", zap.Error(err))
 	}
 }
 
+func (s *space) onObjectClose(id string) {
+	s.treesUsed.Add(-1)
+	_ = s.handleQueue.CloseThread(id)
+}
+
 func (s *space) Close() error {
 	if s.isClosed.Swap(true) {
 		log.Warn("call space.Close on closed space", zap.String("id", s.id))

commonspace/spaceservice.go

@@ -143,6 +143,7 @@ func (s *spaceService) NewSpace(ctx context.Context, id string) (Space, error) {
 		configuration: lastConfiguration,
 		peerManager:   peerManager,
 		storage:       st,
+		treesUsed:     &atomic.Int32{},
 		isClosed:      spaceIsClosed,
 	}
 	return sp, nil

net/dialer/dialer.go

@@ -100,7 +100,7 @@ func (d *dialer) Dial(ctx context.Context, peerId string) (p peer.Peer, err erro
 }
 
 func (d *dialer) handshake(ctx context.Context, addr string) (conn drpc.Conn, sc sec.SecureConn, err error) {
-	tcpConn, err := net.Dial("tcp", addr)
+	tcpConn, err := net.DialTimeout("tcp", addr, time.Second)
 	if err != nil {
 		return
 	}