Repository: ipfs/boxo

Issue:
bitswap/httpnet: Disconnect from peers eventually
This was mentioned but I don't think it is tracked in boxo so opening this for tracking.

Currently, when an http-provider record is found for something, we open a connection to the endpoint and request what we need to request.

Unless things are 500-ing or giving any sort of that ends up as a "fatal" error, we don't disconnect and keep "peered" to the endpoints, which start getting optimistic-bitswap requests for everything.

State of things:

- A MsgSender is a peer ID with one several http urls attached (from provider records). Requests are sent to this urls sorted by serverErrors, cooldown periods.
- Each url tracks a count of "serverErrors". When all urls have exhausted their max serverErrors, bestURL() returns an error which is converted into a fatalError, which triggers a disconnect from the peer.
- Currently, clientErrors like NotFound, Gone, Forbidden and redirects do not cause disconnects regardless of how many times they happen.

Desired goal:

- We should stop making optimistic http requests to http endpoints at some point, while ensuring we keep 

Options:

- We could not make optimistic requests at all, and only make them based on a provider-record basis. The connections will die per their idle timeouts. Implementation wise this is ugly because at the bitswap-network layer we have no information on whether a peer and a CID are related by a provider-record and we just get wantlists that need to be sent to a peer. We could track cid->peer relations by augmenting the Connect() method, which gets called directly from the provider query manager.
- We could set a per-peer timeout which gets reset on a successful request and otherwise disconnects peers after a while.
- We could set a per-error timeout which disconnects peers when a global clientError threshold is crossed.

I think the simplest approach is the latter (per-error):

- Easy to implement
- On a node with a lot of activity making requests to and endpoint that doesn't have anything  we request, we ensure we don't make too many requests for no reason.
- On a node with very low activity making a few requests to an endpoint, we let the http idle timeouts do their work or otherwise keep peered.
- Note that "Disconnecting" means removing HTTP  addresses associated to a peer, stop pinging the peer and telling bitswap to not send anymore wantlists to the peer. It does not immediately stop requests, there might be wantlists in process of being sent that will still be sent. However, given average wantlist size of ~2 during normal conditions, this means not meany requests will be leaked after crossing the limit.
- Downsides: if the error limit is 100, a peer with 1% success rate will still be connected forever. Down the line we probably want to introduce configuration options per peer ID, so that we can effectively configure thresholds per endpoint.

cc https://github.com/ipshipyard/roadmaps/issues/9

Discussion:
@aschmahmann: > Easy to implement
> Downsides: if the error limit is 100, a peer with 1% success rate will still be connected forever

If it's easy seems reasonable to test with given the current alternative isn't great. However, this could come back to bite us if effectively the "optimistic request / spamming-if-your-guesses-are-bad" part of the system ends up breaking downloading sessions by getting us a pile of consecutive errors.

My understanding is that given we're going to be doing spam reduction / being better at optimistic guessing anyway (https://github.com/ipshipyard/roadmaps/issues/10) that the thing that's missing here is effectively that the pinger code shouldn't be active on peers who we're not going to send optimistic requests to and in the HTTP case that'll let Go handle when the connection gets cleaned up.

cc @gammazero it might help to take what we have so far regarding spam and put it in an issue to see if there's sane overlap or not.@lidel: Triage question:
- @hsanjuan  was this closed by #918 or more work needs to happen ehre

Solution patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index be7fed4b..5bf70daf 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -16,6 +16,13 @@ The following emojis are used to highlight certain changes:
 
 ### Added
 
+- `bitswap/httpnet`: New option `WithMaxDontHaveErrors(int)` (defaults to 100)
+  will trigger disconnections from bitswap peers that cannot provide any of
+  the blocks that are requested for the given number of requests in a
+  row.. This is meant to limit bitswap HTTP-based optimistic requests for
+  blocks to discovered endpoints, which were before considered permanently
+  peered upon discovery.
+
 ### Changed
 
 ### Removed
diff --git a/bitswap/network/httpnet/error_tracker.go b/bitswap/network/httpnet/error_tracker.go
new file mode 100644
index 00000000..72eb6ae1
--- /dev/null
+++ b/bitswap/network/httpnet/error_tracker.go
@@ -0,0 +1,48 @@
+package httpnet
+
+import (
+	"errors"
+	"sync"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+)
+
+var errThresholdCrossed = errors.New("the peer crossed the error threshold")
+
+type errorTracker struct {
+	ht *Network
+
+	mux    sync.RWMutex
+	errors map[peer.ID]int
+}
+
+func newErrorTracker(ht *Network) *errorTracker {
+	return &errorTracker{
+		ht:     ht,
+		errors: make(map[peer.ID]int),
+	}
+}
+
+func (et *errorTracker) stopTracking(p peer.ID) {
+	et.mux.Lock()
+	delete(et.errors, p)
+	et.mux.Unlock()
+}
+
+// logErrors adds n to the current error count for p. If the total count is above the threshold, then an error is returned. If n is 0, the the total count is reset to 0.
+func (et *errorTracker) logErrors(p peer.ID, n int, threshold int) error {
+	et.mux.Lock()
+	defer et.mux.Unlock()
+
+	if n == 0 { // reset error count
+		delete(et.errors, p)
+		return nil
+	}
+	count := et.errors[p]
+	total := count + n
+	et.errors[p] = total
+	if total > threshold {
+		return errThresholdCrossed
+	}
+	return nil
+}
diff --git a/bitswap/network/httpnet/httpnet.go b/bitswap/network/httpnet/httpnet.go
index 7f5dc108..939e4e4a 100644
--- a/bitswap/network/httpnet/httpnet.go
+++ b/bitswap/network/httpnet/httpnet.go
@@ -51,6 +51,7 @@ const (
 	DefaultInsecureSkipVerify            = false
 	DefaultMaxBackoff                    = time.Minute
 	DefaultMaxHTTPAddressesPerPeer       = 10
+	DefaultMaxDontHaveErrors             = 100
 	DefaultHTTPWorkers                   = 64
 )
 
@@ -155,6 +156,19 @@ func WithHTTPWorkers(n int) Option {
 	}
 }
 
+// WithMaxDontHaveErrors sets the maximum number of client errors that a peer
+// can cause in a row before we disconnect. For example, if set to 50, and a
+// peer returns 404 to 50 requests in a row, we will disconnect and signal the
+// upper layers to stop making requests to this peer and its endpoints. It may
+// be that pending requests will still happen. The HTTP connection might be
+// kept until it times-out per the IdleConnTimeout. Requests will resume if a
+// provider record is found causing us to "reconnect" to the peer.
+func WithMaxDontHaveErrors(threshold int) Option {
+	return func(net *Network) {
+		net.maxDontHaveErrors = threshold
+	}
+}
+
 type Network struct {
 	// NOTE: Stats must be at the top of the heap allocation to ensure 64bit
 	// alignment.
@@ -168,6 +182,7 @@ type Network struct {
 	receivers       []network.Receiver
 	connEvtMgr      *network.ConnectEventManager
 	pinger          *pinger
+	errorTracker    *errorTracker
 	requestTracker  *requestTracker
 	cooldownTracker *cooldownTracker
 
@@ -180,6 +195,7 @@ type Network struct {
 	maxIdleConns            int
 	insecureSkipVerify      bool
 	maxHTTPAddressesPerPeer int
+	maxDontHaveErrors       int
 	httpWorkers             int
 	allowlist               map[string]struct{}
 	denylist                map[string]struct{}
@@ -215,6 +231,7 @@ func New(host host.Host, opts ...Option) network.BitSwapNetwork {
 		maxIdleConns:            DefaultMaxIdleConns,
 		insecureSkipVerify:      DefaultInsecureSkipVerify,
 		maxHTTPAddressesPerPeer: DefaultMaxHTTPAddressesPerPeer,
+		maxDontHaveErrors:       DefaultMaxDontHaveErrors,
 		httpWorkers:             DefaultHTTPWorkers,
 		httpRequests:            make(chan httpRequestInfo),
 	}
@@ -293,6 +310,9 @@ func New(host host.Host, opts ...Option) network.BitSwapNetwork {
 	pinger := newPinger(htnet, pingCid)
 	htnet.pinger = pinger
 
+	et := newErrorTracker(htnet)
+	htnet.errorTracker = et
+
 	for i := 0; i < htnet.httpWorkers; i++ {
 		go htnet.httpWorker(i)
 	}
@@ -520,6 +540,7 @@ func (ht *Network) DisconnectFrom(ctx context.Context, p peer.ID) error {
 		ht.host.Peerstore().SetAddrs(p, bsaddrs.Addrs, peerstore.TempAddrTTL)
 	}
 	ht.pinger.stopPinging(p)
+	ht.errorTracker.stopTracking(p)
 
 	// coolDownTracker: we leave untouched. We want to keep
 	// ongoing cooldowns there in case we reconnect to this peer.
diff --git a/bitswap/network/httpnet/msg_sender.go b/bitswap/network/httpnet/msg_sender.go
index f4457b64..0b44b93f 100644
--- a/bitswap/network/httpnet/msg_sender.go
+++ b/bitswap/network/httpnet/msg_sender.go
@@ -502,6 +502,7 @@ WANTLIST_LOOP:
 	go func() {
 		bsresp := bsmsg.New(false)
 		totalResponses := 0
+		totalClientErrors := 0
 
 		for result := range resultsCollector {
 			// Record total request time.
@@ -521,12 +522,13 @@ WANTLIST_LOOP:
 				// error handling
 				switch result.err.Type {
 				case typeFatal:
-					log.Errorf("Disconnecting from %s: %s", sender.peer, result.err.Err)
+					log.Errorf("fatal error. Disconnecting from %s: %s", sender.peer, result.err.Err)
 					sender.ht.DisconnectFrom(ctx, sender.peer)
 					err = result.err
 					// continue processing responses as workers
 					// might have done other requests in parallel
 				case typeClient:
+					totalClientErrors++
 					if entry.SendDontHave {
 						bsresp.AddDontHave(entry.Cid)
 					}
@@ -545,6 +547,13 @@ WANTLIST_LOOP:
 			}
 		}
 
+		// if totalClientErrors == 0, count is reset.
+		if err := sender.ht.errorTracker.logErrors(sender.peer, totalClientErrors, sender.ht.maxDontHaveErrors); err != nil {
+			log.Debugf("too many client errors. Disconnecting from %s", sender.peer)
+			sender.ht.DisconnectFrom(ctx, sender.peer)
+
+		}
+
 		// We return a special "cancel" function that we need to call
 		// explicitally. This cleans up our request-tracker.
 		for _, cancel := range entryCancels {



Test patch:
diff --git a/bitswap/network/httpnet/error_tracker_test.go b/bitswap/network/httpnet/error_tracker_test.go
new file mode 100644
index 00000000..fd7bec11
--- /dev/null
+++ b/bitswap/network/httpnet/error_tracker_test.go
@@ -0,0 +1,98 @@
+package httpnet
+
+// Write tests for the errorTracker implementation found in watcher.go
+import (
+	"sync"
+	"testing"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+)
+
+func TestErrorTracker_StopTracking(t *testing.T) {
+	et := newErrorTracker(&Network{})
+	p := peer.ID("testpeer")
+
+	// Stop tracking the peer
+	et.stopTracking(p)
+
+	// Check if the error count is removed
+	if _, ok := et.errors[p]; ok {
+		t.Errorf("Expected peer %s to be untracked but it was still tracked", p)
+	}
+}
+
+func TestErrorTracker_LogErrors_Reset(t *testing.T) {
+	et := newErrorTracker(&Network{})
+	p := peer.ID("testpeer")
+
+	// Log some errors
+	err := et.logErrors(p, 5, 10)
+	if err != nil {
+		t.Errorf("Expected no error when logging errors but got %v", err)
+	}
+
+	// Reset error count
+	err = et.logErrors(p, 0, 10)
+	if err != nil {
+		t.Errorf("Expected no error when resetting error count but got %v", err)
+	}
+
+	// Check if the error count is reset to 0
+	count := et.errors[p]
+	if count != 0 {
+		t.Errorf("Expected error count for peer %s to be 0 after reset but got %d", p, count)
+	}
+}
+
+func TestErrorTracker_LogErrors_ThresholdCrossed(t *testing.T) {
+	et := newErrorTracker(&Network{})
+	p := peer.ID("testpeer")
+
+	// Log errors until threshold is crossed
+	err := et.logErrors(p, 11, 10)
+	if err != errThresholdCrossed {
+		t.Errorf("Expected errorThresholdCrossed when logging errors above threshold but got %v", err)
+	}
+
+	// Check if the error count reflects the logged errors
+	count, ok := et.errors[p]
+	if !ok {
+		t.Errorf("Expected peer %s to be tracked but it was not", p)
+	}
+	if count != 11 {
+		t.Errorf("Expected error count for peer %s to be 10 after logging errors above threshold but got %d", p, count)
+	}
+}
+
+// Write a test that tests concurrent access to the methods
+func TestErrorTracker_ConcurrentAccess(t *testing.T) {
+	et := newErrorTracker(&Network{})
+	p := peer.ID("testpeer")
+
+	var wg sync.WaitGroup
+	numRoutines := 10
+	threshold := 100
+
+	for i := 0; i < numRoutines; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for j := 0; j < int(threshold)/numRoutines; j++ {
+				et.logErrors(p, 1, threshold)
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	// Check if the error count is correct
+	count, ok := et.errors[p]
+	if !ok {
+		t.Errorf("Expected peer %s to be tracked but it was not", p)
+	}
+	expectedCount := threshold
+	actualCount := count
+	if actualCount != expectedCount {
+		t.Errorf("Expected error count for peer %s to be %d after concurrent logging but got %d", p, expectedCount, actualCount)
+	}
+}
diff --git a/bitswap/network/httpnet/httpnet_test.go b/bitswap/network/httpnet/httpnet_test.go
index d900b9e7..8ed06781 100644
--- a/bitswap/network/httpnet/httpnet_test.go
+++ b/bitswap/network/httpnet/httpnet_test.go
@@ -34,10 +34,12 @@ var backoffCid = cid.MustParse("bafkreid6g5qrufgqj46djic7ntjnppaj5bg4urppjoyywrx
 var _ network.Receiver = (*mockRecv)(nil)
 
 type mockRecv struct {
-	blocks    map[cid.Cid]struct{}
-	haves     map[cid.Cid]struct{}
-	donthaves map[cid.Cid]struct{}
-	waitCh    chan struct{}
+	blocks             map[cid.Cid]struct{}
+	haves              map[cid.Cid]struct{}
+	donthaves          map[cid.Cid]struct{}
+	waitCh             chan struct{}
+	waitConnectedCh    chan struct{}
+	waitDisconnectedCh chan struct{}
 }
 
 func (recv *mockRecv) ReceiveMessage(ctx context.Context, sender peer.ID, incoming bsmsg.BitSwapMessage) {
@@ -56,8 +58,8 @@ func (recv *mockRecv) ReceiveMessage(ctx context.Context, sender peer.ID, incomi
 	recv.waitCh <- struct{}{}
 }
 
-func (recv *mockRecv) wait(seconds time.Duration) error {
-	ctx, cancel := context.WithTimeout(context.Background(), seconds*time.Second)
+func (recv *mockRecv) wait(seconds int) error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(seconds)*time.Second)
 	defer cancel()
 	select {
 	case <-ctx.Done():
@@ -67,25 +69,49 @@ func (recv *mockRecv) wait(seconds time.Duration) error {
 	}
 }
 
+func (recv *mockRecv) waitConnected(seconds int) error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(seconds)*time.Second)
+	defer cancel()
+	select {
+	case <-ctx.Done():
+		return errors.New("receiver waited too long without receiving a connect event")
+	case <-recv.waitConnectedCh:
+		return nil
+	}
+}
+
+func (recv *mockRecv) waitDisconnected(seconds int) error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(seconds)*time.Second)
+	defer cancel()
+	select {
+	case <-ctx.Done():
+		return errors.New("receiver waited too long without receiving a disconnect event")
+	case <-recv.waitDisconnectedCh:
+		return nil
+	}
+}
+
 func (recv *mockRecv) ReceiveError(err error) {
 
 }
 
 func (recv *mockRecv) PeerConnected(p peer.ID) {
-
+	recv.waitConnectedCh <- struct{}{}
 }
 
 func (recv *mockRecv) PeerDisconnected(p peer.ID) {
-
+	recv.waitDisconnectedCh <- struct{}{}
 }
 
 func mockReceiver(t *testing.T) *mockRecv {
 	t.Helper()
 	return &mockRecv{
-		blocks:    make(map[cid.Cid]struct{}),
-		haves:     make(map[cid.Cid]struct{}),
-		donthaves: make(map[cid.Cid]struct{}),
-		waitCh:    make(chan struct{}, 1),
+		blocks:             make(map[cid.Cid]struct{}),
+		haves:              make(map[cid.Cid]struct{}),
+		donthaves:          make(map[cid.Cid]struct{}),
+		waitCh:             make(chan struct{}, 1),
+		waitConnectedCh:    make(chan struct{}, 1),
+		waitDisconnectedCh: make(chan struct{}, 1),
 	}
 
 }
@@ -559,3 +585,49 @@ func TestBackOff(t *testing.T) {
 		t.Error("no blocks should have been received while on backoff")
 	}
 }
+
+// Write a TestErrorTracking function which tests that a peer is disconnected when the treshold is crossed.
+func TestErrorTracking(t *testing.T) {
+	ctx := context.Background()
+	recv := mockReceiver(t)
+	htnet, mn := mockNetwork(t, recv, WithMaxDontHaveErrors(1))
+
+	peer, err := mn.GenPeer()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	msrv := makeServer(t, 0, 0)
+	connectToPeer(t, ctx, htnet, peer, msrv)
+
+	err = recv.waitConnected(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	wl := makeCids(t, 0, 1)
+	msg := makeWantsMessage(wl)
+
+	err = htnet.SendMessage(ctx, peer.ID(), msg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	recv.wait(1)
+	err = recv.waitDisconnected(1)
+	if err == nil { // we received a disconnect event
+		t.Fatal("disconnect event not expected")
+	}
+
+	// Threshold was 1. This will trigger a disconnection.
+	err = htnet.SendMessage(ctx, peer.ID(), msg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	recv.wait(1)
+	err = recv.waitDisconnected(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+}