fix leak in blockresponsemanager

SessionManager.ReceiveFrom added HAVEs/DONT_HAVEs even if no sessions were intersted in those, so the following could happen:
Session requests BLOCK from one peer and HAVE from another. BLOCK is received before HAVE. Session cleans up its interest on BLOCK,
but the following HAVE is still added to BlockResponseManager leaving it there forever.

The solution is to add a CID to BlockResponseManager only if there a session interested in it.

Author: Wondertan

bitswap/client/internal/session/session.go

@@ -189,13 +189,6 @@ func (s *Session) Shutdown() {
 
 // ReceiveFrom receives incoming blocks from the given peer.
 func (s *Session) ReceiveFrom(from peer.ID, ks []cid.Cid, haves []cid.Cid, dontHaves []cid.Cid) {
-	// The SessionManager tells each Session about all keys that it may be
-	// interested in. Here the Session filters the keys to the ones that this
-	// particular Session is interested in.
-	interestedRes := s.sim.FilterSessionInterested(s.id, ks, haves, dontHaves)
-	ks = interestedRes[0]
-	haves = interestedRes[1]
-	dontHaves = interestedRes[2]
 	s.logReceiveFrom(from, ks, haves, dontHaves)
 
 	// Inform the session want sender that a message has been received

bitswap/client/internal/sessioninterestmanager/sessioninterestmanager.go

@@ -175,27 +175,36 @@ func (sim *SessionInterestManager) SplitWantedUnwanted(blks []blocks.Block) ([]b
 
 // When the SessionManager receives a message it calls InterestedSessions() to
 // find out which sessions are interested in the message.
-func (sim *SessionInterestManager) InterestedSessions(blks []cid.Cid, haves []cid.Cid, dontHaves []cid.Cid) []uint64 {
+func (sim *SessionInterestManager) InterestedSessions(sets ...[]cid.Cid) map[uint64][][]cid.Cid {
 	sim.lk.RLock()
 	defer sim.lk.RUnlock()
 
-	ks := make([]cid.Cid, 0, len(blks)+len(haves)+len(dontHaves))
-	ks = append(ks, blks...)
-	ks = append(ks, haves...)
-	ks = append(ks, dontHaves...)
-
 	// Create a set of sessions that are interested in the keys
-	sesSet := make(map[uint64]struct{})
-	for _, c := range ks {
-		for s := range sim.wants[c] {
-			sesSet[s] = struct{}{}
+	sesSet := make(map[uint64][][]cid.Cid)
+	for i, set := range sets {
+		for _, c := range set {
+			sessions, ok := sim.wants[c]
+			if !ok {
+				continue
+			}
+
+			for ses, ok := range sessions {
+				if !ok {
+					continue
+				}
+
+				if _, ok := sesSet[ses]; !ok {
+					sessionSets := make([][]cid.Cid, len(sets))
+					for i, set := range sets {
+						sets[i] = make([]cid.Cid, 0, len(set))
+					}
+					sesSet[ses] = sessionSets
+				}
+
+				sesSet[ses][i] = append(sesSet[ses][i], c)
+			}
 		}
 	}
 
-	// Convert the set into a list
-	ses := make([]uint64, 0, len(sesSet))
-	for s := range sesSet {
-		ses = append(ses, s)
-	}
-	return ses
+	return sesSet
 }

bitswap/client/internal/sessionmanager/sessionmanager.go

@@ -154,11 +154,8 @@ func (sm *SessionManager) GetNextSessionID() uint64 {
 
 // ReceiveFrom is called when a new message is received
 func (sm *SessionManager) ReceiveFrom(ctx context.Context, p peer.ID, blks []cid.Cid, haves []cid.Cid, dontHaves []cid.Cid) {
-	// Record block presence for HAVE / DONT_HAVE
-	sm.blockPresenceManager.ReceiveFrom(p, haves, dontHaves)
-
 	// Notify each session that is interested in the blocks / HAVEs / DONT_HAVEs
-	for _, id := range sm.sessionInterestManager.InterestedSessions(blks, haves, dontHaves) {
+	for id, keys := range sm.sessionInterestManager.InterestedSessions(blks, haves, dontHaves) {
 		sm.sessLk.Lock()
 		if sm.sessions == nil { // check if SessionManager was shutdown
 			sm.sessLk.Unlock()
@@ -168,6 +165,13 @@ func (sm *SessionManager) ReceiveFrom(ctx context.Context, p peer.ID, blks []cid
 		sm.sessLk.Unlock()
 
 		if ok {
+			blks = keys[0]
+			haves = keys[1]
+			dontHaves = keys[2]
+			// Record block presence for HAVE / DONT_HAVE
+			// must be called before Seession.ReceiveFrom
+			sm.blockPresenceManager.ReceiveFrom(p, haves, dontHaves)
+
 			sess.ReceiveFrom(p, blks, haves, dontHaves)
 		}
 	}