p2p: rate limit outgoing p2p http connections

catchup/ledgerFetcher.go

@@ -33,6 +33,7 @@ import (
 	"github.com/algorand/go-algorand/ledger/encoded"
 	"github.com/algorand/go-algorand/logging"
 	"github.com/algorand/go-algorand/network"
+	"github.com/algorand/go-algorand/network/addr"
 	"github.com/algorand/go-algorand/rpcs"
 	"github.com/algorand/go-algorand/util"
 )
@@ -75,11 +76,11 @@ func makeLedgerFetcher(net network.GossipNode, accessor ledger.CatchpointCatchup
 
 func (lf *ledgerFetcher) requestLedger(ctx context.Context, peer network.HTTPPeer, round basics.Round, method string) (*http.Response, error) {
 	var ledgerURL string
-	if network.IsMultiaddr(peer.GetAddress()) {
+	if addr.IsMultiaddr(peer.GetAddress()) {
 		ledgerURL = network.SubstituteGenesisID(lf.net, "/v1/{genesisID}/ledger/"+strconv.FormatUint(uint64(round), 36))
 	} else {
 
-		parsedURL, err := network.ParseHostOrURL(peer.GetAddress())
+		parsedURL, err := addr.ParseHostOrURL(peer.GetAddress())
 		if err != nil {
 			return nil, err
 		}

catchup/universalFetcher.go

@@ -32,6 +32,7 @@ import (
 	"github.com/algorand/go-algorand/data/bookkeeping"
 	"github.com/algorand/go-algorand/logging"
 	"github.com/algorand/go-algorand/network"
+	"github.com/algorand/go-algorand/network/addr"
 	"github.com/algorand/go-algorand/protocol"
 	"github.com/algorand/go-algorand/rpcs"
 )
@@ -221,10 +222,10 @@ type HTTPFetcher struct {
 func (hf *HTTPFetcher) getBlockBytes(ctx context.Context, r basics.Round) (data []byte, err error) {
 	var blockURL string
 
-	if network.IsMultiaddr(hf.rootURL) {
+	if addr.IsMultiaddr(hf.rootURL) {
 		blockURL = rpcs.FormatBlockQuery(uint64(r), "", hf.net)
 	} else {
-		if parsedURL, err0 := network.ParseHostOrURL(hf.rootURL); err0 == nil {
+		if parsedURL, err0 := addr.ParseHostOrURL(hf.rootURL); err0 == nil {
 			parsedURL.Path = rpcs.FormatBlockQuery(uint64(r), parsedURL.Path, hf.net)
 			blockURL = parsedURL.String()
 		} else {

cmd/algod/main.go

@@ -32,7 +32,7 @@ import (
 	"github.com/algorand/go-algorand/data/bookkeeping"
 	"github.com/algorand/go-algorand/logging"
 	"github.com/algorand/go-algorand/logging/telemetryspec"
-	"github.com/algorand/go-algorand/network"
+	"github.com/algorand/go-algorand/network/addr"
 	"github.com/algorand/go-algorand/protocol"
 	toolsnet "github.com/algorand/go-algorand/tools/network"
 	"github.com/algorand/go-algorand/util"
@@ -276,7 +276,7 @@ func run() int {
 
 		// make sure that the format of each entry is valid:
 		for idx, peer := range peerOverrideArray {
-			addr, addrErr := network.ParseHostOrURLOrMultiaddr(peer)
+			addr, addrErr := addr.ParseHostOrURLOrMultiaddr(peer)
 			if addrErr != nil {
 				fmt.Fprintf(os.Stderr, "Provided command line parameter '%s' is not a valid host:port pair\n", peer)
 				return 1

cmd/goal/node.go

@@ -39,7 +39,7 @@ import (
 	"github.com/algorand/go-algorand/daemon/algod/api/server/v2/generated/model"
 	"github.com/algorand/go-algorand/ledger/ledgercore"
 	"github.com/algorand/go-algorand/libgoal"
-	"github.com/algorand/go-algorand/network"
+	naddr "github.com/algorand/go-algorand/network/addr"
 	"github.com/algorand/go-algorand/nodecontrol"
 	"github.com/algorand/go-algorand/util"
 	"github.com/algorand/go-algorand/util/tokens"
@@ -751,7 +751,7 @@ func verifyPeerDialArg() bool {
 
 	// make sure that the format of each entry is valid:
 	for _, peer := range strings.Split(peerDial, ";") {
-		_, err := network.ParseHostOrURLOrMultiaddr(peer)
+		_, err := naddr.ParseHostOrURLOrMultiaddr(peer)
 		if err != nil {
 			reportErrorf("Provided peer '%s' is not a valid peer address : %v", peer, err)
 			return false

network/addr.go

@@ -17,82 +17,17 @@
 package network
 
 import (
-	"errors"
-	"net/url"
 	"path"
-	"regexp"
 	"strings"
 
-	"github.com/multiformats/go-multiaddr"
+	"github.com/algorand/go-algorand/network/addr"
 )
 
-var errURLNoHost = errors.New("could not parse a host from url")
-
-var errURLColonHost = errors.New("host name starts with a colon")
-
-// HostColonPortPattern matches "^[-a-zA-Z0-9.]+:\\d+$" e.g. "foo.com.:1234"
-var HostColonPortPattern = regexp.MustCompile(`^[-a-zA-Z0-9.]+:\d+$`)
-
-// ParseHostOrURL handles "host:port" or a full URL.
-// Standard library net/url.Parse chokes on "host:port".
-func ParseHostOrURL(addr string) (*url.URL, error) {
-	// If the entire addr is "host:port" grab that right away.
-	// Don't try url.Parse() because that will grab "host:" as if it were "scheme:"
-	if HostColonPortPattern.MatchString(addr) {
-		return &url.URL{Scheme: "http", Host: addr}, nil
-	}
-	parsed, err := url.Parse(addr)
-	if err == nil {
-		if parsed.Host == "" {
-			return nil, errURLNoHost
-		}
-		return parsed, nil
-	}
-	if strings.HasPrefix(addr, "http:") || strings.HasPrefix(addr, "https:") || strings.HasPrefix(addr, "ws:") || strings.HasPrefix(addr, "wss:") || strings.HasPrefix(addr, "://") || strings.HasPrefix(addr, "//") {
-		return parsed, err
-	}
-	// This turns "[::]:4601" into "http://[::]:4601" which url.Parse can do
-	parsed, e2 := url.Parse("http://" + addr)
-	if e2 == nil {
-		// https://datatracker.ietf.org/doc/html/rfc1123#section-2
-		// first character is relaxed to allow either a letter or a digit
-		if parsed.Host[0] == ':' && (len(parsed.Host) < 2 || parsed.Host[1] != ':') {
-			return nil, errURLColonHost
-		}
-		return parsed, nil
-	}
-	return parsed, err /* return original err, not our prefix altered try */
-}
-
-// IsMultiaddr returns true if the provided string is a valid multiaddr.
-func IsMultiaddr(addr string) bool {
-	if strings.HasPrefix(addr, "/") && !strings.HasPrefix(addr, "//") { // multiaddr starts with '/' but not '//' which is possible for scheme relative URLS
-		_, err := multiaddr.NewMultiaddr(addr)
-		return err == nil
-	}
-	return false
-}
-
-// ParseHostOrURLOrMultiaddr returns an error if it could not parse the provided
-// string as a valid "host:port", full URL, or multiaddr. If no error, it returns
-// a host:port address, or a multiaddr.
-func ParseHostOrURLOrMultiaddr(addr string) (string, error) {
-	if strings.HasPrefix(addr, "/") && !strings.HasPrefix(addr, "//") { // multiaddr starts with '/' but not '//' which is possible for scheme relative URLS
-		_, err := multiaddr.NewMultiaddr(addr)
-		return addr, err
-	}
-	url, err := ParseHostOrURL(addr)
-	if err != nil {
-		return "", err
-	}
-	return url.Host, nil
-}
-
 // addrToGossipAddr parses host:port or a URL and returns the URL to the websocket interface at that address.
-func (wn *WebsocketNetwork) addrToGossipAddr(addr string) (string, error) {
-	parsedURL, err := ParseHostOrURL(addr)
+func (wn *WebsocketNetwork) addrToGossipAddr(a string) (string, error) {
+	parsedURL, err := addr.ParseHostOrURL(a)
 	if err != nil {
-		wn.log.Warnf("could not parse addr %#v: %s", addr, err)
+		wn.log.Warnf("could not parse addr %#v: %s", a, err)
 		return "", errBadAddr
 	}
 	parsedURL.Scheme = websocketsScheme[parsedURL.Scheme]

network/addr/addr.go

@@ -0,0 +1,88 @@
+// 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/>.
+
+package addr
+
+import (
+	"errors"
+	"net/url"
+	"regexp"
+	"strings"
+
+	"github.com/multiformats/go-multiaddr"
+)
+
+var errURLNoHost = errors.New("could not parse a host from url")
+
+var errURLColonHost = errors.New("host name starts with a colon")
+
+// HostColonPortPattern matches "^[-a-zA-Z0-9.]+:\\d+$" e.g. "foo.com.:1234"
+var HostColonPortPattern = regexp.MustCompile(`^[-a-zA-Z0-9.]+:\d+$`)
+
+// ParseHostOrURL handles "host:port" or a full URL.
+// Standard library net/url.Parse chokes on "host:port".
+func ParseHostOrURL(addr string) (*url.URL, error) {
+	// If the entire addr is "host:port" grab that right away.
+	// Don't try url.Parse() because that will grab "host:" as if it were "scheme:"
+	if HostColonPortPattern.MatchString(addr) {
+		return &url.URL{Scheme: "http", Host: addr}, nil
+	}
+	parsed, err := url.Parse(addr)
+	if err == nil {
+		if parsed.Host == "" {
+			return nil, errURLNoHost
+		}
+		return parsed, nil
+	}
+	if strings.HasPrefix(addr, "http:") || strings.HasPrefix(addr, "https:") || strings.HasPrefix(addr, "ws:") || strings.HasPrefix(addr, "wss:") || strings.HasPrefix(addr, "://") || strings.HasPrefix(addr, "//") {
+		return parsed, err
+	}
+	// This turns "[::]:4601" into "http://[::]:4601" which url.Parse can do
+	parsed, e2 := url.Parse("http://" + addr)
+	if e2 == nil {
+		// https://datatracker.ietf.org/doc/html/rfc1123#section-2
+		// first character is relaxed to allow either a letter or a digit
+		if parsed.Host[0] == ':' && (len(parsed.Host) < 2 || parsed.Host[1] != ':') {
+			return nil, errURLColonHost
+		}
+		return parsed, nil
+	}
+	return parsed, err /* return original err, not our prefix altered try */
+}
+
+// IsMultiaddr returns true if the provided string is a valid multiaddr.
+func IsMultiaddr(addr string) bool {
+	if strings.HasPrefix(addr, "/") && !strings.HasPrefix(addr, "//") { // multiaddr starts with '/' but not '//' which is possible for scheme relative URLS
+		_, err := multiaddr.NewMultiaddr(addr)
+		return err == nil
+	}
+	return false
+}
+
+// ParseHostOrURLOrMultiaddr returns an error if it could not parse the provided
+// string as a valid "host:port", full URL, or multiaddr. If no error, it returns
+// a host:port address, or a multiaddr.
+func ParseHostOrURLOrMultiaddr(addr string) (string, error) {
+	if strings.HasPrefix(addr, "/") && !strings.HasPrefix(addr, "//") { // multiaddr starts with '/' but not '//' which is possible for scheme relative URLS
+		_, err := multiaddr.NewMultiaddr(addr)
+		return addr, err
+	}
+	url, err := ParseHostOrURL(addr)
+	if err != nil {
+		return "", err
+	}
+	return url.Host, nil
+}

network/addr_test.go → network/addr/addr_test.go

@@ -14,7 +14,7 @@
 // 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/>.
 
-package network
+package addr
 
 import (
 	"net/url"

network/dialer.go → network/limitcaller/dialer.go

@@ -14,13 +14,14 @@
 // 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/>.
 
-package network
+package limitcaller
 
 import (
 	"context"
 	"net"
 	"time"
 
+	"github.com/algorand/go-algorand/network/phonebook"
 	"github.com/algorand/go-algorand/tools/network/dnssec"
 	"github.com/algorand/go-algorand/util"
 )
@@ -31,14 +32,13 @@ type netDialer interface {
 
 // Dialer establish tcp-level connection with the destination
 type Dialer struct {
-	phonebook   Phonebook
+	phonebook   phonebook.Phonebook
 	innerDialer netDialer
-	resolver    *net.Resolver
 }
 
-// makeRateLimitingDialer creates a rate limiting dialer that would limit the connections
+// MakeRateLimitingDialer creates a rate limiting dialer that would limit the connections
 // according to the entries in the phonebook.
-func makeRateLimitingDialer(phonebook Phonebook, resolver dnssec.ResolverIf) Dialer {
+func MakeRateLimitingDialer(phonebook phonebook.Phonebook, resolver dnssec.ResolverIf) Dialer {
 	var innerDialer netDialer = &net.Dialer{
 		Timeout:   30 * time.Second,
 		KeepAlive: 30 * time.Second,

network/rateLimitingTransport.go → network/limitcaller/rateLimitingTransport.go

@@ -14,7 +14,7 @@
 // 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/>.
 
-package network
+package limitcaller
 
 import (
 	"errors"
@@ -24,22 +24,32 @@ import (
 	"github.com/algorand/go-algorand/util"
 )
 
-// rateLimitingTransport is the transport for execute a single HTTP transaction, obtaining the Response for a given Request.
-type rateLimitingTransport struct {
-	phonebook       Phonebook
-	innerTransport  *http.Transport
+// ConnectionTimeStore is a subset of the phonebook that is used to store the connection times.
+type ConnectionTimeStore interface {
+	GetConnectionWaitTime(addrOrInfo interface{}) (bool, time.Duration, time.Time)
+	UpdateConnectionTime(addrOrInfo interface{}, provisionalTime time.Time) bool
+}
+
+// RateLimitingTransport is the transport for execute a single HTTP transaction, obtaining the Response for a given Request.
+type RateLimitingTransport struct {
+	phonebook       ConnectionTimeStore
+	innerTransport  http.RoundTripper
 	queueingTimeout time.Duration
+	targetAddr      interface{} // target address for the p2p http request
 }
 
+// DefaultQueueingTimeout is the default timeout for queueing the request.
+const DefaultQueueingTimeout = 10 * time.Second
+
 // ErrConnectionQueueingTimeout indicates that we've exceeded the time allocated for
 // queueing the current request before the request attempt could be made.
 var ErrConnectionQueueingTimeout = errors.New("rateLimitingTransport: queueing timeout")
 
-// makeRateLimitingTransport creates a rate limiting http transport that would limit the requests rate
+// MakeRateLimitingTransport creates a rate limiting http transport that would limit the requests rate
 // according to the entries in the phonebook.
-func makeRateLimitingTransport(phonebook Phonebook, queueingTimeout time.Duration, dialer *Dialer, maxIdleConnsPerHost int) rateLimitingTransport {
+func MakeRateLimitingTransport(phonebook ConnectionTimeStore, queueingTimeout time.Duration, dialer *Dialer, maxIdleConnsPerHost int) RateLimitingTransport {
 	defaultTransport := http.DefaultTransport.(*http.Transport)
-	return rateLimitingTransport{
+	return RateLimitingTransport{
 		phonebook: phonebook,
 		innerTransport: &http.Transport{
 			Proxy:                 defaultTransport.Proxy,
@@ -54,14 +64,30 @@ func makeRateLimitingTransport(phonebook Phonebook, queueingTimeout time.Duratio
 	}
 }
 
+// MakeRateLimitingTransportWithTransport creates a rate limiting http transport that would limit the requests rate
+// according to the entries in the phonebook.
+func MakeRateLimitingTransportWithTransport(phonebook ConnectionTimeStore, queueingTimeout time.Duration, rt http.RoundTripper, target interface{}, maxIdleConnsPerHost int) RateLimitingTransport {
+	return RateLimitingTransport{
+		phonebook:       phonebook,
+		innerTransport:  rt,
+		queueingTimeout: queueingTimeout,
+		targetAddr:      target,
+	}
+}
+
 // RoundTrip connects to the address on the named network using the provided context.
 // It waits if needed not to exceed connectionsRateLimitingCount.
-func (r *rateLimitingTransport) RoundTrip(req *http.Request) (res *http.Response, err error) {
+func (r *RateLimitingTransport) RoundTrip(req *http.Request) (res *http.Response, err error) {
 	var waitTime time.Duration
 	var provisionalTime time.Time
 	queueingDeadline := time.Now().Add(r.queueingTimeout)
+	var host interface{} = req.Host
+	if len(req.Host) == 0 && req.URL != nil && len(req.URL.Host) == 0 {
+		// p2p/http clients have per-connection transport and address info so use that
+		host = r.targetAddr
+	}
 	for {
-		_, waitTime, provisionalTime = r.phonebook.GetConnectionWaitTime(req.Host)
+		_, waitTime, provisionalTime = r.phonebook.GetConnectionWaitTime(host)
 		if waitTime == 0 {
 			break // break out of the loop and proceed to the connection
 		}
@@ -73,6 +99,6 @@ func (r *rateLimitingTransport) RoundTrip(req *http.Request) (res *http.Response
 		return nil, ErrConnectionQueueingTimeout
 	}
 	res, err = r.innerTransport.RoundTrip(req)
-	r.phonebook.UpdateConnectionTime(req.Host, provisionalTime)
+	r.phonebook.UpdateConnectionTime(host, provisionalTime)
 	return
 }