net test

peer: gc active sub conn

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

@@ -95,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
@@ -222,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

@@ -105,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()
@@ -118,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)

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,8 +16,6 @@ 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"
@@ -52,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 {
@@ -76,8 +80,6 @@ func (s *secureService) Init(a *app.App) (err error) {
 		return
 	}
 
-	s.protoVersion = ProtoVersion
-
 	log.Info("secure service init", zap.String("peerId", account.Account().PeerId))
 	return 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

@@ -106,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
@@ -152,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"
+}