chore: Change from net to go-p2p (#4051)

## Relevant issue(s)

Resolves #4048
Resolves #4031

## Description

This PR removes the `defradb/net` package in favour of
`sourcenetwork/go-p2p`. It brings in the latest changes from `go-p2p`
which implies making changes to the `client.Host` interface.

Author: fredcarle

cbindings/node.go

@@ -21,8 +21,9 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/sourcenetwork/go-p2p"
+
 	"github.com/sourcenetwork/defradb/internal/db"
-	netConfig "github.com/sourcenetwork/defradb/net/config"
 	"github.com/sourcenetwork/defradb/node"
 )
 
@@ -55,7 +56,7 @@ func NewNode(cOptions C.NodeInitOptions) C.NewNodeResult {
 		node.WithLensRuntime(node.Wazero),
 	}
 	if len(listeningAddresses) > 0 {
-		opts = append(opts, netConfig.WithListenAddresses(listeningAddresses...))
+		opts = append(opts, p2p.WithListenAddresses(listeningAddresses...))
 	}
 	maxTxnRetries := gocOptions.MaxTransactionRetries
 	if maxTxnRetries > 0 {
@@ -74,7 +75,7 @@ func NewNode(cOptions C.NodeInitOptions) C.NewNodeResult {
 	}
 	peers := splitCommaSeparatedString(gocOptions.Peers)
 	if len(peers) > 0 {
-		opts = append(opts, netConfig.WithBootstrapPeers(peers...))
+		opts = append(opts, p2p.WithBootstrapPeers(peers...))
 	}
 	if gocOptions.Identity != nil {
 		opts = append(opts, db.WithNodeIdentity(gocOptions.Identity))

cbindings/p2p.go

@@ -18,10 +18,7 @@ import "C"
 
 import (
 	"context"
-	"encoding/json"
 	"time"
-
-	"github.com/sourcenetwork/defradb/client"
 )
 
 //export P2PInfo
@@ -30,8 +27,11 @@ func P2PInfo(nodePtr C.uintptr_t) C.Result {
 	if err != nil {
 		return returnC(returnGoC(1, err.Error(), ""))
 	}
-	info := node.DB.PeerInfo()
-	return returnC(marshalJSONToGoCResult(info))
+	addresses, err := node.DB.PeerInfo()
+	if err != nil {
+		return returnC(returnGoC(1, err.Error(), ""))
+	}
+	return returnC(marshalJSONToGoCResult(addresses))
 }
 
 //export P2PgetAllReplicators
@@ -49,41 +49,32 @@ func P2PgetAllReplicators(nodePtr C.uintptr_t) C.Result {
 }
 
 //export P2PsetReplicator
-func P2PsetReplicator(nodePtr C.uintptr_t, collections *C.char, peerInfo *C.char) C.Result {
+func P2PsetReplicator(nodePtr C.uintptr_t, collections *C.char, addresses *C.char) C.Result {
 	ctx := context.Background()
 	colArgs := splitCommaSeparatedString(C.GoString(collections))
-
-	var info client.PeerInfo
-	if err := json.Unmarshal([]byte(C.GoString(peerInfo)), &info); err != nil {
-		return returnC(returnGoC(1, err.Error(), ""))
-	}
+	addressesArgs := splitCommaSeparatedString(C.GoString(addresses))
 
 	node, err := getNodeFromPointer(nodePtr)
 	if err != nil {
 		return returnC(returnGoC(1, err.Error(), ""))
 	}
-	err = node.DB.SetReplicator(ctx, info, colArgs...)
+	err = node.DB.SetReplicator(ctx, addressesArgs, colArgs...)
 	if err != nil {
 		return returnC(returnGoC(1, err.Error(), ""))
 	}
 	return returnC(returnGoC(0, "", ""))
 }
 
 //export P2PdeleteReplicator
-func P2PdeleteReplicator(nodePtr C.uintptr_t, collections *C.char, peerInfo *C.char) C.Result {
+func P2PdeleteReplicator(nodePtr C.uintptr_t, collections *C.char, id *C.char) C.Result {
 	ctx := context.Background()
 	colArgs := splitCommaSeparatedString(C.GoString(collections))
 
-	var info client.PeerInfo
-	if err := json.Unmarshal([]byte(C.GoString(peerInfo)), &info); err != nil {
-		return returnC(returnGoC(1, err.Error(), ""))
-	}
-
 	node, err := getNodeFromPointer(nodePtr)
 	if err != nil {
 		return returnC(returnGoC(1, err.Error(), ""))
 	}
-	err = node.DB.DeleteReplicator(ctx, info, colArgs...)
+	err = node.DB.DeleteReplicator(ctx, C.GoString(id), colArgs...)
 	if err != nil {
 		return returnC(returnGoC(1, err.Error(), ""))
 	}
@@ -217,16 +208,14 @@ func P2PdocumentSync(nodePtr C.uintptr_t, collection *C.char, docIDs *C.char, ti
 }
 
 //export P2Pconnect
-func P2Pconnect(nodePtr C.uintptr_t, peerID *C.char, peerAddresses *C.char) C.Result {
+func P2Pconnect(nodePtr C.uintptr_t, peerAddresses *C.char) C.Result {
 	ctx := context.Background()
 	node, err := getNodeFromPointer(nodePtr)
 	if err != nil {
 		return returnC(returnGoC(1, err.Error(), ""))
 	}
-	var info client.PeerInfo
-	info.ID = C.GoString(peerID)
-	info.Addresses = splitCommaSeparatedString(C.GoString(peerAddresses))
-	err = node.DB.Connect(ctx, info)
+	addresses := splitCommaSeparatedString(C.GoString(peerAddresses))
+	err = node.DB.Connect(ctx, addresses)
 	if err != nil {
 		return returnC(returnGoC(1, err.Error(), ""))
 	}

cbindings/wrapper.go

@@ -40,12 +40,12 @@ extern NewNodeResult NewNode(NodeInitOptions cOptions);
 extern Result NodeClose(uintptr_t nodePtr);
 extern Result P2PInfo(uintptr_t nodePtr);
 extern Result P2PgetAllReplicators(uintptr_t nodePtr);
-extern Result P2PsetReplicator(uintptr_t nodePtr, char* collections, char* peerInfo);
-extern Result P2PdeleteReplicator(uintptr_t nodePtr, char* collections, char* peerInfo);
+extern Result P2PsetReplicator(uintptr_t nodePtr, char* collections, char* addresses);
+extern Result P2PdeleteReplicator(uintptr_t nodePtr, char* collections, char* id);
 extern Result P2PcollectionAdd(uintptr_t nodePtr, char* collections);
 extern Result P2PcollectionRemove(uintptr_t nodePtr, char* collections);
 extern Result P2PcollectionGetAll(uintptr_t nodePtr);
-extern Result P2Pconnect(uintptr_t nodePtr, char* peerID, char* peerAddresses);
+extern Result P2Pconnect(uintptr_t nodePtr, char* peerAddresses);
 extern Result P2PdocumentAdd(uintptr_t nodePtr, char* collections);
 extern Result P2PdocumentRemove(uintptr_t nodePtr, char* collections);
 extern Result P2PdocumentGetAll(uintptr_t nodePtr);
@@ -104,41 +104,41 @@ func NewCWrapper(node *node.Node) (*CWrapper, error) {
 	}, nil
 }
 
-func (w *CWrapper) PeerInfo() client.PeerInfo {
+func (w *CWrapper) PeerInfo() ([]string, error) {
 	res := ConvertAndFreeCResult(C.P2PInfo(C.uintptr_t(w.handle)))
 
 	if res.Status != 0 {
-		return client.PeerInfo{}
+		return nil, errors.New(res.Error)
 	}
 
-	addrInfo, err := unmarshalResult[client.PeerInfo](res.Value)
+	addresses, err := unmarshalResult[[]string](res.Value)
 	if err != nil {
-		return client.PeerInfo{}
+		return nil, err
 	}
-	return addrInfo
+	return addresses, nil
 }
 
-func (w *CWrapper) SetReplicator(ctx context.Context, info client.PeerInfo, collections ...string) error {
-	peerStr := C.CString(info.String())
+func (w *CWrapper) SetReplicator(ctx context.Context, addresses []string, collections ...string) error {
+	addrStr := C.CString(strings.Join(addresses, ","))
 	colStr := C.CString(strings.Join(collections, ","))
-	defer C.free(unsafe.Pointer(peerStr))
+	defer C.free(unsafe.Pointer(addrStr))
 	defer C.free(unsafe.Pointer(colStr))
 
-	res := ConvertAndFreeCResult(C.P2PsetReplicator(C.uintptr_t(w.handle), colStr, peerStr))
+	res := ConvertAndFreeCResult(C.P2PsetReplicator(C.uintptr_t(w.handle), colStr, addrStr))
 
 	if res.Status != 0 {
 		return errors.New(res.Error)
 	}
 	return nil
 }
 
-func (w *CWrapper) DeleteReplicator(ctx context.Context, info client.PeerInfo, collections ...string) error {
-	peerStr := C.CString(info.String())
+func (w *CWrapper) DeleteReplicator(ctx context.Context, id string, collections ...string) error {
+	peerID := C.CString(id)
 	colStr := C.CString(strings.Join(collections, ","))
-	defer C.free(unsafe.Pointer(peerStr))
+	defer C.free(unsafe.Pointer(peerID))
 	defer C.free(unsafe.Pointer(colStr))
 
-	res := ConvertAndFreeCResult(C.P2PdeleteReplicator(C.uintptr_t(w.handle), colStr, peerStr))
+	res := ConvertAndFreeCResult(C.P2PdeleteReplicator(C.uintptr_t(w.handle), colStr, peerID))
 
 	if res.Status != 0 {
 		return errors.New(res.Error)
@@ -822,13 +822,11 @@ func (w *CWrapper) PrintDump(ctx context.Context) error {
 	panic("not implemented")
 }
 
-func (w *CWrapper) Connect(ctx context.Context, addr client.PeerInfo) error {
-	cPeerID := C.CString(addr.ID)
-	cPeerAddresses := C.CString(strings.Join(addr.Addresses, ","))
-	defer C.free(unsafe.Pointer(cPeerID))
+func (w *CWrapper) Connect(ctx context.Context, addresses []string) error {
+	cPeerAddresses := C.CString(strings.Join(addresses, ","))
 	defer C.free(unsafe.Pointer(cPeerAddresses))
 	callHandle := getNodeOrTxnHandle(w.handle, ctx)
-	res := ConvertAndFreeCResult(C.P2Pconnect(callHandle, cPeerID, cPeerAddresses))
+	res := ConvertAndFreeCResult(C.P2Pconnect(callHandle, cPeerAddresses))
 	if res.Status != 0 {
 		return errors.New(res.Error)
 	}

cli/p2p_connect.go

@@ -11,30 +11,24 @@
 package cli
 
 import (
-	"encoding/json"
-
 	"github.com/spf13/cobra"
-
-	"github.com/sourcenetwork/defradb/client"
 )
 
 func MakeP2PConnectCommand() *cobra.Command {
 	var cmd = &cobra.Command{
-		Use:   "connect <peerInfo>",
-		Short: "Connect to a peer",
-		Long: `Connect to a peer with the given ID and addresses
-Example: 
-  defradb client p2p connect '{"ID": "12D3", "Addrs": ["/ip4/0.0.0.0/tcp/9171"]}'
-  		`,
-		Args: cobra.ExactArgs(1),
+		Use:   "connect <addresses...>",
+		Short: "Connect to one or more peers",
+		Long: `Connect to one or more peers with the given addresses
+
+Example: Connect to a peer
+  defradb client p2p connect /ip4/0.0.0.0/tcp/9171/p2p/12D3KooW...
+  
+Example: Connect to multiple peers
+  defradb client p2p connect /ip4/0.0.0.0/tcp/9171/p2p/12D3KooW... /ip4/0.0.0.0/tcp/9172/p2p/1543LKs...
+		`,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			cliClient := mustGetContextCLIClient(cmd)
-
-			var info client.PeerInfo
-			if err := json.Unmarshal([]byte(args[0]), &info); err != nil {
-				return err
-			}
-			return cliClient.Connect(cmd.Context(), info)
+			return cliClient.Connect(cmd.Context(), args)
 		},
 	}
 	return cmd

cli/p2p_info.go

@@ -21,7 +21,11 @@ func MakeP2PInfoCommand() *cobra.Command {
 		Long:  `Get peer info from a DefraDB node`,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			cliClient := mustGetContextCLIClient(cmd)
-			return writeJSON(cmd, cliClient.PeerInfo())
+			addresses, err := cliClient.PeerInfo()
+			if err != nil {
+				return err
+			}
+			return writeJSON(cmd, addresses)
 		},
 	}
 	return cmd

cli/p2p_replicator_delete.go

@@ -11,33 +11,24 @@
 package cli
 
 import (
-	"encoding/json"
-
 	"github.com/spf13/cobra"
-
-	"github.com/sourcenetwork/defradb/client"
 )
 
 func MakeP2PReplicatorDeleteCommand() *cobra.Command {
 	var collections []string
 	var cmd = &cobra.Command{
-		Use:   "delete [-c, --collection] <peer>",
+		Use:   "delete [-c, --collection] <peerID>",
 		Short: "Delete replicator(s) and stop synchronization",
 		Long: `Delete replicator(s) and stop synchronization.
-A replicator synchronizes one or all collection(s) from this node to another.
+A replicator synchronizes one or all collection(s) from this instance to another.
 		
 Example:		
-  defradb client p2p replicator delete -c Users '{"ID": "12D3", "Addrs": ["/ip4/0.0.0.0/tcp/9171"]}'
+  defradb client p2p replicator delete -c Users 12D3...
 		`,
 		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			cliClient := mustGetContextCLIClient(cmd)
-
-			var info client.PeerInfo
-			if err := json.Unmarshal([]byte(args[0]), &info); err != nil {
-				return err
-			}
-			return cliClient.DeleteReplicator(cmd.Context(), info, collections...)
+			return cliClient.DeleteReplicator(cmd.Context(), args[0], collections...)
 		},
 	}
 	cmd.Flags().StringSliceVarP(&collections, "collection", "c",

cli/p2p_replicator_getall.go

@@ -19,7 +19,7 @@ func MakeP2PReplicatorGetAllCommand() *cobra.Command {
 		Use:   "getall",
 		Short: "Get all replicators",
 		Long: `Get all the replicators active in the P2P data sync system.
-A replicator synchronizes one or all collection(s) from this node to another.
+A replicator synchronizes one or all collection(s) from this instance to another.
 
 Example:
   defradb client p2p replicator getall

cli/p2p_replicator_set.go

@@ -11,33 +11,26 @@
 package cli
 
 import (
-	"encoding/json"
-
 	"github.com/spf13/cobra"
-
-	"github.com/sourcenetwork/defradb/client"
 )
 
 func MakeP2PReplicatorSetCommand() *cobra.Command {
 	var collections []string
 	var cmd = &cobra.Command{
-		Use:   "set [-c, --collection] <peer>",
+		Use:   "set [-c, --collection] <addresses...>",
 		Short: "Add replicator(s) and start synchronization",
 		Long: `Add replicator(s) and start synchronization.
-A replicator synchronizes one or all collection(s) from this node to another.
+A replicator synchronizes one or all collection(s) from this instance to another.
 
-Example:
-  defradb client p2p replicator set -c Users '{"ID": "12D3", "Addrs": ["/ip4/0.0.0.0/tcp/9171"]}'
+Example: Add a replicator to replicate the "Users" collection to a peer at the given address
+  defradb client p2p replicator set -c Users /ip4/0.0.0.0/tcp/9171/p2p/12D3KooW...
+
+Example: Add a replicator to replicate the "Orders" collection to multiple peers at the given addresses
+  defradb client p2p replicator set -c Orders /ip4/0.0.0.0/tcp/9171/p2p/12D3KooW... /ip4/0.0.0.0/tcp/9172/p2p/1543LKs...
 `,
-		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			cliClient := mustGetContextCLIClient(cmd)
-
-			var info client.PeerInfo
-			if err := json.Unmarshal([]byte(args[0]), &info); err != nil {
-				return err
-			}
-			return cliClient.SetReplicator(cmd.Context(), info, collections...)
+			return cliClient.SetReplicator(cmd.Context(), args, collections...)
 		},
 	}
 

cli/start.go

@@ -20,6 +20,7 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
+	"github.com/sourcenetwork/go-p2p"
 	"github.com/sourcenetwork/immutable"
 
 	"github.com/sourcenetwork/defradb/acp/identity"
@@ -31,7 +32,6 @@ import (
 	"github.com/sourcenetwork/defradb/internal/db"
 	"github.com/sourcenetwork/defradb/internal/telemetry"
 	"github.com/sourcenetwork/defradb/keyring"
-	netConfig "github.com/sourcenetwork/defradb/net/config"
 	"github.com/sourcenetwork/defradb/node"
 	"github.com/sourcenetwork/defradb/version"
 )
@@ -102,10 +102,10 @@ func MakeStartCommand() *cobra.Command {
 				db.WithMaxRetries(cfg.GetInt("datastore.MaxTxnRetries")),
 				db.WithRetryInterval(replicatorRetryIntervals),
 				// net node options
-				netConfig.WithListenAddresses(cfg.GetStringSlice("net.p2pAddresses")...),
-				netConfig.WithEnablePubSub(cfg.GetBool("net.pubSubEnabled")),
-				netConfig.WithEnableRelay(cfg.GetBool("net.relayEnabled")),
-				netConfig.WithBootstrapPeers(cfg.GetStringSlice("net.peers")...),
+				p2p.WithListenAddresses(cfg.GetStringSlice("net.p2pAddresses")...),
+				p2p.WithEnablePubSub(cfg.GetBool("net.pubSubEnabled")),
+				p2p.WithEnableRelay(cfg.GetBool("net.relayEnabled")),
+				p2p.WithBootstrapPeers(cfg.GetStringSlice("net.peers")...),
 
 				// http server options
 				http.WithAddress(cfg.GetString("api.address")),
@@ -407,7 +407,7 @@ func getOrCreatePeerKey(kr keyring.Keyring, opts []node.Option) ([]node.Option,
 	} else if err != nil {
 		return nil, err
 	}
-	return append(opts, netConfig.WithPrivateKey(peerKey)), nil
+	return append(opts, p2p.WithPrivateKey(peerKey)), nil
 }
 
 func getOrCreateIdentity(kr keyring.Keyring, opts []node.Option, cfg *viper.Viper) ([]node.Option, error) {