Merge pull request #36 from anyproto/subconn-limit

peer sub connections limit/throttling

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

@@ -89,7 +89,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 {
@@ -262,6 +262,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 {

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/objecttree/rawloader.go

@@ -2,10 +2,11 @@ package objecttree
 
 import (
 	"context"
+	"time"
+
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
 	"github.com/anyproto/any-sync/util/slice"
-	"time"
 )
 
 type rawChangeLoader struct {
@@ -22,6 +23,7 @@ type rawCacheEntry struct {
 	change    *Change
 	rawChange *treechangeproto.RawTreeChangeWithId
 	position  int
+	removed   bool
 }
 
 func newStorageLoader(treeStorage treestorage.TreeStorage, changeBuilder ChangeBuilder) *rawChangeLoader {
@@ -126,7 +128,6 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 		if err != nil {
 			continue
 		}
-		entry.position = -1
 		r.cache[b] = entry
 		existingBreakpoints = append(existingBreakpoints, b)
 	}
@@ -135,8 +136,7 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 	dfs := func(
 		commonSnapshot string,
 		heads []string,
-		startCounter int,
-		shouldVisit func(counter int, mapExists bool) bool,
+		shouldVisit func(entry rawCacheEntry, mapExists bool) bool,
 		visit func(entry rawCacheEntry) rawCacheEntry) bool {
 
 		// resetting stack
@@ -150,7 +150,7 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 			r.idStack = r.idStack[:len(r.idStack)-1]
 
 			entry, exists := r.cache[id]
-			if !shouldVisit(entry.position, exists) {
+			if !shouldVisit(entry, exists) {
 				continue
 			}
 			if id == commonSnapshot {
@@ -159,7 +159,6 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 			}
 			if !exists {
 				entry, err = r.loadEntry(id)
-				entry.position = -1
 				if err != nil {
 					continue
 				}
@@ -174,7 +173,7 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 					break
 				}
 				prevEntry, exists := r.cache[prev]
-				if !shouldVisit(prevEntry.position, exists) {
+				if !shouldVisit(prevEntry, exists) {
 					continue
 				}
 				r.idStack = append(r.idStack, prev)
@@ -187,8 +186,8 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 	r.idStack = append(r.idStack, heads...)
 	var buffer []*treechangeproto.RawTreeChangeWithId
 
-	rootVisited := dfs(commonSnapshot, heads, 0,
-		func(counter int, mapExists bool) bool {
+	rootVisited := dfs(commonSnapshot, heads,
+		func(_ rawCacheEntry, mapExists bool) bool {
 			return !mapExists
 		},
 		func(entry rawCacheEntry) rawCacheEntry {
@@ -213,11 +212,13 @@ func (r *rawChangeLoader) loadFromStorage(commonSnapshot string, heads, breakpoi
 	}
 
 	// marking all visited as nil
-	dfs(commonSnapshot, existingBreakpoints, len(buffer),
-		func(counter int, mapExists bool) bool {
-			return !mapExists || counter < len(buffer)
+	dfs(commonSnapshot, existingBreakpoints,
+		func(entry rawCacheEntry, mapExists bool) bool {
+			// only going through already loaded changes
+			return mapExists && !entry.removed
 		},
 		func(entry rawCacheEntry) rawCacheEntry {
+			entry.removed = true
 			if entry.position != -1 {
 				buffer[entry.position] = nil
 			}
@@ -248,6 +249,7 @@ func (r *rawChangeLoader) loadEntry(id string) (entry rawCacheEntry, err error)
 	entry = rawCacheEntry{
 		change:    change,
 		rawChange: rawChange,
+		position:  -1,
 	}
 	return
 }

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

@@ -71,7 +71,6 @@ type BuildDeps struct {
 	SyncStatus      syncstatus.StatusUpdater
 	PeerGetter      ResponsiblePeersGetter
 	BuildObjectTree objecttree.BuildObjectTreeFunc
-	RetryTimeout    time.Duration
 }
 
 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

@@ -3,22 +3,17 @@ package synctree
 import (
 	"context"
 	"errors"
-	"time"
 
 	"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"
 )
 
 var (
-	newRequestTimeout = 1 * time.Second
-
-	ErrRetryTimeout       = errors.New("failed to retry request")
 	ErrNoResponsiblePeers = errors.New("no responsible peers")
 )
 
@@ -65,31 +60,13 @@ func (t treeRemoteGetter) treeRequest(ctx context.Context, peerId string) (msg *
 	return
 }
 
-func (t treeRemoteGetter) treeRequestLoop(ctx context.Context, retryTimeout time.Duration) (msg *treechangeproto.TreeSyncMessage, err error) {
-	peerIdx := 0
-	retryCtx, cancel := context.WithTimeout(ctx, retryTimeout)
-	defer cancel()
-	for {
-		availablePeers, err := t.getPeers(ctx)
-		if err != nil {
-			if retryTimeout == 0 {
-				return nil, err
-			}
-		} else {
-			peerIdx = peerIdx % len(availablePeers)
-			msg, err = t.treeRequest(ctx, availablePeers[peerIdx])
-			if err == nil || retryTimeout == 0 {
-				return msg, err
-			}
-			peerIdx++
-		}
-		select {
-		case <-time.After(newRequestTimeout):
-			break
-		case <-retryCtx.Done():
-			return nil, ErrRetryTimeout
-		}
+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) {
@@ -111,22 +88,15 @@ func (t treeRemoteGetter) getTree(ctx context.Context) (treeStorage treestorage.
 	}
 
 	isRemote = true
-	resp, err := t.treeRequestLoop(ctx, t.deps.RetryTimeout)
+	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

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"testing"
-	"time"
 
 	"github.com/anyproto/any-sync/commonspace/object/tree/synctree/mock_synctree"
 	"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
@@ -49,8 +48,6 @@ func (fx *treeRemoteGetterFixture) stop() {
 }
 
 func TestTreeRemoteGetter(t *testing.T) {
-	newRequestTimeout = 20 * time.Millisecond
-	retryTimeout := 2 * newRequestTimeout
 	ctx := context.Background()
 	peerId := "peerId"
 	treeRequest := &treechangeproto.TreeSyncMessage{}
@@ -70,18 +67,7 @@ func TestTreeRemoteGetter(t *testing.T) {
 		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, 0)
-		require.NoError(t, err)
-		require.Equal(t, "id", resp.RootChange.Id)
-	})
-
-	t.Run("request peerId from context", func(t *testing.T) {
-		fx := newTreeRemoteGetterFixture(t)
-		defer fx.stop()
-		ctx := peer.CtxWithPeerId(ctx, peerId)
-		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, 0)
+		resp, err := fx.treeGetter.treeRequestLoop(ctx)
 		require.NoError(t, err)
 		require.Equal(t, "id", resp.RootChange.Id)
 	})
@@ -89,54 +75,13 @@ func TestTreeRemoteGetter(t *testing.T) {
 	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).Return(nil, fmt.Errorf("failed"))
-		_, err := fx.treeGetter.treeRequestLoop(ctx, 0)
-		require.Error(t, err)
-		require.NotEqual(t, ErrRetryTimeout, err)
-	})
-
-	t.Run("retry request success", 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).AnyTimes().Return([]peer.Peer{mockPeer}, nil)
-		fx.syncClientMock.EXPECT().CreateNewTreeRequest().AnyTimes().Return(treeRequest)
-		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).Times(1).Return(nil, fmt.Errorf("some"))
-		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).Times(1).Return(objectResponse, nil)
-		resp, err := fx.treeGetter.treeRequestLoop(ctx, retryTimeout)
-		require.NoError(t, err)
-		require.Equal(t, "id", resp.RootChange.Id)
-	})
-
-	t.Run("retry get peers success", 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).Times(1).Return([]peer.Peer{}, nil)
-		fx.peerGetterMock.EXPECT().GetResponsiblePeers(ctx).Times(1).Return([]peer.Peer{mockPeer}, nil)
-		fx.syncClientMock.EXPECT().CreateNewTreeRequest().AnyTimes().Return(treeRequest)
-		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).Times(1).Return(objectResponse, nil)
-		resp, err := fx.treeGetter.treeRequestLoop(ctx, retryTimeout)
-		require.NoError(t, err)
-		require.Equal(t, "id", resp.RootChange.Id)
-	})
-
-	t.Run("retry request fail", 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).AnyTimes().Return([]peer.Peer{mockPeer}, nil)
-		fx.syncClientMock.EXPECT().CreateNewTreeRequest().AnyTimes().Return(treeRequest)
 		fx.syncClientMock.EXPECT().SendRequest(ctx, peerId, fx.treeGetter.treeId, treeRequest).AnyTimes().Return(nil, fmt.Errorf("some"))
-		_, err := fx.treeGetter.treeRequestLoop(ctx, retryTimeout)
-		require.Equal(t, ErrRetryTimeout, err)
+		_, 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

@@ -5,7 +5,6 @@ import (
 	"context"
 	"errors"
 	"sync/atomic"
-	"time"
 
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
@@ -28,9 +27,8 @@ import (
 )
 
 type BuildTreeOpts struct {
-	Listener     updatelistener.UpdateListener
-	RetryTimeout time.Duration
-	TreeBuilder  objecttree.BuildObjectTreeFunc
+	Listener    updatelistener.UpdateListener
+	TreeBuilder objecttree.BuildObjectTreeFunc
 }
 
 const CName = "common.commonspace.objecttreebuilder"
@@ -121,7 +119,6 @@ func (t *treeBuilder) BuildTree(ctx context.Context, id string, opts BuildTreeOp
 		SpaceStorage:    t.spaceStorage,
 		OnClose:         t.onClose,
 		SyncStatus:      t.syncStatus,
-		RetryTimeout:    opts.RetryTimeout,
 		PeerGetter:      t.peerManager,
 		BuildObjectTree: treeBuilder,
 	}

commonspace/requestmanager/requestmanager.go

@@ -10,6 +10,7 @@ import (
 	"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"
@@ -123,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
 }

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/peer/limiter.go

@@ -0,0 +1,18 @@
+package peer
+
+import (
+	"time"
+)
+
+type limiter struct {
+	startThreshold int
+	slowDownStep   time.Duration
+}
+
+func (l limiter) wait(count int) <-chan time.Time {
+	if count > l.startThreshold {
+		wait := l.slowDownStep * time.Duration(count-l.startThreshold)
+		return time.After(wait)
+	}
+	return nil
+}

net/peer/peer.go

@@ -19,6 +19,7 @@ import (
 	"storj.io/drpc/drpcstream"
 	"storj.io/drpc/drpcwire"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -35,8 +36,15 @@ func NewPeer(mc transport.MultiConn, ctrl connCtrl) (p Peer, err error) {
 		active:    map[*subConn]struct{}{},
 		MultiConn: mc,
 		ctrl:      ctrl,
-		created:   time.Now(),
+		limiter: limiter{
+			// start throttling after 10 sub conns
+			startThreshold: 10,
+			slowDownStep:   time.Millisecond * 100,
+		},
+		subConnRelease: make(chan drpc.Conn),
+		created:        time.Now(),
 	}
+	pr.acceptCtx, pr.acceptCtxCancel = context.WithCancel(context.Background())
 	if pr.id, err = CtxPeerId(ctx); err != nil {
 		return
 	}
@@ -70,13 +78,22 @@ type peer struct {
 	ctrl connCtrl
 
 	// drpc conn pool
-	inactive []*subConn
-	active   map[*subConn]struct{}
+	// outgoing
+	inactive         []*subConn
+	active           map[*subConn]struct{}
+	subConnRelease   chan drpc.Conn
+	openingWaitCount atomic.Int32
+
+	incomingCount atomic.Int32
+	acceptCtx     context.Context
+
+	acceptCtxCancel context.CancelFunc
+
+	limiter limiter
 
 	mu sync.Mutex
 
 	created time.Time
-
 	transport.MultiConn
 }
 
@@ -87,7 +104,20 @@ func (p *peer) Id() string {
 func (p *peer) AcquireDrpcConn(ctx context.Context) (drpc.Conn, error) {
 	p.mu.Lock()
 	if len(p.inactive) == 0 {
+		wait := p.limiter.wait(len(p.active) + int(p.openingWaitCount.Load()))
 		p.mu.Unlock()
+		if wait != nil {
+			p.openingWaitCount.Add(1)
+			defer p.openingWaitCount.Add(-1)
+			// throttle new connection opening
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case dconn := <-p.subConnRelease:
+				return dconn, nil
+			case <-wait:
+			}
+		}
 		dconn, err := p.openDrpcConn(ctx)
 		if err != nil {
 			return nil, err
@@ -110,6 +140,21 @@ func (p *peer) AcquireDrpcConn(ctx context.Context) (drpc.Conn, error) {
 }
 
 func (p *peer) ReleaseDrpcConn(conn drpc.Conn) {
+	// do nothing if it's closed connection
+	select {
+	case <-conn.Closed():
+		return
+	default:
+	}
+
+	// try to send this connection to acquire if anyone is waiting for it
+	select {
+	case p.subConnRelease <- conn:
+		return
+	default:
+	}
+
+	// return to pool
 	p.mu.Lock()
 	defer p.mu.Unlock()
 	sc, ok := conn.(*subConn)
@@ -162,12 +207,21 @@ func (p *peer) acceptLoop() {
 		}
 	}()
 	for {
+		if wait := p.limiter.wait(int(p.incomingCount.Load())); wait != nil {
+			select {
+			case <-wait:
+			case <-p.acceptCtx.Done():
+				return
+			}
+		}
 		conn, err := p.Accept()
 		if err != nil {
 			exitErr = err
 			return
 		}
 		go func() {
+			p.incomingCount.Add(1)
+			defer p.incomingCount.Add(-1)
 			serveErr := p.serve(conn)
 			if serveErr != io.EOF && serveErr != transport.ErrConnClosed {
 				log.InfoCtx(p.Context(), "serve connection error", zap.Error(serveErr))

net/peer/peer_test.go

@@ -12,6 +12,8 @@ import (
 	"io"
 	"net"
 	_ "net/http/pprof"
+	"storj.io/drpc"
+	"storj.io/drpc/drpcconn"
 	"testing"
 	"time"
 )
@@ -19,32 +21,86 @@ import (
 var ctx = context.Background()
 
 func TestPeer_AcquireDrpcConn(t *testing.T) {
+	t.Run("generic", func(t *testing.T) {
+		fx := newFixture(t, "p1")
+		defer fx.finish()
+		in, out := net.Pipe()
+		go func() {
+			handshake.IncomingProtoHandshake(ctx, out, defaultProtoChecker)
+		}()
+		defer out.Close()
+		fx.mc.EXPECT().Open(gomock.Any()).Return(in, nil)
+		dc, err := fx.AcquireDrpcConn(ctx)
+		require.NoError(t, err)
+		assert.NotEmpty(t, dc)
+		defer dc.Close()
+
+		assert.Len(t, fx.active, 1)
+		assert.Len(t, fx.inactive, 0)
+
+		fx.ReleaseDrpcConn(dc)
+
+		assert.Len(t, fx.active, 0)
+		assert.Len(t, fx.inactive, 1)
+
+		dc, err = fx.AcquireDrpcConn(ctx)
+		require.NoError(t, err)
+		assert.NotEmpty(t, dc)
+		assert.Len(t, fx.active, 1)
+		assert.Len(t, fx.inactive, 0)
+	})
+	t.Run("closed sub conn", func(t *testing.T) {
+		fx := newFixture(t, "p1")
+		defer fx.finish()
+
+		closedIn, _ := net.Pipe()
+		dc := drpcconn.New(closedIn)
+		fx.ReleaseDrpcConn(&subConn{Conn: dc})
+		dc.Close()
+
+		in, out := net.Pipe()
+		go func() {
+			handshake.IncomingProtoHandshake(ctx, out, defaultProtoChecker)
+		}()
+		defer out.Close()
+		fx.mc.EXPECT().Open(gomock.Any()).Return(in, nil)
+		_, err := fx.AcquireDrpcConn(ctx)
+		require.NoError(t, err)
+	})
+}
+
+func TestPeer_DrpcConn_OpenThrottling(t *testing.T) {
 	fx := newFixture(t, "p1")
 	defer fx.finish()
-	in, out := net.Pipe()
+
+	acquire := func() (func(), drpc.Conn, error) {
+		in, out := net.Pipe()
+		go func() {
+			_, err := handshake.IncomingProtoHandshake(ctx, out, defaultProtoChecker)
+			require.NoError(t, err)
+		}()
+
+		fx.mc.EXPECT().Open(gomock.Any()).Return(in, nil)
+		dconn, err := fx.AcquireDrpcConn(ctx)
+		return func() { out.Close() }, dconn, err
+	}
+
+	var conCount = fx.limiter.startThreshold + 3
+	var conns []drpc.Conn
+	for i := 0; i < conCount; i++ {
+		cc, dc, err := acquire()
+		require.NoError(t, err)
+		defer cc()
+		conns = append(conns, dc)
+	}
+
 	go func() {
-		handshake.IncomingProtoHandshake(ctx, out, defaultProtoChecker)
+		time.Sleep(fx.limiter.slowDownStep)
+		fx.ReleaseDrpcConn(conns[0])
+		conns = conns[1:]
 	}()
-	defer out.Close()
-	fx.mc.EXPECT().Open(gomock.Any()).Return(in, nil)
-	dc, err := fx.AcquireDrpcConn(ctx)
+	_, err := fx.AcquireDrpcConn(ctx)
 	require.NoError(t, err)
-	assert.NotEmpty(t, dc)
-	defer dc.Close()
-
-	assert.Len(t, fx.active, 1)
-	assert.Len(t, fx.inactive, 0)
-
-	fx.ReleaseDrpcConn(dc)
-
-	assert.Len(t, fx.active, 0)
-	assert.Len(t, fx.inactive, 1)
-
-	dc, err = fx.AcquireDrpcConn(ctx)
-	require.NoError(t, err)
-	assert.NotEmpty(t, dc)
-	assert.Len(t, fx.active, 1)
-	assert.Len(t, fx.inactive, 0)
 }
 
 func TestPeerAccept(t *testing.T) {
@@ -63,6 +119,26 @@ func TestPeerAccept(t *testing.T) {
 	assert.NoError(t, <-outHandshakeCh)
 }
 
+func TestPeer_DrpcConn_AcceptThrottling(t *testing.T) {
+	fx := newFixture(t, "p1")
+	defer fx.finish()
+
+	var conCount = fx.limiter.startThreshold + 3
+	for i := 0; i < conCount; i++ {
+		in, out := net.Pipe()
+		defer out.Close()
+
+		var outHandshakeCh = make(chan error)
+		go func() {
+			outHandshakeCh <- handshake.OutgoingProtoHandshake(ctx, out, handshakeproto.ProtoType_DRPC)
+		}()
+		fx.acceptCh <- acceptedConn{conn: in}
+		cn := <-fx.testCtrl.serveConn
+		assert.Equal(t, in, cn)
+		assert.NoError(t, <-outHandshakeCh)
+	}
+}
+
 func TestPeer_TryClose(t *testing.T) {
 	t.Run("not close in first minute", func(t *testing.T) {
 		fx := newFixture(t, "p1")

net/pool/poolservice.go

@@ -49,8 +49,8 @@ func (p *poolService) Init(a *app.App) (err error) {
 			return p.dialer.Dial(ctx, id)
 		},
 		ocache.WithLogger(log.Sugar()),
-		ocache.WithGCPeriod(time.Minute),
-		ocache.WithTTL(time.Minute*5),
+		ocache.WithGCPeriod(time.Minute/2),
+		ocache.WithTTL(time.Minute),
 		ocache.WithPrometheus(p.metricReg, "netpool", "outgoing"),
 	)
 	p.pool.incoming = ocache.New(
@@ -58,8 +58,8 @@ func (p *poolService) Init(a *app.App) (err error) {
 			return nil, ocache.ErrNotExists
 		},
 		ocache.WithLogger(log.Sugar()),
-		ocache.WithGCPeriod(time.Minute),
-		ocache.WithTTL(time.Minute*5),
+		ocache.WithGCPeriod(time.Minute/2),
+		ocache.WithTTL(time.Minute),
 		ocache.WithPrometheus(p.metricReg, "netpool", "incoming"),
 	)
 	return nil

net/rpc/rpctest/multiconntest/multiconntest.go

@@ -0,0 +1,29 @@
+package multiconntest
+
+import (
+	"context"
+	"github.com/anyproto/any-sync/net/connutil"
+	"github.com/anyproto/any-sync/net/transport"
+	yamux2 "github.com/anyproto/any-sync/net/transport/yamux"
+	"github.com/hashicorp/yamux"
+	"net"
+)
+
+func MultiConnPair(peerServCtx, peerClientCtx context.Context) (serv, client transport.MultiConn) {
+	sc, cc := net.Pipe()
+	var servConn = make(chan transport.MultiConn, 1)
+	go func() {
+		sess, err := yamux.Server(sc, yamux.DefaultConfig())
+		if err != nil {
+			panic(err)
+		}
+		servConn <- yamux2.NewMultiConn(peerServCtx, connutil.NewLastUsageConn(sc), "", sess)
+	}()
+	sess, err := yamux.Client(cc, yamux.DefaultConfig())
+	if err != nil {
+		panic(err)
+	}
+	client = yamux2.NewMultiConn(peerClientCtx, connutil.NewLastUsageConn(cc), "", sess)
+	serv = <-servConn
+	return
+}

net/rpc/rpctest/peer.go

@@ -2,29 +2,11 @@ package rpctest
 
 import (
 	"context"
-	"github.com/anyproto/any-sync/net/connutil"
 	"github.com/anyproto/any-sync/net/peer"
+	"github.com/anyproto/any-sync/net/rpc/rpctest/multiconntest"
 	"github.com/anyproto/any-sync/net/transport"
-	yamux2 "github.com/anyproto/any-sync/net/transport/yamux"
-	"github.com/hashicorp/yamux"
-	"net"
 )
 
 func MultiConnPair(peerIdServ, peerIdClient string) (serv, client transport.MultiConn) {
-	sc, cc := net.Pipe()
-	var servConn = make(chan transport.MultiConn, 1)
-	go func() {
-		sess, err := yamux.Server(sc, yamux.DefaultConfig())
-		if err != nil {
-			panic(err)
-		}
-		servConn <- yamux2.NewMultiConn(peer.CtxWithPeerId(context.Background(), peerIdServ), connutil.NewLastUsageConn(sc), "", sess)
-	}()
-	sess, err := yamux.Client(cc, yamux.DefaultConfig())
-	if err != nil {
-		panic(err)
-	}
-	client = yamux2.NewMultiConn(peer.CtxWithPeerId(context.Background(), peerIdClient), connutil.NewLastUsageConn(cc), "", sess)
-	serv = <-servConn
-	return
+	return multiconntest.MultiConnPair(peer.CtxWithPeerId(context.Background(), peerIdServ), peer.CtxWithPeerId(context.Background(), peerIdClient))
 }

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,13 +3,16 @@ 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"
+	"io"
 	"net"
 	"time"
 )
 
 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,
@@ -46,7 +49,7 @@ func (y *yamuxConn) Addr() string {
 
 func (y *yamuxConn) Accept() (conn net.Conn, err error) {
 	if conn, err = y.Session.Accept(); err != nil {
-		if err == yamux.ErrSessionShutdown {
+		if err == yamux.ErrSessionShutdown || err == io.EOF {
 			err = transport.ErrConnClosed
 		}
 		return

net/transport/yamux/yamux_test.go

@@ -30,8 +30,12 @@ func TestYamuxTransport_Dial(t *testing.T) {
 
 	mcC, err := fxC.Dial(ctx, fxS.addr)
 	require.NoError(t, err)
-	require.Len(t, fxS.accepter.mcs, 1)
-	mcS := <-fxS.accepter.mcs
+	var mcS transport.MultiConn
+	select {
+	case mcS = <-fxS.accepter.mcs:
+	case <-time.After(time.Second * 5):
+		require.True(t, false, "timeout")
+	}
 
 	var (
 		sData     string
@@ -69,11 +73,11 @@ func TestYamuxTransport_Dial(t *testing.T) {
 // no deadline - 69100 rps
 // common write deadline - 66700 rps
 // subconn write deadline - 67100 rps
-func TestWriteBench(t *testing.T) {
+func TestWriteBenchReuse(t *testing.T) {
 	t.Skip()
 	var (
 		numSubConn = 10
-		numWrites  = 100000
+		numWrites  = 10000
 	)
 
 	fxS := newFixture(t)
@@ -124,6 +128,63 @@ func TestWriteBench(t *testing.T) {
 	t.Logf("%.2f req per sec", float64(numWrites*numSubConn)/dur.Seconds())
 }
 
+func TestWriteBenchNew(t *testing.T) {
+	t.Skip()
+	var (
+		numSubConn = 10
+		numWrites  = 10000
+	)
+
+	fxS := newFixture(t)
+	defer fxS.finish(t)
+	fxC := newFixture(t)
+	defer fxC.finish(t)
+
+	mcC, err := fxC.Dial(ctx, fxS.addr)
+	require.NoError(t, err)
+	mcS := <-fxS.accepter.mcs
+
+	go func() {
+		for i := 0; i < numSubConn; i++ {
+			require.NoError(t, err)
+			go func() {
+				var b = make([]byte, 1024)
+				for {
+					conn, _ := mcS.Accept()
+					n, _ := conn.Read(b)
+					if n > 0 {
+						conn.Write(b[:n])
+					} else {
+						_ = conn.Close()
+						break
+					}
+					conn.Close()
+				}
+			}()
+		}
+	}()
+
+	var wg sync.WaitGroup
+	wg.Add(numSubConn)
+	st := time.Now()
+	for i := 0; i < numSubConn; i++ {
+		go func() {
+			defer wg.Done()
+			for j := 0; j < numWrites; j++ {
+				sc, err := mcC.Open(ctx)
+				require.NoError(t, err)
+				var b = []byte("some data some data some data some data some data some data some data some data some data")
+				sc.Write(b)
+				sc.Read(b)
+				sc.Close()
+			}
+		}()
+	}
+	wg.Wait()
+	dur := time.Since(st)
+	t.Logf("%.2f req per sec", float64(numWrites*numSubConn)/dur.Seconds())
+}
+
 type fixture struct {
 	*yamuxTransport
 	a            *app.App