Persistent DHT and Peerstore

Makefile

@@ -49,6 +49,20 @@ test-go-parallel:
 	go test ./... -race -timeout 30s
 
 
+.PHONY: test-key-value-stores
+test-key-value-stores: test-key-value-stores-go test-key-value-stores-wasm
+
+
+.PHONY: test-key-value-stores-go
+test-key-value-stores-go:
+	ENABLE_KEY_VALUE_TESTS=true go test ./db
+
+
+.PHONY: test-key-value-stores-wasm
+test-key-value-stores-wasm:
+	WASM_INIT_FILE="$$(pwd)/packages/mesh-browser-shim/dist/browser_shim.js" GOOS=js GOARCH=wasm ENABLE_KEY_VALUE_TESTS=true go test ./db -timeout 20m -tags=browser -exec="$$GOPATH/bin/wasmbrowsertest"
+
+
 .PHONY: test-go-serial
 test-go-serial:
 	go test ./zeroex/ordervalidator ./zeroex/orderwatch ./core -race -timeout 90s -p=1 --serial

cmd/mesh-bootstrap/main.go

@@ -12,12 +12,13 @@ import (
 	"strings"
 	"time"
 
+	meshdb "github.com/0xProject/0x-mesh/db"
 	"github.com/0xProject/0x-mesh/loghooks"
 	"github.com/0xProject/0x-mesh/p2p"
 	"github.com/0xProject/0x-mesh/p2p/banner"
-	sqlds "github.com/0xProject/sql-datastore"
 	"github.com/ipfs/go-datastore"
 	leveldbStore "github.com/ipfs/go-ds-leveldb"
+	sqlds "github.com/ipfs/go-ds-sql"
 	libp2p "github.com/libp2p/go-libp2p"
 	autonat "github.com/libp2p/go-libp2p-autonat-svc"
 	circuit "github.com/libp2p/go-libp2p-circuit"
@@ -172,7 +173,12 @@ func main() {
 		newDHT = func(h host.Host) (routing.PeerRouting, error) {
 			var err error
 			dhtDir := getDHTDir(config)
-			kadDHT, err = p2p.NewDHT(ctx, dhtDir, h)
+			// Set up the DHT to use LevelDB.
+			store, err := leveldbStore.NewDatastore(dhtDir, nil)
+			if err != nil {
+				return nil, err
+			}
+			kadDHT, err = dht.New(ctx, h, dhtopts.Datastore(store), dhtopts.Protocols(p2p.DHTProtocolID))
 			if err != nil {
 				log.WithField("error", err).Fatal("could not create DHT")
 			}
@@ -203,7 +209,7 @@ func main() {
 
 		newDHT = func(h host.Host) (routing.PeerRouting, error) {
 			var err error
-			dstore := sqlds.NewDatastore(db, sqlds.NewQueriesForTable(dhtTableName))
+			dstore := sqlds.NewDatastore(db, meshdb.NewPostgreSQLQueriesForTable(dhtTableName))
 
 			kadDHT, err = NewDHTWithDatastore(ctx, dstore, h)
 			if err != nil {
@@ -213,7 +219,7 @@ func main() {
 			return kadDHT, err
 		}
 
-		pstore := sqlds.NewDatastore(db, sqlds.NewQueriesForTable(peerStoreTableName))
+		pstore := sqlds.NewDatastore(db, meshdb.NewPostgreSQLQueriesForTable(peerStoreTableName))
 		peerStore, err = pstoreds.NewPeerstore(ctx, pstore, pstoreds.DefaultOpts())
 		if err != nil {
 			log.WithField("error", err).Fatal("could not create peerStore")

core/core.go

@@ -627,7 +627,7 @@ func (app *App) Start() error {
 		RendezvousPoints:       rendezvousPoints,
 		UseBootstrapList:       app.config.UseBootstrapList,
 		BootstrapList:          bootstrapList,
-		DataDir:                filepath.Join(app.config.DataDir, "p2p"),
+		DB:                     app.db,
 		CustomMessageValidator: app.orderFilter.ValidatePubSubMessage,
 		MaxBytesPerSecond:      app.config.MaxBytesPerSecond,
 	}

core/core_test.go

@@ -246,7 +246,7 @@ func TestOrderSync(t *testing.T) {
 	}
 	for i, testCase := range testCases {
 		testCaseName := fmt.Sprintf("%s (test case %d)", testCase.name, i)
-		t.Run(testCaseName, runOrdersyncTestCase(t, testCase))
+		t.Run(testCaseName, runOrdersyncTestCase(testCase))
 	}
 }
 
@@ -259,7 +259,7 @@ type ordersyncTestCase struct {
 
 const defaultOrderFilter = "{}"
 
-func runOrdersyncTestCase(t *testing.T, testCase ordersyncTestCase) func(t *testing.T) {
+func runOrdersyncTestCase(testCase ordersyncTestCase) func(t *testing.T) {
 	return func(t *testing.T) {
 		teardownSubTest := setupSubTest(t)
 		defer teardownSubTest(t)

core/ordersync/ordersync_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/0xProject/0x-mesh/constants"
+	"github.com/0xProject/0x-mesh/db"
 	"github.com/0xProject/0x-mesh/ethereum"
 	"github.com/0xProject/0x-mesh/p2p"
 	"github.com/0xProject/0x-mesh/scenario"
@@ -30,12 +31,14 @@ func TestCalculateDelayWithJitters(t *testing.T) {
 }
 
 func TestHandleRawRequest(t *testing.T) {
+	db, err := db.New(context.Background(), db.TestOptions())
+	require.NoError(t, err)
 	n, err := p2p.New(
 		context.Background(),
 		p2p.Config{
 			MessageHandler:   &noopMessageHandler{},
 			RendezvousPoints: []string{"/test-rendezvous-point"},
-			DataDir:          "/tmp",
+			DB:               db,
 		},
 	)
 	require.NoError(t, err)

db/common.go

@@ -12,6 +12,7 @@ import (
 	"github.com/0xProject/0x-mesh/zeroex"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/gibson042/canonicaljson-go"
+	ds "github.com/ipfs/go-datastore"
 )
 
 const (
@@ -48,6 +49,8 @@ type Database interface {
 	GetMetadata() (*types.Metadata, error)
 	SaveMetadata(metadata *types.Metadata) error
 	UpdateMetadata(updateFunc func(oldmetadata *types.Metadata) (newMetadata *types.Metadata)) error
+	PeerStore() ds.Batching
+	DHTStore() ds.Batching
 }
 
 type Options struct {

db/db_test.go

@@ -19,12 +19,32 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/math"
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
+	dstest "github.com/ipfs/go-datastore/test"
+	"github.com/plaid/go-envvar/envvar"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 var contractAddresses = ethereum.GanacheAddresses
 
+// Since the key value store tests take a very long time to run, we disable them
+// by default. These tests can be enabled with the `ENABLE_KEY_VALUE_TESTS`
+// environment variable.
+var keyValueTestsEnabled bool
+
+type TestingFlags struct {
+	EnableKeyValueTests bool `envvar:"ENABLE_KEY_VALUE_TESTS" default:"false"`
+}
+
+func init() {
+	var flags TestingFlags
+	if err := envvar.Parse(&flags); err != nil {
+		panic(fmt.Sprintf("could not parse environment variables: %s", err.Error()))
+	}
+	keyValueTestsEnabled = flags.EnableKeyValueTests
+	testing.Init()
+}
+
 func TestAddOrders(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -1818,6 +1838,66 @@ func makeMiniHeaderFilterTestCases(t *testing.T, db *DB) ([]*types.MiniHeader, [
 	return storedMiniHeaders, testCases
 }
 
+func TestPeerStoreBasic(t *testing.T) {
+	if !keyValueTestsEnabled {
+		t.Skip("Key-value store tests are disabled. You can enable them with the ENABLE_KEY_VALUE_TESTS environment variable")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	db := newTestDB(t, ctx)
+	peerstore := db.PeerStore()
+
+	for _, test := range dstest.BasicSubtests {
+		test(t, peerstore)
+	}
+}
+
+func TestPeerStoreBatch(t *testing.T) {
+	if !keyValueTestsEnabled {
+		t.Skip("Key-value store tests are disabled. You can enable them with the ENABLE_KEY_VALUE_TESTS environment variable")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	db := newTestDB(t, ctx)
+	peerstore := db.PeerStore()
+
+	for _, test := range dstest.BatchSubtests {
+		test(t, peerstore)
+	}
+}
+
+func TestDHTStoreBasic(t *testing.T) {
+	if !keyValueTestsEnabled {
+		t.Skip("Key-value store tests are disabled. You can enable them with the ENABLE_KEY_VALUE_TESTS environment variable")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	db := newTestDB(t, ctx)
+	dhtstore := db.DHTStore()
+
+	for _, test := range dstest.BasicSubtests {
+		test(t, dhtstore)
+	}
+}
+
+func TestDHTStoreBatch(t *testing.T) {
+	if !keyValueTestsEnabled {
+		t.Skip("Key-value store tests are disabled. You can enable them with the ENABLE_KEY_VALUE_TESTS environment variable")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	db := newTestDB(t, ctx)
+	dhtstore := db.DHTStore()
+
+	for _, test := range dstest.BatchSubtests {
+		test(t, dhtstore)
+	}
+}
+
 // safeSubsliceOrders returns a (shallow) subslice of orders without modifying
 // the original slice. Uses the same semantics as slice expressions: low is
 // inclusive, hi is exclusive. The returned slice still contains pointers, it