feat: markets: Integrate index ingest protocol and retrieve by any CID

.circleci/config.yml

@@ -785,6 +785,11 @@ workflows:
           suite: itest-deals_512mb
           target: "./itests/deals_512mb_test.go"
 
+      - test:
+          name: test-itest-deals_anycid
+          suite: itest-deals_anycid
+          target: "./itests/deals_anycid_test.go"
+
       - test:
           name: test-itest-deals_concurrent
           suite: itest-deals_concurrent

api/api_net.go

@@ -51,6 +51,10 @@ type Net interface {
 	NetBlockRemove(ctx context.Context, acl NetBlockList) error //perm:admin
 	NetBlockList(ctx context.Context) (NetBlockList, error)     //perm:read
 
+	NetProtectAdd(ctx context.Context, acl []peer.ID) error    //perm:admin
+	NetProtectRemove(ctx context.Context, acl []peer.ID) error //perm:admin
+	NetProtectList(ctx context.Context) ([]peer.ID, error)     //perm:read
+
 	// ResourceManager API
 	NetStat(ctx context.Context, scope string) (NetStat, error)          //perm:read
 	NetLimit(ctx context.Context, scope string) (NetLimit, error)        //perm:read

api/api_storage.go

@@ -219,6 +219,19 @@ type StorageMiner interface {
 	// DagstoreGC runs garbage collection on the DAG store.
 	DagstoreGC(ctx context.Context) ([]DagstoreShardResult, error) //perm:admin
 
+	// IndexerAnnounceDeal informs indexer nodes that a new deal was received,
+	// so they can download its index
+	IndexerAnnounceDeal(ctx context.Context, proposalCid cid.Cid) error //perm:admin
+
+	// IndexerAnnounceAllDeals informs the indexer nodes aboutall active deals.
+	IndexerAnnounceAllDeals(ctx context.Context) error //perm:admin
+
+	// DagstorePieceIndexSize returns the size of the piece index.
+	DagstorePieceIndexSize(ctx context.Context) (int64, error) //perm:admin
+
+	// DagstoreLookupPieces returns information about shards that contain the given CID.
+	DagstoreLookupPieces(ctx context.Context, cid cid.Cid) ([]DagstoreShardInfo, error) //perm:admin
+
 	// RuntimeSubsystems returns the subsystems that are enabled
 	// in this instance.
 	RuntimeSubsystems(ctx context.Context) (MinerSubsystems, error) //perm:read

build/params_shared_funcs.go

@@ -13,6 +13,9 @@ 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 {
+	return "/indexer/ingest/" + string(netName)
+}
 func DhtProtocolName(netName dtypes.NetworkName) protocol.ID {
 	return protocol.ID("/fil/kad/" + string(netName))
 }

chain/sub/incoming.go

@@ -444,3 +444,24 @@ func recordFailure(ctx context.Context, metric *stats.Int64Measure, failureType
 	)
 	stats.Record(ctx, metric.M(1))
 }
+
+type IndexerMessageValidator struct {
+	self peer.ID
+}
+
+func NewIndexerMessageValidator(self peer.ID) *IndexerMessageValidator {
+	return &IndexerMessageValidator{self: self}
+}
+
+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 miner-nodes or index publishers.  If a node appears to be
+	// local, reject it.
+	if pid == v.self {
+		log.Warnf("refusing to relay indexer message from self")
+		stats.Record(ctx, metrics.IndexerMessageValidationFailure.M(1))
+		return pubsub.ValidationReject
+	}
+	stats.Record(ctx, metrics.IndexerMessageValidationSuccess.M(1))
+	return pubsub.ValidationAccept
+}