fix: pubsub discovery

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/aead/ecdh v0.2.0
 	github.com/berty/emitter-go v0.0.0-20221031144724-5dae963c3622
 	github.com/berty/go-libp2p-mock v1.0.1
+	github.com/berty/go-libp2p-pubsub v0.9.4-0.20230706070911-6e35c0f470b8
 	github.com/berty/go-libp2p-rendezvous v0.5.0
 	github.com/buicongtan1997/protoc-gen-swagger-config v0.0.0-20200705084907-1342b78c1a7e
 	github.com/daixiang0/gci v0.8.2

pkg/ipfsutil/mobile.go

@@ -54,6 +54,16 @@ func (o *MobileOptions) fillDefault() {
 	if o.IpfsConfigPatch == nil {
 		o.IpfsConfigPatch = defaultIpfsConfigPatch
 	}
+
+	// apply default extras
+	if o.ExtraOpts == nil {
+		o.ExtraOpts = make(map[string]bool)
+	}
+
+	//  if not set, disable pubsub by default to avoid collision
+	if _, ok := o.ExtraOpts["pubsub"]; !ok {
+		o.ExtraOpts["pubsub"] = false
+	}
 }
 
 func NewIPFSMobile(ctx context.Context, repo *ipfs_mobile.RepoMobile, opts *MobileOptions) (*ipfs_mobile.IpfsMobile, error) {

service.go

@@ -9,7 +9,9 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+	"unsafe"
 
+	pubsub_fix "github.com/berty/go-libp2p-pubsub"
 	ds "github.com/ipfs/go-datastore"
 	ds_sync "github.com/ipfs/go-datastore/sync"
 	ipfs_interface "github.com/ipfs/interface-go-ipfs-core"
@@ -18,6 +20,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/event"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
+	backoff "github.com/libp2p/go-libp2p/p2p/discovery/backoff"
 	"github.com/libp2p/go-libp2p/p2p/host/eventbus"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -29,6 +32,7 @@ import (
 	"berty.tech/go-orbit-db/baseorbitdb"
 	"berty.tech/go-orbit-db/iface"
 	"berty.tech/go-orbit-db/pubsub/directchannel"
+	"berty.tech/go-orbit-db/pubsub/pubsubraw"
 	"berty.tech/weshnet/internal/bertyversion"
 	"berty.tech/weshnet/internal/datastoreutil"
 	"berty.tech/weshnet/pkg/bertyvcissuer"
@@ -135,6 +139,8 @@ func (opts *Opts) applyDefaults(ctx context.Context) error {
 		opts.Logger = zap.NewNop()
 	}
 
+	rng := mrand.New(mrand.NewSource(srand.MustSecure())) // nolint:gosec // we need to use math/rand here, but it is seeded from crypto/rand
+
 	opts.applyDefaultsGetDatastore()
 
 	if err := opts.applyPushDefaults(); err != nil {
@@ -166,11 +172,7 @@ func (opts *Opts) applyDefaults(ctx context.Context) error {
 		}
 
 		mrepo := ipfs_mobile.NewRepoMobile("", repo)
-		mnode, err = ipfsutil.NewIPFSMobile(ctx, mrepo, &ipfsutil.MobileOptions{
-			ExtraOpts: map[string]bool{
-				"pubsub": true,
-			},
-		})
+		mnode, err = ipfsutil.NewIPFSMobile(ctx, mrepo, &ipfsutil.MobileOptions{})
 		if err != nil {
 			return err
 		}
@@ -197,7 +199,6 @@ func (opts *Opts) applyDefaults(ctx context.Context) error {
 
 	// setup default tinder service
 	if opts.TinderService == nil {
-		rng := mrand.New(mrand.NewSource(srand.MustSecure())) // nolint:gosec // we need to use math/rand here, but it is seeded from crypto/rand
 		drivers := []tinder.IDriver{}
 
 		// setup loac disc
@@ -218,15 +219,51 @@ func (opts *Opts) applyDefaults(ctx context.Context) error {
 		}
 	}
 
+	if opts.PubSub == nil {
+		var err error
+
+		popts := []pubsub_fix.Option{
+			pubsub_fix.WithMessageSigning(true),
+			pubsub_fix.WithPeerExchange(true),
+		}
+
+		backoffstrat := backoff.NewExponentialBackoff(
+			time.Second*10, time.Hour,
+			backoff.FullJitter,
+			time.Second, 10.0, 0, rng)
+
+		cacheSize := 100
+		dialTimeout := time.Second * 20
+		backoffconnector := func(host host.Host) (*backoff.BackoffConnector, error) {
+			return backoff.NewBackoffConnector(host, cacheSize, dialTimeout, backoffstrat)
+		}
+
+		adaptater := tinder.NewDiscoveryAdaptater(opts.Logger.Named("disc"), opts.TinderService)
+		popts = append(popts, pubsub_fix.WithDiscovery(adaptater, pubsub_fix.WithDiscoverConnector(backoffconnector)))
+
+		// pubsub.DiscoveryPollInterval = m.Node.Protocol.PollInterval
+		ps, err := pubsub_fix.NewGossipSub(ctx, opts.Host, popts...)
+		if err != nil {
+			return fmt.Errorf("unable to init gossipsub: %w", err)
+		}
+
+		// @NOTE(gfanton): we need to force cast here until our fix is push
+		// upstream on the original go-libp2p-pubsub
+		// see: https://github.com/gfanton/go-libp2p-pubsub/commit/8f4fd394f8dfcb3a5eb724a03f9e4e1e33194cbd
+		opts.PubSub = (*pubsub.PubSub)(unsafe.Pointer(ps))
+	}
+
 	if opts.OrbitDB == nil {
 		orbitDirectory := InMemoryDirectory
 		if opts.DatastoreDir != InMemoryDirectory {
 			orbitDirectory = filepath.Join(opts.DatastoreDir, NamespaceOrbitDBDirectory)
 		}
 
+		pubsub := pubsubraw.NewPubSub(opts.PubSub, opts.Host.ID(), opts.Logger, nil)
 		odbOpts := &NewOrbitDBOptions{
 			NewOrbitDBOptions: baseorbitdb.NewOrbitDBOptions{
 				Directory: &orbitDirectory,
+				PubSub:    pubsub,
 				Logger:    opts.Logger,
 			},
 			PrometheusRegister:     opts.PrometheusRegister,

service_client.go

@@ -99,11 +99,7 @@ func NewPersistentServiceClient(path string) (ServiceClient, error) {
 	}
 
 	mrepo := ipfs_mobile.NewRepoMobile(path, repo)
-	mnode, err := ipfsutil.NewIPFSMobile(context.TODO(), mrepo, &ipfsutil.MobileOptions{
-		ExtraOpts: map[string]bool{
-			"pubsub": true,
-		},
-	})
+	mnode, err := ipfsutil.NewIPFSMobile(context.TODO(), mrepo, &ipfsutil.MobileOptions{})
 	if err != nil {
 		return nil, err
 	}