Merge branch 'main' into fix/sessioninterestmanager

Author: guillaumemichel

CHANGELOG.md

@@ -25,6 +25,13 @@ The following emojis are used to highlight certain changes:
 ### Security
 
 
+## [v0.27.3]
+
+### Added
+
+- `provider`: Added `ReprovideInterval` and `LastRun` stats to the Reprovider [#815](https://github.com/ipfs/boxo/pull/815)
+
+
 ## [v0.27.2]
 
 ### Fixed

bitswap/bitswap.go

@@ -9,16 +9,14 @@ import (
 	"github.com/ipfs/boxo/bitswap/network"
 	"github.com/ipfs/boxo/bitswap/server"
 	"github.com/ipfs/boxo/bitswap/tracer"
-	"github.com/ipfs/go-metrics-interface"
-
 	blockstore "github.com/ipfs/boxo/blockstore"
 	exchange "github.com/ipfs/boxo/exchange"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
+	"github.com/ipfs/go-metrics-interface"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/routing"
-
 	"go.uber.org/multierr"
 )
 

bitswap/client/internal/getter/getter.go

@@ -6,11 +6,10 @@ import (
 
 	"github.com/ipfs/boxo/bitswap/client/internal"
 	notifications "github.com/ipfs/boxo/bitswap/client/internal/notifications"
-	logging "github.com/ipfs/go-log/v2"
-
 	blocks "github.com/ipfs/go-block-format"
 	cid "github.com/ipfs/go-cid"
 	ipld "github.com/ipfs/go-ipld-format"
+	logging "github.com/ipfs/go-log/v2"
 )
 
 var log = logging.Logger("bitswap/client/getter")

bitswap/client/internal/peermanager/peermanager.go

@@ -4,10 +4,9 @@ import (
 	"context"
 	"sync"
 
+	cid "github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/ipfs/go-metrics-interface"
-
-	cid "github.com/ipfs/go-cid"
 	peer "github.com/libp2p/go-libp2p/core/peer"
 )
 

bitswap/client/internal/peermanager/peerwantmanager.go

@@ -158,7 +158,6 @@ func (pwm *peerWantManager) broadcastWantHaves(wantHaves []cid.Cid) {
 // sendWants only sends the peer the want-blocks and want-haves that have not
 // already been sent to it.
 func (pwm *peerWantManager) sendWants(p peer.ID, wantBlocks []cid.Cid, wantHaves []cid.Cid) {
-
 	// Get the existing want-blocks and want-haves for the peer
 	pws, ok := pwm.peerWants[p]
 	if !ok {
@@ -284,58 +283,51 @@ func (pwm *peerWantManager) sendCancels(cancelKs []cid.Cid) {
 		}
 	}
 
-	if len(broadcastCancels) > 0 {
-		// If a broadcast want is being cancelled, send the cancel to all
-		// peers
-		for p, pws := range pwm.peerWants {
-			send(p, pws)
-		}
-	} else {
-		// Only send cancels to peers that received a corresponding want
-		cancelPeers := make(map[peer.ID]struct{}, len(pwm.wantPeers[cancelKs[0]]))
-		for _, c := range cancelKs {
-			for p := range pwm.wantPeers[c] {
-				cancelPeers[p] = struct{}{}
-			}
-		}
-		for p := range cancelPeers {
-			pws, ok := pwm.peerWants[p]
-			if !ok {
-				// Should never happen but check just in case
-				log.Errorf("sendCancels - peerWantManager index missing peer %s", p)
-				continue
-			}
-
-			send(p, pws)
-		}
-	}
-
-	// Decrement the wants gauges
-	for _, c := range cancelKs {
+	clearWantsForCID := func(c cid.Cid) {
 		peerCnts := peerCounts[c]
-
 		// If there were any peers that had a pending want-block for the key
 		if peerCnts.wantBlock > 0 {
 			// Decrement the want-block gauge
 			pwm.wantBlockGauge.Dec()
 		}
-
 		// If there was a peer that had a pending want or it was a broadcast want
 		if peerCnts.wanted() {
 			// Decrement the total wants gauge
 			pwm.wantGauge.Dec()
 		}
+		delete(pwm.wantPeers, c)
 	}
 
-	// Remove cancelled broadcast wants
-	for _, c := range broadcastCancels {
-		pwm.broadcastWants.Remove(c)
-	}
+	if len(broadcastCancels) > 0 {
+		// If a broadcast want is being cancelled, send the cancel to all
+		// peers
+		for p, pws := range pwm.peerWants {
+			send(p, pws)
+		}
 
-	// Batch-remove the reverse-index. There's no need to clear this index
-	// peer-by-peer.
-	for _, c := range cancelKs {
-		delete(pwm.wantPeers, c)
+		// Remove cancelled broadcast wants
+		for _, c := range broadcastCancels {
+			pwm.broadcastWants.Remove(c)
+		}
+
+		for _, c := range cancelKs {
+			clearWantsForCID(c)
+		}
+	} else {
+		// Only send cancels to peers that received a corresponding want
+		for _, c := range cancelKs {
+			for p := range pwm.wantPeers[c] {
+				pws, ok := pwm.peerWants[p]
+				if !ok {
+					// Should never happen but check just in case
+					log.Errorf("sendCancels - peerWantManager index missing peer %s", p)
+					continue
+				}
+				send(p, pws)
+			}
+
+			clearWantsForCID(c)
+		}
 	}
 }
 

bitswap/client/internal/session/session_test.go

@@ -145,6 +145,7 @@ func (pm *fakePeerManager) UnregisterSession(uint64)              {}
 func (pm *fakePeerManager) SendWants(context.Context, peer.ID, []cid.Cid, []cid.Cid) bool {
 	return true
 }
+
 func (pm *fakePeerManager) BroadcastWantHaves(ctx context.Context, cids []cid.Cid) {
 	select {
 	case pm.wantReqs <- wantReq{cids}:
@@ -287,7 +288,7 @@ func TestSessionFindMorePeers(t *testing.T) {
 		// Make sure the first block is not included because it has already
 		// been received
 		for _, c := range receivedWantReq.cids {
-			require.False(t, c.Equals(cids[0]), "should not braodcast block that was already received")
+			require.False(t, c.Equals(cids[0]), "should not broadcast block that was already received")
 		}
 	case <-ctx.Done():
 		t.Fatal("Never rebroadcast want list")

bitswap/client/internal/session/sessionwantsender.go

@@ -4,7 +4,6 @@ import (
 	"context"
 
 	bsbpm "github.com/ipfs/boxo/bitswap/client/internal/blockpresencemanager"
-
 	cid "github.com/ipfs/go-cid"
 	peer "github.com/libp2p/go-libp2p/core/peer"
 )

bitswap/client/internal/sessioninterestmanager/sessioninterestmanager.go

@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	blocks "github.com/ipfs/go-block-format"
-
 	cid "github.com/ipfs/go-cid"
 )
 

bitswap/client/internal/sessionmanager/sessionmanager.go

@@ -6,18 +6,17 @@ import (
 	"sync"
 	"time"
 
-	cid "github.com/ipfs/go-cid"
-	delay "github.com/ipfs/go-ipfs-delay"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
-
 	"github.com/ipfs/boxo/bitswap/client/internal"
 	bsbpm "github.com/ipfs/boxo/bitswap/client/internal/blockpresencemanager"
 	notifications "github.com/ipfs/boxo/bitswap/client/internal/notifications"
 	bssession "github.com/ipfs/boxo/bitswap/client/internal/session"
 	bssim "github.com/ipfs/boxo/bitswap/client/internal/sessioninterestmanager"
 	exchange "github.com/ipfs/boxo/exchange"
+	cid "github.com/ipfs/go-cid"
+	delay "github.com/ipfs/go-ipfs-delay"
 	peer "github.com/libp2p/go-libp2p/core/peer"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 // Session is a session that is managed by the session manager

bitswap/client/internal/sessionpeermanager/sessionpeermanager.go

@@ -5,7 +5,6 @@ import (
 	"sync"
 
 	logging "github.com/ipfs/go-log/v2"
-
 	peer "github.com/libp2p/go-libp2p/core/peer"
 )