Chain nodes relay indexer pubsub messages.

This includes message validation and rate-limiting.

build/params_shared_funcs.go

@@ -13,6 +13,16 @@ 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/idxmsg/message.go

@@ -0,0 +1,40 @@
+// Package idxmsg is a copy of the message and codec from go-legs v0.3.0. It is
+// copied here because having a dependency on go-legs v0.3.0 brings other
+// incompatible dependencies.  The code here was copied from:
+//
+// https://github.com/filecoin-project/go-legs/tree/main/dtsync
+package idxmsg
+
+import (
+	"errors"
+
+	"github.com/ipfs/go-cid"
+	"github.com/multiformats/go-multiaddr"
+)
+
+var ErrBadEncoding = errors.New("invalid message encoding")
+
+type Message struct {
+	Cid       cid.Cid
+	Addrs     [][]byte
+	ExtraData []byte `json:",omitempty"`
+}
+
+func (m *Message) SetAddrs(addrs []multiaddr.Multiaddr) {
+	m.Addrs = make([][]byte, 0, len(addrs))
+	for _, a := range addrs {
+		m.Addrs = append(m.Addrs, a.Bytes())
+	}
+}
+
+func (m *Message) GetAddrs() ([]multiaddr.Multiaddr, error) {
+	addrs := make([]multiaddr.Multiaddr, 0, len(m.Addrs))
+	for _, a := range m.Addrs {
+		p, err := multiaddr.NewMultiaddrBytes(a)
+		if err != nil {
+			return nil, err
+		}
+		addrs = append(addrs, p)
+	}
+	return addrs, nil
+}