p2p: enable p2p http txsync

components/mocks/mockNetwork.go

@@ -18,6 +18,7 @@ package mocks
 
 import (
 	"context"
+	"errors"
 	"net"
 	"net/http"
 
@@ -76,11 +77,6 @@ func (network *MockNetwork) GetPeers(options ...network.PeerOption) []network.Pe
 	return nil
 }
 
-// GetRoundTripper -- returns the network round tripper
-func (network *MockNetwork) GetRoundTripper(peer network.Peer) http.RoundTripper {
-	return http.DefaultTransport
-}
-
 // Ready - always ready
 func (network *MockNetwork) Ready() chan struct{} {
 	c := make(chan struct{})
@@ -115,3 +111,8 @@ func (network *MockNetwork) GetGenesisID() string {
 	}
 	return network.GenesisID
 }
+
+// GetHTTPClient returns a http.Client with a suitable for the network
+func (network *MockNetwork) GetHTTPClient(p network.HTTPPeer) (*http.Client, error) {
+	return nil, errors.New("not implemented")
+}

network/addr_test.go

@@ -89,10 +89,12 @@ func TestParseHostOrURL(t *testing.T) {
 		t.Run(addr, func(t *testing.T) {
 			_, err := ParseHostOrURL(addr)
 			require.Error(t, err, "url should fail", addr)
+			require.False(t, IsMultiaddr(addr))
 		})
 		t.Run(addr+"-multiaddr", func(t *testing.T) {
 			_, err := ParseHostOrURLOrMultiaddr(addr)
 			require.Error(t, err, "url should fail", addr)
+			require.False(t, IsMultiaddr(addr))
 		})
 	}
 

network/gossipNode.go

@@ -83,8 +83,9 @@ type GossipNode interface {
 	// ClearHandlers deregisters all the existing message handlers.
 	ClearHandlers()
 
-	// GetRoundTripper returns a Transport that would limit the number of outgoing connections.
-	GetRoundTripper(peer Peer) http.RoundTripper
+	// GetHTTPClient returns a http.Client with a suitable for the network Transport
+	// that would also limit the number of outgoing connections.
+	GetHTTPClient(peer HTTPPeer) (*http.Client, error)
 
 	// OnNetworkAdvance notifies the network library that the agreement protocol was able to make a notable progress.
 	// this is the only indication that we have that we haven't formed a clique, where all incoming messages

network/hybridNetwork.go

@@ -180,14 +180,14 @@
 	n.wsNetwork.ClearHandlers()
 }
 
-// GetRoundTripper returns a Transport that would limit the number of outgoing connections.
-func (n *HybridP2PNetwork) GetRoundTripper(peer Peer) http.RoundTripper {
-	// TODO today this is used by HTTPTxSync.Sync after calling GetPeers(network.PeersPhonebookRelays)
-	switch p := peer.(type) {
+// GetHTTPClient returns a http.Client with a suitable for the network Transport
+// that would also limit the number of outgoing connections.
+func (n *HybridP2PNetwork) GetHTTPClient(peer HTTPPeer) (*http.Client, error) {
+	switch peer.(type) {
 	case *wsPeer:
-		return p.net.GetRoundTripper(peer)
-	case gossipSubPeer:
-		return p.net.GetRoundTripper(peer)
+		return n.wsNetwork.GetHTTPClient(peer)
+	case *wsPeerCore:
+		return n.p2pNetwork.GetHTTPClient(peer)
 	default:
 		panic("unrecognized peer type")
 	}

network/p2p/capabilities_test.go

@@ -98,7 +98,7 @@ func setupDHTHosts(t *testing.T, numHosts int) []*dht.IpfsDHT {
 		require.NoError(t, err)
 		// this is a workaround for the following issue
 		// "failed to negotiate security protocol: error reading handshake message: noise: message is too short"
-		// it appears simultenous connectino attempts (dht.New() attempts to connect) causes this handshake error.
+		// it appears simultaneous connection attempts (dht.New() attempts to connect) causes this handshake error.
 		// https://github.com/libp2p/go-libp2p-noise/issues/70
 		time.Sleep(200 * time.Millisecond)
 
@@ -158,7 +158,7 @@ func setupCapDiscovery(t *testing.T, numHosts int, numBootstrapPeers int) []*Cap
 		require.NoError(t, err)
 		// this is a workaround for the following issue
 		// "failed to negotiate security protocol: error reading handshake message: noise: message is too short"
-		// it appears simultenous connectino attempts (dht.New() attempts to connect) causes this handshake error.
+		// it appears simultaneous connection attempts (dht.New() attempts to connect) causes this handshake error.
 		// https://github.com/libp2p/go-libp2p-noise/issues/70
 		time.Sleep(200 * time.Millisecond)
 

network/p2p/http.go

@@ -18,17 +18,47 @@
 
 import (
 	"net/http"
+	"sync"
 
+	"github.com/gorilla/mux"
 	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/peer"
 	libp2phttp "github.com/libp2p/go-libp2p/p2p/http"
 )
 
-// MakeHTTPClient creates a http.Client that uses libp2p transport for a goven protocol and peer address.
-func MakeHTTPClient(protocolID string, addrInfo peer.AddrInfo) (http.Client, error) {
+// algorandP2pHTTPProtocol defines a libp2p protocol name for algorand's http over p2p messages
+const algorandP2pHTTPProtocol = "/algorand-http/1.0.0"
+
+// HTTPServer is a wrapper around libp2phttp.Host that allows registering http handlers with path parameters.
+type HTTPServer struct {
+	libp2phttp.Host
+	p2phttpMux              *mux.Router
+	p2phttpMuxRegistrarOnce sync.Once
+}
+
+// MakeHTTPServer creates a new HTTPServer
+func MakeHTTPServer(streamHost host.Host) *HTTPServer {
+	httpServer := HTTPServer{
+		Host:       libp2phttp.Host{StreamHost: streamHost},
+		p2phttpMux: mux.NewRouter(),
+	}
+	return &httpServer
+}
+
+// RegisterHTTPHandler registers a http handler with a given path.
+func (s *HTTPServer) RegisterHTTPHandler(path string, handler http.Handler) {
+	s.p2phttpMux.Handle(path, handler)
+	s.p2phttpMuxRegistrarOnce.Do(func() {
+		s.Host.SetHTTPHandlerAtPath(algorandP2pHTTPProtocol, "/", s.p2phttpMux)
+	})
+}
+
+// MakeHTTPClient creates a http.Client that uses libp2p transport for a given protocol and peer address.
+func MakeHTTPClient(addrInfo *peer.AddrInfo) (*http.Client, error) {
 	clientStreamHost, err := libp2p.New(libp2p.NoListenAddrs)
 	if err != nil {
-		return http.Client{}, err
+		return nil, err
 	}
 
 	client := libp2phttp.Host{StreamHost: clientStreamHost}
@@ -37,10 +67,10 @@
 	// to make a NamespaceRoundTripper that limits to specific URL paths.
 	// First, we do not want make requests when listing peers (the main MakeHTTPClient invoker).
 	// Secondly, this makes unit testing easier - no need to register fake handlers.
-	rt, err := client.NewConstrainedRoundTripper(addrInfo)
+	rt, err := client.NewConstrainedRoundTripper(*addrInfo)
 	if err != nil {
-		return http.Client{}, err
+		return nil, err
 	}
 
-	return http.Client{Transport: rt}, nil
+	return &http.Client{Transport: rt}, nil
 }

network/p2p/p2p.go

@@ -75,9 +75,6 @@ type serviceImpl struct {
 // AlgorandWsProtocol defines a libp2p protocol name for algorand's websockets messages
 const AlgorandWsProtocol = "/algorand-ws/1.0.0"
 
-// AlgorandP2pHTTPProtocol defines a libp2p protocol name for algorand's http over p2p messages
-const AlgorandP2pHTTPProtocol = "/algorand-http/1.0.0"
-
 const dialTimeout = 30 * time.Second
 
 // MakeHost creates a libp2p host but does not start listening.

network/p2p/streams.go

@@ -80,7 +80,7 @@
 				if stream.Stat().Direction == network.DirUnknown {
 					n.log.Warnf("Unknown direction for a steam %s to/from %s", stream.ID(), remotePeer)
 				} else {
-					n.log.Warnf("Unexpected outgoing sream in streamHandler for connection %s (%s): %s vs %s stream", stream.Conn().ID(), remotePeer, stream.Conn().Stat().Direction, stream.Stat().Direction.String())
+					n.log.Warnf("Unexpected outgoing stream in streamHandler for connection %s (%s): %s vs %s stream", stream.Conn().ID(), remotePeer, stream.Conn().Stat().Direction, stream.Stat().Direction.String())
 				}
 			}
 			n.handler(n.ctx, remotePeer, stream, incoming)
@@ -98,7 +98,7 @@
 		if stream.Stat().Direction == network.DirUnknown {
 			n.log.Warnf("streamHandler: unknown direction for a steam %s to/from %s", stream.ID(), remotePeer)
 		} else {
-			n.log.Warnf("Unexpected outgoing sream in streamHandler for connection %s (%s): %s vs %s stream", stream.Conn().ID(), remotePeer, stream.Conn().Stat().Direction, stream.Stat().Direction.String())
+			n.log.Warnf("Unexpected outgoing stream in streamHandler for connection %s (%s): %s vs %s stream", stream.Conn().ID(), remotePeer, stream.Conn().Stat().Direction, stream.Stat().Direction.String())
 		}
 	}
 	n.handler(n.ctx, remotePeer, stream, incoming)
@@ -141,7 +141,7 @@
 	// a new stream created above, expected direction is outbound
 	incoming := stream.Stat().Direction == network.DirInbound
 	if incoming {
-		n.log.Warnf("Unexpected incoming sream in streamHandler for connection %s (%s): %s vs %s stream", stream.Conn().ID(), remotePeer, stream.Conn().Stat().Direction, stream.Stat().Direction.String())
+		n.log.Warnf("Unexpected incoming stream in streamHandler for connection %s (%s): %s vs %s stream", stream.Conn().ID(), remotePeer, stream.Conn().Stat().Direction, stream.Stat().Direction.String())
 	} else {
 		if stream.Stat().Direction == network.DirUnknown {
 			n.log.Warnf("Connected: unknown direction for a steam %s to/from %s", stream.ID(), remotePeer)

network/p2p/testing/httpNode.go

@@ -0,0 +1,116 @@
+// Copyright (C) 2019-2024 Algorand, Inc.
+// This file is part of go-algorand
+//
+// go-algorand is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// go-algorand is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with go-algorand.  If not, see <https://www.gnu.org/licenses/>.
+
+// This package wraps and re-exports the libp2p functions on order to keep
+// all go-libp2p imports in one place.
+
+package p2p
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/algorand/go-algorand/components/mocks"
+	"github.com/algorand/go-algorand/network"
+	"github.com/algorand/go-algorand/network/p2p"
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/stretchr/testify/require"
+)
+
+// HTTPNode is a mock network node that uses libp2p and http.
+type HTTPNode struct {
+	mocks.MockNetwork
+	host.Host
+	httpServer *p2p.HTTPServer
+	peers      []network.Peer
+	tb         testing.TB
+	genesisID  string
+}
+
+// MakeHTTPNode returns a new P2PHTTPNode node.
+func MakeHTTPNode(tb testing.TB) *HTTPNode {
+	p2pHost, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0"))
+	require.NoError(tb, err)
+
+	return &HTTPNode{
+		Host:       p2pHost,
+		httpServer: p2p.MakeHTTPServer(p2pHost),
+		tb:         tb,
+	}
+}
+
+// RegisterHTTPHandler registers a http handler with a given path.
+func (p *HTTPNode) RegisterHTTPHandler(path string, handler http.Handler) {
+	p.httpServer.RegisterHTTPHandler(path, handler)
+}
+
+// RegisterHandlers not implemented.
+func (p *HTTPNode) RegisterHandlers(dispatch []network.TaggedMessageHandler) {}
+
+// Start starts http service
+func (p *HTTPNode) Start() error {
+	go func() {
+		err := p.httpServer.Serve()
+		require.NoError(p.tb, err)
+	}()
+	return nil
+}
+
+// Stop stops http service
+func (p *HTTPNode) Stop() {
+	p.httpServer.Close()
+	p.Host.Close()
+}
+
+// GetGenesisID returns genesisID
+func (p *HTTPNode) GetGenesisID() string { return p.genesisID }
+
+// SetGenesisID sets genesisID
+func (p *HTTPNode) SetGenesisID(genesisID string) { p.genesisID = genesisID }
+
+type httpPeer struct {
+	addrInfo peer.AddrInfo
+	tb       testing.TB
+}
+
+// GetAddress implements HTTPPeer interface returns the address of the peer
+func (p httpPeer) GetAddress() string {
+	mas, err := peer.AddrInfoToP2pAddrs(&p.addrInfo)
+	require.NoError(p.tb, err)
+	require.Len(p.tb, mas, 1)
+	return mas[0].String()
+}
+
+// GetAddress implements HTTPPeer interface and returns the http client for a peer
+func (p httpPeer) GetHTTPClient() *http.Client {
+	c, err := p2p.MakeHTTPClient(&p.addrInfo)
+	require.NoError(p.tb, err)
+	return c
+}
+
+// SetPeers sets peers
+func (p *HTTPNode) SetPeers(other *HTTPNode) {
+	addrInfo := peer.AddrInfo{ID: other.ID(), Addrs: other.Addrs()}
+	hpeer := httpPeer{addrInfo, p.tb}
+	p.peers = append(p.peers, hpeer)
+}
+
+// GetPeers returns peers
+func (p *HTTPNode) GetPeers(options ...network.PeerOption) []network.Peer {
+	return p.peers
+}