Remove support for graphsync

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/ipfs/go-log/v2 v2.5.1
 	github.com/ipld/go-car/v2 v2.14.2
 	github.com/ipld/go-ipld-prime v0.21.0
-	github.com/ipni/go-libipni v0.5.27
+	github.com/ipni/go-libipni v0.6.15
 	github.com/libp2p/go-libp2p v0.38.1
 	github.com/mattn/go-isatty v0.0.20
 	github.com/montanaflynn/stats v0.7.0
@@ -27,7 +27,6 @@ require (
 require (
 	github.com/benbjohnson/clock v1.3.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/bep/debounce v1.2.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/containerd/cgroups v1.1.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
@@ -37,48 +36,30 @@ require (
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/elastic/gosigar v0.14.3 // indirect
-	github.com/filecoin-project/go-cbor-util v0.0.1 // indirect
-	github.com/filecoin-project/go-data-transfer/v2 v2.0.0-rc8 // indirect
-	github.com/filecoin-project/go-ds-versioning v0.1.2 // indirect
-	github.com/filecoin-project/go-statemachine v1.0.2 // indirect
-	github.com/filecoin-project/go-statestore v0.2.0 // indirect
 	github.com/flynn/noise v1.1.0 // indirect
 	github.com/francoispqt/gojay v1.2.13 // indirect
 	github.com/gammazero/chanqueue v1.0.0 // indirect
 	github.com/gammazero/deque v1.0.0 // indirect
-	github.com/go-logr/logr v1.4.2 // indirect
-	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/gopacket v1.1.19 // indirect
 	github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/websocket v1.5.3 // indirect
-	github.com/hannahhoward/cbor-gen-for v0.0.0-20230214144701-5d17c9d5243c // indirect
-	github.com/hannahhoward/go-pubsub v0.0.0-20200423002714-8d62886cc36e // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-retryablehttp v0.7.7 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/huin/goupnp v1.3.0 // indirect
 	github.com/ipfs/go-block-format v0.2.0 // indirect
-	github.com/ipfs/go-blockservice v0.5.1 // indirect
-	github.com/ipfs/go-graphsync v0.17.0 // indirect
-	github.com/ipfs/go-ipfs-pq v0.0.3 // indirect
 	github.com/ipfs/go-ipfs-util v0.0.3 // indirect
 	github.com/ipfs/go-ipld-cbor v0.1.0 // indirect
 	github.com/ipfs/go-ipld-format v0.6.0 // indirect
-	github.com/ipfs/go-ipld-legacy v0.2.1 // indirect
-	github.com/ipfs/go-libipfs v0.6.2 // indirect
-	github.com/ipfs/go-log v1.0.5 // indirect
-	github.com/ipfs/go-peertaskqueue v0.8.1 // indirect
-	github.com/ipld/go-codec-dagpb v1.6.0 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
 	github.com/jbenet/goprocess v0.1.4 // indirect
-	github.com/jpillora/backoff v1.0.0 // indirect
 	github.com/klauspost/compress v1.17.11 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
 	github.com/koron/go-ssdp v0.0.4 // indirect
@@ -108,7 +89,6 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/onsi/ginkgo/v2 v2.22.0 // indirect
 	github.com/opencontainers/runtime-spec v1.2.0 // indirect
-	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect
 	github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9 // indirect
 	github.com/pion/datachannel v1.5.10 // indirect
@@ -145,10 +125,6 @@ require (
 	github.com/whyrusleeping/cbor-gen v0.2.0 // indirect
 	github.com/wlynxg/anet v0.0.5 // indirect
 	github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 // indirect
-	go.opentelemetry.io/otel v1.13.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.11.1 // indirect
-	go.opentelemetry.io/otel/trace v1.13.0 // indirect
-	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/dig v1.18.0 // indirect
 	go.uber.org/fx v1.23.0 // indirect
 	go.uber.org/mock v0.5.0 // indirect

pkg/adpub/client.go

@@ -35,7 +35,6 @@ type client struct {
 	publisher peer.AddrInfo
 	host      host.Host
 	ownsHost  bool
-	topic     string
 
 	store *ClientStore
 	sub   *dagsync.Subscriber
@@ -69,12 +68,11 @@ func NewClient(addrInfo peer.AddrInfo, options ...Option) (Client, error) {
 		publisher: addrInfo,
 		host:      opts.p2pHost,
 		ownsHost:  ownsHost,
-		topic:     opts.topic,
 
 		store: newClientStore(),
 	}
 
-	c.sub, err = dagsync.NewSubscriber(c.host, c.store.Batching, c.store.LinkSystem, c.topic, dagsync.HttpTimeout(opts.httpTimeout))
+	c.sub, err = dagsync.NewSubscriber(c.host, c.store.LinkSystem, dagsync.HttpTimeout(opts.httpTimeout))
 	if err != nil {
 		return nil, err
 	}

pkg/adpub/options.go

@@ -19,7 +19,6 @@ type config struct {
 	maxSyncRetry      uint64
 	p2pHost           host.Host
 	syncRetryBackoff  time.Duration
-	topic             string
 }
 
 // Option is a function that sets a value in a config.
@@ -31,7 +30,6 @@ func getOpts(opts []Option) (config, error) {
 		entriesDepthLimit: defaultEntriesDepthLimit,
 		httpTimeout:       defaultHttpTimeout,
 		syncRetryBackoff:  500 * time.Millisecond,
-		topic:             "/indexer/ingest/mainnet",
 	}
 
 	for i, opt := range opts {
@@ -68,15 +66,6 @@ func WithLibp2pHost(h host.Host) Option {
 	}
 }
 
-// WithTopicName sets the topic name on which the provider announces advertised
-// content. Defaults to '/indexer/ingest/mainnet'.
-func WithTopicName(topic string) Option {
-	return func(c *config) error {
-		c.topic = topic
-		return nil
-	}
-}
-
 // WithEntriesDepthLimit sets the depth limit when syncing an
 // advertisement entries chain. Setting to 0 means no limit.
 func WithEntriesDepthLimit(depthLimit int64) Option {

pkg/ads/dist.go

@@ -39,7 +39,6 @@ var adsDistFlags = []cli.Flag{
 		Aliases: []string{"dl"},
 		Value:   5000,
 	},
-	topicFlag,
 }
 
 func adsDistAction(cctx *cli.Context) error {
@@ -54,7 +53,6 @@ func adsDistAction(cctx *cli.Context) error {
 	}
 
 	adDist, err := dtrack.NewAdDistance(
-		dtrack.WithTopic(cctx.String("topic")),
 		dtrack.WithDepthLimit(cctx.Int64("dist-limit")))
 	if err != nil {
 		return err

pkg/ads/flags.go

@@ -23,10 +23,3 @@ var timeoutFlag = &cli.DurationFlag{
 	Value:       10 * time.Second,
 	DefaultText: "10s",
 }
-
-var topicFlag = &cli.StringFlag{
-	Name:    "topic",
-	Usage:   "Topic on which index advertisements are published. Only needed if connecting via Graphsync with non-standard topic.",
-	Value:   "/indexer/ingest/mainnet",
-	Aliases: []string{"t"},
-}

pkg/ads/get.go

@@ -65,7 +65,6 @@ var adsGetFlags = []cli.Flag{
 		DefaultText: "100 (set to '0' for unlimited)",
 	},
 	timeoutFlag,
-	topicFlag,
 }
 
 func adsGetAction(cctx *cli.Context) error {
@@ -125,7 +124,6 @@ func adsGetAction(cctx *cli.Context) error {
 	}
 
 	pubClient, err := adpub.NewClient(*addrInfo,
-		adpub.WithTopicName(cctx.String("topic")),
 		adpub.WithEntriesDepthLimit(cctx.Int64("entries-depth-limit")),
 		adpub.WithHttpTimeout(cctx.Duration("timeout")))
 	if err != nil {

pkg/ads/list.go

@@ -35,7 +35,6 @@ var adsListFlags = []cli.Flag{
 		Required: true,
 	},
 	timeoutFlag,
-	topicFlag,
 }
 
 func adsListAction(cctx *cli.Context) error {
@@ -44,8 +43,7 @@ func adsListAction(cctx *cli.Context) error {
 		return fmt.Errorf("bad pub-addr-info: %w", err)
 	}
 
-	provClient, err := adpub.NewClient(*addrInfo, adpub.WithTopicName(cctx.String("topic")),
-		adpub.WithHttpTimeout(cctx.Duration("timeout")))
+	provClient, err := adpub.NewClient(*addrInfo, adpub.WithHttpTimeout(cctx.Duration("timeout")))
 	if err != nil {
 		return err
 	}

pkg/dtrack/ad_distance.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-datastore"
-	dssync "github.com/ipfs/go-datastore/sync"
 	"github.com/ipld/go-ipld-prime"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
 	"github.com/ipni/go-libipni/dagsync"
@@ -48,8 +47,7 @@ func NewAdDistance(options ...Option) (*AdDistance, error) {
 	}
 
 	store := newCountStore()
-	gsds := dssync.MutexWrap(datastore.NewMapDatastore())
-	sub, err := dagsync.NewSubscriber(p2pHost, gsds, store.LinkSystem, opts.topic)
+	sub, err := dagsync.NewSubscriber(p2pHost, store.LinkSystem)
 	if err != nil {
 		return nil, err
 	}

pkg/dtrack/option.go

@@ -7,7 +7,6 @@ import (
 type config struct {
 	depthLimit int64
 	p2pHost    host.Host
-	topic      string
 }
 
 type Option func(*config)
@@ -16,7 +15,6 @@ type Option func(*config)
 func getOpts(opts []Option) config {
 	cfg := config{
 		depthLimit: 5000,
-		topic:      "/indexer/ingest/mainnet",
 	}
 	for _, opt := range opts {
 		opt(&cfg)
@@ -38,13 +36,3 @@ func WithP2pHost(p2pHost host.Host) Option {
 		c.p2pHost = p2pHost
 	}
 }
-
-// WithTopic configures the topic name used to get the head advertisement when
-// using data-transfer/graphsync.
-func WithTopic(topic string) Option {
-	return func(c *config) {
-		if topic != "" {
-			c.topic = topic
-		}
-	}
-}

pkg/provider/provider.go

@@ -114,11 +114,6 @@ var providerFlags = []cli.Flag{
 		Aliases: []string{"pub"},
 		Usage:   "Only print publisher address info.",
 	},
-	&cli.StringFlag{
-		Name:  "topic",
-		Usage: "Topic on which index advertisements are published. Only needed to get head advertisement via Graphsync with non-standard topic.",
-		Value: "/indexer/ingest/mainnet",
-	},
 }
 
 func providerAction(cctx *cli.Context) error {
@@ -302,7 +297,7 @@ func followDistance(cctx *cli.Context, include, exclude map[peer.ID]struct{}, pc
 	fmt.Fprintln(os.Stderr, "Showing provider distance updates, ctrl-c to cancel...")
 	limit := cctx.Int64("ad-depth-limit")
 	updates, err := dtrack.RunDistanceTracker(cctx.Context, include, exclude, pc, trackUpdateIn, timeout,
-		dtrack.WithDepthLimit(limit), dtrack.WithTopic(cctx.String("topic")))
+		dtrack.WithDepthLimit(limit))
 	if err != nil {
 		return err
 	}
@@ -457,7 +452,6 @@ func getLastSeenDistance(cctx *cli.Context, pinfo *model.ProviderInfo, p2pHost h
 	}
 	adDist, err := dtrack.NewAdDistance(
 		dtrack.WithDepthLimit(cctx.Int64("ad-depth-limit")),
-		dtrack.WithTopic(cctx.String("topic")),
 		dtrack.WithP2pHost(p2pHost))
 	if err != nil {
 		return 0, cid.Undef, err

pkg/random/random.go

@@ -59,12 +59,6 @@ var randomFlags = []cli.Flag{
 		Usage:   "Only print multihashes and do not print descriptive output.",
 		Aliases: []string{"q"},
 	},
-	&cli.StringFlag{
-		Name:    "topic",
-		Usage:   "Topic on which index advertisements are published. Only needed if connecting via Graphsync with non-standard topic.",
-		Value:   "/indexer/ingest/mainnet",
-		Aliases: []string{"t"},
-	},
 }
 
 func randomAction(cctx *cli.Context) error {
@@ -104,7 +98,7 @@ func randomAction(cctx *cli.Context) error {
 			fmt.Fprintf(os.Stderr, "Provider %s has no publisher\n", peerID)
 			continue
 		}
-		err = RandomMultihashes(cctx.Context, *prov.Publisher, cctx.String("topic"), adCount, mhsCount, cctx.Bool("quiet"))
+		err = RandomMultihashes(cctx.Context, *prov.Publisher, adCount, mhsCount, cctx.Bool("quiet"))
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "Cannot get random multihashes from provider %s: %s\n", peerID, err)
 			continue
@@ -130,9 +124,8 @@ func getProvider(cctx *cli.Context, pc *pcache.ProviderCache, peerID peer.ID) (*
 	return prov, nil
 }
 
-func RandomMultihashes(ctx context.Context, addrInfo peer.AddrInfo, topic string, adCount, mhsCount int, quiet bool) error {
+func RandomMultihashes(ctx context.Context, addrInfo peer.AddrInfo, adCount, mhsCount int, quiet bool) error {
 	provClient, err := adpub.NewClient(addrInfo,
-		adpub.WithTopicName(topic),
 		adpub.WithEntriesDepthLimit(1),
 	)
 	if err != nil {

pkg/verify/ingest.go

@@ -42,11 +42,9 @@ sources:
 - Path to a CAR file (i.e. --from-car)
 - Path to a CARv2 index file in iterable multihash format (i.e. --from-car-index)
 
-If fetching multihashes from an advertisement publisher, then the "topic" flag can specify the topic
-name is the advertisements are published on a non-standard topic. The user may optionally specify an
-advertisement CID, or to use the latest advertisement seen by the indexer, as the source of
-multihash entries. If not specified, the latest advertisement is fetched from the publisher and its
-entries are used as the source of multihashes.
+The user may optionally specify an advertisement CID, or to use the latestadvertisement seen by the
+indexer, as the source of multihash entries. If a CID is not specified then the latest advertisement
+is fetched from the publisher and its entries are used as the source of multihashes.
 
 The path to CAR files may point to any CAR version (CARv1 or CARv2). The list of multihashes are
 generated automatically from the CAR payload if no suitable index is present.
@@ -176,12 +174,6 @@ var verifyIngestFlags = []cli.Flag{
 			"A smaller batch size will increase the number of requests to the indexer but may avoid timing out waiting for a response.",
 		Value: 4096,
 	},
-	&cli.StringFlag{
-		Name:    "topic",
-		Usage:   "The topic name on which advertisements are published. Only takes effect if multihashes read from publisher.",
-		Value:   "/indexer/ingest/mainnet",
-		Aliases: []string{"t"},
-	},
 	&cli.BoolFlag{
 		Name:        "print-unindexed-mhs",
 		Usage:       "Print multihashes that are not indexed by the indexer. Only printed if the indexer is successfully contacted.",
@@ -331,11 +323,9 @@ func verifyIngestFromProvider(cctx *cli.Context, provID peer.ID) error {
 	}
 	fmt.Println("Publisher:", pubAddrInfo.String())
 	fmt.Printf("Ads/Entries depth: %s/%d\n", adDepthLimitStr, cctx.Int64("entries-depth-limit"))
-	fmt.Println("Topic:", cctx.String("topic"))
 	fmt.Println("Last ad seen by indexer:", provInfo.LastAdvertisement.String())
 
 	pubClient, err := adpub.NewClient(pubAddrInfo,
-		adpub.WithTopicName(cctx.String("topic")),
 		adpub.WithEntriesDepthLimit(cctx.Int64("entries-depth-limit")))
 	if err != nil {
 		return err