chore: remove IPNI advertisement relay over pubsub via Lotus node

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
 

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))
 }

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
-}