Merge pull request #15 from anyproto/new-sync-protocol

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/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) {
@@ -127,8 +132,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
 }

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"
@@ -176,6 +177,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

@@ -21,6 +21,12 @@ 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{}
 }
@@ -70,6 +76,8 @@ 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/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)