net test

peer: gc active sub conn

net/config.go

@@ -5,21 +5,3 @@ import "errors"
 var (
 	ErrUnableToConnect = errors.New("unable to connect")
 )
-
-type ConfigGetter interface {
-	GetNet() Config
-}
-
-type Config struct {
-	Server ServerConfig `yaml:"server"`
-	Stream StreamConfig `yaml:"stream"`
-}
-
-type ServerConfig struct {
-	ListenAddrs []string `yaml:"listenAddrs"`
-}
-
-type StreamConfig struct {
-	TimeoutMilliseconds int `yaml:"timeoutMilliseconds"`
-	MaxMsgSizeMb        int `yaml:"maxMsgSizeMb"`
-}

net/connutil/timeout.go

@@ -21,7 +21,13 @@ func NewTimeout(conn net.Conn, timeout time.Duration) *TimeoutConn {
 	return &TimeoutConn{conn, timeout}
 }
 
-func (c *TimeoutConn) Write(p []byte) (n int, err error) {
+func (c *TimeoutConn) Write(p []byte) (n int, retErr error) {
+	log.Debug("start write", zap.Int("n", len(p)))
+	defer func() {
+		if retErr != nil {
+			log.Debug("conn write error", zap.Int("n", n), zap.Error(retErr))
+		}
+	}()
 	for {
 		if c.timeout != 0 {
 			if e := c.Conn.SetWriteDeadline(time.Now().Add(c.timeout)); e != nil {
@@ -47,6 +53,7 @@ func (c *TimeoutConn) Write(p []byte) (n int, err error) {
 			}
 			log.Debug("connection timed out", zap.String("remoteAddr", c.RemoteAddr().String()))
 		}
+		log.Debug("conn write", zap.Int("n", n))
 		return n, err
 	}
 }

net/peer/peer.go

@@ -6,6 +6,7 @@ import (
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/app/ocache"
 	"github.com/anyproto/any-sync/net/connutil"
+	"github.com/anyproto/any-sync/net/rpc"
 	"github.com/anyproto/any-sync/net/secureservice/handshake"
 	"github.com/anyproto/any-sync/net/secureservice/handshake/handshakeproto"
 	"github.com/anyproto/any-sync/net/transport"
@@ -14,6 +15,9 @@ import (
 	"net"
 	"storj.io/drpc"
 	"storj.io/drpc/drpcconn"
+	"storj.io/drpc/drpcmanager"
+	"storj.io/drpc/drpcstream"
+	"storj.io/drpc/drpcwire"
 	"sync"
 	"time"
 )
@@ -22,6 +26,7 @@ var log = logger.NewNamed("common.net.peer")
 
 type connCtrl interface {
 	ServeConn(ctx context.Context, conn net.Conn) (err error)
+	DrpcConfig() rpc.Config
 }
 
 func NewPeer(mc transport.MultiConn, ctrl connCtrl) (p Peer, err error) {
@@ -90,6 +95,12 @@ func (p *peer) AcquireDrpcConn(ctx context.Context) (drpc.Conn, error) {
 	idx := len(p.inactive) - 1
 	res := p.inactive[idx]
 	p.inactive = p.inactive[:idx]
+	select {
+	case <-res.Closed():
+		p.mu.Unlock()
+		return p.AcquireDrpcConn(ctx)
+	default:
+	}
 	p.active[res] = struct{}{}
 	p.mu.Unlock()
 	return res, nil
@@ -127,8 +138,14 @@ func (p *peer) openDrpcConn(ctx context.Context) (dconn *subConn, err error) {
 		return nil, err
 	}
 	tconn := connutil.NewLastUsageConn(conn)
+	bufSize := p.ctrl.DrpcConfig().Stream.MaxMsgSizeMb * (1 << 20)
 	return &subConn{
-		Conn:          drpcconn.New(tconn),
+		Conn: drpcconn.NewWithOptions(conn, drpcconn.Options{
+			Manager: drpcmanager.Options{
+				Reader: drpcwire.ReaderOptions{MaximumBufferSize: bufSize},
+				Stream: drpcstream.Options{MaximumBufferSize: bufSize},
+			},
+		}),
 		LastUsageConn: tconn,
 	}, nil
 }
@@ -211,8 +228,15 @@ func (p *peer) gc(ttl time.Duration) {
 		select {
 		case <-act.Closed():
 			delete(p.active, act)
+			continue
 		default:
 		}
+		if act.LastUsage().Before(minLastUsage) {
+			log.Warn("close active connection because no activity", zap.String("peerId", p.id), zap.String("addr", p.Addr()))
+			_ = act.Close()
+			delete(p.active, act)
+			continue
+		}
 	}
 }
 

net/peer/peer_test.go

@@ -2,6 +2,7 @@ package peer
 
 import (
 	"context"
+	"github.com/anyproto/any-sync/net/rpc"
 	"github.com/anyproto/any-sync/net/secureservice/handshake"
 	"github.com/anyproto/any-sync/net/secureservice/handshake/handshakeproto"
 	"github.com/anyproto/any-sync/net/transport/mock_transport"
@@ -104,6 +105,7 @@ func TestPeer_TryClose(t *testing.T) {
 		}()
 		defer out2.Close()
 		fx.mc.EXPECT().Open(gomock.Any()).Return(in2, nil)
+		fx.mc.EXPECT().Addr().Return("")
 		dc2, err := fx.AcquireDrpcConn(ctx)
 		require.NoError(t, err)
 		_ = dc2.Close()
@@ -117,6 +119,18 @@ func TestPeer_TryClose(t *testing.T) {
 		fx.mc.EXPECT().Open(gomock.Any()).Return(in3, nil)
 		dc3, err := fx.AcquireDrpcConn(ctx)
 		require.NoError(t, err)
+
+		// make another active, should be removed by ttl
+		in4, out4 := net.Pipe()
+		go func() {
+			handshake.IncomingProtoHandshake(ctx, out4, defaultProtoChecker)
+		}()
+		defer out4.Close()
+		fx.mc.EXPECT().Open(gomock.Any()).Return(in4, nil)
+		dc4, err := fx.AcquireDrpcConn(ctx)
+		require.NoError(t, err)
+		defer dc4.Close()
+
 		fx.ReleaseDrpcConn(dc3)
 		_ = dc3.Close()
 		fx.ReleaseDrpcConn(dc)
@@ -176,6 +190,10 @@ type testCtrl struct {
 	closeCh   chan struct{}
 }
 
+func (t *testCtrl) DrpcConfig() rpc.Config {
+	return rpc.Config{Stream: rpc.StreamConfig{MaxMsgSizeMb: 10}}
+}
+
 func (t *testCtrl) ServeConn(ctx context.Context, conn net.Conn) (err error) {
 	t.serveConn <- conn
 	<-t.closeCh

net/rpc/debugserver/config.go

@@ -0,0 +1,9 @@
+package debugserver
+
+type configGetter interface {
+	GetDebugServer() Config
+}
+
+type Config struct {
+	ListenAddr string `yaml:"listenAddr"`
+}

net/rpc/debugserver/debugserver.go

@@ -0,0 +1,70 @@
+package debugserver
+
+import (
+	"context"
+	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/net/rpc"
+	"net"
+	"storj.io/drpc"
+	"storj.io/drpc/drpcmanager"
+	"storj.io/drpc/drpcmux"
+	"storj.io/drpc/drpcserver"
+	"storj.io/drpc/drpcstream"
+	"storj.io/drpc/drpcwire"
+)
+
+const CName = "net.rpc.debugserver"
+
+func New() DebugServer {
+	return &debugServer{}
+}
+
+type DebugServer interface {
+	app.ComponentRunnable
+	drpc.Mux
+}
+
+type debugServer struct {
+	drpcServer *drpcserver.Server
+	*drpcmux.Mux
+	drpcConf     rpc.Config
+	config       Config
+	runCtx       context.Context
+	runCtxCancel context.CancelFunc
+}
+
+func (d *debugServer) Init(a *app.App) (err error) {
+	d.drpcConf = a.MustComponent("config").(rpc.ConfigGetter).GetDrpc()
+	d.config = a.MustComponent("config").(configGetter).GetDebugServer()
+	d.Mux = drpcmux.New()
+	bufSize := d.drpcConf.Stream.MaxMsgSizeMb * (1 << 20)
+	d.drpcServer = drpcserver.NewWithOptions(d, drpcserver.Options{Manager: drpcmanager.Options{
+		Reader: drpcwire.ReaderOptions{MaximumBufferSize: bufSize},
+		Stream: drpcstream.Options{MaximumBufferSize: bufSize},
+	}})
+	return nil
+}
+
+func (d *debugServer) Name() (name string) {
+	return CName
+}
+
+func (d *debugServer) Run(ctx context.Context) (err error) {
+	if d.config.ListenAddr == "" {
+		return
+	}
+	lis, err := net.Listen("tcp", d.config.ListenAddr)
+	if err != nil {
+		return
+	}
+	d.runCtx, d.runCtxCancel = context.WithCancel(context.Background())
+	go d.drpcServer.Serve(d.runCtx, lis)
+	return
+}
+
+func (d *debugServer) Close(ctx context.Context) (err error) {
+	if d.runCtx != nil {
+		d.runCtxCancel()
+	}
+	return nil
+}

net/rpc/drpcconfig.go

@@ -0,0 +1,13 @@
+package rpc
+
+type ConfigGetter interface {
+	GetDrpc() Config
+}
+
+type Config struct {
+	Stream StreamConfig `yaml:"stream"`
+}
+
+type StreamConfig struct {
+	MaxMsgSizeMb int `yaml:"maxMsgSizeMb"`
+}

net/rpc/rpctest/pool.go

@@ -0,0 +1,76 @@
+package rpctest
+
+import (
+	"context"
+	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/net"
+	"github.com/anyproto/any-sync/net/peer"
+	"github.com/anyproto/any-sync/net/pool"
+	"sync"
+)
+
+func NewTestPool() *TestPool {
+	return &TestPool{peers: map[string]peer.Peer{}}
+}
+
+type TestPool struct {
+	peers map[string]peer.Peer
+	mu    sync.Mutex
+	ts    *TestServer
+}
+
+func (t *TestPool) Init(a *app.App) (err error) {
+	return nil
+}
+
+func (t *TestPool) Name() (name string) {
+	return pool.CName
+}
+
+func (t *TestPool) WithServer(ts *TestServer) *TestPool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.ts = ts
+	return t
+}
+
+func (t *TestPool) Run(ctx context.Context) (err error) {
+	return nil
+}
+
+func (t *TestPool) Close(ctx context.Context) (err error) {
+	return nil
+}
+
+func (t *TestPool) Get(ctx context.Context, id string) (peer.Peer, error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if p, ok := t.peers[id]; ok {
+		return p, nil
+	}
+	if t.ts == nil {
+		return nil, net.ErrUnableToConnect
+	}
+	return t.ts.Dial(id)
+}
+
+func (t *TestPool) GetOneOf(ctx context.Context, peerIds []string) (peer.Peer, error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	for _, peerId := range peerIds {
+		if p, ok := t.peers[peerId]; ok {
+			return p, nil
+		}
+	}
+	if t.ts == nil || len(peerIds) == 0 {
+		return nil, net.ErrUnableToConnect
+	}
+	return t.ts.Dial(peerIds[0])
+}
+
+func (t *TestPool) AddPeer(ctx context.Context, p peer.Peer) (err error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.peers[p.Id()] = p
+	return nil
+}

net/rpc/rpctest/server.go

@@ -3,12 +3,25 @@ package rpctest
 import (
 	"context"
 	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/net/peer"
+	"github.com/anyproto/any-sync/net/rpc"
 	"github.com/anyproto/any-sync/net/rpc/server"
 	"net"
 	"storj.io/drpc/drpcmux"
 	"storj.io/drpc/drpcserver"
 )
 
+type mockCtrl struct {
+}
+
+func (m mockCtrl) ServeConn(ctx context.Context, conn net.Conn) (err error) {
+	return nil
+}
+
+func (m mockCtrl) DrpcConfig() rpc.Config {
+	return rpc.Config{}
+}
+
 func NewTestServer() *TestServer {
 	ts := &TestServer{
 		Mux: drpcmux.New(),
@@ -22,22 +35,37 @@ type TestServer struct {
 	*drpcserver.Server
 }
 
-func (ts *TestServer) Init(a *app.App) (err error) {
+func (s *TestServer) Init(a *app.App) (err error) {
 	return nil
 }
 
-func (ts *TestServer) Name() (name string) {
+func (s *TestServer) Name() (name string) {
 	return server.CName
 }
 
-func (ts *TestServer) Run(ctx context.Context) (err error) {
+func (s *TestServer) Run(ctx context.Context) (err error) {
 	return nil
 }
 
-func (ts *TestServer) Close(ctx context.Context) (err error) {
+func (s *TestServer) Close(ctx context.Context) (err error) {
 	return nil
 }
 
 func (s *TestServer) ServeConn(ctx context.Context, conn net.Conn) (err error) {
 	return s.Server.ServeOne(ctx, conn)
 }
+
+func (s *TestServer) DrpcConfig() rpc.Config {
+	return rpc.Config{Stream: rpc.StreamConfig{MaxMsgSizeMb: 10}}
+}
+
+func (s *TestServer) Dial(peerId string) (clientPeer peer.Peer, err error) {
+	mcS, mcC := MultiConnPair(peerId+"server", peerId)
+	// NewPeer runs the accept loop
+	_, err = peer.NewPeer(mcS, s)
+	if err != nil {
+		return
+	}
+	// and we ourselves don't call server methods on accept
+	return peer.NewPeer(mcC, mockCtrl{})
+}

net/rpc/server/drpcserver.go

@@ -5,13 +5,14 @@ import (
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
 	"github.com/anyproto/any-sync/metric"
-	anyNet "github.com/anyproto/any-sync/net"
+	"github.com/anyproto/any-sync/net/rpc"
 	"go.uber.org/zap"
 	"net"
 	"storj.io/drpc"
 	"storj.io/drpc/drpcmanager"
 	"storj.io/drpc/drpcmux"
 	"storj.io/drpc/drpcserver"
+	"storj.io/drpc/drpcstream"
 	"storj.io/drpc/drpcwire"
 )
 
@@ -25,6 +26,7 @@ func New() DRPCServer {
 
 type DRPCServer interface {
 	ServeConn(ctx context.Context, conn net.Conn) (err error)
+	DrpcConfig() rpc.Config
 	app.Component
 	drpc.Mux
 }
@@ -32,7 +34,7 @@ type DRPCServer interface {
 type drpcServer struct {
 	drpcServer *drpcserver.Server
 	*drpcmux.Mux
-	config anyNet.Config
+	config rpc.Config
 	metric metric.Metric
 }
 
@@ -43,7 +45,7 @@ func (s *drpcServer) Name() (name string) {
 }
 
 func (s *drpcServer) Init(a *app.App) (err error) {
-	s.config = a.MustComponent("config").(anyNet.ConfigGetter).GetNet()
+	s.config = a.MustComponent("config").(rpc.ConfigGetter).GetDrpc()
 	s.metric, _ = a.Component(metric.CName).(metric.Metric)
 	s.Mux = drpcmux.New()
 
@@ -52,8 +54,10 @@ func (s *drpcServer) Init(a *app.App) (err error) {
 	if s.metric != nil {
 		handler = s.metric.WrapDRPCHandler(s)
 	}
+	bufSize := s.config.Stream.MaxMsgSizeMb * (1 << 20)
 	s.drpcServer = drpcserver.NewWithOptions(handler, drpcserver.Options{Manager: drpcmanager.Options{
-		Reader: drpcwire.ReaderOptions{MaximumBufferSize: s.config.Stream.MaxMsgSizeMb * (1 << 20)},
+		Reader: drpcwire.ReaderOptions{MaximumBufferSize: bufSize},
+		Stream: drpcstream.Options{MaximumBufferSize: bufSize},
 	}})
 	return
 }
@@ -63,3 +67,7 @@ func (s *drpcServer) ServeConn(ctx context.Context, conn net.Conn) (err error) {
 	l.Debug("drpc serve peer")
 	return s.drpcServer.ServeOne(ctx, conn)
 }
+
+func (s *drpcServer) DrpcConfig() rpc.Config {
+	return s.config
+}

net/secureservice/secureservice.go

@@ -3,6 +3,9 @@ package secureservice
 import (
 	"context"
 	"crypto/tls"
+	"io"
+	"net"
+
 	commonaccount "github.com/anyproto/any-sync/accountservice"
 	"github.com/anyproto/any-sync/app"
 	"github.com/anyproto/any-sync/app/logger"
@@ -13,14 +16,18 @@ import (
 	"github.com/libp2p/go-libp2p/core/crypto"
 	libp2ptls "github.com/libp2p/go-libp2p/p2p/security/tls"
 	"go.uber.org/zap"
-	"io"
-	"net"
 )
 
 const CName = "common.net.secure"
 
 var log = logger.NewNamed(CName)
 
+const (
+	// ProtoVersion 0 - first any-sync version with raw tcp connections
+	// ProtoVersion 1 - version with yamux over tcp and quic
+	ProtoVersion = 1
+)
+
 func New() SecureService {
 	return &secureService{}
 }
@@ -46,6 +53,9 @@ type secureService struct {
 }
 
 func (s *secureService) Init(a *app.App) (err error) {
+	if s.protoVersion == 0 {
+		s.protoVersion = ProtoVersion
+	}
 	account := a.MustComponent(commonaccount.CName).(commonaccount.Service)
 	peerKey, err := account.Account().PeerKey.Raw()
 	if err != nil {

net/secureservice/secureservice_test.go

@@ -57,7 +57,7 @@ func TestHandshake(t *testing.T) {
 
 func TestHandshakeIncompatibleVersion(t *testing.T) {
 	nc := testnodeconf.GenNodeConfig(2)
-	fxS := newFixture(t, nc, nc.GetAccountService(0), 0)
+	fxS := newFixture(t, nc, nc.GetAccountService(0), 1)
 	defer fxS.Finish(t)
 	sc, cc := net.Pipe()
 
@@ -72,7 +72,7 @@ func TestHandshakeIncompatibleVersion(t *testing.T) {
 		ar.ctx, ar.err = fxS.SecureInbound(ctx, sc)
 		resCh <- ar
 	}()
-	fxC := newFixture(t, nc, nc.GetAccountService(1), 1)
+	fxC := newFixture(t, nc, nc.GetAccountService(1), 2)
 	defer fxC.Finish(t)
 	_, err := fxC.SecureOutbound(ctx, cc)
 	require.Equal(t, handshake.ErrIncompatibleVersion, err)

net/transport/yamux/conn.go

@@ -29,6 +29,7 @@ func (y *yamuxConn) Open(ctx context.Context) (conn net.Conn, err error) {
 	if conn, err = y.Session.Open(); err != nil {
 		return
 	}
+	//conn = connutil.NewTimeout(conn, time.Second*10)
 	return
 }
 
@@ -51,5 +52,6 @@ func (y *yamuxConn) Accept() (conn net.Conn, err error) {
 		}
 		return
 	}
+	//conn = connutil.NewTimeout(conn, time.Second*10)
 	return
 }

net/transport/yamux/yamux.go

@@ -11,6 +11,7 @@ import (
 	"github.com/hashicorp/yamux"
 	"go.uber.org/zap"
 	"net"
+	"sync"
 	"time"
 )
 
@@ -25,6 +26,7 @@ func New() Yamux {
 // Yamux implements transport.Transport with tcp+yamux
 type Yamux interface {
 	transport.Transport
+	AddListener(lis net.Listener)
 	app.ComponentRunnable
 }
 
@@ -37,15 +39,23 @@ type yamuxTransport struct {
 	listCtx       context.Context
 	listCtxCancel context.CancelFunc
 	yamuxConf     *yamux.Config
+	mu            sync.Mutex
 }
 
 func (y *yamuxTransport) Init(a *app.App) (err error) {
 	y.secure = a.MustComponent(secureservice.CName).(secureservice.SecureService)
 	y.conf = a.MustComponent("config").(configGetter).GetYamux()
+	if y.conf.DialTimeoutSec <= 0 {
+		y.conf.DialTimeoutSec = 10
+	}
+	if y.conf.WriteTimeoutSec <= 0 {
+		y.conf.WriteTimeoutSec = 10
+	}
 	y.yamuxConf = yamux.DefaultConfig()
 	y.yamuxConf.EnableKeepAlive = false
 	y.yamuxConf.StreamOpenTimeout = time.Duration(y.conf.DialTimeoutSec) * time.Second
 	y.yamuxConf.ConnectionWriteTimeout = time.Duration(y.conf.WriteTimeoutSec) * time.Second
+	y.listCtx, y.listCtxCancel = context.WithCancel(context.Background())
 	return
 }
 
@@ -57,6 +67,8 @@ func (y *yamuxTransport) Run(ctx context.Context) (err error) {
 	if y.accepter == nil {
 		return fmt.Errorf("can't run service without accepter")
 	}
+	y.mu.Lock()
+	defer y.mu.Unlock()
 	for _, listAddr := range y.conf.ListenAddrs {
 		list, err := net.Listen("tcp", listAddr)
 		if err != nil {
@@ -64,7 +76,6 @@ func (y *yamuxTransport) Run(ctx context.Context) (err error) {
 		}
 		y.listeners = append(y.listeners, list)
 	}
-	y.listCtx, y.listCtxCancel = context.WithCancel(context.Background())
 	for _, list := range y.listeners {
 		go y.acceptLoop(y.listCtx, list)
 	}
@@ -75,6 +86,13 @@ func (y *yamuxTransport) SetAccepter(accepter transport.Accepter) {
 	y.accepter = accepter
 }
 
+func (y *yamuxTransport) AddListener(lis net.Listener) {
+	y.mu.Lock()
+	defer y.mu.Unlock()
+	y.listeners = append(y.listeners, lis)
+	go y.acceptLoop(y.listCtx, lis)
+}
+
 func (y *yamuxTransport) Dial(ctx context.Context, addr string) (mc transport.MultiConn, err error) {
 	dialTimeout := time.Duration(y.conf.DialTimeoutSec) * time.Second
 	conn, err := net.DialTimeout("tcp", addr, dialTimeout)
@@ -88,7 +106,7 @@ func (y *yamuxTransport) Dial(ctx context.Context, addr string) (mc transport.Mu
 		_ = conn.Close()
 		return nil, err
 	}
-	luc := connutil.NewLastUsageConn(conn)
+	luc := connutil.NewLastUsageConn(connutil.NewTimeout(conn, time.Duration(y.conf.WriteTimeoutSec)*time.Second))
 	sess, err := yamux.Client(luc, y.yamuxConf)
 	if err != nil {
 		return
@@ -134,7 +152,7 @@ func (y *yamuxTransport) accept(conn net.Conn) {
 		log.Warn("incoming connection handshake error", zap.Error(err))
 		return
 	}
-	luc := connutil.NewLastUsageConn(conn)
+	luc := connutil.NewLastUsageConn(connutil.NewTimeout(conn, time.Duration(y.conf.WriteTimeoutSec)*time.Second))
 	sess, err := yamux.Server(luc, y.yamuxConf)
 	if err != nil {
 		log.Warn("incoming connection yamux session error", zap.Error(err))

testutil/nettest/net_test.go

@@ -0,0 +1,114 @@
+package nettest
+
+import (
+	"context"
+	"github.com/anyproto/any-sync/app"
+	"github.com/anyproto/any-sync/app/logger"
+	"github.com/anyproto/any-sync/coordinator/coordinatorclient"
+	"github.com/anyproto/any-sync/coordinator/coordinatorproto"
+	"github.com/anyproto/any-sync/coordinator/nodeconfsource"
+	"github.com/anyproto/any-sync/net/peerservice"
+	"github.com/anyproto/any-sync/net/pool"
+	"github.com/anyproto/any-sync/net/rpc"
+	"github.com/anyproto/any-sync/net/rpc/server"
+	"github.com/anyproto/any-sync/net/secureservice"
+	"github.com/anyproto/any-sync/net/transport/yamux"
+	"github.com/anyproto/any-sync/nodeconf"
+	"github.com/anyproto/any-sync/nodeconf/nodeconfstore"
+	"github.com/anyproto/any-sync/testutil/accounttest"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"net/http"
+	_ "net/http/pprof"
+	"testing"
+	"time"
+)
+
+var ctx = context.Background()
+
+func init() {
+	go http.ListenAndServe(":6061", nil)
+}
+
+func TestNet(t *testing.T) {
+	l := logger.NewNamed("test")
+	pl := pool.New()
+	a := new(app.App)
+	a.Register(&accounttest.AccountTestService{}).
+		Register(pl).
+		Register(peerservice.New()).
+		Register(yamux.New()).
+		Register(nodeconf.New()).
+		Register(nodeconfstore.New()).
+		Register(nodeconfsource.New()).
+		Register(server.New()).
+		Register(secureservice.New()).
+		Register(coordinatorclient.New()).
+		Register(&testConfig{})
+	require.NoError(t, a.Start(ctx))
+	defer a.Close(ctx)
+
+	p, err := pl.Get(ctx, "12D3KooWB4hmEo7YAdWzAaFpjyk4npkcwrPm2kRigsWu3MP9Xdmg")
+	require.NoError(t, err)
+
+	for i := 0; i < 1000000; i++ {
+		dc, err := p.AcquireDrpcConn(ctx)
+		require.NoError(t, err)
+		cl := coordinatorproto.NewDRPCCoordinatorClient(dc)
+		time.Sleep(time.Second)
+		res, err := cl.NetworkConfiguration(ctx, &coordinatorproto.NetworkConfigurationRequest{})
+		if err != nil {
+			l.Warn("req error", zap.Error(err))
+		} else {
+
+			l.Info("req success", zap.String("nid", res.NetworkId))
+		}
+		p.ReleaseDrpcConn(dc)
+	}
+
+}
+
+type testConfig struct {
+}
+
+func (t *testConfig) GetDrpc() rpc.Config {
+	return rpc.Config{
+		Stream: rpc.StreamConfig{
+			MaxMsgSizeMb: 1,
+		},
+	}
+}
+
+func (t *testConfig) GetNodeConfStorePath() string {
+	return "/tmp"
+}
+
+func (t *testConfig) GetNodeConf() nodeconf.Configuration {
+	return nodeconf.Configuration{
+		Id:        "",
+		NetworkId: "",
+		Nodes: []nodeconf.Node{
+			{
+				PeerId:    "12D3KooWB4hmEo7YAdWzAaFpjyk4npkcwrPm2kRigsWu3MP9Xdmg",
+				Addresses: []string{"ec2-3-65-13-48.eu-central-1.compute.amazonaws.com:443"},
+				Types:     []nodeconf.NodeType{nodeconf.NodeTypeTree},
+			},
+		},
+		CreationTime: time.Now(),
+	}
+}
+
+func (t *testConfig) GetYamux() yamux.Config {
+	return yamux.Config{
+		WriteTimeoutSec: 10,
+		DialTimeoutSec:  10,
+	}
+}
+
+func (t *testConfig) Init(a *app.App) (err error) {
+	return nil
+}
+
+func (t *testConfig) Name() (name string) {
+	return "config"
+}