p2p: remove support for multiple transports and endpoints (#8420)

internal/p2p/address.go

@@ -97,7 +97,7 @@ func ParseNodeAddress(urlString string) (NodeAddress, error) {
 
 // Resolve resolves a NodeAddress into a set of Endpoints, by expanding
 // out a DNS hostname to IP addresses.
-func (a NodeAddress) Resolve(ctx context.Context) ([]Endpoint, error) {
+func (a NodeAddress) Resolve(ctx context.Context) ([]*Endpoint, error) {
 	if a.Protocol == "" {
 		return nil, errors.New("address has no protocol")
 	}
@@ -109,7 +109,7 @@ func (a NodeAddress) Resolve(ctx context.Context) ([]Endpoint, error) {
 		if a.NodeID == "" {
 			return nil, errors.New("local address has no node ID")
 		}
-		return []Endpoint{{
+		return []*Endpoint{{
 			Protocol: a.Protocol,
 			Path:     string(a.NodeID),
 		}}, nil
@@ -119,9 +119,9 @@ func (a NodeAddress) Resolve(ctx context.Context) ([]Endpoint, error) {
 	if err != nil {
 		return nil, err
 	}
-	endpoints := make([]Endpoint, len(ips))
+	endpoints := make([]*Endpoint, len(ips))
 	for i, ip := range ips {
-		endpoints[i] = Endpoint{
+		endpoints[i] = &Endpoint{
 			Protocol: a.Protocol,
 			IP:       ip,
 			Port:     a.Port,

internal/p2p/address_test.go

@@ -210,71 +210,71 @@ func TestNodeAddress_Resolve(t *testing.T) {
 
 	testcases := []struct {
 		address p2p.NodeAddress
-		expect  p2p.Endpoint
+		expect  *p2p.Endpoint
 		ok      bool
 	}{
 		// Valid networked addresses (with hostname).
 		{
 			p2p.NodeAddress{Protocol: "tcp", Hostname: "127.0.0.1", Port: 80, Path: "/path"},
-			p2p.Endpoint{Protocol: "tcp", IP: net.IPv4(127, 0, 0, 1), Port: 80, Path: "/path"},
+			&p2p.Endpoint{Protocol: "tcp", IP: net.IPv4(127, 0, 0, 1), Port: 80, Path: "/path"},
 			true,
 		},
 		{
 			p2p.NodeAddress{Protocol: "tcp", Hostname: "localhost", Port: 80, Path: "/path"},
-			p2p.Endpoint{Protocol: "tcp", IP: net.IPv4(127, 0, 0, 1), Port: 80, Path: "/path"},
+			&p2p.Endpoint{Protocol: "tcp", IP: net.IPv4(127, 0, 0, 1), Port: 80, Path: "/path"},
 			true,
 		},
 		{
 			p2p.NodeAddress{Protocol: "tcp", Hostname: "localhost", Port: 80, Path: "/path"},
-			p2p.Endpoint{Protocol: "tcp", IP: net.IPv6loopback, Port: 80, Path: "/path"},
+			&p2p.Endpoint{Protocol: "tcp", IP: net.IPv6loopback, Port: 80, Path: "/path"},
 			true,
 		},
 		{
 			p2p.NodeAddress{Protocol: "tcp", Hostname: "127.0.0.1"},
-			p2p.Endpoint{Protocol: "tcp", IP: net.IPv4(127, 0, 0, 1)},
+			&p2p.Endpoint{Protocol: "tcp", IP: net.IPv4(127, 0, 0, 1)},
 			true,
 		},
 		{
 			p2p.NodeAddress{Protocol: "tcp", Hostname: "::1"},
-			p2p.Endpoint{Protocol: "tcp", IP: net.IPv6loopback},
+			&p2p.Endpoint{Protocol: "tcp", IP: net.IPv6loopback},
 			true,
 		},
 		{
 			p2p.NodeAddress{Protocol: "tcp", Hostname: "8.8.8.8"},
-			p2p.Endpoint{Protocol: "tcp", IP: net.IPv4(8, 8, 8, 8)},
+			&p2p.Endpoint{Protocol: "tcp", IP: net.IPv4(8, 8, 8, 8)},
 			true,
 		},
 		{
 			p2p.NodeAddress{Protocol: "tcp", Hostname: "2001:0db8::ff00:0042:8329"},
-			p2p.Endpoint{Protocol: "tcp", IP: []byte{
+			&p2p.Endpoint{Protocol: "tcp", IP: []byte{
 				0x20, 0x01, 0x0d, 0xb8, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xff, 0x00, 0x00, 0x42, 0x83, 0x29}},
 			true,
 		},
 		{
 			p2p.NodeAddress{Protocol: "tcp", Hostname: "some.missing.host.tendermint.com"},
-			p2p.Endpoint{},
+			&p2p.Endpoint{},
 			false,
 		},
 
 		// Valid non-networked addresses.
 		{
 			p2p.NodeAddress{Protocol: "memory", NodeID: id},
-			p2p.Endpoint{Protocol: "memory", Path: string(id)},
+			&p2p.Endpoint{Protocol: "memory", Path: string(id)},
 			true,
 		},
 		{
 			p2p.NodeAddress{Protocol: "memory", NodeID: id, Path: string(id)},
-			p2p.Endpoint{Protocol: "memory", Path: string(id)},
+			&p2p.Endpoint{Protocol: "memory", Path: string(id)},
 			true,
 		},
 
 		// Invalid addresses.
-		{p2p.NodeAddress{}, p2p.Endpoint{}, false},
-		{p2p.NodeAddress{Hostname: "127.0.0.1"}, p2p.Endpoint{}, false},
-		{p2p.NodeAddress{Protocol: "tcp", Hostname: "127.0.0.1:80"}, p2p.Endpoint{}, false},
-		{p2p.NodeAddress{Protocol: "memory"}, p2p.Endpoint{}, false},
-		{p2p.NodeAddress{Protocol: "memory", Path: string(id)}, p2p.Endpoint{}, false},
-		{p2p.NodeAddress{Protocol: "tcp", Hostname: "💥"}, p2p.Endpoint{}, false},
+		{p2p.NodeAddress{}, &p2p.Endpoint{}, false},
+		{p2p.NodeAddress{Hostname: "127.0.0.1"}, &p2p.Endpoint{}, false},
+		{p2p.NodeAddress{Protocol: "tcp", Hostname: "127.0.0.1:80"}, &p2p.Endpoint{}, false},
+		{p2p.NodeAddress{Protocol: "memory"}, &p2p.Endpoint{}, false},
+		{p2p.NodeAddress{Protocol: "memory", Path: string(id)}, &p2p.Endpoint{}, false},
+		{p2p.NodeAddress{Protocol: "tcp", Hostname: "💥"}, &p2p.Endpoint{}, false},
 	}
 	for _, tc := range testcases {
 		tc := tc

internal/p2p/mocks/transport.go

@@ -62,11 +62,11 @@ func (_m *Transport) Close() error {
 }
 
 // Dial provides a mock function with given fields: _a0, _a1
-func (_m *Transport) Dial(_a0 context.Context, _a1 p2p.Endpoint) (p2p.Connection, error) {
+func (_m *Transport) Dial(_a0 context.Context, _a1 *p2p.Endpoint) (p2p.Connection, error) {
 	ret := _m.Called(_a0, _a1)
 
 	var r0 p2p.Connection
-	if rf, ok := ret.Get(0).(func(context.Context, p2p.Endpoint) p2p.Connection); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, *p2p.Endpoint) p2p.Connection); ok {
 		r0 = rf(_a0, _a1)
 	} else {
 		if ret.Get(0) != nil {
@@ -75,7 +75,7 @@ func (_m *Transport) Dial(_a0 context.Context, _a1 p2p.Endpoint) (p2p.Connection
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, p2p.Endpoint) error); ok {
+	if rf, ok := ret.Get(1).(func(context.Context, *p2p.Endpoint) error); ok {
 		r1 = rf(_a0, _a1)
 	} else {
 		r1 = ret.Error(1)
@@ -84,28 +84,35 @@ func (_m *Transport) Dial(_a0 context.Context, _a1 p2p.Endpoint) (p2p.Connection
 	return r0, r1
 }
 
-// Endpoints provides a mock function with given fields:
-func (_m *Transport) Endpoints() []p2p.Endpoint {
+// Endpoint provides a mock function with given fields:
+func (_m *Transport) Endpoint() (*p2p.Endpoint, error) {
 	ret := _m.Called()
 
-	var r0 []p2p.Endpoint
-	if rf, ok := ret.Get(0).(func() []p2p.Endpoint); ok {
+	var r0 *p2p.Endpoint
+	if rf, ok := ret.Get(0).(func() *p2p.Endpoint); ok {
 		r0 = rf()
 	} else {
 		if ret.Get(0) != nil {
-			r0 = ret.Get(0).([]p2p.Endpoint)
+			r0 = ret.Get(0).(*p2p.Endpoint)
 		}
 	}
 
-	return r0
+	var r1 error
+	if rf, ok := ret.Get(1).(func() error); ok {
+		r1 = rf()
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
 }
 
 // Listen provides a mock function with given fields: _a0
-func (_m *Transport) Listen(_a0 p2p.Endpoint) error {
+func (_m *Transport) Listen(_a0 *p2p.Endpoint) error {
 	ret := _m.Called(_a0)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(p2p.Endpoint) error); ok {
+	if rf, ok := ret.Get(0).(func(*p2p.Endpoint) error); ok {
 		r0 = rf(_a0)
 	} else {
 		r0 = ret.Error(0)

internal/p2p/p2ptest/network.go

@@ -247,7 +247,9 @@ func (n *Network) MakeNode(ctx context.Context, t *testing.T, opts NodeOptions)
 	}
 
 	transport := n.memoryNetwork.CreateTransport(nodeID)
-	require.Len(t, transport.Endpoints(), 1, "transport not listening on 1 endpoint")
+	ep, err := transport.Endpoint()
+	require.NoError(t, err)
+	require.NotNil(t, ep, "transport not listening an endpoint")
 
 	peerManager, err := p2p.NewPeerManager(nodeID, dbm.NewMemDB(), p2p.PeerManagerOptions{
 		MinRetryTime:    10 * time.Millisecond,
@@ -264,8 +266,8 @@ func (n *Network) MakeNode(ctx context.Context, t *testing.T, opts NodeOptions)
 		privKey,
 		peerManager,
 		func() *types.NodeInfo { return &nodeInfo },
-		[]p2p.Transport{transport},
-		transport.Endpoints(),
+		transport,
+		ep,
 		p2p.RouterOptions{DialSleep: func(_ context.Context) {}},
 	)
 
@@ -284,7 +286,7 @@ func (n *Network) MakeNode(ctx context.Context, t *testing.T, opts NodeOptions)
 	return &Node{
 		NodeID:      nodeID,
 		NodeInfo:    nodeInfo,
-		NodeAddress: transport.Endpoints()[0].NodeAddress(nodeID),
+		NodeAddress: ep.NodeAddress(nodeID),
 		PrivKey:     privKey,
 		Router:      router,
 		PeerManager: peerManager,

internal/p2p/router.go

@@ -148,15 +148,14 @@ type Router struct {
 	*service.BaseService
 	logger log.Logger
 
-	metrics            *Metrics
-	options            RouterOptions
-	privKey            crypto.PrivKey
-	peerManager        *PeerManager
-	chDescs            []*ChannelDescriptor
-	transports         []Transport
-	endpoints          []Endpoint
-	connTracker        connectionTracker
-	protocolTransports map[Protocol]Transport
+	metrics     *Metrics
+	options     RouterOptions
+	privKey     crypto.PrivKey
+	peerManager *PeerManager
+	chDescs     []*ChannelDescriptor
+	transport   Transport
+	endpoint    *Endpoint
+	connTracker connectionTracker
 
 	peerMtx    sync.RWMutex
 	peerQueues map[types.NodeID]queue // outbound messages per peer for all channels
@@ -182,8 +181,8 @@ func NewRouter(
 	privKey crypto.PrivKey,
 	peerManager *PeerManager,
 	nodeInfoProducer func() *types.NodeInfo,
-	transports []Transport,
-	endpoints []Endpoint,
+	transport Transport,
+	endpoint *Endpoint,
 	options RouterOptions,
 ) (*Router, error) {
 
@@ -200,28 +199,19 @@ func NewRouter(
 			options.MaxIncomingConnectionAttempts,
 			options.IncomingConnectionWindow,
 		),
-		chDescs:            make([]*ChannelDescriptor, 0),
-		transports:         transports,
-		endpoints:          endpoints,
-		protocolTransports: map[Protocol]Transport{},
-		peerManager:        peerManager,
-		options:            options,
-		channelQueues:      map[ChannelID]queue{},
-		channelMessages:    map[ChannelID]proto.Message{},
-		peerQueues:         map[types.NodeID]queue{},
-		peerChannels:       make(map[types.NodeID]ChannelIDSet),
+		chDescs:         make([]*ChannelDescriptor, 0),
+		transport:       transport,
+		endpoint:        endpoint,
+		peerManager:     peerManager,
+		options:         options,
+		channelQueues:   map[ChannelID]queue{},
+		channelMessages: map[ChannelID]proto.Message{},
+		peerQueues:      map[types.NodeID]queue{},
+		peerChannels:    make(map[types.NodeID]ChannelIDSet),
 	}
 
 	router.BaseService = service.NewBaseService(logger, "router", router)
 
-	for _, transport := range transports {
-		for _, protocol := range transport.Protocols() {
-			if _, ok := router.protocolTransports[protocol]; !ok {
-				router.protocolTransports[protocol] = transport
-			}
-		}
-	}
-
 	return router, nil
 }
 
@@ -286,9 +276,7 @@ func (r *Router) OpenChannel(ctx context.Context, chDesc *ChannelDescriptor) (*C
 	// add the channel to the nodeInfo if it's not already there.
 	r.nodeInfoProducer().AddChannel(uint16(chDesc.ID))
 
-	for _, t := range r.transports {
-		t.AddChannelDescriptors([]*ChannelDescriptor{chDesc})
-	}
+	r.transport.AddChannelDescriptors([]*ChannelDescriptor{chDesc})
 
 	go func() {
 		defer func() {
@@ -670,12 +658,6 @@ func (r *Router) dialPeer(ctx context.Context, address NodeAddress) (Connection,
 	}
 
 	for _, endpoint := range endpoints {
-		transport, ok := r.protocolTransports[endpoint.Protocol]
-		if !ok {
-			r.logger.Error("no transport found for protocol", "endpoint", endpoint)
-			continue
-		}
-
 		dialCtx := ctx
 		if r.options.DialTimeout > 0 {
 			var cancel context.CancelFunc
@@ -690,7 +672,7 @@ func (r *Router) dialPeer(ctx context.Context, address NodeAddress) (Connection,
 		// by the peer's endpoint, since e.g. a peer on 192.168.0.0 can reach us
 		// on a private address on this endpoint, but a peer on the public
 		// Internet can't and needs a different public address.
-		conn, err := transport.Dial(dialCtx, endpoint)
+		conn, err := r.transport.Dial(dialCtx, endpoint)
 		if err != nil {
 			r.logger.Error("failed to dial endpoint", "peer", address.NodeID, "endpoint", endpoint, "err", err)
 		} else {
@@ -950,12 +932,8 @@ func (r *Router) OnStart(ctx context.Context) error {
 		return err
 	}
 
-	for _, transport := range r.transports {
-		for _, endpoint := range r.endpoints {
-			if err := transport.Listen(endpoint); err != nil {
-				return err
-			}
-		}
+	if err := r.transport.Listen(r.endpoint); err != nil {
+		return err
 	}
 
 	nodeInfo := r.nodeInfoProducer()
@@ -964,15 +942,11 @@ func (r *Router) OnStart(ctx context.Context) error {
 		"node_id", nodeInfo.NodeID,
 		"channels", nodeInfo.Channels,
 		"listen_addr", nodeInfo.ListenAddr,
-		"transports", len(r.transports),
 	)
 
 	go r.dialPeers(ctx)
 	go r.evictPeers(ctx)
-
-	for _, transport := range r.transports {
-		go r.acceptPeers(ctx, transport)
-	}
+	go r.acceptPeers(ctx, r.transport)
 
 	return nil
 }
@@ -985,10 +959,8 @@ func (r *Router) OnStart(ctx context.Context) error {
 // sender's responsibility.
 func (r *Router) OnStop() {
 	// Close transport listeners (unblocks Accept calls).
-	for _, transport := range r.transports {
-		if err := transport.Close(); err != nil {
-			r.logger.Error("failed to close transport", "transport", transport, "err", err)
-		}
+	if err := r.transport.Close(); err != nil {
+		r.logger.Error("failed to close transport", "err", err)
 	}
 
 	// Collect all remaining queues, and wait for them to close.

internal/p2p/router_test.go

@@ -105,14 +105,16 @@ func TestRouter_Channel_Basic(t *testing.T) {
 	peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
 	require.NoError(t, err)
 
+	testnet := p2ptest.MakeNetwork(ctx, t, p2ptest.NetworkOptions{NumNodes: 1})
+
 	router, err := p2p.NewRouter(
 		log.NewNopLogger(),
 		p2p.NopMetrics(),
 		selfKey,
 		peerManager,
 		func() *types.NodeInfo { return &selfInfo },
-		nil,
-		nil,
+		testnet.RandomNode().Transport,
+		&p2p.Endpoint{},
 		p2p.RouterOptions{},
 	)
 	require.NoError(t, err)
@@ -396,10 +398,10 @@ func TestRouter_AcceptPeers(t *testing.T) {
 
 			mockTransport := &mocks.Transport{}
 			mockTransport.On("String").Maybe().Return("mock")
-			mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
 			mockTransport.On("Close").Return(nil).Maybe()
 			mockTransport.On("Accept", mock.Anything).Once().Return(mockConnection, nil)
 			mockTransport.On("Accept", mock.Anything).Maybe().Return(nil, io.EOF)
+			mockTransport.On("Listen", mock.Anything).Return(nil)
 
 			// Set up and start the router.
 			peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
@@ -413,7 +415,7 @@ func TestRouter_AcceptPeers(t *testing.T) {
 				selfKey,
 				peerManager,
 				func() *types.NodeInfo { return &selfInfo },
-				[]p2p.Transport{mockTransport},
+				mockTransport,
 				nil,
 				p2p.RouterOptions{},
 			)
@@ -453,9 +455,9 @@ func TestRouter_AcceptPeers_Error(t *testing.T) {
 	// the router from calling Accept again.
 	mockTransport := &mocks.Transport{}
 	mockTransport.On("String").Maybe().Return("mock")
-	mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
 	mockTransport.On("Accept", mock.Anything).Once().Return(nil, errors.New("boom"))
 	mockTransport.On("Close").Return(nil)
+	mockTransport.On("Listen", mock.Anything).Return(nil)
 
 	// Set up and start the router.
 	peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
@@ -467,7 +469,7 @@ func TestRouter_AcceptPeers_Error(t *testing.T) {
 		selfKey,
 		peerManager,
 		func() *types.NodeInfo { return &selfInfo },
-		[]p2p.Transport{mockTransport},
+		mockTransport,
 		nil,
 		p2p.RouterOptions{},
 	)
@@ -490,9 +492,9 @@ func TestRouter_AcceptPeers_ErrorEOF(t *testing.T) {
 	// the router from calling Accept again.
 	mockTransport := &mocks.Transport{}
 	mockTransport.On("String").Maybe().Return("mock")
-	mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
 	mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF)
 	mockTransport.On("Close").Return(nil)
+	mockTransport.On("Listen", mock.Anything).Return(nil)
 
 	// Set up and start the router.
 	peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
@@ -504,7 +506,7 @@ func TestRouter_AcceptPeers_ErrorEOF(t *testing.T) {
 		selfKey,
 		peerManager,
 		func() *types.NodeInfo { return &selfInfo },
-		[]p2p.Transport{mockTransport},
+		mockTransport,
 		nil,
 		p2p.RouterOptions{},
 	)
@@ -538,12 +540,12 @@ func TestRouter_AcceptPeers_HeadOfLineBlocking(t *testing.T) {
 
 	mockTransport := &mocks.Transport{}
 	mockTransport.On("String").Maybe().Return("mock")
-	mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
 	mockTransport.On("Close").Return(nil)
 	mockTransport.On("Accept", mock.Anything).Times(3).Run(func(_ mock.Arguments) {
 		acceptCh <- true
 	}).Return(mockConnection, nil)
 	mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF)
+	mockTransport.On("Listen", mock.Anything).Return(nil)
 
 	// Set up and start the router.
 	peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
@@ -555,7 +557,7 @@ func TestRouter_AcceptPeers_HeadOfLineBlocking(t *testing.T) {
 		selfKey,
 		peerManager,
 		func() *types.NodeInfo { return &selfInfo },
-		[]p2p.Transport{mockTransport},
+		mockTransport,
 		nil,
 		p2p.RouterOptions{},
 	)
@@ -611,7 +613,7 @@ func TestRouter_DialPeers(t *testing.T) {
 			defer cancel()
 
 			address := p2p.NodeAddress{Protocol: "mock", NodeID: tc.dialID}
-			endpoint := p2p.Endpoint{Protocol: "mock", Path: string(tc.dialID)}
+			endpoint := &p2p.Endpoint{Protocol: "mock", Path: string(tc.dialID)}
 
 			// Set up a mock transport that handshakes.
 			connCtx, connCancel := context.WithCancel(context.Background())
@@ -629,8 +631,8 @@ func TestRouter_DialPeers(t *testing.T) {
 
 			mockTransport := &mocks.Transport{}
 			mockTransport.On("String").Maybe().Return("mock")
-			mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
 			mockTransport.On("Close").Return(nil).Maybe()
+			mockTransport.On("Listen", mock.Anything).Return(nil)
 			mockTransport.On("Accept", mock.Anything).Maybe().Return(nil, io.EOF)
 			if tc.dialErr == nil {
 				mockTransport.On("Dial", mock.Anything, endpoint).Once().Return(mockConnection, nil)
@@ -658,7 +660,7 @@ func TestRouter_DialPeers(t *testing.T) {
 				selfKey,
 				peerManager,
 				func() *types.NodeInfo { return &selfInfo },
-				[]p2p.Transport{mockTransport},
+				mockTransport,
 				nil,
 				p2p.RouterOptions{},
 			)
@@ -711,11 +713,11 @@ func TestRouter_DialPeers_Parallel(t *testing.T) {
 
 	mockTransport := &mocks.Transport{}
 	mockTransport.On("String").Maybe().Return("mock")
-	mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
 	mockTransport.On("Close").Return(nil)
+	mockTransport.On("Listen", mock.Anything).Return(nil)
 	mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF)
 	for _, address := range []p2p.NodeAddress{a, b, c} {
-		endpoint := p2p.Endpoint{Protocol: address.Protocol, Path: string(address.NodeID)}
+		endpoint := &p2p.Endpoint{Protocol: address.Protocol, Path: string(address.NodeID)}
 		mockTransport.On("Dial", mock.Anything, endpoint).Run(func(_ mock.Arguments) {
 			dialCh <- true
 		}).Return(mockConnection, nil)
@@ -743,7 +745,7 @@ func TestRouter_DialPeers_Parallel(t *testing.T) {
 		selfKey,
 		peerManager,
 		func() *types.NodeInfo { return &selfInfo },
-		[]p2p.Transport{mockTransport},
+		mockTransport,
 		nil,
 		p2p.RouterOptions{
 			DialSleep: func(_ context.Context) {},
@@ -800,10 +802,10 @@ func TestRouter_EvictPeers(t *testing.T) {
 
 	mockTransport := &mocks.Transport{}
 	mockTransport.On("String").Maybe().Return("mock")
-	mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
 	mockTransport.On("Close").Return(nil)
 	mockTransport.On("Accept", mock.Anything).Once().Return(mockConnection, nil)
 	mockTransport.On("Accept", mock.Anything).Maybe().Return(nil, io.EOF)
+	mockTransport.On("Listen", mock.Anything).Return(nil)
 
 	// Set up and start the router.
 	peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
@@ -817,7 +819,7 @@ func TestRouter_EvictPeers(t *testing.T) {
 		selfKey,
 		peerManager,
 		func() *types.NodeInfo { return &selfInfo },
-		[]p2p.Transport{mockTransport},
+		mockTransport,
 		nil,
 		p2p.RouterOptions{},
 	)
@@ -864,10 +866,10 @@ func TestRouter_ChannelCompatability(t *testing.T) {
 
 	mockTransport := &mocks.Transport{}
 	mockTransport.On("String").Maybe().Return("mock")
-	mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
 	mockTransport.On("Close").Return(nil)
 	mockTransport.On("Accept", mock.Anything).Once().Return(mockConnection, nil)
 	mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF)
+	mockTransport.On("Listen", mock.Anything).Return(nil)
 
 	// Set up and start the router.
 	peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
@@ -879,7 +881,7 @@ func TestRouter_ChannelCompatability(t *testing.T) {
 		selfKey,
 		peerManager,
 		func() *types.NodeInfo { return &selfInfo },
-		[]p2p.Transport{mockTransport},
+		mockTransport,
 		nil,
 		p2p.RouterOptions{},
 	)
@@ -917,10 +919,10 @@ func TestRouter_DontSendOnInvalidChannel(t *testing.T) {
 	mockTransport := &mocks.Transport{}
 	mockTransport.On("AddChannelDescriptors", mock.Anything).Return()
 	mockTransport.On("String").Maybe().Return("mock")
-	mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
 	mockTransport.On("Close").Return(nil)
 	mockTransport.On("Accept", mock.Anything).Once().Return(mockConnection, nil)
 	mockTransport.On("Accept", mock.Anything).Maybe().Return(nil, io.EOF)
+	mockTransport.On("Listen", mock.Anything).Return(nil)
 
 	// Set up and start the router.
 	peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
@@ -934,7 +936,7 @@ func TestRouter_DontSendOnInvalidChannel(t *testing.T) {
 		selfKey,
 		peerManager,
 		func() *types.NodeInfo { return &selfInfo },
-		[]p2p.Transport{mockTransport},
+		mockTransport,
 		nil,
 		p2p.RouterOptions{},
 	)

internal/p2p/transport.go

@@ -24,7 +24,7 @@ type Protocol string
 // Transport is a connection-oriented mechanism for exchanging data with a peer.
 type Transport interface {
 	// Listen starts the transport on the specified endpoint.
-	Listen(Endpoint) error
+	Listen(*Endpoint) error
 
 	// Protocols returns the protocols supported by the transport. The Router
 	// uses this to pick a transport for an Endpoint.
@@ -34,7 +34,7 @@ type Transport interface {
 	//
 	// How to listen is transport-dependent, e.g. MConnTransport uses Listen() while
 	// MemoryTransport starts listening via MemoryNetwork.CreateTransport().
-	Endpoints() []Endpoint
+	Endpoint() (*Endpoint, error)
 
 	// Accept waits for the next inbound connection on a listening endpoint, blocking
 	// until either a connection is available or the transport is closed. On closure,
@@ -42,7 +42,7 @@ type Transport interface {
 	Accept(context.Context) (Connection, error)
 
 	// Dial creates an outbound connection to an endpoint.
-	Dial(context.Context, Endpoint) (Connection, error)
+	Dial(context.Context, *Endpoint) (Connection, error)
 
 	// Close stops accepting new connections, but does not close active connections.
 	Close() error
@@ -129,13 +129,13 @@ type Endpoint struct {
 }
 
 // NewEndpoint constructs an Endpoint from a types.NetAddress structure.
-func NewEndpoint(addr string) (Endpoint, error) {
+func NewEndpoint(addr string) (*Endpoint, error) {
 	ip, port, err := types.ParseAddressString(addr)
 	if err != nil {
-		return Endpoint{}, err
+		return nil, err
 	}
 
-	return Endpoint{
+	return &Endpoint{
 		Protocol: MConnProtocol,
 		IP:       ip,
 		Port:     port,

internal/p2p/transport_mconn.go

@@ -78,25 +78,25 @@ func (m *MConnTransport) Protocols() []Protocol {
 	return []Protocol{MConnProtocol, TCPProtocol}
 }
 
-// Endpoints implements Transport.
-func (m *MConnTransport) Endpoints() []Endpoint {
+// Endpoint implements Transport.
+func (m *MConnTransport) Endpoint() (*Endpoint, error) {
 	if m.listener == nil {
-		return []Endpoint{}
+		return nil, errors.New("listenter not defined")
 	}
 	select {
 	case <-m.doneCh:
-		return []Endpoint{}
+		return nil, errors.New("transport closed")
 	default:
 	}
 
-	endpoint := Endpoint{
+	endpoint := &Endpoint{
 		Protocol: MConnProtocol,
 	}
 	if addr, ok := m.listener.Addr().(*net.TCPAddr); ok {
 		endpoint.IP = addr.IP
 		endpoint.Port = uint16(addr.Port)
 	}
-	return []Endpoint{endpoint}
+	return endpoint, nil
 }
 
 // Listen asynchronously listens for inbound connections on the given endpoint.
@@ -106,7 +106,7 @@ func (m *MConnTransport) Endpoints() []Endpoint {
 // FIXME: Listen currently only supports listening on a single endpoint, it
 // might be useful to support listening on multiple addresses (e.g. IPv4 and
 // IPv6, or a private and public address) via multiple Listen() calls.
-func (m *MConnTransport) Listen(endpoint Endpoint) error {
+func (m *MConnTransport) Listen(endpoint *Endpoint) error {
 	if m.listener != nil {
 		return errors.New("transport is already listening")
 	}
@@ -170,7 +170,7 @@ func (m *MConnTransport) Accept(ctx context.Context) (Connection, error) {
 }
 
 // Dial implements Transport.
-func (m *MConnTransport) Dial(ctx context.Context, endpoint Endpoint) (Connection, error) {
+func (m *MConnTransport) Dial(ctx context.Context, endpoint *Endpoint) (Connection, error) {
 	if err := m.validateEndpoint(endpoint); err != nil {
 		return nil, err
 	}
@@ -217,7 +217,7 @@ func (m *MConnTransport) AddChannelDescriptors(channelDesc []*ChannelDescriptor)
 }
 
 // validateEndpoint validates an endpoint.
-func (m *MConnTransport) validateEndpoint(endpoint Endpoint) error {
+func (m *MConnTransport) validateEndpoint(endpoint *Endpoint) error {
 	if err := endpoint.Validate(); err != nil {
 		return err
 	}

internal/p2p/transport_mconn_test.go

@@ -25,7 +25,7 @@ func init() {
 			[]*p2p.ChannelDescriptor{{ID: chID, Priority: 1}},
 			p2p.MConnTransportOptions{},
 		)
-		err := transport.Listen(p2p.Endpoint{
+		err := transport.Listen(&p2p.Endpoint{
 			Protocol: p2p.MConnProtocol,
 			IP:       net.IPv4(127, 0, 0, 1),
 			Port:     0, // assign a random port
@@ -73,13 +73,14 @@ func TestMConnTransport_AcceptMaxAcceptedConnections(t *testing.T) {
 	t.Cleanup(func() {
 		_ = transport.Close()
 	})
-	err := transport.Listen(p2p.Endpoint{
+	err := transport.Listen(&p2p.Endpoint{
 		Protocol: p2p.MConnProtocol,
 		IP:       net.IPv4(127, 0, 0, 1),
 	})
 	require.NoError(t, err)
-	require.NotEmpty(t, transport.Endpoints())
-	endpoint := transport.Endpoints()[0]
+	endpoint, err := transport.Endpoint()
+	require.NoError(t, err)
+	require.NotNil(t, endpoint)
 
 	// Start a goroutine to just accept any connections.
 	acceptCh := make(chan p2p.Connection, 10)
@@ -132,20 +133,20 @@ func TestMConnTransport_Listen(t *testing.T) {
 	defer cancel()
 
 	testcases := []struct {
-		endpoint p2p.Endpoint
+		endpoint *p2p.Endpoint
 		ok       bool
 	}{
 		// Valid v4 and v6 addresses, with mconn and tcp protocols.
-		{p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv4zero}, true},
-		{p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv4(127, 0, 0, 1)}, true},
-		{p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv6zero}, true},
-		{p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv6loopback}, true},
-		{p2p.Endpoint{Protocol: p2p.TCPProtocol, IP: net.IPv4zero}, true},
+		{&p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv4zero}, true},
+		{&p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv4(127, 0, 0, 1)}, true},
+		{&p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv6zero}, true},
+		{&p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv6loopback}, true},
+		{&p2p.Endpoint{Protocol: p2p.TCPProtocol, IP: net.IPv4zero}, true},
 
 		// Invalid endpoints.
-		{p2p.Endpoint{}, false},
-		{p2p.Endpoint{Protocol: p2p.MConnProtocol, Path: "foo"}, false},
-		{p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv4zero, Path: "foo"}, false},
+		{&p2p.Endpoint{}, false},
+		{&p2p.Endpoint{Protocol: p2p.MConnProtocol, Path: "foo"}, false},
+		{&p2p.Endpoint{Protocol: p2p.MConnProtocol, IP: net.IPv4zero, Path: "foo"}, false},
 	}
 	for _, tc := range testcases {
 		tc := tc
@@ -160,10 +161,12 @@ func TestMConnTransport_Listen(t *testing.T) {
 			)
 
 			// Transport should not listen on any endpoints yet.
-			require.Empty(t, transport.Endpoints())
+			endpoint, err := transport.Endpoint()
+			require.Error(t, err)
+			require.Nil(t, endpoint)
 
 			// Start listening, and check any expected errors.
-			err := transport.Listen(tc.endpoint)
+			err = transport.Listen(tc.endpoint)
 			if !tc.ok {
 				require.Error(t, err)
 				return
@@ -171,9 +174,9 @@ func TestMConnTransport_Listen(t *testing.T) {
 			require.NoError(t, err)
 
 			// Check the endpoint.
-			endpoints := transport.Endpoints()
-			require.Len(t, endpoints, 1)
-			endpoint := endpoints[0]
+			endpoint, err = transport.Endpoint()
+			require.NoError(t, err)
+			require.NotNil(t, endpoint)
 
 			require.Equal(t, p2p.MConnProtocol, endpoint.Protocol)
 			if tc.endpoint.IP.IsUnspecified() {

internal/p2p/transport_memory.go

@@ -119,7 +119,7 @@ func (t *MemoryTransport) String() string {
 	return string(MemoryProtocol)
 }
 
-func (*MemoryTransport) Listen(Endpoint) error { return nil }
+func (*MemoryTransport) Listen(*Endpoint) error { return nil }
 
 func (t *MemoryTransport) AddChannelDescriptors([]*ChannelDescriptor) {}
 
@@ -129,19 +129,19 @@ func (t *MemoryTransport) Protocols() []Protocol {
 }
 
 // Endpoints implements Transport.
-func (t *MemoryTransport) Endpoints() []Endpoint {
+func (t *MemoryTransport) Endpoint() (*Endpoint, error) {
 	if n := t.network.GetTransport(t.nodeID); n == nil {
-		return []Endpoint{}
+		return nil, errors.New("node not defined")
 	}
 
-	return []Endpoint{{
+	return &Endpoint{
 		Protocol: MemoryProtocol,
 		Path:     string(t.nodeID),
 		// An arbitrary IP and port is used in order for the pex
 		// reactor to be able to send addresses to one another.
 		IP:   net.IPv4zero,
 		Port: 0,
-	}}
+	}, nil
 }
 
 // Accept implements Transport.
@@ -158,7 +158,7 @@ func (t *MemoryTransport) Accept(ctx context.Context) (Connection, error) {
 }
 
 // Dial implements Transport.
-func (t *MemoryTransport) Dial(ctx context.Context, endpoint Endpoint) (Connection, error) {
+func (t *MemoryTransport) Dial(ctx context.Context, endpoint *Endpoint) (Connection, error) {
 	if endpoint.Protocol != MemoryProtocol {
 		return nil, fmt.Errorf("invalid protocol %q", endpoint.Protocol)
 	}

internal/p2p/transport_test.go

@@ -87,9 +87,9 @@ func TestTransport_DialEndpoints(t *testing.T) {
 
 	withTransports(ctx, t, func(ctx context.Context, t *testing.T, makeTransport transportFactory) {
 		a := makeTransport(t)
-		endpoints := a.Endpoints()
-		require.NotEmpty(t, endpoints)
-		endpoint := endpoints[0]
+		endpoint, err := a.Endpoint()
+		require.NoError(t, err)
+		require.NotNil(t, endpoint)
 
 		// Spawn a goroutine to simply accept any connections until closed.
 		go func() {
@@ -108,19 +108,19 @@ func TestTransport_DialEndpoints(t *testing.T) {
 		require.NoError(t, conn.Close())
 
 		// Dialing empty endpoint should error.
-		_, err = a.Dial(ctx, p2p.Endpoint{})
+		_, err = a.Dial(ctx, &p2p.Endpoint{})
 		require.Error(t, err)
 
 		// Dialing without protocol should error.
-		noProtocol := endpoint
+		noProtocol := *endpoint
 		noProtocol.Protocol = ""
-		_, err = a.Dial(ctx, noProtocol)
+		_, err = a.Dial(ctx, &noProtocol)
 		require.Error(t, err)
 
 		// Dialing with invalid protocol should error.
-		fooProtocol := endpoint
+		fooProtocol := *endpoint
 		fooProtocol.Protocol = "foo"
-		_, err = a.Dial(ctx, fooProtocol)
+		_, err = a.Dial(ctx, &fooProtocol)
 		require.Error(t, err)
 
 		// Tests for networked endpoints (with IP).
@@ -129,11 +129,12 @@ func TestTransport_DialEndpoints(t *testing.T) {
 				tc := tc
 				t.Run(tc.ip.String(), func(t *testing.T) {
 					e := endpoint
+					require.NotNil(t, e)
 					e.IP = tc.ip
 					conn, err := a.Dial(ctx, e)
 					if tc.ok {
-						require.NoError(t, conn.Close())
 						require.NoError(t, err)
+						require.NoError(t, conn.Close())
 					} else {
 						require.Error(t, err, "endpoint=%s", e)
 					}
@@ -167,16 +168,18 @@ func TestTransport_Dial(t *testing.T) {
 		a := makeTransport(t)
 		b := makeTransport(t)
 
-		require.NotEmpty(t, a.Endpoints())
-		require.NotEmpty(t, b.Endpoints())
-		aEndpoint := a.Endpoints()[0]
-		bEndpoint := b.Endpoints()[0]
+		aEndpoint, err := a.Endpoint()
+		require.NoError(t, err)
+		require.NotNil(t, aEndpoint)
+		bEndpoint, err := b.Endpoint()
+		require.NoError(t, err)
+		require.NotNil(t, bEndpoint)
 
 		// Context cancellation should error. We can't test timeouts since we'd
 		// need a non-responsive endpoint.
 		cancelCtx, cancel := context.WithCancel(ctx)
 		cancel()
-		_, err := a.Dial(cancelCtx, bEndpoint)
+		_, err = a.Dial(cancelCtx, bEndpoint)
 		require.Error(t, err)
 
 		// Unavailable endpoint should error.
@@ -210,21 +213,26 @@ func TestTransport_Endpoints(t *testing.T) {
 		b := makeTransport(t)
 
 		// Both transports return valid and different endpoints.
-		aEndpoints := a.Endpoints()
-		bEndpoints := b.Endpoints()
-		require.NotEmpty(t, aEndpoints)
-		require.NotEmpty(t, bEndpoints)
-		require.NotEqual(t, aEndpoints, bEndpoints)
-		for _, endpoint := range append(aEndpoints, bEndpoints...) {
+		aEndpoint, err := a.Endpoint()
+		require.NoError(t, err)
+		require.NotNil(t, aEndpoint)
+		bEndpoint, err := b.Endpoint()
+		require.NoError(t, err)
+		require.NotNil(t, bEndpoint)
+		require.NotEqual(t, aEndpoint, bEndpoint)
+		for _, endpoint := range []*p2p.Endpoint{aEndpoint, bEndpoint} {
 			err := endpoint.Validate()
 			require.NoError(t, err, "invalid endpoint %q", endpoint)
 		}
 
 		// When closed, the transport should no longer return any endpoints.
-		err := a.Close()
+		require.NoError(t, a.Close())
+		aEndpoint, err = a.Endpoint()
+		require.Error(t, err)
+		require.Nil(t, aEndpoint)
+		bEndpoint, err = b.Endpoint()
 		require.NoError(t, err)
-		require.Empty(t, a.Endpoints())
-		require.NotEmpty(t, b.Endpoints())
+		require.NotNil(t, bEndpoint)
 	})
 }
 
@@ -235,13 +243,12 @@ func TestTransport_Protocols(t *testing.T) {
 	withTransports(ctx, t, func(ctx context.Context, t *testing.T, makeTransport transportFactory) {
 		a := makeTransport(t)
 		protocols := a.Protocols()
-		endpoints := a.Endpoints()
+		endpoint, err := a.Endpoint()
+		require.NoError(t, err)
 		require.NotEmpty(t, protocols)
-		require.NotEmpty(t, endpoints)
+		require.NotNil(t, endpoint)
 
-		for _, endpoint := range endpoints {
-			require.Contains(t, protocols, endpoint.Protocol)
-		}
+		require.Contains(t, protocols, endpoint.Protocol)
 	})
 }
 
@@ -595,8 +602,9 @@ func TestEndpoint_Validate(t *testing.T) {
 func dialAccept(ctx context.Context, t *testing.T, a, b p2p.Transport) (p2p.Connection, p2p.Connection) {
 	t.Helper()
 
-	endpoints := b.Endpoints()
-	require.NotEmpty(t, endpoints, "peer not listening on any endpoints")
+	endpoint, err := b.Endpoint()
+	require.NoError(t, err)
+	require.NotNil(t, endpoint, "peer not listening on any endpoints")
 
 	ctx, cancel := context.WithTimeout(ctx, time.Second)
 	defer cancel()
@@ -609,7 +617,7 @@ func dialAccept(ctx context.Context, t *testing.T, a, b p2p.Transport) (p2p.Conn
 		acceptCh <- conn
 	}()
 
-	dialConn, err := a.Dial(ctx, endpoints[0])
+	dialConn, err := a.Dial(ctx, endpoint)
 	require.NoError(t, err)
 
 	acceptConn := <-acceptCh

node/setup.go

@@ -310,8 +310,8 @@ func createRouter(
 		nodeKey.PrivKey,
 		peerManager,
 		nodeInfoProducer,
-		[]p2p.Transport{transport},
-		[]p2p.Endpoint{ep},
+		transport,
+		ep,
 		getRouterConfig(cfg, appClient),
 	)
 }