Use refactored go-libipni with ipnisync (#396)

* Use refactored go-libipni with ipnisync
* Mirror only publishes over HTTP

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.19 as build
+FROM golang:1.20 as build
 WORKDIR /go/src/provider
 
 COPY go.mod go.sum ./

cmd/provider/internal/config/ingest.go

@@ -31,7 +31,7 @@ type Ingest struct {
 	// PurgeLinkCache tells whether to purge the link cache on daemon startup.
 	PurgeLinkCache bool
 
-	// HttpPublisher configures the dagsync httpsync publisher.
+	// HttpPublisher configures the dagsync ipnisync publisher.
 	HttpPublisher HttpPublisher
 
 	// PublisherKind specifies which dagsync.Publisher implementation to use.

cmd/provider/mirror.go

@@ -7,7 +7,6 @@ import (
 	"os"
 
 	leveldb "github.com/ipfs/go-ds-leveldb"
-	"github.com/ipld/go-ipld-prime/traversal/selector"
 	"github.com/ipni/index-provider/metrics"
 	"github.com/ipni/index-provider/mirror"
 	"github.com/libp2p/go-libp2p"
@@ -24,7 +23,8 @@ var Mirror struct {
 		source                      *cli.StringFlag
 		syncInterval                *cli.DurationFlag
 		identityPath                *cli.PathFlag
-		listenAddrs                 *cli.StringSliceFlag
+		listenAddr                  *cli.StringFlag
+		p2pListenAddrs              *cli.StringSliceFlag
 		storePath                   *cli.PathFlag
 		initAdRecurLimit            *cli.UintFlag
 		entriesRecurLimit           *cli.UintFlag
@@ -58,9 +58,15 @@ func init() {
 		Usage:       "The path to the file containing the marshalled libp2p private key that the mirror should use as its identity.",
 		DefaultText: "Randomly generated",
 	}
-	Mirror.flags.listenAddrs = &cli.StringSliceFlag{
-		Name:        "listenAddrs",
-		Usage:       "The mirror listen addresses in form of multiaddr.",
+	Mirror.flags.listenAddr = &cli.StringFlag{
+		Name:  "listenAddr",
+		Usage: "The HTTP address:port that the mirror publishes advertisements over HTTP on.",
+		// TODO: when libp2phttp available: "none, publish http over libp2p"
+		DefaultText: "Local host with a random open port",
+	}
+	Mirror.flags.p2pListenAddrs = &cli.StringSliceFlag{
+		Name:        "p2pListenAddrs",
+		Usage:       "The mirror p2p host listen addresses in form of multiaddr.",
 		DefaultText: "Local host with a random open port",
 	}
 	Mirror.flags.storePath = &cli.PathFlag{
@@ -126,7 +132,8 @@ func init() {
 			Mirror.flags.source,
 			Mirror.flags.syncInterval,
 			Mirror.flags.identityPath,
-			Mirror.flags.listenAddrs,
+			Mirror.flags.listenAddr,
+			Mirror.flags.p2pListenAddrs,
 			Mirror.flags.storePath,
 			Mirror.flags.initAdRecurLimit,
 			Mirror.flags.entriesRecurLimit,
@@ -154,28 +161,33 @@ func beforeMirror(cctx *cli.Context) error {
 		Mirror.options = append(Mirror.options, mirror.WithSyncInterval(Mirror.flags.syncInterval.Get(cctx)))
 	}
 	var hostOpts []libp2p.Option
+	var pk crypto.PrivKey
 	if cctx.IsSet(Mirror.flags.identityPath.Name) {
 		pkPath := Mirror.flags.identityPath.Get(cctx)
 		pkBytes, err := os.ReadFile(pkPath)
 		if err != nil {
 			return err
 		}
-		pk, err := crypto.UnmarshalPrivateKey(pkBytes)
+		pk, err = crypto.UnmarshalPrivateKey(pkBytes)
 		if err != nil {
 			return err
 		}
 		hostOpts = append(hostOpts, libp2p.Identity(pk))
 	}
-	if cctx.IsSet(Mirror.flags.listenAddrs.Name) {
-		listenAddrs := Mirror.flags.listenAddrs.Get(cctx)
-		hostOpts = append(hostOpts, libp2p.ListenAddrStrings(listenAddrs...))
+	if cctx.IsSet(Mirror.flags.listenAddr.Name) {
+		listenAddr := Mirror.flags.listenAddr.Get(cctx)
+		Mirror.options = append(Mirror.options, mirror.WithHTTPListenAddr(listenAddr))
+	}
+	if cctx.IsSet(Mirror.flags.p2pListenAddrs.Name) {
+		p2pListenAddrs := Mirror.flags.p2pListenAddrs.Get(cctx)
+		hostOpts = append(hostOpts, libp2p.ListenAddrStrings(p2pListenAddrs...))
 	}
 	if len(hostOpts) != 0 {
 		h, err := libp2p.New(hostOpts...)
 		if err != nil {
 			return err
 		}
-		Mirror.options = append(Mirror.options, mirror.WithHost(h))
+		Mirror.options = append(Mirror.options, mirror.WithHost(h, pk))
 	}
 	if cctx.IsSet(Mirror.flags.storePath.Name) {
 		path := Mirror.flags.storePath.Get(cctx)
@@ -186,11 +198,11 @@ func beforeMirror(cctx *cli.Context) error {
 		Mirror.options = append(Mirror.options, mirror.WithDatastore(ds))
 	}
 	if cctx.IsSet(Mirror.flags.initAdRecurLimit.Name) {
-		limit := selector.RecursionLimitDepth(int64(Mirror.flags.initAdRecurLimit.Get(cctx)))
+		limit := int64(Mirror.flags.initAdRecurLimit.Get(cctx))
 		Mirror.options = append(Mirror.options, mirror.WithInitialAdRecursionLimit(limit))
 	}
 	if cctx.IsSet(Mirror.flags.entriesRecurLimit.Name) {
-		limit := selector.RecursionLimitDepth(int64(Mirror.flags.entriesRecurLimit.Get(cctx)))
+		limit := int64(Mirror.flags.entriesRecurLimit.Get(cctx))
 		Mirror.options = append(Mirror.options, mirror.WithEntriesRecursionLimit(limit))
 	}
 

e2e_retrieve_test.go

@@ -11,7 +11,6 @@ import (
 
 	datatransfer "github.com/filecoin-project/go-data-transfer/v2"
 	retrievaltypes "github.com/filecoin-project/go-retrieval-types"
-	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
 	"github.com/ipfs/go-graphsync/storeutil"
@@ -120,7 +119,7 @@ func testRetrievalRoundTripWithTestCase(t *testing.T, tc testCase) {
 		ID:    server.h.ID(),
 		Addrs: []multiaddr.Multiaddr{server.publisherAddr},
 	}
-	headCid, err := sub.Sync(ctx, serverInfo, cid.Undef, nil)
+	headCid, err := sub.SyncAdChain(ctx, serverInfo)
 	require.NoError(t, err)
 	require.Equal(t, advCid, headCid)
 
@@ -200,7 +199,7 @@ func testReimportCarWtihTestCase(t *testing.T, tc testCase) {
 		ID:    server.h.ID(),
 		Addrs: []multiaddr.Multiaddr{server.publisherAddr},
 	}
-	headCid, err := sub.Sync(ctx, serverInfo, cid.Undef, nil)
+	headCid, err := sub.SyncAdChain(ctx, serverInfo)
 	require.NoError(t, err)
 	require.Equal(t, advCid, headCid)
 
@@ -228,7 +227,7 @@ func testReimportCarWtihTestCase(t *testing.T, tc testCase) {
 	require.NoError(t, err)
 
 	// Sync the new advertisement
-	headCid, err = sub.Sync(ctx, serverInfo, cid.Undef, nil)
+	headCid, err = sub.SyncAdChain(ctx, serverInfo)
 	require.NoError(t, err)
 	require.Equal(t, advCid2, headCid)
 

engine/engine.go

@@ -23,7 +23,7 @@ import (
 	"github.com/ipni/go-libipni/announce/p2psender"
 	"github.com/ipni/go-libipni/dagsync"
 	"github.com/ipni/go-libipni/dagsync/dtsync"
-	"github.com/ipni/go-libipni/dagsync/httpsync"
+	"github.com/ipni/go-libipni/dagsync/ipnisync"
 	"github.com/ipni/go-libipni/ingest/schema"
 	"github.com/ipni/go-libipni/metadata"
 	provider "github.com/ipni/index-provider"
@@ -148,12 +148,17 @@ func (e *Engine) newPublisher() (dagsync.Publisher, error) {
 		log.Info("Remote announcements disabled; all advertisements will only be stored locally.")
 		return nil, nil
 	case HttpPublisher:
-		var httpPub *httpsync.Publisher
+		var httpPub *ipnisync.Publisher
 		var err error
 		if e.pubHttpWithoutServer {
-			httpPub, err = httpsync.NewPublisherWithoutServer(e.pubHttpListenAddr, e.pubHttpHandlerPath, e.lsys, e.key)
+			httpPub, err = ipnisync.NewPublisher(e.pubHttpListenAddr, e.lsys, e.key,
+				ipnisync.WithHeadTopic(e.pubTopicName),
+				ipnisync.WithHandlerPath(e.pubHttpHandlerPath),
+				ipnisync.WithServer(false))
 		} else {
-			httpPub, err = httpsync.NewPublisher(e.pubHttpListenAddr, e.lsys, e.key)
+			httpPub, err = ipnisync.NewPublisher(e.pubHttpListenAddr, e.lsys, e.key,
+				ipnisync.WithHeadTopic(e.pubTopicName),
+				ipnisync.WithServer(true))
 		}
 		if err != nil {
 			return nil, fmt.Errorf("cannot create http publisher: %w", err)
@@ -478,7 +483,7 @@ func (e *Engine) GetPublisherHttpFunc() (http.HandlerFunc, error) {
 	if !e.pubHttpWithoutServer {
 		return nil, errors.New("HttpPublisherWithoutServer option not set")
 	}
-	hp, ok := e.publisher.(*httpsync.Publisher)
+	hp, ok := e.publisher.(*ipnisync.Publisher)
 	if !ok {
 		return nil, errors.New("publisher is not an http publisher")
 	}

go.mod

@@ -10,16 +10,16 @@ require (
 	github.com/ipfs/go-cid v0.4.1
 	github.com/ipfs/go-datastore v0.6.0
 	github.com/ipfs/go-ds-leveldb v0.5.0
-	github.com/ipfs/go-graphsync v0.14.6
+	github.com/ipfs/go-graphsync v0.14.7
 	github.com/ipfs/go-ipfs-blockstore v1.3.0
 	github.com/ipfs/go-log/v2 v2.5.1
 	github.com/ipfs/kubo v0.21.0
 	github.com/ipld/go-car/v2 v2.10.1
 	github.com/ipld/go-codec-dagpb v1.6.0
 	github.com/ipld/go-ipld-adl-hamt v0.0.0-20220616142416-9004dbd839e0
 	github.com/ipld/go-ipld-prime v0.20.0
-	github.com/ipni/go-libipni v0.3.1
-	github.com/libp2p/go-libp2p v0.29.1
+	github.com/ipni/go-libipni v0.4.0
+	github.com/libp2p/go-libp2p v0.29.2
 	github.com/libp2p/go-libp2p-pubsub v0.9.3
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/multiformats/go-multiaddr v0.10.1
@@ -53,7 +53,7 @@ require (
 	github.com/quic-go/qpack v0.4.0 // indirect
 	github.com/quic-go/qtls-go1-19 v0.3.3 // indirect
 	github.com/quic-go/qtls-go1-20 v0.2.3 // indirect
-	github.com/quic-go/quic-go v0.36.3 // indirect
+	github.com/quic-go/quic-go v0.36.4 // indirect
 	github.com/quic-go/webtransport-go v0.5.3 // indirect
 	github.com/samber/lo v1.36.0 // indirect
 	github.com/whyrusleeping/cbor-gen v0.0.0-20230418232409-daab9ece03a0 // indirect

go.sum

@@ -279,8 +279,8 @@ github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ
 github.com/ipfs/go-detect-race v0.0.1/go.mod h1:8BNT7shDZPo99Q74BpGMK+4D8Mn4j46UU0LZ723meps=
 github.com/ipfs/go-ds-leveldb v0.5.0 h1:s++MEBbD3ZKc9/8/njrn4flZLnCuY9I79v94gBUNumo=
 github.com/ipfs/go-ds-leveldb v0.5.0/go.mod h1:d3XG9RUDzQ6V4SHi8+Xgj9j1XuEk1z82lquxrVbml/Q=
-github.com/ipfs/go-graphsync v0.14.6 h1:NPxvuUy4Z08Mg8dwpBzwgbv/PGLIufSJ1sle6iAX8yo=
-github.com/ipfs/go-graphsync v0.14.6/go.mod h1:yT0AfjFgicOoWdAlUJ96tQ5AkuGI4r1taIQX/aHbBQo=
+github.com/ipfs/go-graphsync v0.14.7 h1:V90NORSdCpUHAgqQhApU/bmPSLOnwtSHM2v7R90k9Do=
+github.com/ipfs/go-graphsync v0.14.7/go.mod h1:yT0AfjFgicOoWdAlUJ96tQ5AkuGI4r1taIQX/aHbBQo=
 github.com/ipfs/go-hamt-ipld v0.1.1/go.mod h1:1EZCr2v0jlCnhpa+aZ0JZYp8Tt2w16+JJOAVz17YcDk=
 github.com/ipfs/go-ipfs-blockstore v1.3.0 h1:m2EXaWgwTzAfsmt5UdJ7Is6l4gJcaM/A12XwJyvYvMM=
 github.com/ipfs/go-ipfs-blockstore v1.3.0/go.mod h1:KgtZyc9fq+P2xJUiCAzbRdhhqJHvsw8u2Dlqy2MyRTE=
@@ -344,8 +344,8 @@ github.com/ipld/go-ipld-adl-hamt v0.0.0-20220616142416-9004dbd839e0/go.mod h1:od
 github.com/ipld/go-ipld-prime v0.20.0 h1:Ud3VwE9ClxpO2LkCYP7vWPc0Fo+dYdYzgxUJZ3uRG4g=
 github.com/ipld/go-ipld-prime v0.20.0/go.mod h1:PzqZ/ZR981eKbgdr3y2DJYeD/8bgMawdGVlJDE8kK+M=
 github.com/ipld/go-ipld-prime/storage/bsadapter v0.0.0-20230102063945-1a409dc236dd h1:gMlw/MhNr2Wtp5RwGdsW23cs+yCuj9k2ON7i9MiJlRo=
-github.com/ipni/go-libipni v0.3.1 h1:zCq9UXrvVL4NAxyumPGHboUAGraNfmkDC16BAoQGrww=
-github.com/ipni/go-libipni v0.3.1/go.mod h1:2zyo+mgV+E1T0n6eGmef7+uEt0awOpIhqONfz9ZPtNo=
+github.com/ipni/go-libipni v0.4.0 h1:zZ8OU2N0D4iYt0E9jInbDapeh9bG10b5sBgqvScflNw=
+github.com/ipni/go-libipni v0.4.0/go.mod h1:LxH6NUmEVruK3FjV2bFWfXKougX7AIe7wVjvPqITrDI=
 github.com/ipsn/go-secp256k1 v0.0.0-20180726113642-9d62b9f0bc52 h1:QG4CGBqCeuBo6aZlGAamSkxWdgWfZGeE49eUOWJPA4c=
 github.com/ipsn/go-secp256k1 v0.0.0-20180726113642-9d62b9f0bc52/go.mod h1:fdg+/X9Gg4AsAIzWpEHwnqd+QY3b7lajxyjE1m4hkq4=
 github.com/jackpal/go-nat-pmp v1.0.2 h1:KzKSgb7qkJvOUTqYl9/Hg/me3pWgBmERKrTGD7BdWus=
@@ -393,8 +393,8 @@ github.com/libp2p/go-cidranger v1.1.0 h1:ewPN8EZ0dd1LSnrtuwd4709PXVcITVeuwbag38y
 github.com/libp2p/go-cidranger v1.1.0/go.mod h1:KWZTfSr+r9qEo9OkI9/SIEeAtw+NNoU0dXIXt15Okic=
 github.com/libp2p/go-flow-metrics v0.1.0 h1:0iPhMI8PskQwzh57jB9WxIuIOQ0r+15PChFGkx3Q3WM=
 github.com/libp2p/go-flow-metrics v0.1.0/go.mod h1:4Xi8MX8wj5aWNDAZttg6UPmc0ZrnFNsMtpsYUClFtro=
-github.com/libp2p/go-libp2p v0.29.1 h1:yNeg6XgP8gbdc4YSrwiIt5T1TGOrVjH8dzl8h0GIOfQ=
-github.com/libp2p/go-libp2p v0.29.1/go.mod h1:20El+LLy3/YhdUYIvGbLnvVJN32nMdqY6KXBENRAfLY=
+github.com/libp2p/go-libp2p v0.29.2 h1:uPw/c8hOxoLP/KhFnzlc5Ejqf+OmAL1dwIsqE31WBtY=
+github.com/libp2p/go-libp2p v0.29.2/go.mod h1:OU7nSq0aEZMsV2wY8nXn1+XNNt9q2UiR8LjW3Kmp2UE=
 github.com/libp2p/go-libp2p-asn-util v0.3.0 h1:gMDcMyYiZKkocGXDQ5nsUQyquC9+H+iLEQHwOCZ7s8s=
 github.com/libp2p/go-libp2p-asn-util v0.3.0/go.mod h1:B1mcOrKUE35Xq/ASTmQ4tN3LNzVVaMNmq2NACuqyB9w=
 github.com/libp2p/go-libp2p-gostream v0.6.0 h1:QfAiWeQRce6pqnYfmIVWJFXNdDyfiR/qkCnjyaZUPYU=
@@ -541,8 +541,8 @@ github.com/quic-go/qtls-go1-19 v0.3.3 h1:wznEHvJwd+2X3PqftRha0SUKmGsnb6dfArMhy9P
 github.com/quic-go/qtls-go1-19 v0.3.3/go.mod h1:ySOI96ew8lnoKPtSqx2BlI5wCpUVPT05RMAlajtnyOI=
 github.com/quic-go/qtls-go1-20 v0.2.3 h1:m575dovXn1y2ATOb1XrRFcrv0F+EQmlowTkoraNkDPI=
 github.com/quic-go/qtls-go1-20 v0.2.3/go.mod h1:JKtK6mjbAVcUTN/9jZpvLbGxvdWIKS8uT7EiStoU1SM=
-github.com/quic-go/quic-go v0.36.3 h1:f+yOqeGhMoRX7/M3wmEw/djhzKWr15FtQysox85/834=
-github.com/quic-go/quic-go v0.36.3/go.mod h1:qxQumdeKw5GmWs1OsTZZnOxzSI+RJWuhf1O8FN35L2o=
+github.com/quic-go/quic-go v0.36.4 h1:CXn/ZLN5Vntlk53fjR+kUMC8Jt7flfQe+I5Ty5A+k0o=
+github.com/quic-go/quic-go v0.36.4/go.mod h1:qxQumdeKw5GmWs1OsTZZnOxzSI+RJWuhf1O8FN35L2o=
 github.com/quic-go/webtransport-go v0.5.3 h1:5XMlzemqB4qmOlgIus5zB45AcZ2kCgCy2EptUrfOPWU=
 github.com/quic-go/webtransport-go v0.5.3/go.mod h1:OhmmgJIzTTqXK5xvtuX0oBpLV2GkLWNDA+UeTGJXErU=
 github.com/raulk/go-watchdog v1.3.0 h1:oUmdlHxdkXRJlwfG0O9omj8ukerm8MEQavSiDTEtBsk=