| 1 | package libp2p |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "log/slog" |
| 7 | |
| 8 | "github.com/ipfs/go-datastore" |
| 9 | logging "github.com/ipfs/go-log/v2" |
| 10 | pubsub "github.com/libp2p/go-libp2p-pubsub" |
| 11 | "github.com/libp2p/go-libp2p/core/discovery" |
| 12 | "github.com/libp2p/go-libp2p/core/host" |
| 13 | "github.com/libp2p/go-libp2p/core/peer" |
| 14 | "go.uber.org/fx" |
| 15 | |
| 16 | "github.com/ipfs/kubo/core/node/helpers" |
| 17 | "github.com/ipfs/kubo/repo" |
| 18 | ) |
| 19 | |
| 20 | type pubsubParams struct { |
| 21 | fx.In |
| 22 | |
| 23 | Repo repo.Repo |
| 24 | Host host.Host |
| 25 | Discovery discovery.Discovery |
| 26 | } |
| 27 | |
| 28 | func FloodSub(pubsubOptions ...pubsub.Option) any { |
| 29 | return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, params pubsubParams) (service *pubsub.PubSub, err error) { |
| 30 | return pubsub.NewFloodSub( |
| 31 | helpers.LifecycleCtx(mctx, lc), |
| 32 | params.Host, |
| 33 | append(pubsubOptions, |
| 34 | pubsub.WithDiscovery(params.Discovery), |
| 35 | pubsub.WithDefaultValidator(newSeqnoValidator(params.Repo.Datastore())))..., |
| 36 | ) |
| 37 | } |
| 38 | } |
| 39 | |
| 40 | func GossipSub(pubsubOptions ...pubsub.Option) any { |
| 41 | return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, params pubsubParams) (service *pubsub.PubSub, err error) { |
| 42 | return pubsub.NewGossipSub( |
| 43 | helpers.LifecycleCtx(mctx, lc), |
| 44 | params.Host, |
| 45 | append(pubsubOptions, |
| 46 | pubsub.WithDiscovery(params.Discovery), |
| 47 | pubsub.WithFloodPublish(true), // flood own publications to all peers for reliable IPNS delivery |
| 48 | pubsub.WithDefaultValidator(newSeqnoValidator(params.Repo.Datastore())))..., |
| 49 | ) |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | func newSeqnoValidator(ds datastore.Datastore) pubsub.ValidatorEx { |
| 54 | return pubsub.NewBasicSeqnoValidator(&seqnoStore{ds: ds}, slog.New(logging.SlogHandler()).With("logger", "pubsub")) |
| 55 | } |
| 56 | |
| 57 | // SeqnoStorePrefix is the datastore prefix for pubsub seqno validator state. |
| 58 | const SeqnoStorePrefix = "/pubsub/seqno/" |
| 59 | |
| 60 | // seqnoStore implements pubsub.PeerMetadataStore using the repo datastore. |
| 61 | // It stores the maximum seen sequence number per peer to prevent message |
| 62 | // cycles when network diameter exceeds the timecache span. |
| 63 | type seqnoStore struct { |
| 64 | ds datastore.Datastore |
| 65 | } |
| 66 | |
| 67 | var _ pubsub.PeerMetadataStore = (*seqnoStore)(nil) |
| 68 | |
| 69 | // Get returns the stored seqno for a peer, or (nil, nil) if the peer is unknown. |
| 70 | // Returning (nil, nil) for unknown peers allows BasicSeqnoValidator to accept |
| 71 | // the first message from any peer. |
| 72 | func (s *seqnoStore) Get(ctx context.Context, p peer.ID) ([]byte, error) { |
| 73 | key := datastore.NewKey(SeqnoStorePrefix + p.String()) |
| 74 | val, err := s.ds.Get(ctx, key) |
| 75 | if errors.Is(err, datastore.ErrNotFound) { |
| 76 | return nil, nil |
| 77 | } |
| 78 | return val, err |
| 79 | } |
| 80 | |
| 81 | // Put stores the seqno for a peer. |
| 82 | func (s *seqnoStore) Put(ctx context.Context, p peer.ID, val []byte) error { |
| 83 | key := datastore.NewKey(SeqnoStorePrefix + p.String()) |
| 84 | return s.ds.Put(ctx, key, val) |
| 85 | } |