Add queue to handle quickly updating records

common/commonspace/synctree/synctreehandler.go

@@ -8,20 +8,34 @@ import (
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/pkg/acl/treechangeproto"
 	"github.com/anytypeio/go-anytype-infrastructure-experiments/common/util/slice"
 	"github.com/gogo/protobuf/proto"
+	"go.uber.org/zap"
+	"sync"
 )
 
 type syncTreeHandler struct {
 	objTree     tree.ObjectTree
 	syncClient  SyncClient
+	handlerLock sync.Mutex
+	handlerMap  map[string][]treeMsg
+}
+
+const maxQueueSize = 5
+
+type treeMsg struct {
+	replyId     string
+	syncMessage *treechangeproto.TreeSyncMessage
 }
 
 func newSyncTreeHandler(objTree tree.ObjectTree, syncClient SyncClient) synchandler.SyncHandler {
 	return &syncTreeHandler{
 		objTree:    objTree,
 		syncClient: syncClient,
+		handlerMap: map[string][]treeMsg{},
 	}
 }
 
+type sendFunc = func() error
+
 func (s *syncTreeHandler) HandleMessage(ctx context.Context, senderId string, msg *spacesyncproto.ObjectSyncMessage) (err error) {
 	unmarshalled := &treechangeproto.TreeSyncMessage{}
 	err = proto.Unmarshal(msg.Payload, unmarshalled)
@@ -29,23 +43,69 @@ func (s *syncTreeHandler) HandleMessage(ctx context.Context, senderId string, ms
 		return
 	}
 	
+	s.handlerLock.Lock()
+	queue := s.handlerMap[senderId]
+	queueFull := len(queue) >= maxQueueSize
+	queue = append(queue, treeMsg{msg.ReplyId, unmarshalled})
+	s.handlerMap[senderId] = queue
+	s.handlerLock.Unlock()
+
+	if queueFull {
+		return
+	}
+
+	actions, err := s.handleMessage(ctx, senderId)
+	if err != nil {
+		log.With(zap.Error(err)).Debug("handling message finished with error")
+	}
+	for _, action := range actions {
+		err := action()
+		if err != nil {
+			log.With(zap.Error(err)).Debug("error while sending action")
+		}
+	}
+
+	return
+}
+
+func (s *syncTreeHandler) handleMessage(ctx context.Context, senderId string) (actions []sendFunc, err error) {
+	s.objTree.Lock()
+	defer s.objTree.Unlock()
+	s.handlerLock.Lock()
+	treeMessage := s.handlerMap[senderId][0]
+	unmarshalled := treeMessage.syncMessage
+	replyId := treeMessage.replyId
+	s.handlerLock.Unlock()
+
+	defer func() {
+		s.handlerLock.Lock()
+		defer s.handlerLock.Unlock()
+		queue := s.handlerMap[senderId]
+		excessLen := len(queue) - maxQueueSize + 1
+		if excessLen <= 0 {
+			excessLen = 1
+		}
+		queue = queue[excessLen:]
+		s.handlerMap[senderId] = queue
+	}()
+
 	content := unmarshalled.GetContent()
 	switch {
 	case content.GetHeadUpdate() != nil:
-		return s.handleHeadUpdate(ctx, senderId, content.GetHeadUpdate(), msg.ReplyId)
+		return s.handleHeadUpdate(ctx, senderId, content.GetHeadUpdate(), replyId)
 	case content.GetFullSyncRequest() != nil:
-		return s.handleFullSyncRequest(ctx, senderId, content.GetFullSyncRequest(), msg.ReplyId)
+		return s.handleFullSyncRequest(ctx, senderId, content.GetFullSyncRequest(), replyId)
 	case content.GetFullSyncResponse() != nil:
 		return s.handleFullSyncResponse(ctx, senderId, content.GetFullSyncResponse())
 	}
-	return nil
+	return
 }
 
 func (s *syncTreeHandler) handleHeadUpdate(
 	ctx context.Context,
 	senderId string,
 	update *treechangeproto.TreeHeadUpdate,
-	replyId string) (err error) {
+	replyId string) (actions []sendFunc, err error) {
 	log.With("senderId", senderId).
 		With("heads", update.Heads).
 		With("treeId", s.objTree.ID()).
@@ -57,24 +117,32 @@ func (s *syncTreeHandler) handleHeadUpdate(
 		objTree       = s.objTree
 		addResult     tree.AddResult
 	)
-
+	defer func() {
-	err = func() error {
+		if headUpdate != nil {
-		objTree.Lock()
+			actions = append(actions, func() error {
-		defer objTree.Unlock()
+				return s.syncClient.BroadcastAsync(headUpdate)
+			})
+		}
+		if fullRequest != nil {
+			actions = append(actions, func() error {
+				return s.syncClient.SendAsync(senderId, fullRequest, replyId)
+			})
+		}
+	}()
 
 	// isEmptyUpdate is sent when the tree is brought up from cache
 	if isEmptyUpdate {
 		log.With("treeId", objTree.ID()).Debug("is empty update")
 		if slice.UnsortedEquals(objTree.Heads(), update.Heads) {
-				return nil
+			return
 		}
 		// we need to sync in any case
 		fullRequest, err = s.syncClient.CreateFullSyncRequest(objTree, update.Heads, update.SnapshotPath)
-			return err
+		return
 	}
 
 	if s.alreadyHasHeads(objTree, update.Heads) {
-			return nil
+		return
 	}
 
 	addResult, err = objTree.AddRawChanges(ctx, tree.RawChangesPayload{
@@ -82,35 +150,29 @@ func (s *syncTreeHandler) handleHeadUpdate(
 		RawChanges: update.Changes,
 	})
 	if err != nil {
-			return err
+		return
 	}
 	if addResult.Mode != tree.Nothing {
 		headUpdate = s.syncClient.CreateHeadUpdate(objTree, addResult.Added)
 	}
 
 	if s.alreadyHasHeads(objTree, update.Heads) {
-			return nil
+		return
 	}
 
 	fullRequest, err = s.syncClient.CreateFullSyncRequest(objTree, update.Heads, update.SnapshotPath)
-		return err
-	}()
-
-	if headUpdate != nil {
-		s.syncClient.BroadcastAsync(headUpdate)
-	}
 
 	if fullRequest != nil {
 		log.With("senderId", senderId).
 			With("heads", objTree.Heads()).
 			With("treeId", objTree.ID()).
 			Debug("sending full sync request")
-		return s.syncClient.SendAsync(senderId, fullRequest, replyId)
+	} else {
-	}
 		log.With("senderId", senderId).
 			With("heads", update.Heads).
 			With("treeId", objTree.ID()).
 			Debug("head update finished correctly")
+	}
 	return
 }
 
@@ -118,7 +180,7 @@ func (s *syncTreeHandler) handleFullSyncRequest(
 	ctx context.Context,
 	senderId string,
 	request *treechangeproto.TreeFullSyncRequest,
-	replyId string) (err error) {
+	replyId string) (actions []sendFunc, err error) {
 	log.With("senderId", senderId).
 		With("heads", request.Heads).
 		With("treeId", s.objTree.ID()).
@@ -133,43 +195,44 @@ func (s *syncTreeHandler) handleFullSyncRequest(
 	)
 	defer func() {
 		if err != nil {
-			s.syncClient.SendAsync(senderId, treechangeproto.WrapError(err, header), replyId)
+			actions = append(actions, func() error {
+				return s.syncClient.SendAsync(senderId, treechangeproto.WrapError(err, header), replyId)
+			})
+			return
+		}
+
+		if headUpdate != nil {
+			actions = append(actions, func() error {
+				return s.syncClient.BroadcastAsync(headUpdate)
+			})
+		}
+		if fullResponse != nil {
+			actions = append(actions, func() error {
+				return s.syncClient.SendAsync(senderId, fullResponse, replyId)
+			})
 		}
 	}()
 
-	err = func() error {
-		objTree.Lock()
-		defer objTree.Unlock()
-
 	if len(request.Changes) != 0 && !s.alreadyHasHeads(objTree, request.Heads) {
 		addResult, err = objTree.AddRawChanges(ctx, tree.RawChangesPayload{
 			NewHeads:   request.Heads,
 			RawChanges: request.Changes,
 		})
 		if err != nil {
-				return err
+			return
 		}
 		if addResult.Mode != tree.Nothing {
 			headUpdate = s.syncClient.CreateHeadUpdate(objTree, addResult.Added)
 		}
 	}
 	fullResponse, err = s.syncClient.CreateFullSyncResponse(objTree, request.Heads, request.SnapshotPath)
-		return err
-	}()
-	if headUpdate != nil {
-		s.syncClient.BroadcastAsync(headUpdate)
-	}
-
-	if err != nil {
 	return
 }
-	return s.syncClient.SendAsync(senderId, fullResponse, replyId)
-}
 
 func (s *syncTreeHandler) handleFullSyncResponse(
 	ctx context.Context,
 	senderId string,
-	response *treechangeproto.TreeFullSyncResponse) (err error) {
+	response *treechangeproto.TreeFullSyncResponse) (actions []sendFunc, err error) {
 	log.With("senderId", senderId).
 		With("heads", response.Heads).
 		With("treeId", s.objTree.ID()).
@@ -179,12 +242,16 @@ func (s *syncTreeHandler) handleFullSyncResponse(
 		addResult  tree.AddResult
 		headUpdate *treechangeproto.TreeSyncMessage
 	)
-	err = func() error {
+	defer func() {
-		objTree.Lock()
+		if headUpdate != nil {
-		defer objTree.Unlock()
+			actions = append(actions, func() error {
+				return s.syncClient.BroadcastAsync(headUpdate)
+			})
+		}
+	}()
 
 	if s.alreadyHasHeads(objTree, response.Heads) {
-			return nil
+		return
 	}
 
 	addResult, err = objTree.AddRawChanges(ctx, tree.RawChangesPayload{
@@ -192,22 +259,16 @@ func (s *syncTreeHandler) handleFullSyncResponse(
 		RawChanges: response.Changes,
 	})
 	if err != nil {
-			return err
+		return
 	}
 	if addResult.Mode != tree.Nothing {
 		headUpdate = s.syncClient.CreateHeadUpdate(objTree, addResult.Added)
 	}
-		return err
-	}()
-	if headUpdate != nil {
-		s.syncClient.BroadcastAsync(headUpdate)
-	}
 
 	log.With("error", err != nil).
 		With("heads", response.Heads).
 		With("treeId", s.objTree.ID()).
 		Debug("finished full sync response")
-
 	return
 }