Add requests and more stuff

commonspace/sync/handlemessage.go

@@ -0,0 +1,30 @@
+package sync
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
+	"github.com/anyproto/any-sync/metric"
+)
+
+type HandleMessage struct {
+	Id                uint64
+	ReceiveTime       time.Time
+	StartHandlingTime time.Time
+	Deadline          time.Time
+	SenderId          string
+	Message           *spacesyncproto.ObjectSyncMessage
+	PeerCtx           context.Context
+}
+
+func (m HandleMessage) LogFields(fields ...zap.Field) []zap.Field {
+	return append(fields,
+		metric.SpaceId(m.Message.SpaceId),
+		metric.ObjectId(m.Message.ObjectId),
+		metric.QueueDur(m.StartHandlingTime.Sub(m.ReceiveTime)),
+		metric.TotalDur(time.Since(m.ReceiveTime)),
+	)
+}

commonspace/sync/objectsync/request.go

@@ -0,0 +1,57 @@
+package objectsync
+
+import (
+	"github.com/gogo/protobuf/proto"
+
+	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
+)
+
+type InnerRequest interface {
+	Marshall() ([]byte, error)
+}
+
+type Request struct {
+	peerId   string
+	spaceId  string
+	objectId string
+	Inner    InnerRequest
+	Bytes    []byte
+}
+
+func NewByteRequest(peerId, spaceId, objectId string, message []byte) *Request {
+	return &Request{
+		peerId:   peerId,
+		spaceId:  spaceId,
+		objectId: objectId,
+		Bytes:    message,
+	}
+}
+
+func NewRequest(peerId, spaceId, objectId string, inner InnerRequest) *Request {
+	return &Request{
+		peerId:   peerId,
+		spaceId:  spaceId,
+		objectId: objectId,
+		Inner:    inner,
+	}
+}
+
+func (r *Request) PeerId() string {
+	return r.peerId
+}
+
+func (r *Request) ObjectId() string {
+	return r.objectId
+}
+
+func (r *Request) Proto() (proto.Message, error) {
+	msg, err := r.Inner.Marshall()
+	if err != nil {
+		return nil, err
+	}
+	return &spacesyncproto.ObjectSyncMessage{
+		SpaceId:  r.spaceId,
+		Payload:  msg,
+		ObjectId: r.objectId,
+	}, nil
+}

commonspace/sync/requestmanager.go

@@ -16,6 +16,7 @@ type RequestManager interface {
 	QueueRequest(rq syncdeps.Request) error
 	SendRequest(ctx context.Context, rq syncdeps.Request, collector syncdeps.ResponseCollector) error
 	HandleStreamRequest(ctx context.Context, rq syncdeps.Request, stream drpc.Stream) error
+	Close()
 }
 
 type StreamResponse struct {
@@ -79,6 +80,10 @@ func (r *requestManager) HandleStreamRequest(ctx context.Context, rq syncdeps.Re
 	return nil
 }
 
+func (r *requestManager) Close() {
+	r.requestPool.Close()
+}
+
 func fullId(peerId, objectId string) string {
 	return strings.Join([]string{peerId, objectId}, "-")
 }

commonspace/sync/requestpool.go

@@ -9,6 +9,7 @@ type RequestPool interface {
 	TryTake(peerId, objectId string) bool
 	Release(peerId, objectId string)
 	QueueRequestAction(peerId, objectId string, action func(ctx context.Context)) (err error)
+	Close()
 }
 
 type requestPool struct {

commonspace/sync/requeststat.go

@@ -0,0 +1,132 @@
+package sync
+
+import (
+	"sync"
+
+	"golang.org/x/exp/slices"
+
+	"github.com/anyproto/any-sync/app/debugstat"
+	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
+)
+
+type requestStat struct {
+	sync.Mutex
+	peerStats map[string]peerStat
+	spaceId   string
+}
+
+func newRequestStat(spaceId string) *requestStat {
+	return &requestStat{
+		peerStats: make(map[string]peerStat),
+		spaceId:   spaceId,
+	}
+}
+
+type spaceQueueStat struct {
+	SpaceId   string     `json:"space_id"`
+	TotalSize int64      `json:"total_size"`
+	PeerStats []peerStat `json:"peer_stats,omitempty"`
+}
+
+type summaryStat struct {
+	TotalSize  int64            `json:"total_size"`
+	QueueStats []spaceQueueStat `json:"sorted_stats,omitempty"`
+}
+
+type peerStat struct {
+	QueueCount int    `json:"queue_count"`
+	SyncCount  int    `json:"sync_count"`
+	QueueSize  int64  `json:"queue_size"`
+	SyncSize   int64  `json:"sync_size"`
+	PeerId     string `json:"peer_id"`
+}
+
+func (r *requestStat) AddQueueRequest(peerId string, req *spacesyncproto.ObjectSyncMessage) {
+	r.Lock()
+	defer r.Unlock()
+	stat := r.peerStats[peerId]
+	stat.QueueCount++
+	stat.QueueSize += int64(req.Size())
+	r.peerStats[peerId] = stat
+}
+
+func (r *requestStat) AddSyncRequest(peerId string, req *spacesyncproto.ObjectSyncMessage) {
+	r.Lock()
+	defer r.Unlock()
+	stat := r.peerStats[peerId]
+	stat.SyncCount++
+	stat.SyncSize += int64(req.Size())
+	r.peerStats[peerId] = stat
+}
+
+func (r *requestStat) RemoveSyncRequest(peerId string, req *spacesyncproto.ObjectSyncMessage) {
+	r.Lock()
+	defer r.Unlock()
+	stat := r.peerStats[peerId]
+	stat.SyncCount--
+	stat.SyncSize -= int64(req.Size())
+	r.peerStats[peerId] = stat
+}
+
+func (r *requestStat) RemoveQueueRequest(peerId string, req *spacesyncproto.ObjectSyncMessage) {
+	r.Lock()
+	defer r.Unlock()
+	stat := r.peerStats[peerId]
+	stat.QueueCount--
+	stat.QueueSize -= int64(req.Size())
+	r.peerStats[peerId] = stat
+}
+
+func (r *requestStat) QueueStat() spaceQueueStat {
+	r.Lock()
+	defer r.Unlock()
+	var totalSize int64
+	var peerStats []peerStat
+	for peerId, stat := range r.peerStats {
+		totalSize += stat.QueueSize
+		stat.PeerId = peerId
+		peerStats = append(peerStats, stat)
+	}
+	slices.SortFunc(peerStats, func(first, second peerStat) int {
+		firstTotalSize := first.QueueSize + first.SyncSize
+		secondTotalSize := second.QueueSize + second.SyncSize
+		if firstTotalSize > secondTotalSize {
+			return -1
+		} else if firstTotalSize == secondTotalSize {
+			return 0
+		} else {
+			return 1
+		}
+	})
+	return spaceQueueStat{
+		SpaceId:   r.spaceId,
+		TotalSize: totalSize,
+		PeerStats: peerStats,
+	}
+}
+
+func (r *requestStat) Aggregate(values []debugstat.StatValue) summaryStat {
+	var totalSize int64
+	var stats []spaceQueueStat
+	for _, v := range values {
+		stat, ok := v.Value.(spaceQueueStat)
+		if !ok {
+			continue
+		}
+		totalSize += stat.TotalSize
+		stats = append(stats, stat)
+	}
+	slices.SortFunc(stats, func(first, second spaceQueueStat) int {
+		if first.TotalSize > second.TotalSize {
+			return -1
+		} else if first.TotalSize == second.TotalSize {
+			return 0
+		} else {
+			return 1
+		}
+	})
+	return summaryStat{
+		TotalSize:  totalSize,
+		QueueStats: stats,
+	}
+}

commonspace/sync/sync.go

@@ -2,6 +2,7 @@ package sync
 
 import (
 	"context"
+	"errors"
 
 	"github.com/cheggaaa/mb/v3"
 	"go.uber.org/zap"
@@ -47,6 +48,10 @@ func NewSyncService() SyncService {
 	return &syncService{}
 }
 
+func (s *syncService) Name() (name string) {
+	return CName
+}
+
 func (s *syncService) Init(a *app.App) (err error) {
 	s.handler = a.MustComponent(syncdeps.CName).(syncdeps.SyncHandler)
 	s.sendQueueProvider = multiqueue.NewQueueProvider[drpc.Message](100, s.handleOutgoingMessage)
@@ -59,8 +64,18 @@ func (s *syncService) Init(a *app.App) (err error) {
 	return nil
 }
 
-func (s *syncService) Name() (name string) {
-	return CName
+func (s *syncService) Run(ctx context.Context) (err error) {
+	return nil
+}
+
+func (s *syncService) Close(ctx context.Context) (err error) {
+	receiveErr := s.receiveQueue.Close()
+	providerErr := s.sendQueueProvider.Close()
+	if receiveErr != nil || providerErr != nil {
+		err = errors.Join(receiveErr, providerErr)
+	}
+	s.manager.Close()
+	return
 }
 
 func (s *syncService) BroadcastMessage(ctx context.Context, msg drpc.Message) error {
@@ -83,6 +98,12 @@ func (s *syncService) handleIncomingMessage(msg msgCtx) {
 	if err != nil {
 		log.Error("failed to queue request", zap.Error(err))
 	}
+	//msg.StartHandlingTime = time.Now()
+	//ctx := peer.CtxWithPeerId(context.Background(), msg.SenderId)
+	//ctx = logger.CtxWithFields(ctx, zap.Uint64("msgId", msg.Id), zap.String("senderId", msg.SenderId))
+	//s.metric.RequestLog(msg.PeerCtx, "space.streamOp", msg.LogFields(
+	//	zap.Error(err),
+	//)...)
 }
 
 func (s *syncService) GetQueue(peerId string) *multiqueue.Queue[drpc.Message] {
@@ -96,10 +117,15 @@ func (s *syncService) NewReadMessage() drpc.Message {
 
 func (s *syncService) HandleMessage(ctx context.Context, peerId string, msg drpc.Message) error {
 	// TODO: make this queue per object and add closing of the individual queues
-	return s.receiveQueue.Add(ctx, peerId, msgCtx{
+	err := s.receiveQueue.Add(ctx, peerId, msgCtx{
 		ctx:     ctx,
 		Message: msg,
 	})
+	if errors.Is(err, mb.ErrOverflowed) {
+		log.Info("queue overflowed", zap.String("peerId", peerId))
+		return nil
+	}
+	return err
 }
 
 func (s *syncService) QueueRequest(ctx context.Context, rq syncdeps.Request) error {

commonspace/sync/synctest/peerglobalpool.go

@@ -64,8 +64,8 @@ func (p *PeerGlobalPool) MakePeers() {
 	}
 }
 
-func (c *PeerGlobalPool) GetPeerIds() (peerIds []string) {
-	return c.peerIds
+func (p *PeerGlobalPool) GetPeerIds() (peerIds []string) {
+	return p.peerIds
 }
 
 func (p *PeerGlobalPool) AddCtrl(peerId string, addCtrl connCtrl) {