Periodically persist routing table snapshots

dht.go

@@ -30,6 +30,7 @@ import (
 	logging "github.com/ipfs/go-log"
 	"github.com/jbenet/goprocess"
 	goprocessctx "github.com/jbenet/goprocess/context"
+	periodicproc "github.com/jbenet/goprocess/periodic"
 	"github.com/multiformats/go-base32"
 	ma "github.com/multiformats/go-multiaddr"
 	"github.com/multiformats/go-multihash"
@@ -180,6 +181,18 @@ func New(ctx context.Context, h host.Host, options ...Option) (*IpfsDHT, error)
 		}
 	}
 
+	// schedule periodic snapshots
+	snapshotter := cfg.persist.snapshotter
+	snapshotInterval := cfg.persist.snapshotInterval
+	sproc := periodicproc.Tick(snapshotInterval, func(proc goprocess.Process) {
+		logger.Debug("persisting Routing Table snapshot")
+		err := snapshotter.Store(h, dht.routingTable)
+		if err != nil {
+			logger.Errorw("failed to persist Routing Table snapshot", "err", err)
+		}
+	})
+	dht.proc.AddChild(sproc)
+
 	// register for event bus and network notifications
 	sn, err := newSubscriberNotifiee(dht)
 	if err != nil {

dht_options.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/libp2p/go-libp2p-kad-dht/persist"
+
 	ds "github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
 	"github.com/ipfs/go-ipns"
@@ -59,6 +61,11 @@ type config struct {
 		peerFilter          RouteTableFilterFunc
 	}
 
+	persist struct {
+		snapshotter      persist.Snapshotter
+		snapshotInterval time.Duration
+	}
+
 	// set to true if we're operating in v1 dht compatible mode
 	v1CompatibleMode bool
 	bootstrapPeers   []peer.AddrInfo
@@ -93,6 +100,15 @@ func (c *config) applyFallbacks(h host.Host) error {
 			return fmt.Errorf("the default validator was changed without being marked as changed")
 		}
 	}
+
+	if c.persist.snapshotter == nil {
+		s, err := persist.NewDatastoreSnapshotter(c.datastore, persist.DefaultSnapshotNS)
+		if err != nil {
+			return fmt.Errorf("failed to create snapshotter %w", err)
+		}
+		c.persist.snapshotter = s
+	}
+
 	return nil
 }
 
@@ -124,6 +140,8 @@ var defaults = func(o *config) error {
 
 	o.v1CompatibleMode = true
 
+	o.persist.snapshotInterval = 5 * time.Minute
+
 	return nil
 }
 
@@ -408,3 +426,16 @@ func BootstrapPeers(addrs ...ma.Multiaddr) Option {
 		return nil
 	}
 }
+
+// Snapshotter configures the snapshotter and the interval to use to periodically
+// persist the Routing Table. See the documentation for the `persist.Snapshotter` interface
+// for more details.
+// If you configure this, please ensure that you configure the DHT with a persistent Datastore
+// to ensure that the routing table snapshots are persistent across DHT restarts.
+func Snapshotter(s persist.Snapshotter, interval time.Duration) Option {
+	return func(c *config) error {
+		c.persist.snapshotter = s
+		c.persist.snapshotInterval = interval
+		return nil
+	}
+}

dht_test.go

@@ -21,12 +21,15 @@ import (
 
 	test "github.com/libp2p/go-libp2p-kad-dht/internal/testing"
 	pb "github.com/libp2p/go-libp2p-kad-dht/pb"
+	"github.com/libp2p/go-libp2p-kad-dht/persist"
 	kb "github.com/libp2p/go-libp2p-kbucket"
 	record "github.com/libp2p/go-libp2p-record"
 	swarmt "github.com/libp2p/go-libp2p-swarm/testing"
 	bhost "github.com/libp2p/go-libp2p/p2p/host/basic"
 
 	"github.com/ipfs/go-cid"
+	"github.com/ipfs/go-datastore"
+	dssync "github.com/ipfs/go-datastore/sync"
 	detectrace "github.com/ipfs/go-detect-race"
 	u "github.com/ipfs/go-ipfs-util"
 	ma "github.com/multiformats/go-multiaddr"
@@ -387,6 +390,56 @@ func TestValueSetInvalid(t *testing.T) {
 	testSetGet("valid", true, "newer", nil)
 }
 
+func TestRoutingTableSnapshot(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	sdstore := dssync.MutexWrap(datastore.NewMapDatastore())
+	snapShotter, err := persist.NewDatastoreSnapshotter(sdstore, "rand")
+	require.NoError(t, err)
+	snapShotterOpt := Snapshotter(snapShotter, 100*time.Millisecond)
+
+	// start 4 other dht's we can connect to & add to our RT
+	dhts := setupDHTS(t, ctx, 4)
+	defer func() {
+		for i := 0; i < 4; i++ {
+			dhts[i].Close()
+			defer dhts[i].host.Close()
+		}
+	}()
+
+	// connect them with each other
+	connect(t, ctx, dhts[0], dhts[1])
+	connect(t, ctx, dhts[1], dhts[2])
+	connect(t, ctx, dhts[2], dhts[3])
+
+	// create dht with snapshotter  & connect it to one of the above dht's
+	selfDht := setupDHT(ctx, t, false, snapShotterOpt)
+	require.True(t, selfDht.routingTable.Size() == 0)
+	defer selfDht.Close()
+	defer selfDht.host.Close()
+
+	connect(t, ctx, selfDht, dhts[0])
+
+	for selfDht.routingTable.Size() != 4 {
+		<-selfDht.ForceRefresh()
+	}
+
+	// assert snapshot & close dht
+	time.Sleep(500 * time.Millisecond) // wait for one snapshot
+	snapshotPeers, err := snapShotter.Load()
+	require.Len(t, snapshotPeers, 4)
+	var peerIds []peer.ID
+	for i := range snapshotPeers {
+		peerIds = append(peerIds, snapshotPeers[i].ID)
+	}
+	require.Len(t, peerIds, 4)
+
+	require.NoError(t, err)
+	for _, d := range dhts {
+		require.Contains(t, peerIds, d.self)
+	}
+}
+
 func TestContextShutDown(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

go.mod

@@ -17,8 +17,9 @@ require (
 	github.com/jbenet/goprocess v0.1.4
 	github.com/libp2p/go-eventbus v0.1.0
 	github.com/libp2p/go-libp2p v0.8.2
+	github.com/libp2p/go-libp2p-blankhost v0.1.4
 	github.com/libp2p/go-libp2p-core v0.5.4
-	github.com/libp2p/go-libp2p-kbucket v0.4.2
+	github.com/libp2p/go-libp2p-kbucket v0.4.3-0.20200515080035-b37bee77ff8d
 	github.com/libp2p/go-libp2p-peerstore v0.2.4
 	github.com/libp2p/go-libp2p-record v0.1.2
 	github.com/libp2p/go-libp2p-routing-helpers v0.2.3

go.sum

@@ -195,6 +195,8 @@ github.com/libp2p/go-libp2p-discovery v0.3.0 h1:+JnYBRLzZQtRq0mK3xhyjBwHytLmJXMT
 github.com/libp2p/go-libp2p-discovery v0.3.0/go.mod h1:o03drFnz9BVAZdzC/QUQ+NeQOu38Fu7LJGEOK2gQltw=
 github.com/libp2p/go-libp2p-kbucket v0.4.2 h1:wg+VPpCtY61bCasGRexCuXOmEmdKjN+k1w+JtTwu9gA=
 github.com/libp2p/go-libp2p-kbucket v0.4.2/go.mod h1:7sCeZx2GkNK1S6lQnGUW5JYZCFPnXzAZCCBBS70lytY=
+github.com/libp2p/go-libp2p-kbucket v0.4.3-0.20200515080035-b37bee77ff8d h1:tLb9EZQonRbrMI3lo7/0LkhV0tiNIpddEK0hAk70dC4=
+github.com/libp2p/go-libp2p-kbucket v0.4.3-0.20200515080035-b37bee77ff8d/go.mod h1:7sCeZx2GkNK1S6lQnGUW5JYZCFPnXzAZCCBBS70lytY=
 github.com/libp2p/go-libp2p-loggables v0.1.0 h1:h3w8QFfCt2UJl/0/NW4K829HX/0S4KD31PQ7m8UXXO8=
 github.com/libp2p/go-libp2p-loggables v0.1.0/go.mod h1:EyumB2Y6PrYjr55Q3/tiJ/o3xoDasoRYM7nOzEpoa90=
 github.com/libp2p/go-libp2p-mplex v0.2.0/go.mod h1:Ejl9IyjvXJ0T9iqUTE1jpYATQ9NM3g+OtR+EMMODbKo=
@@ -266,6 +268,7 @@ github.com/libp2p/go-netroute v0.1.2 h1:UHhB35chwgvcRI392znJA3RCBtZ3MpE3ahNCN5MR
 github.com/libp2p/go-netroute v0.1.2/go.mod h1:jZLDV+1PE8y5XxBySEBgbuVAXbhtuHSdmLPL2n9MKbk=
 github.com/libp2p/go-openssl v0.0.2/go.mod h1:v8Zw2ijCSWBQi8Pq5GAixw6DbFfa9u6VIYDXnvOXkc0=
 github.com/libp2p/go-openssl v0.0.3/go.mod h1:unDrJpgy3oFr+rqXsarWifmJuNnJR4chtO1HmaZjggc=
+github.com/libp2p/go-openssl v0.0.4 h1:d27YZvLoTyMhIN4njrkr8zMDOM4lfpHIp6A+TK9fovg=
 github.com/libp2p/go-openssl v0.0.4/go.mod h1:unDrJpgy3oFr+rqXsarWifmJuNnJR4chtO1HmaZjggc=
 github.com/libp2p/go-reuseport v0.0.1 h1:7PhkfH73VXfPJYKQ6JwS5I/eVcoyYi9IMNGc6FWpFLw=
 github.com/libp2p/go-reuseport v0.0.1/go.mod h1:jn6RmB1ufnQwl0Q1f+YxAj8isJgDCQzaaxIFYDhcYEA=
@@ -375,6 +378,7 @@ github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/smola/gocompat v0.2.0/go.mod h1:1B0MlxbmoZNo3h8guHp8HztB3BSYR5itql9qtVc0ypY=
 github.com/spacemonkeygo/openssl v0.0.0-20181017203307-c2dcc5cca94a/go.mod h1:7AyxJNCJ7SBZ1MfVQCWD6Uqo2oubI2Eq2y2eqf+A5r0=
+github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572 h1:RC6RW7j+1+HkWaX/Yh71Ee5ZHaHYt7ZP4sQgUrm6cDU=
 github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572/go.mod h1:w0SWMsp6j9O/dk4/ZpIhL+3CkG8ofA2vuv7k+ltqUMc=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=