Add peer filters per Request

pkg/retriever/protocolsplitter.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/filecoin-project/lassie/pkg/types"
+	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/multiformats/go-multicodec"
 )
 
@@ -18,21 +19,48 @@ func NewProtocolSplitter(protocols []multicodec.Code) types.CandidateSplitter[mu
 }
 
 func (ps *ProtocolSplitter) SplitRetrievalRequest(ctx context.Context, request types.RetrievalRequest, events func(types.RetrievalEvent)) types.RetrievalSplitter[multicodec.Code] {
+	filteredPeers := make(map[peer.ID]peerFilter, len(request.FilteredPeers))
+	for _, filteredPeer := range request.FilteredPeers {
+		existing := filteredPeers[filteredPeer.Peer]
+		excludeAll := existing.excludeAll || filteredPeer.ExcludeAll
+		protocolSet := existing.protocolsSet
+		if protocolSet == nil {
+			protocolSet = make(map[multicodec.Code]struct{})
+		}
+		for _, protocol := range filteredPeer.ExcludedProtocols {
+			protocolSet[protocol] = struct{}{}
+		}
+		filteredPeers[filteredPeer.Peer] = peerFilter{excludeAll, protocolSet}
+	}
+	return &retrievalProtocolSplitter{ps, request.GetSupportedProtocols(ps.protocols), filteredPeers}
+}
 
-	return &retrievalProtocolSplitter{ps, request.GetSupportedProtocols(ps.protocols)}
+type peerFilter struct {
+	excludeAll   bool
+	protocolsSet map[multicodec.Code]struct{}
 }
 
 type retrievalProtocolSplitter struct {
 	*ProtocolSplitter
-	protocols []multicodec.Code
+	protocols     []multicodec.Code
+	filteredPeers map[peer.ID]peerFilter
 }
 
 func (rps *retrievalProtocolSplitter) SplitCandidates(candidates []types.RetrievalCandidate) (map[multicodec.Code][]types.RetrievalCandidate, error) {
 	protocolCandidates := make(map[multicodec.Code][]types.RetrievalCandidate, len(rps.protocols))
 	for _, candidate := range candidates {
+		filter, isFiltered := rps.filteredPeers[candidate.MinerPeer.ID]
+		if isFiltered && filter.excludeAll {
+			continue
+		}
 		candidateProtocolsArr := candidate.Metadata.Protocols()
 		candidateProtocolsSet := make(map[multicodec.Code]struct{})
 		for _, candidateProtocol := range candidateProtocolsArr {
+			if isFiltered {
+				if _, ok := filter.protocolsSet[candidateProtocol]; ok {
+					continue
+				}
+			}
 			candidateProtocolsSet[candidateProtocol] = struct{}{}
 		}
 		for _, protocol := range rps.protocols {

pkg/server/http/ipfs.go

@@ -145,6 +145,33 @@ func ipfsHandler(lassie *lassie.Lassie, cfg HttpServerConfig) func(http.Response
 			}
 		}
 
+		var filteredPeers []types.PeerFilter
+		if req.URL.Query().Has("peer-filter") {
+			peerFilterStrings := req.URL.Query()["peer-filter"]
+			for _, peerFilterString := range peerFilterStrings {
+				peerString, protocolsString, hasProtocols := strings.Cut(peerFilterString, "~")
+				peerID, err := peer.Decode(peerString)
+				if err != nil {
+					statusLogger.logStatus(http.StatusBadRequest, "Invalid peer-filter parameter")
+					res.WriteHeader(http.StatusBadRequest)
+					return
+				}
+				var filteredProtocols []multicodec.Code
+				if hasProtocols {
+					filteredProtocols, err = types.ParseProtocolsString(protocolsString)
+					if err != nil {
+						statusLogger.logStatus(http.StatusBadRequest, "Invalid peer-filter  parameter")
+						res.WriteHeader(http.StatusBadRequest)
+						return
+					}
+				}
+				filteredPeers = append(filteredPeers, types.PeerFilter{
+					Peer:              peerID,
+					ExcludeAll:        !hasProtocols,
+					ExcludedProtocols: filteredProtocols,
+				})
+			}
+		}
 		// for setting Content-Disposition header based on filename url parameter
 		var filename string
 		if req.URL.Query().Has("filename") {
@@ -210,6 +237,7 @@ func ipfsHandler(lassie *lassie.Lassie, cfg HttpServerConfig) func(http.Response
 			http.Error(res, msg, http.StatusInternalServerError)
 			return
 		}
+		request.FilteredPeers = filteredPeers
 		request.RetrievalID = retrievalId
 		// setup preload storage for bitswap, the temporary CAR store can set up a
 		// separate preload space in its storage

pkg/types/request.go

@@ -45,6 +45,12 @@ func (id *RetrievalID) UnmarshalText(data []byte) error {
 	return (*uuid.UUID)(id).UnmarshalText(data)
 }
 
+type PeerFilter struct {
+	Peer              peer.ID
+	ExcludeAll        bool
+	ExcludedProtocols []multicodec.Code
+}
+
 // RetrievalRequest is the top level parameters for a request --
 // this should be left unchanged as you move down a retriever tree
 type RetrievalRequest struct {
@@ -56,6 +62,7 @@ type RetrievalRequest struct {
 	PreloadLinkSystem ipld.LinkSystem
 	MaxBlocks         uint64
 	FixedPeers        []peer.AddrInfo
+	FilteredPeers     []PeerFilter
 }
 
 // NewRequestForPath creates a new RetrievalRequest from the provided parameters
@@ -151,7 +158,6 @@ func ParseProtocolsString(v string) ([]multicodec.Code, error) {
 func ParseProviderStrings(v string) ([]peer.AddrInfo, error) {
 	vs := strings.Split(v, ",")
 	providerAddrInfos := make([]peer.AddrInfo, 0, len(vs))
-
 	for _, v := range vs {
 		providerAddrInfo, err := peer.AddrInfoFromString(v)
 		if err != nil {