chore: remove IPNI advertisement relay over pubsub via Lotus node (#1…

…2768) The initial implementation of IPNI used GossipSub as a way to propagate IPNI advertisement chain. To do this the propagation had to be relayed through the Lotus node due to strict Filecoin GossipSub validation rules. Since then IPNI has moved on to roll out its own sync protocol that works over HTTP, and HTTP-over-libp2p. This move has been the official way of advertising content to IPNI federation over a year now. Therefore, remove the ad relay over pubsub via Lotus node as it is now considered to have reached its EOL as a mechanism for advertising to IPNI.
filecoin-project · Dec 10, 2024 · 9bfd0b4 · 9bfd0b4
1 parent 8b3de48
commit 9bfd0b4
Show file tree

Hide file tree

Showing 10 changed files with 9 additions and 428 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -10,6 +10,7 @@
 # UNRELEASED
 
 - Add json output of tipsets to `louts chain list`. ([filecoin-project/lotus#12691](https://github.com/filecoin-project/lotus/pull/12691))
+- Remove IPNI advertisement relay over pubsub via Lotus node as it now has been deprecated. ([filecoin-project/lotus#12768](https://github.com/filecoin-project/lotus/pull/12768)
 
 # UNRELEASED v.1.32.0
 

diff --git a/build/params_shared_funcs.go b/build/params_shared_funcs.go
@@ -14,17 +14,6 @@ import (
 
 func BlocksTopic(netName dtypes.NetworkName) string   { return "/fil/blocks/" + string(netName) }
 func MessagesTopic(netName dtypes.NetworkName) string { return "/fil/msgs/" + string(netName) }
-func IndexerIngestTopic(netName dtypes.NetworkName) string {
-
-	nn := string(netName)
-	// The network name testnetnet is here for historical reasons.
-	// Going forward we aim to use the name `mainnet` where possible.
-	if nn == "testnetnet" {
-		nn = "mainnet"
-	}
-
-	return "/indexer/ingest/" + nn
-}
 func DhtProtocolName(netName dtypes.NetworkName) protocol.ID {
 	return protocol.ID("/fil/kad/" + string(netName))
 }

diff --git a/chain/sub/incoming.go b/chain/sub/incoming.go
@@ -1,19 +1,15 @@
 package sub
 
 import (
-	"bytes"
 	"context"
-	"encoding/binary"
 	"errors"
-	"sync"
 	"time"
 
 	lru "github.com/hashicorp/golang-lru/v2"
 	bserv "github.com/ipfs/boxo/blockservice"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
-	"github.com/ipni/go-libipni/announce/message"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/connmgr"
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -30,10 +26,8 @@ import (
 	"github.com/filecoin-project/lotus/chain/consensus"
 	"github.com/filecoin-project/lotus/chain/messagepool"
 	"github.com/filecoin-project/lotus/chain/store"
-	"github.com/filecoin-project/lotus/chain/sub/ratelimit"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/metrics"
-	"github.com/filecoin-project/lotus/node/impl/full"
 )
 
 var log = logging.Logger("sub")
@@ -470,167 +464,3 @@ func recordFailure(ctx context.Context, metric *stats.Int64Measure, failureType
 	)
 	stats.Record(ctx, metric.M(1))
 }
-
-type peerMsgInfo struct {
-	peerID    peer.ID
-	lastCid   cid.Cid
-	lastSeqno uint64
-	rateLimit *ratelimit.Window
-	mutex     sync.Mutex
-}
-
-type IndexerMessageValidator struct {
-	self peer.ID
-
-	peerCache *lru.TwoQueueCache[address.Address, *peerMsgInfo]
-	chainApi  full.ChainModuleAPI
-	stateApi  full.StateModuleAPI
-}
-
-func NewIndexerMessageValidator(self peer.ID, chainApi full.ChainModuleAPI, stateApi full.StateModuleAPI) *IndexerMessageValidator {
-	peerCache, _ := lru.New2Q[address.Address, *peerMsgInfo](8192)
-
-	return &IndexerMessageValidator{
-		self:      self,
-		peerCache: peerCache,
-		chainApi:  chainApi,
-		stateApi:  stateApi,
-	}
-}
-
-func (v *IndexerMessageValidator) Validate(ctx context.Context, pid peer.ID, msg *pubsub.Message) pubsub.ValidationResult {
-	// This chain-node should not be publishing its own messages.  These are
-	// relayed from market-nodes.  If a node appears to be local, reject it.
-	if pid == v.self {
-		log.Debug("ignoring indexer message from self")
-		stats.Record(ctx, metrics.IndexerMessageValidationFailure.M(1))
-		return pubsub.ValidationIgnore
-	}
-	originPeer := msg.GetFrom()
-	if originPeer == v.self {
-		log.Debug("ignoring indexer message originating from self")
-		stats.Record(ctx, metrics.IndexerMessageValidationFailure.M(1))
-		return pubsub.ValidationIgnore
-	}
-
-	idxrMsg := message.Message{}
-	err := idxrMsg.UnmarshalCBOR(bytes.NewBuffer(msg.Data))
-	if err != nil {
-		log.Errorw("Could not decode indexer pubsub message", "err", err)
-		return pubsub.ValidationReject
-	}
-	if len(idxrMsg.ExtraData) == 0 {
-		log.Debugw("ignoring message missing miner id", "peer", originPeer)
-		return pubsub.ValidationIgnore
-	}
-
-	// Get miner info from lotus
-	minerAddr, err := address.NewFromBytes(idxrMsg.ExtraData)
-	if err != nil {
-		log.Warnw("cannot parse extra data as miner address", "err", err, "extraData", idxrMsg.ExtraData)
-		return pubsub.ValidationReject
-	}
-
-	msgCid := idxrMsg.Cid
-
-	msgInfo, cached := v.peerCache.Get(minerAddr)
-	if !cached {
-		msgInfo = &peerMsgInfo{}
-	}
-
-	// Lock this peer's message info.
-	msgInfo.mutex.Lock()
-	defer msgInfo.mutex.Unlock()
-
-	var seqno uint64
-	if cached {
-		// Reject replayed messages.
-		seqno = binary.BigEndian.Uint64(msg.Message.GetSeqno())
-		if seqno <= msgInfo.lastSeqno {
-			log.Debugf("ignoring replayed indexer message")
-			return pubsub.ValidationIgnore
-		}
-	}
-
-	if !cached || originPeer != msgInfo.peerID {
-		// Check that the miner ID maps to the peer that sent the message.
-		err = v.authenticateMessage(ctx, minerAddr, originPeer)
-		if err != nil {
-			log.Warnw("cannot authenticate message", "err", err, "peer", originPeer, "minerID", minerAddr)
-			stats.Record(ctx, metrics.IndexerMessageValidationFailure.M(1))
-			return pubsub.ValidationReject
-		}
-		msgInfo.peerID = originPeer
-		if !cached {
-			// Add msgInfo to cache only after being authenticated.  If two
-			// messages from the same peer are handled concurrently, there is a
-			// small chance that one msgInfo could replace the other here when
-			// the info is first cached.  This is OK, so no need to prevent it.
-			v.peerCache.Add(minerAddr, msgInfo)
-		}
-	}
-
-	// Update message info cache with the latest message's sequence number.
-	msgInfo.lastSeqno = seqno
-
-	// See if message needs to be ignored due to rate limiting.
-	if v.rateLimitPeer(msgInfo, msgCid) {
-		return pubsub.ValidationIgnore
-	}
-
-	stats.Record(ctx, metrics.IndexerMessageValidationSuccess.M(1))
-	return pubsub.ValidationAccept
-}
-
-func (v *IndexerMessageValidator) rateLimitPeer(msgInfo *peerMsgInfo, msgCid cid.Cid) bool {
-	const (
-		msgLimit        = 5
-		msgTimeLimit    = 10 * time.Second
-		repeatTimeLimit = 2 * time.Hour
-	)
-
-	timeWindow := msgInfo.rateLimit
-
-	// Check overall message rate.
-	if timeWindow == nil {
-		timeWindow = ratelimit.NewWindow(msgLimit, msgTimeLimit)
-		msgInfo.rateLimit = timeWindow
-	} else if msgInfo.lastCid == msgCid {
-		// Check if this is a repeat of the previous message data.
-		if time.Since(timeWindow.Newest()) < repeatTimeLimit {
-			log.Warnw("ignoring repeated indexer message", "sender", msgInfo.peerID)
-			return true
-		}
-	}
-
-	err := timeWindow.Add()
-	if err != nil {
-		log.Warnw("ignoring indexer message", "sender", msgInfo.peerID, "err", err)
-		return true
-	}
-
-	msgInfo.lastCid = msgCid
-
-	return false
-}
-
-func (v *IndexerMessageValidator) authenticateMessage(ctx context.Context, minerAddress address.Address, peerID peer.ID) error {
-	ts, err := v.chainApi.ChainHead(ctx)
-	if err != nil {
-		return err
-	}
-
-	minerInfo, err := v.stateApi.StateMinerInfo(ctx, minerAddress, ts.Key())
-	if err != nil {
-		return err
-	}
-
-	if minerInfo.PeerId == nil {
-		return xerrors.New("no peer id for miner")
-	}
-	if *minerInfo.PeerId != peerID {
-		return xerrors.New("miner id does not map to peer that sent message")
-	}
-
-	return nil
-}