fix typo

Component can provide optional fields for the start/stop logger

Makefile

@@ -25,7 +25,6 @@ deps:
 	go mod download
 	go build -o deps storj.io/drpc/cmd/protoc-gen-go-drpc
 	go build -o deps github.com/gogo/protobuf/protoc-gen-gogofaster
-	go build -o deps github.com/golang/mock/mockgen
 
 test:
 	go test ./... --cover

app/app.go

@@ -4,14 +4,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/anyproto/any-sync/app/logger"
-	"go.uber.org/zap"
 	"os"
 	"runtime"
 	"runtime/debug"
 	"strings"
 	"sync"
 	"time"
+
+	"github.com/anyproto/any-sync/app/logger"
+	"go.uber.org/zap"
 )
 
 var (
@@ -48,6 +49,12 @@ type ComponentRunnable interface {
 	Close(ctx context.Context) (err error)
 }
 
+type ComponentLogFieldsGetter interface {
+	// GetLogFields returns additional useful fields for logs to debug long app start/stop duration or something else in the future
+	// You don't need to provide the component name in the field's Key, because it will be added automatically
+	GetLogFields() []zap.Field
+}
+
 type ComponentStatable interface {
 	StateChange(state int)
 }
@@ -89,7 +96,7 @@ func (app *App) VersionName() string {
 	if app.versionName != "" {
 		return app.versionName
 	}
-	return AppName + ":" + GitSummary + "/any-sync:" + app.anySyncVersion
+	return AppName + ":" + GitSummary + "/any-sync:" + app.AnySyncVersion()
 }
 
 type Stat struct {
@@ -219,7 +226,7 @@ func (app *App) Start(ctx context.Context) (err error) {
 		case <-done:
 			return
 		case <-time.After(StartWarningAfter):
-			l := statLogger(app.stopStat, log).With(zap.String("in_progress", currentComponentStarting))
+			l := app.statLogger(app.startStat, log).With(zap.String("in_progress", currentComponentStarting))
 			l.Warn("components start in progress")
 		}
 	}()
@@ -254,7 +261,8 @@ func (app *App) Start(ctx context.Context) (err error) {
 	}
 
 	close(done)
-	l := statLogger(app.stopStat, log)
+	l := app.statLogger(app.startStat, log)
+
 	if app.startStat.SpentMsTotal > StartWarningAfter.Milliseconds() {
 		l.Warn("all components started")
 	}
@@ -262,6 +270,15 @@ func (app *App) Start(ctx context.Context) (err error) {
 	return
 }
 
+// IterateComponents iterates over all registered components. It's safe for concurrent use.
+func (app *App) IterateComponents(fn func(Component)) {
+	app.mu.RLock()
+	defer app.mu.RUnlock()
+	for _, s := range app.components {
+		fn(s)
+	}
+}
+
 func stackAllGoroutines() []byte {
 	buf := make([]byte, 1024)
 	for {
@@ -273,13 +290,22 @@ func stackAllGoroutines() []byte {
 	}
 }
 
-func statLogger(stat Stat, ctxLogger logger.CtxLogger) logger.CtxLogger {
+func (app *App) statLogger(stat Stat, ctxLogger logger.CtxLogger) logger.CtxLogger {
 	l := ctxLogger
 	for k, v := range stat.SpentMsPerComp {
 		l = l.With(zap.Int64(k, v))
 	}
 	l = l.With(zap.Int64("total", stat.SpentMsTotal))
 
+	for _, s := range app.components {
+		// get optional log fields from components
+		if compLog, ok := s.(ComponentLogFieldsGetter); ok {
+			for _, val := range compLog.GetLogFields() {
+				val.Key = s.Name() + "_" + val.Key
+				l = l.With(val)
+			}
+		}
+	}
 	return l
 }
 
@@ -298,7 +324,7 @@ func (app *App) Close(ctx context.Context) error {
 		case <-done:
 			return
 		case <-time.After(StopWarningAfter):
-			statLogger(app.stopStat, log).
+			app.statLogger(app.stopStat, log).
 				With(zap.String("in_progress", currentComponentStopping)).
 				Warn("components close in progress")
 		}
@@ -332,7 +358,7 @@ func (app *App) Close(ctx context.Context) error {
 		return errors.New(strings.Join(errs, "\n"))
 	}
 
-	l := statLogger(app.stopStat, log)
+	l := app.statLogger(app.stopStat, log)
 	if app.stopStat.SpentMsTotal > StopWarningAfter.Milliseconds() {
 		l.Warn("all components have been closed")
 	}

app/app_test.go

@@ -55,6 +55,21 @@ func TestAppServiceRegistry(t *testing.T) {
 	})
 }
 
+func TestApp_IterateComponents(t *testing.T) {
+	app := new(App)
+
+	app.Register(newTestService(testTypeRunnable, "c1", nil, nil))
+	app.Register(newTestService(testTypeRunnable, "r1", nil, nil))
+	app.Register(newTestService(testTypeComponent, "s1", nil, nil))
+
+	var got []string
+	app.IterateComponents(func(s Component) {
+		got = append(got, s.Name())
+	})
+
+	assert.ElementsMatch(t, []string{"c1", "r1", "s1"}, got)
+}
+
 func TestAppStart(t *testing.T) {
 	t.Run("SuccessStartStop", func(t *testing.T) {
 		app := new(App)

app/ocache/entry.go

@@ -2,9 +2,10 @@ package ocache
 
 import (
 	"context"
-	"go.uber.org/zap"
 	"sync"
 	"time"
+
+	"go.uber.org/zap"
 )
 
 type entryState int
@@ -25,6 +26,7 @@ type entry struct {
 	value     Object
 	close     chan struct{}
 	mx        sync.Mutex
+	cancel    context.CancelFunc
 }
 
 func newEntry(id string, value Object, state entryState) *entry {
@@ -49,6 +51,20 @@ func (e *entry) isClosing() bool {
 	return e.state == entryStateClosed || e.state == entryStateClosing
 }
 
+func (e *entry) setCancel(cancel context.CancelFunc) {
+	e.mx.Lock()
+	defer e.mx.Unlock()
+	e.cancel = cancel
+}
+
+func (e *entry) cancelLoad() {
+	e.mx.Lock()
+	defer e.mx.Unlock()
+	if e.cancel != nil {
+		e.cancel()
+	}
+}
+
 func (e *entry) waitLoad(ctx context.Context, id string) (value Object, err error) {
 	select {
 	case <-ctx.Done():

app/ocache/ocache.go

@@ -3,10 +3,11 @@ package ocache
 import (
 	"context"
 	"errors"
-	"github.com/anyproto/any-sync/app/logger"
-	"go.uber.org/zap"
 	"sync"
 	"time"
+
+	"github.com/anyproto/any-sync/app/logger"
+	"go.uber.org/zap"
 )
 
 var (
@@ -157,7 +158,10 @@ func (c *oCache) Pick(ctx context.Context, id string) (value Object, err error)
 
 func (c *oCache) load(ctx context.Context, id string, e *entry) {
 	defer close(e.load)
+	ctx, cancel := context.WithCancel(ctx)
+	e.setCancel(cancel)
 	value, err := c.loadFunc(ctx, id)
+	cancel()
 
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -315,6 +319,7 @@ func (c *oCache) Close() (err error) {
 	close(c.closeCh)
 	var toClose []*entry
 	for _, e := range c.data {
+		e.cancelLoad()
 		toClose = append(toClose, e)
 	}
 	c.mu.Unlock()

app/ocache/ocache_test.go

@@ -386,6 +386,25 @@ func Test_OCache_Remove(t *testing.T) {
 	})
 }
 
+func TestOCacheCancelWhenRemove(t *testing.T) {
+	c := New(func(ctx context.Context, id string) (value Object, err error) {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}, WithTTL(time.Millisecond*10))
+	stopLoad := make(chan struct{})
+	var err error
+	go func() {
+		_, err = c.Get(context.TODO(), "id")
+		stopLoad <- struct{}{}
+	}()
+	time.Sleep(time.Millisecond * 10)
+	c.Close()
+	<-stopLoad
+	require.Equal(t, context.Canceled, err)
+}
+
 func TestOCacheFuzzy(t *testing.T) {
 	t.Run("test many objects gc, get and remove simultaneously, close after", func(t *testing.T) {
 		tryCloseIds := make(map[string]bool)

commonspace/object/tree/synctree/protocolintegration_test.go

@@ -2,6 +2,10 @@ package synctree
 
 import (
 	"context"
+	"math/rand"
+	"testing"
+	"time"
+
 	"github.com/anyproto/any-sync/commonspace/object/accountdata"
 	"github.com/anyproto/any-sync/commonspace/object/acl/list"
 	"github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
@@ -10,9 +14,6 @@ import (
 	"github.com/anyproto/any-sync/util/slice"
 	"github.com/gogo/protobuf/proto"
 	"github.com/stretchr/testify/require"
-	"math/rand"
-	"testing"
-	"time"
 )
 
 func TestEmptyClientGetsFullHistory(t *testing.T) {

commonspace/object/tree/synctree/synctree.go

@@ -59,19 +59,18 @@ type ResponsiblePeersGetter interface {
 }
 
 type BuildDeps struct {
-	SpaceId            string
-	SyncClient         SyncClient
-	Configuration      nodeconf.NodeConf
-	HeadNotifiable     HeadNotifiable
-	Listener           updatelistener.UpdateListener
-	AclList            list.AclList
-	SpaceStorage       spacestorage.SpaceStorage
-	TreeStorage        treestorage.TreeStorage
-	OnClose            func(id string)
-	SyncStatus         syncstatus.StatusUpdater
-	PeerGetter         ResponsiblePeersGetter
-	BuildObjectTree    objecttree.BuildObjectTreeFunc
-	WaitTreeRemoteSync bool
+	SpaceId         string
+	SyncClient      SyncClient
+	Configuration   nodeconf.NodeConf
+	HeadNotifiable  HeadNotifiable
+	Listener        updatelistener.UpdateListener
+	AclList         list.AclList
+	SpaceStorage    spacestorage.SpaceStorage
+	TreeStorage     treestorage.TreeStorage
+	OnClose         func(id string)
+	SyncStatus      syncstatus.StatusUpdater
+	PeerGetter      ResponsiblePeersGetter
+	BuildObjectTree objecttree.BuildObjectTreeFunc
 }
 
 func BuildSyncTreeOrGetRemote(ctx context.Context, id string, deps BuildDeps) (t SyncTree, err error) {
@@ -206,18 +205,27 @@ func (s *syncTree) Delete() (err error) {
 }
 
 func (s *syncTree) TryClose(objectTTL time.Duration) (bool, error) {
-	return true, s.Close()
+	if !s.TryLock() {
+		return false, nil
+	}
+	log.Debug("closing sync tree", zap.String("id", s.Id()))
+	return true, s.close()
 }
 
 func (s *syncTree) Close() (err error) {
 	log.Debug("closing sync tree", zap.String("id", s.Id()))
+	s.Lock()
+	return s.close()
+}
+
+func (s *syncTree) close() (err error) {
+	defer s.Unlock()
 	defer func() {
 		log.Debug("closed sync tree", zap.Error(err), zap.String("id", s.Id()))
 	}()
-	s.Lock()
-	defer s.Unlock()
 	if s.isClosed {
-		return ErrSyncTreeClosed
+		err = ErrSyncTreeClosed
+		return
 	}
 	s.onClose(s.Id())
 	s.isClosed = true

commonspace/object/tree/synctree/synctreehandler.go

@@ -3,18 +3,21 @@ package synctree
 import (
 	"context"
 	"errors"
+	"sync"
+
 	"github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	"github.com/anyproto/any-sync/commonspace/objectsync/synchandler"
 	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
 	"github.com/anyproto/any-sync/commonspace/syncstatus"
+	"github.com/anyproto/any-sync/util/slice"
 	"github.com/gogo/protobuf/proto"
-	"sync"
 )
 
 var (
 	ErrMessageIsRequest    = errors.New("message is request")
 	ErrMessageIsNotRequest = errors.New("message is not request")
+	ErrMoreThanOneRequest  = errors.New("more than one request for same peer")
 )
 
 type syncTreeHandler struct {
@@ -22,21 +25,23 @@ type syncTreeHandler struct {
 	syncClient   SyncClient
 	syncProtocol TreeSyncProtocol
 	syncStatus   syncstatus.StatusUpdater
-	handlerLock  sync.Mutex
 	spaceId      string
-	queue        ReceiveQueue
+
+	handlerLock     sync.Mutex
+	pendingRequests map[string]struct{}
+	heads           []string
 }
 
 const maxQueueSize = 5
 
 func newSyncTreeHandler(spaceId string, objTree objecttree.ObjectTree, syncClient SyncClient, syncStatus syncstatus.StatusUpdater) synchandler.SyncHandler {
 	return &syncTreeHandler{
-		objTree:      objTree,
-		syncProtocol: newTreeSyncProtocol(spaceId, objTree, syncClient),
-		syncClient:   syncClient,
-		syncStatus:   syncStatus,
-		spaceId:      spaceId,
-		queue:        newReceiveQueue(maxQueueSize),
+		objTree:         objTree,
+		syncProtocol:    newTreeSyncProtocol(spaceId, objTree, syncClient),
+		syncClient:      syncClient,
+		syncStatus:      syncStatus,
+		spaceId:         spaceId,
+		pendingRequests: make(map[string]struct{}),
 	}
 }
 
@@ -48,17 +53,35 @@ func (s *syncTreeHandler) HandleRequest(ctx context.Context, senderId string, re
 	}
 	fullSyncRequest := unmarshalled.GetContent().GetFullSyncRequest()
 	if fullSyncRequest == nil {
-		err = ErrMessageIsNotRequest
-		return
+		return nil, ErrMessageIsNotRequest
 	}
-	s.syncStatus.HeadsReceive(senderId, request.ObjectId, treechangeproto.GetHeads(unmarshalled))
+	// setting pending requests
+	s.handlerLock.Lock()
+	_, exists := s.pendingRequests[senderId]
+	if exists {
+		s.handlerLock.Unlock()
+		return nil, ErrMoreThanOneRequest
+	}
+	s.pendingRequests[senderId] = struct{}{}
+	s.handlerLock.Unlock()
+
+	response, err = s.handleRequest(ctx, senderId, fullSyncRequest)
+
+	// removing pending requests
+	s.handlerLock.Lock()
+	delete(s.pendingRequests, senderId)
+	s.handlerLock.Unlock()
+	return
+}
+
+func (s *syncTreeHandler) handleRequest(ctx context.Context, senderId string, fullSyncRequest *treechangeproto.TreeFullSyncRequest) (response *spacesyncproto.ObjectSyncMessage, err error) {
 	s.objTree.Lock()
 	defer s.objTree.Unlock()
 	treeResp, err := s.syncProtocol.FullSyncRequest(ctx, senderId, fullSyncRequest)
 	if err != nil {
 		return
 	}
-	response, err = MarshallTreeMessage(treeResp, s.spaceId, request.ObjectId, "")
+	response, err = MarshallTreeMessage(treeResp, s.spaceId, s.objTree.Id(), "")
 	return
 }
 
@@ -68,28 +91,41 @@ func (s *syncTreeHandler) HandleMessage(ctx context.Context, senderId string, ms
 	if err != nil {
 		return
 	}
-	s.syncStatus.HeadsReceive(senderId, msg.ObjectId, treechangeproto.GetHeads(unmarshalled))
-
-	queueFull := s.queue.AddMessage(senderId, unmarshalled, msg.RequestId)
-	if queueFull {
+	heads := treechangeproto.GetHeads(unmarshalled)
+	s.syncStatus.HeadsReceive(senderId, msg.ObjectId, heads)
+	s.handlerLock.Lock()
+	// if the update has same heads then returning not to hang on a lock
+	if unmarshalled.GetContent().GetHeadUpdate() != nil && slice.UnsortedEquals(heads, s.heads) {
+		s.handlerLock.Unlock()
 		return
 	}
-
-	return s.handleMessage(ctx, senderId)
+	s.handlerLock.Unlock()
+	return s.handleMessage(ctx, unmarshalled, senderId)
 }
 
-func (s *syncTreeHandler) handleMessage(ctx context.Context, senderId string) (err error) {
+func (s *syncTreeHandler) handleMessage(ctx context.Context, msg *treechangeproto.TreeSyncMessage, senderId string) (err error) {
 	s.objTree.Lock()
 	defer s.objTree.Unlock()
-	msg, _, err := s.queue.GetMessage(senderId)
-	if err != nil {
-		return
-	}
+	var (
+		copyHeads = make([]string, 0, len(s.objTree.Heads()))
+		treeId    = s.objTree.Id()
+		content   = msg.GetContent()
+	)
 
-	defer s.queue.ClearQueue(senderId)
+	// getting old heads
+	copyHeads = append(copyHeads, s.objTree.Heads()...)
+	defer func() {
+		// checking if something changed
+		if !slice.UnsortedEquals(copyHeads, s.objTree.Heads()) {
+			s.handlerLock.Lock()
+			defer s.handlerLock.Unlock()
+			s.heads = s.heads[:0]
+			for _, h := range s.objTree.Heads() {
+				s.heads = append(s.heads, h)
+			}
+		}
+	}()
 
-	treeId := s.objTree.Id()
-	content := msg.GetContent()
 	switch {
 	case content.GetHeadUpdate() != nil:
 		var syncReq *treechangeproto.TreeSyncMessage

commonspace/object/tree/synctree/synctreehandler_test.go

@@ -2,15 +2,15 @@ package synctree
 
 import (
 	"context"
-	"github.com/anyproto/any-sync/commonspace/object/tree/synctree/mock_synctree"
-	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
-	"github.com/stretchr/testify/require"
 	"sync"
 	"testing"
 
 	"github.com/anyproto/any-sync/commonspace/object/tree/objecttree/mock_objecttree"
+	"github.com/anyproto/any-sync/commonspace/object/tree/synctree/mock_synctree"
+	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	"github.com/anyproto/any-sync/commonspace/syncstatus"
 	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/require"
 )
 
 type testObjTreeMock struct {
@@ -52,7 +52,6 @@ type syncHandlerFixture struct {
 	ctrl             *gomock.Controller
 	syncClientMock   *mock_synctree.MockSyncClient
 	objectTreeMock   *testObjTreeMock
-	receiveQueueMock ReceiveQueue
 	syncProtocolMock *mock_synctree.MockTreeSyncProtocol
 	spaceId          string
 	senderId         string
@@ -67,20 +66,18 @@ func newSyncHandlerFixture(t *testing.T) *syncHandlerFixture {
 	syncClientMock := mock_synctree.NewMockSyncClient(ctrl)
 	syncProtocolMock := mock_synctree.NewMockTreeSyncProtocol(ctrl)
 	spaceId := "spaceId"
-	receiveQueue := newReceiveQueue(5)
 
 	syncHandler := &syncTreeHandler{
-		objTree:      objectTreeMock,
-		syncClient:   syncClientMock,
-		syncProtocol: syncProtocolMock,
-		spaceId:      spaceId,
-		queue:        receiveQueue,
-		syncStatus:   syncstatus.NewNoOpSyncStatus(),
+		objTree:         objectTreeMock,
+		syncClient:      syncClientMock,
+		syncProtocol:    syncProtocolMock,
+		spaceId:         spaceId,
+		syncStatus:      syncstatus.NewNoOpSyncStatus(),
+		pendingRequests: map[string]struct{}{},
 	}
 	return &syncHandlerFixture{
 		ctrl:             ctrl,
 		objectTreeMock:   objectTreeMock,
-		receiveQueueMock: receiveQueue,
 		syncProtocolMock: syncProtocolMock,
 		syncClientMock:   syncClientMock,
 		syncHandler:      syncHandler,
@@ -97,38 +94,68 @@ func (fx *syncHandlerFixture) stop() {
 func TestSyncTreeHandler_HandleMessage(t *testing.T) {
 	ctx := context.Background()
 
-	t.Run("handle head update message", func(t *testing.T) {
+	t.Run("handle head update message, heads not equal, request returned", func(t *testing.T) {
 		fx := newSyncHandlerFixture(t)
 		defer fx.stop()
 		treeId := "treeId"
 		chWithId := &treechangeproto.RawTreeChangeWithId{}
-		headUpdate := &treechangeproto.TreeHeadUpdate{}
+		headUpdate := &treechangeproto.TreeHeadUpdate{
+			Heads: []string{"h3"},
+		}
 		treeMsg := treechangeproto.WrapHeadUpdate(headUpdate, chWithId)
 		objectMsg, _ := MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
 
 		syncReq := &treechangeproto.TreeSyncMessage{}
+		fx.syncHandler.heads = []string{"h2"}
 		fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(fx.treeId)
+		fx.objectTreeMock.EXPECT().Heads().Times(2).Return([]string{"h2"})
+		fx.objectTreeMock.EXPECT().Heads().Times(2).Return([]string{"h3"})
 		fx.syncProtocolMock.EXPECT().HeadUpdate(ctx, fx.senderId, gomock.Any()).Return(syncReq, nil)
 		fx.syncClientMock.EXPECT().QueueRequest(fx.senderId, fx.treeId, syncReq).Return(nil)
 
 		err := fx.syncHandler.HandleMessage(ctx, fx.senderId, objectMsg)
 		require.NoError(t, err)
+		require.Equal(t, []string{"h3"}, fx.syncHandler.heads)
 	})
 
-	t.Run("handle head update message, empty sync request", func(t *testing.T) {
+	t.Run("handle head update message, heads equal", func(t *testing.T) {
 		fx := newSyncHandlerFixture(t)
 		defer fx.stop()
 		treeId := "treeId"
 		chWithId := &treechangeproto.RawTreeChangeWithId{}
-		headUpdate := &treechangeproto.TreeHeadUpdate{}
+		headUpdate := &treechangeproto.TreeHeadUpdate{
+			Heads: []string{"h1"},
+		}
 		treeMsg := treechangeproto.WrapHeadUpdate(headUpdate, chWithId)
 		objectMsg, _ := MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
 
+		fx.syncHandler.heads = []string{"h1"}
 		fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(fx.treeId)
+
+		err := fx.syncHandler.HandleMessage(ctx, fx.senderId, objectMsg)
+		require.NoError(t, err)
+	})
+
+	t.Run("handle head update message, empty sync request returned", func(t *testing.T) {
+		fx := newSyncHandlerFixture(t)
+		defer fx.stop()
+		treeId := "treeId"
+		chWithId := &treechangeproto.RawTreeChangeWithId{}
+		headUpdate := &treechangeproto.TreeHeadUpdate{
+			Heads: []string{"h3"},
+		}
+		treeMsg := treechangeproto.WrapHeadUpdate(headUpdate, chWithId)
+		objectMsg, _ := MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
+
+		fx.syncHandler.heads = []string{"h2"}
+		fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(fx.treeId)
+		fx.objectTreeMock.EXPECT().Heads().Times(2).Return([]string{"h2"})
+		fx.objectTreeMock.EXPECT().Heads().Times(2).Return([]string{"h3"})
 		fx.syncProtocolMock.EXPECT().HeadUpdate(ctx, fx.senderId, gomock.Any()).Return(nil, nil)
 
 		err := fx.syncHandler.HandleMessage(ctx, fx.senderId, objectMsg)
 		require.NoError(t, err)
+		require.Equal(t, []string{"h3"}, fx.syncHandler.heads)
 	})
 
 	t.Run("handle full sync request returns error", func(t *testing.T) {
@@ -136,11 +163,15 @@ func TestSyncTreeHandler_HandleMessage(t *testing.T) {
 		defer fx.stop()
 		treeId := "treeId"
 		chWithId := &treechangeproto.RawTreeChangeWithId{}
-		fullRequest := &treechangeproto.TreeFullSyncRequest{}
+		fullRequest := &treechangeproto.TreeFullSyncRequest{
+			Heads: []string{"h3"},
+		}
 		treeMsg := treechangeproto.WrapFullRequest(fullRequest, chWithId)
 		objectMsg, _ := MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
 
+		fx.syncHandler.heads = []string{"h2"}
 		fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(fx.treeId)
+		fx.objectTreeMock.EXPECT().Heads().Times(3).Return([]string{"h2"})
 
 		err := fx.syncHandler.HandleMessage(ctx, fx.senderId, objectMsg)
 		require.Equal(t, err, ErrMessageIsRequest)
@@ -151,11 +182,16 @@ func TestSyncTreeHandler_HandleMessage(t *testing.T) {
 		defer fx.stop()
 		treeId := "treeId"
 		chWithId := &treechangeproto.RawTreeChangeWithId{}
-		fullSyncResponse := &treechangeproto.TreeFullSyncResponse{}
+		fullSyncResponse := &treechangeproto.TreeFullSyncResponse{
+			Heads: []string{"h3"},
+		}
 		treeMsg := treechangeproto.WrapFullResponse(fullSyncResponse, chWithId)
 		objectMsg, _ := MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
 
+		fx.syncHandler.heads = []string{"h2"}
 		fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(fx.treeId)
+		fx.objectTreeMock.EXPECT().Heads().Times(2).Return([]string{"h2"})
+		fx.objectTreeMock.EXPECT().Heads().Times(2).Return([]string{"h3"})
 		fx.syncProtocolMock.EXPECT().FullSyncResponse(ctx, fx.senderId, gomock.Any()).Return(nil)
 
 		err := fx.syncHandler.HandleMessage(ctx, fx.senderId, objectMsg)
@@ -184,24 +220,6 @@ func TestSyncTreeHandler_HandleRequest(t *testing.T) {
 		require.NotNil(t, res)
 	})
 
-	t.Run("handle request", func(t *testing.T) {
-		fx := newSyncHandlerFixture(t)
-		defer fx.stop()
-		treeId := "treeId"
-		chWithId := &treechangeproto.RawTreeChangeWithId{}
-		fullRequest := &treechangeproto.TreeFullSyncRequest{}
-		treeMsg := treechangeproto.WrapFullRequest(fullRequest, chWithId)
-		objectMsg, _ := MarshallTreeMessage(treeMsg, "spaceId", treeId, "")
-
-		syncResp := &treechangeproto.TreeSyncMessage{}
-		fx.objectTreeMock.EXPECT().Id().AnyTimes().Return(fx.treeId)
-		fx.syncProtocolMock.EXPECT().FullSyncRequest(ctx, fx.senderId, gomock.Any()).Return(syncResp, nil)
-
-		res, err := fx.syncHandler.HandleRequest(ctx, fx.senderId, objectMsg)
-		require.NoError(t, err)
-		require.NotNil(t, res)
-	})
-
 	t.Run("handle other message", func(t *testing.T) {
 		fx := newSyncHandlerFixture(t)
 		defer fx.stop()

commonspace/object/tree/synctree/treeremotegetter.go

@@ -2,16 +2,19 @@ package synctree
 
 import (
 	"context"
-	"fmt"
+	"errors"
+
 	"github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
 	"github.com/anyproto/any-sync/commonspace/spacestorage"
 	"github.com/anyproto/any-sync/net/peer"
-	"github.com/anyproto/any-sync/net/rpc/rpcerr"
 	"github.com/gogo/protobuf/proto"
 	"go.uber.org/zap"
-	"time"
+)
+
+var (
+	ErrNoResponsiblePeers = errors.New("no responsible peers")
 )
 
 type treeRemoteGetter struct {
@@ -36,7 +39,7 @@ func (t treeRemoteGetter) getPeers(ctx context.Context) (peerIds []string, err e
 		return
 	}
 	if len(respPeers) == 0 {
-		err = fmt.Errorf("no responsible peers")
+		err = ErrNoResponsiblePeers
 		return
 	}
 	for _, p := range respPeers {
@@ -57,37 +60,13 @@ func (t treeRemoteGetter) treeRequest(ctx context.Context, peerId string) (msg *
 	return
 }
 
-func (t treeRemoteGetter) treeRequestLoop(ctx context.Context, wait bool) (msg *treechangeproto.TreeSyncMessage, err error) {
-	peerIdx := 0
-Loop:
-	for {
-		select {
-		case <-ctx.Done():
-			return nil, fmt.Errorf("waiting for object %s interrupted, context closed", t.treeId)
-		default:
-			break
-		}
-		availablePeers, err := t.getPeers(ctx)
-		if err != nil {
-			if !wait {
-				return nil, err
-			}
-			select {
-			// wait for peers to connect
-			case <-time.After(1 * time.Second):
-				continue Loop
-			case <-ctx.Done():
-				return nil, fmt.Errorf("waiting for object %s interrupted, context closed", t.treeId)
-			}
-		}
-
-		peerIdx = peerIdx % len(availablePeers)
-		msg, err = t.treeRequest(ctx, availablePeers[peerIdx])
-		if err == nil || !wait {
-			return msg, err
-		}
-		peerIdx++
+func (t treeRemoteGetter) treeRequestLoop(ctx context.Context) (msg *treechangeproto.TreeSyncMessage, err error) {
+	availablePeers, err := t.getPeers(ctx)
+	if err != nil {
+		return
 	}
+	// in future we will try to load from different peers
+	return t.treeRequest(ctx, availablePeers[0])
 }
 
 func (t treeRemoteGetter) getTree(ctx context.Context) (treeStorage treestorage.TreeStorage, isRemote bool, err error) {
@@ -109,22 +88,15 @@ func (t treeRemoteGetter) getTree(ctx context.Context) (treeStorage treestorage.
 	}
 
 	isRemote = true
-	resp, err := t.treeRequestLoop(ctx, t.deps.WaitTreeRemoteSync)
+	resp, err := t.treeRequestLoop(ctx)
 	if err != nil {
 		return
 	}
-	switch {
-	case resp.GetContent().GetErrorResponse() != nil:
-		errResp := resp.GetContent().GetErrorResponse()
-		err = rpcerr.Err(errResp.ErrCode)
-		return
-	case resp.GetContent().GetFullSyncResponse() == nil:
+	fullSyncResp := resp.GetContent().GetFullSyncResponse()
+	if fullSyncResp == nil {
 		err = treechangeproto.ErrUnexpected
 		return
-	default:
-		break
 	}
-	fullSyncResp := resp.GetContent().GetFullSyncResponse()
 
 	payload := treestorage.TreeStorageCreatePayload{
 		RootRawChange: resp.RootChange,

commonspace/object/tree/synctree/treeremotegetter_test.go

@@ -0,0 +1,87 @@
+package synctree
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/anyproto/any-sync/commonspace/object/tree/synctree/mock_synctree"
+	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
+	"github.com/anyproto/any-sync/commonspace/peermanager/mock_peermanager"
+	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
+	"github.com/anyproto/any-sync/net/peer"
+	"github.com/anyproto/any-sync/net/peer/mock_peer"
+	"github.com/gogo/protobuf/proto"
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/require"
+)
+
+type treeRemoteGetterFixture struct {
+	ctrl *gomock.Controller
+
+	treeGetter     treeRemoteGetter
+	syncClientMock *mock_synctree.MockSyncClient
+	peerGetterMock *mock_peermanager.MockPeerManager
+}
+
+func newTreeRemoteGetterFixture(t *testing.T) *treeRemoteGetterFixture {
+	ctrl := gomock.NewController(t)
+	syncClientMock := mock_synctree.NewMockSyncClient(ctrl)
+	peerGetterMock := mock_peermanager.NewMockPeerManager(ctrl)
+	treeGetter := treeRemoteGetter{
+		deps: BuildDeps{
+			SyncClient: syncClientMock,
+			PeerGetter: peerGetterMock,
+		},
+		treeId: "treeId",
+	}
+	return &treeRemoteGetterFixture{
+		ctrl:           ctrl,
+		treeGetter:     treeGetter,
+		syncClientMock: syncClientMock,
+		peerGetterMock: peerGetterMock,
+	}
+}
+
+func (fx *treeRemoteGetterFixture) stop() {
+	fx.ctrl.Finish()
+}
+
+func TestTreeRemoteGetter(t *testing.T) {
+	ctx := context.Background()
+	peerId := "peerId"
+	treeRequest := &treechangeproto.TreeSyncMessage{}
+	treeResponse := &treechangeproto.TreeSyncMessage{
+		RootChange: &treechangeproto.RawTreeChangeWithId{Id: "id"},
+	}
+	marshalled, _ := proto.Marshal(treeResponse)
+	objectResponse := &spacesyncproto.ObjectSyncMessage{
+		Payload: marshalled,
+	}
+
+	t.Run("request works", func(t *testing.T) {
+		fx := newTreeRemoteGetterFixture(t)
+		defer fx.stop()
+		mockPeer := mock_peer.NewMockPeer(fx.ctrl)
+		mockPeer.EXPECT().Id().AnyTimes().Return(peerId)
+		fx.peerGetterMock.EXPECT().GetResponsiblePeers(ctx).Return([]peer.Peer{mockPeer}, nil)
+		fx.syncClientMock.EXPECT().CreateNewTreeRequest().Return(treeRequest)
+		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).Return(objectResponse, nil)
+		resp, err := fx.treeGetter.treeRequestLoop(ctx)
+		require.NoError(t, err)
+		require.Equal(t, "id", resp.RootChange.Id)
+	})
+
+	t.Run("request fails", func(t *testing.T) {
+		fx := newTreeRemoteGetterFixture(t)
+		defer fx.stop()
+		treeRequest := &treechangeproto.TreeSyncMessage{}
+		mockPeer := mock_peer.NewMockPeer(fx.ctrl)
+		mockPeer.EXPECT().Id().AnyTimes().Return(peerId)
+		fx.peerGetterMock.EXPECT().GetResponsiblePeers(ctx).Return([]peer.Peer{mockPeer}, nil)
+		fx.syncClientMock.EXPECT().CreateNewTreeRequest().Return(treeRequest)
+		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).AnyTimes().Return(nil, fmt.Errorf("some"))
+		_, err := fx.treeGetter.treeRequestLoop(ctx)
+		require.Error(t, err)
+	})
+}

commonspace/object/tree/synctree/treesyncprotocol.go

@@ -2,6 +2,7 @@ package synctree
 
 import (
 	"context"
+
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"

commonspace/objectsync/objectsync.go

@@ -4,6 +4,9 @@ package objectsync
 import (
 	"context"
 	"fmt"
+	"sync/atomic"
+	"time"
+
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	"github.com/anyproto/any-sync/commonspace/object/treemanager"
@@ -13,8 +16,6 @@ import (
 	"github.com/anyproto/any-sync/util/multiqueue"
 	"github.com/cheggaaa/mb/v3"
 	"github.com/gogo/protobuf/proto"
-	"sync/atomic"
-	"time"
 
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/commonspace/object/syncobjectgetter"
@@ -79,7 +80,7 @@ func (s *objectSync) Init(a *app.App) (err error) {
 	}
 	s.spaceIsDeleted = sharedData.SpaceIsDeleted
 	s.spaceId = sharedData.SpaceId
-	s.handleQueue = multiqueue.New[HandleMessage](s.processHandleMessage, 100)
+	s.handleQueue = multiqueue.New[HandleMessage](s.processHandleMessage, 30)
 	return nil
 }
 

commonspace/objecttreebuilder/treebuilder.go

@@ -4,6 +4,8 @@ package objecttreebuilder
 import (
 	"context"
 	"errors"
+	"sync/atomic"
+
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/commonspace/headsync"
@@ -22,13 +24,11 @@ import (
 	"github.com/anyproto/any-sync/commonspace/syncstatus"
 	"github.com/anyproto/any-sync/nodeconf"
 	"go.uber.org/zap"
-	"sync/atomic"
 )
 
 type BuildTreeOpts struct {
-	Listener           updatelistener.UpdateListener
-	WaitTreeRemoteSync bool
-	TreeBuilder        objecttree.BuildObjectTreeFunc
+	Listener    updatelistener.UpdateListener
+	TreeBuilder objecttree.BuildObjectTreeFunc
 }
 
 const CName = "common.commonspace.objecttreebuilder"
@@ -110,18 +110,17 @@ func (t *treeBuilder) BuildTree(ctx context.Context, id string, opts BuildTreeOp
 		treeBuilder = t.builder
 	}
 	deps := synctree.BuildDeps{
-		SpaceId:            t.spaceId,
-		SyncClient:         t.syncClient,
-		Configuration:      t.configuration,
-		HeadNotifiable:     t.headsNotifiable,
-		Listener:           opts.Listener,
-		AclList:            t.aclList,
-		SpaceStorage:       t.spaceStorage,
-		OnClose:            t.onClose,
-		SyncStatus:         t.syncStatus,
-		WaitTreeRemoteSync: opts.WaitTreeRemoteSync,
-		PeerGetter:         t.peerManager,
-		BuildObjectTree:    treeBuilder,
+		SpaceId:         t.spaceId,
+		SyncClient:      t.syncClient,
+		Configuration:   t.configuration,
+		HeadNotifiable:  t.headsNotifiable,
+		Listener:        opts.Listener,
+		AclList:         t.aclList,
+		SpaceStorage:    t.spaceStorage,
+		OnClose:         t.onClose,
+		SyncStatus:      t.syncStatus,
+		PeerGetter:      t.peerManager,
+		BuildObjectTree: treeBuilder,
 	}
 	t.treesUsed.Add(1)
 	t.log.Debug("incrementing counter", zap.String("id", id), zap.Int32("trees", t.treesUsed.Load()))

commonspace/requestmanager/requestmanager.go

@@ -2,16 +2,18 @@ package requestmanager
 
 import (
 	"context"
+	"sync"
+
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/commonspace/objectsync"
 	"github.com/anyproto/any-sync/commonspace/spacesyncproto"
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/pool"
+	"github.com/anyproto/any-sync/net/rpc/rpcerr"
 	"github.com/anyproto/any-sync/net/streampool"
 	"go.uber.org/zap"
 	"storj.io/drpc"
-	"sync"
 )
 
 const CName = "common.commonspace.requestmanager"
@@ -122,5 +124,6 @@ func (r *requestManager) doRequest(ctx context.Context, peerId string, msg *spac
 		resp, err = cl.ObjectSync(ctx, msg)
 		return err
 	})
+	err = rpcerr.Unwrap(err)
 	return
 }

commonspace/settings/settings.go

@@ -2,6 +2,8 @@ package settings
 
 import (
 	"context"
+	"sync/atomic"
+
 	"github.com/anyproto/any-sync/accountservice"
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/commonspace/deletionstate"
@@ -16,7 +18,6 @@ import (
 	"github.com/anyproto/any-sync/commonspace/spacestorage"
 	"github.com/anyproto/any-sync/nodeconf"
 	"go.uber.org/zap"
-	"sync/atomic"
 )
 
 const CName = "common.commonspace.settings"
@@ -61,8 +62,7 @@ func (s *settings) Init(a *app.App) (err error) {
 	deps := Deps{
 		BuildFunc: func(ctx context.Context, id string, listener updatelistener.UpdateListener) (t synctree.SyncTree, err error) {
 			res, err := s.treeBuilder.BuildTree(ctx, id, objecttreebuilder.BuildTreeOpts{
-				Listener:           listener,
-				WaitTreeRemoteSync: false,
+				Listener: listener,
 				// space settings document should not have empty data
 				TreeBuilder: objecttree.BuildObjectTree,
 			})

go.mod

@@ -25,26 +25,25 @@ require (
 	github.com/ipfs/go-ipld-format v0.5.0
 	github.com/ipfs/go-merkledag v0.11.0
 	github.com/ipfs/go-unixfs v0.4.6
-	github.com/libp2p/go-libp2p v0.27.5
+	github.com/libp2p/go-libp2p v0.28.0
 	github.com/mr-tron/base58 v1.2.0
 	github.com/multiformats/go-multibase v0.2.0
 	github.com/multiformats/go-multihash v0.2.3
-	github.com/prometheus/client_golang v1.15.1
+	github.com/prometheus/client_golang v1.16.0
 	github.com/stretchr/testify v1.8.4
 	github.com/tyler-smith/go-bip39 v1.1.0
 	github.com/zeebo/blake3 v0.2.3
 	go.uber.org/atomic v1.11.0
 	go.uber.org/zap v1.24.0
-	golang.org/x/crypto v0.9.0
+	golang.org/x/crypto v0.10.0
 	golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1
-	golang.org/x/net v0.10.0
+	golang.org/x/net v0.11.0
 	gopkg.in/yaml.v3 v3.0.1
 	storj.io/drpc v0.0.33
 )
 
 require (
 	github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137 // indirect
-	github.com/benbjohnson/clock v1.3.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/crackcomm/go-gitignore v0.0.0-20170627025303-887ab5e44cc3 // indirect
@@ -56,9 +55,7 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
-	github.com/google/pprof v0.0.0-20230602150820-91b7bce49751 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
-	github.com/huin/goupnp v1.2.0 // indirect
 	github.com/ipfs/bbloom v0.0.4 // indirect
 	github.com/ipfs/go-bitfield v1.1.0 // indirect
 	github.com/ipfs/go-datastore v0.6.0 // indirect
@@ -75,29 +72,25 @@ require (
 	github.com/ipld/go-ipld-prime v0.20.0 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
 	github.com/jbenet/goprocess v0.1.4 // indirect
-	github.com/klauspost/compress v1.16.5 // indirect
-	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.5 // indirect
 	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
 	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
-	github.com/miekg/dns v1.1.54 // indirect
-	github.com/minio/sha256-simd v1.0.0 // indirect
+	github.com/minio/sha256-simd v1.0.1 // indirect
 	github.com/multiformats/go-base32 v0.1.0 // indirect
 	github.com/multiformats/go-base36 v0.2.0 // indirect
 	github.com/multiformats/go-multiaddr v0.9.0 // indirect
 	github.com/multiformats/go-multicodec v0.9.0 // indirect
 	github.com/multiformats/go-multistream v0.4.1 // indirect
 	github.com/multiformats/go-varint v0.0.7 // indirect
-	github.com/onsi/ginkgo/v2 v2.9.7 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/polydawn/refmt v0.0.0-20201211092308-30ac6d18308e // indirect
 	github.com/prometheus/client_model v0.4.0 // indirect
 	github.com/prometheus/common v0.44.0 // indirect
-	github.com/prometheus/procfs v0.10.0 // indirect
+	github.com/prometheus/procfs v0.10.1 // indirect
 	github.com/quic-go/quic-go v0.35.1 // indirect
-	github.com/quic-go/webtransport-go v0.5.3 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f // indirect
 	github.com/zeebo/errs v1.3.0 // indirect
@@ -106,7 +99,7 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/image v0.6.0 // indirect
 	golang.org/x/sync v0.2.0 // indirect
-	golang.org/x/sys v0.8.0 // indirect
+	golang.org/x/sys v0.9.0 // indirect
 	golang.org/x/tools v0.9.3 // indirect
 	google.golang.org/protobuf v1.30.0 // indirect
 	lukechampine.com/blake3 v1.2.1 // indirect

go.sum

@@ -13,7 +13,6 @@ github.com/anyproto/go-slip21 v1.0.0 h1:CI7lUqTIwmPOEGVAj4jyNLoICvueh++0U2HoAi3m
 github.com/anyproto/go-slip21 v1.0.0/go.mod h1:gbIJt7HAdr5DuT4f2pFTKCBSUWYsm/fysHBNqgsuxT0=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/benbjohnson/clock v1.3.5 h1:VvXlSJBzZpA/zum6Sj74hxwYI2DIxRWuNIoXAzHZz5o=
-github.com/benbjohnson/clock v1.3.5/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
@@ -65,7 +64,6 @@ github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
 github.com/google/pprof v0.0.0-20230602150820-91b7bce49751 h1:hR7/MlvK23p6+lIw9SN1TigNLn9ZnF3W4SYRKq2gAHs=
-github.com/google/pprof v0.0.0-20230602150820-91b7bce49751/go.mod h1:Jh3hGz2jkYak8qXPD19ryItVnUgpgeqzdkY/D0EaeuA=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
@@ -82,7 +80,6 @@ github.com/huandu/go-assert v1.1.5/go.mod h1:yOLvuqZwmcHIC5rIzrBhT7D3Q9c3GFnd0Jr
 github.com/huandu/skiplist v1.2.0 h1:gox56QD77HzSC0w+Ws3MH3iie755GBJU1OER3h5VsYw=
 github.com/huandu/skiplist v1.2.0/go.mod h1:7v3iFjLcSAzO4fN5B8dvebvo/qsfumiLiDXMrPiHF9w=
 github.com/huin/goupnp v1.2.0 h1:uOKW26NG1hsSSbXIZ1IR7XP9Gjd1U8pnLaCMgntmkmY=
-github.com/huin/goupnp v1.2.0/go.mod h1:gnGPsThkYa7bFi/KWmEysQRf48l2dvR5bxr2OFckNX8=
 github.com/ipfs/bbloom v0.0.4 h1:Gi+8EGJ2y5qiD5FbsbpX/TMNcJw8gSqr7eyjHa4Fhvs=
 github.com/ipfs/bbloom v0.0.4/go.mod h1:cS9YprKXpoZ9lT0n/Mw/a6/aFV6DTjTLYHeA+gyqMG0=
 github.com/ipfs/go-bitfield v1.1.0 h1:fh7FIo8bSwaJEh6DdTWbCeZ1eqOaOkKFI74SCnsWbGA=
@@ -156,11 +153,9 @@ github.com/jtolds/gls v4.2.1+incompatible h1:fSuqC+Gmlu6l/ZYAoZzx2pyucC8Xza35fpR
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.16.5 h1:IFV2oUNUzZaz+XyusxpLzpzS8Pt5rh0Z16For/djlyI=
-github.com/klauspost/compress v1.16.5/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
-github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.12/go.mod h1:g2LTdtYhdyuGPqyWyv7qRAmj1WBqxuObKfj5c0PQa7c=
-github.com/klauspost/cpuid/v2 v2.2.4 h1:acbojRNwl3o09bUq+yDCtZFc1aiwaAAxtcn8YkZXnvk=
-github.com/klauspost/cpuid/v2 v2.2.4/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY=
+github.com/klauspost/cpuid/v2 v2.2.5 h1:0E5MSMDEoAulmXNFquVs//DdoomxaoTY1kUhbc/qbZg=
+github.com/klauspost/cpuid/v2 v2.2.5/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
 github.com/koron/go-ssdp v0.0.4 h1:1IDwrghSKYM7yLf7XCzbByg2sJ/JcNOZRXS2jczTwz0=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
@@ -171,13 +166,13 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/libp2p/go-buffer-pool v0.1.0 h1:oK4mSFcQz7cTQIfqbe4MIj9gLW+mnanjyFtc6cdF0Y8=
 github.com/libp2p/go-buffer-pool v0.1.0/go.mod h1:N+vh8gMqimBzdKkSMVuydVDq+UV5QTWy5HSiZacSbPg=
 github.com/libp2p/go-cidranger v1.1.0 h1:ewPN8EZ0dd1LSnrtuwd4709PXVcITVeuwbag38yPW7c=
-github.com/libp2p/go-libp2p v0.27.5 h1:KwA7pXKXpz8hG6Cr1fMA7UkgleogcwQj0sxl5qquWRg=
-github.com/libp2p/go-libp2p v0.27.5/go.mod h1:oMfQGTb9CHnrOuSM6yMmyK2lXz3qIhnkn2+oK3B1Y2g=
+github.com/libp2p/go-libp2p v0.28.0 h1:zO8cY98nJiPzZpFv5w5gqqb8aVzt4ukQ0nVOSaaKhJ8=
+github.com/libp2p/go-libp2p v0.28.0/go.mod h1:s3Xabc9LSwOcnv9UD4nORnXKTsWkPMkIMB/JIGXVnzk=
 github.com/libp2p/go-libp2p-asn-util v0.3.0 h1:gMDcMyYiZKkocGXDQ5nsUQyquC9+H+iLEQHwOCZ7s8s=
 github.com/libp2p/go-libp2p-record v0.2.0 h1:oiNUOCWno2BFuxt3my4i1frNrt7PerzB3queqa1NkQ0=
 github.com/libp2p/go-libp2p-testing v0.12.0 h1:EPvBb4kKMWO29qP4mZGyhVzUyR25dvfUIK5WDu6iPUA=
 github.com/libp2p/go-msgio v0.3.0 h1:mf3Z8B1xcFN314sWX+2vOTShIE0Mmn2TXn3YCUQGNj0=
-github.com/libp2p/go-nat v0.1.0 h1:MfVsH6DLcpa04Xr+p8hmVRG4juse0s3J8HyNWYHffXg=
+github.com/libp2p/go-nat v0.2.0 h1:Tyz+bUFAYqGyJ/ppPPymMGbIgNRH+WqC5QrT5fKrrGk=
 github.com/libp2p/go-netroute v0.2.1 h1:V8kVrpD8GK0Riv15/7VN6RbUQ3URNZVosw7H2v9tksU=
 github.com/libp2p/go-yamux/v4 v4.0.0 h1:+Y80dV2Yx/kv7Y7JKu0LECyVdMXm1VUoko+VQ9rBfZQ=
 github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
@@ -186,12 +181,11 @@ github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D
 github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
 github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
 github.com/miekg/dns v1.1.54 h1:5jon9mWcb0sFJGpnI99tOMhCPyJ+RPVz5b63MQG0VWI=
-github.com/miekg/dns v1.1.54/go.mod h1:uInx36IzPl7FYnDcMeVWxj9byh7DutNykX4G9Sj60FY=
 github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1/go.mod h1:pD8RvIylQ358TN4wwqatJ8rNavkEINozVn9DtGI3dfQ=
 github.com/minio/sha256-simd v0.0.0-20190131020904-2d45a736cd16/go.mod h1:2FMWW+8GMoPweT6+pI63m9YE3Lmw4J71hV56Chs1E/U=
 github.com/minio/sha256-simd v0.1.1-0.20190913151208-6de447530771/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=
-github.com/minio/sha256-simd v1.0.0 h1:v1ta+49hkWZyvaKwrQB8elexRqm6Y0aMLjCNsrYxo6g=
-github.com/minio/sha256-simd v1.0.0/go.mod h1:OuYzVNI5vcoYIAmbIvHPl3N3jUzVedXbKy5RFepssQM=
+github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM=
+github.com/minio/sha256-simd v1.0.1/go.mod h1:Pz6AKMiUdngCLpeTL/RJY1M9rUuPMYujV5xJjtbRSN8=
 github.com/mr-tron/base58 v1.1.0/go.mod h1:xcD2VGqlgYjBdcBLw+TuYLr8afG+Hj8g2eTVqeSzSU8=
 github.com/mr-tron/base58 v1.1.3/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
 github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
@@ -221,7 +215,6 @@ github.com/multiformats/go-varint v0.0.7 h1:sWSGR+f/eu5ABZA2ZpYKBILXTTs9JWpdEM/n
 github.com/multiformats/go-varint v0.0.7/go.mod h1:r8PUYw/fD/SjBCiKOoDlGF6QawOELpZAu9eioSos/OU=
 github.com/nfnt/resize v0.0.0-20160724205520-891127d8d1b5 h1:BvoENQQU+fZ9uukda/RzCAL/191HHwJA5b13R6diVlY=
 github.com/onsi/ginkgo/v2 v2.9.7 h1:06xGQy5www2oN160RtEZoTvnP2sPhEfePYmCDc2szss=
-github.com/onsi/ginkgo/v2 v2.9.7/go.mod h1:cxrmXWykAwTwhQsJOPfdIDiJ+l2RYq7U8hFU+M/1uw0=
 github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+1B0VhjKrZUs=
 github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -231,21 +224,20 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/polydawn/refmt v0.0.0-20201211092308-30ac6d18308e h1:ZOcivgkkFRnjfoTcGsDq3UQYiBmekwLA+qg0OjyB/ls=
 github.com/polydawn/refmt v0.0.0-20201211092308-30ac6d18308e/go.mod h1:uIp+gprXxxrWSjjklXD+mN4wed/tMfjMMmN/9+JsA9o=
-github.com/prometheus/client_golang v1.15.1 h1:8tXpTmJbyH5lydzFPoxSIJ0J46jdh3tylbvM1xCv0LI=
-github.com/prometheus/client_golang v1.15.1/go.mod h1:e9yaBhRPU2pPNsZwE+JdQl0KEt1N9XgF6zxWmaC0xOk=
+github.com/prometheus/client_golang v1.16.0 h1:yk/hx9hDbrGHovbci4BY+pRMfSuuat626eFsHb7tmT8=
+github.com/prometheus/client_golang v1.16.0/go.mod h1:Zsulrv/L9oM40tJ7T815tM89lFEugiJ9HzIqaAx4LKc=
 github.com/prometheus/client_model v0.4.0 h1:5lQXD3cAg1OXBf4Wq03gTrXHeaV0TQvGfUooCfx1yqY=
 github.com/prometheus/client_model v0.4.0/go.mod h1:oMQmHW1/JoDwqLtg57MGgP/Fb1CJEYF2imWWhWtMkYU=
 github.com/prometheus/common v0.44.0 h1:+5BrQJwiBB9xsMygAB3TNvpQKOwlkc25LbISbrdOOfY=
 github.com/prometheus/common v0.44.0/go.mod h1:ofAIvZbQ1e/nugmZGz4/qCb9Ap1VoSTIO7x0VV9VvuY=
-github.com/prometheus/procfs v0.10.0 h1:UkG7GPYkO4UZyLnyXjaWYcgOSONqwdBqFUT95ugmt6I=
-github.com/prometheus/procfs v0.10.0/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
+github.com/prometheus/procfs v0.10.1 h1:kYK1Va/YMlutzCGazswoHKo//tZVlFpKYh+PymziUAg=
+github.com/prometheus/procfs v0.10.1/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
 github.com/quic-go/qpack v0.4.0 h1:Cr9BXA1sQS2SmDUWjSofMPNKmvF6IiIfDRmgU0w1ZCo=
 github.com/quic-go/qtls-go1-19 v0.3.2 h1:tFxjCFcTQzK+oMxG6Zcvp4Dq8dx4yD3dDiIiyc86Z5U=
 github.com/quic-go/qtls-go1-20 v0.2.2 h1:WLOPx6OY/hxtTxKV1Zrq20FtXtDEkeY00CGQm8GEa3E=
 github.com/quic-go/quic-go v0.35.1 h1:b0kzj6b/cQAf05cT0CkQubHM31wiA+xH3IBkxP62poo=
 github.com/quic-go/quic-go v0.35.1/go.mod h1:+4CVgVppm0FNjpG3UcX8Joi/frKOH7/ciD5yGcwOO1g=
 github.com/quic-go/webtransport-go v0.5.3 h1:5XMlzemqB4qmOlgIus5zB45AcZ2kCgCy2EptUrfOPWU=
-github.com/quic-go/webtransport-go v0.5.3/go.mod h1:OhmmgJIzTTqXK5xvtuX0oBpLV2GkLWNDA+UeTGJXErU=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM=
@@ -305,8 +297,8 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200602180216-279210d13fed/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
-golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0=
+golang.org/x/crypto v0.10.0 h1:LKqV2xt9+kDzSTfOhx4FrkEBcMrAgHSYgzywV9zcGmM=
+golang.org/x/crypto v0.10.0/go.mod h1:o4eNf7Ede1fv+hwOwZsTHl9EsPFO6q6ZvYR8vYfY45I=
 golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1 h1:k/i9J1pBpvlfR+9QsetwPyERsqu1GIbi967PQMq3Ivc=
 golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1/go.mod h1:V1LtkGg67GoY2N1AnLN78QLrzxkLyJw7RJb1gzOOz9w=
 golang.org/x/image v0.6.0 h1:bR8b5okrPI3g/gyZakLZHeWxAR8Dn5CyxXv1hLH5g/4=
@@ -328,8 +320,8 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
-golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.11.0 h1:Gi2tvZIJyBtO9SDr1q9h5hEQCp/4L2RQ+ar0qjx2oNU=
+golang.org/x/net v0.11.0/go.mod h1:2L/ixqYpgIVXmeoSA/4Lu7BzTG4KIyPIryS4IsOd1oQ=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -350,12 +342,11 @@ golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
-golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.9.0 h1:KS/R3tvhPqvJvwcKfnBHJwwthS11LRhmM5D59eEXa0s=
+golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
@@ -364,7 +355,7 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
-golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
+golang.org/x/text v0.10.0 h1:UpjohKhiEgNc0CSauXmwYftY1+LlaC75SJwh0SgCX58=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190621195816-6e04913cbbac/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=

metric/drpc.go

@@ -2,8 +2,10 @@ package metric
 
 import (
 	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/zap"
 	"storj.io/drpc"
 	"time"
+	"unicode/utf8"
 )
 
 type prometheusDRPC struct {
@@ -14,7 +16,11 @@ type prometheusDRPC struct {
 func (ph *prometheusDRPC) HandleRPC(stream drpc.Stream, rpc string) (err error) {
 	st := time.Now()
 	defer func() {
-		ph.SummaryVec.WithLabelValues(rpc).Observe(time.Since(st).Seconds())
+		if utf8.ValidString(rpc) {
+			ph.SummaryVec.WithLabelValues(rpc).Observe(time.Since(st).Seconds())
+		} else {
+			log.WarnCtx(stream.Context(), "invalid rpc string", zap.String("rpc", rpc))
+		}
 	}()
 	return ph.Handler.HandleRPC(stream, rpc)
 }

net/streampool/stream.go

@@ -2,11 +2,12 @@ package streampool
 
 import (
 	"context"
+	"sync/atomic"
+
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/cheggaaa/mb/v3"
 	"go.uber.org/zap"
 	"storj.io/drpc"
-	"sync/atomic"
 )
 
 type stream struct {
@@ -22,7 +23,7 @@ type stream struct {
 }
 
 func (sr *stream) write(msg drpc.Message) (err error) {
-	return sr.queue.Add(sr.stream.Context(), msg)
+	return sr.queue.TryAdd(msg)
 }
 
 func (sr *stream) readLoop() error {

net/streampool/streampool_test.go

@@ -2,18 +2,19 @@ package streampool
 
 import (
 	"fmt"
+	"sort"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
 	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/rpc/rpctest"
 	"github.com/anyproto/any-sync/net/streampool/testservice"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/context"
-	"sort"
 	"storj.io/drpc"
-	"sync"
-	"sync/atomic"
-	"testing"
-	"time"
 )
 
 var ctx = context.Background()

net/transport/yamux/conn.go

@@ -3,6 +3,7 @@ package yamux
 import (
 	"context"
 	"github.com/anyproto/any-sync/net/connutil"
+	"github.com/anyproto/any-sync/net/peer"
 	"github.com/anyproto/any-sync/net/transport"
 	"github.com/hashicorp/yamux"
 	"net"
@@ -10,6 +11,7 @@ import (
 )
 
 func NewMultiConn(cctx context.Context, luConn *connutil.LastUsageConn, addr string, sess *yamux.Session) transport.MultiConn {
+	cctx = peer.CtxWithPeerAddr(cctx, sess.RemoteAddr().String())
 	return &yamuxConn{
 		ctx:     cctx,
 		luConn:  luConn,