refactor: P2P client interface

cli/p2p_collection_add.go

@@ -22,8 +22,8 @@ func MakeP2PCollectionAddCommand() *cobra.Command {
 The collections are synchronized between nodes of a pubsub network.`,
 		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
-			return store.AddP2PCollection(cmd.Context(), args[0])
+			p2p := mustGetP2PContext(cmd)
+			return p2p.AddP2PCollection(cmd.Context(), args[0])
 		},
 	}
 	return cmd

cli/p2p_collection_getall.go

@@ -22,9 +22,9 @@ func MakeP2PCollectionGetAllCommand() *cobra.Command {
 This is the list of collections of the node that are synchronized on the pubsub network.`,
 		Args: cobra.NoArgs,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
+			p2p := mustGetP2PContext(cmd)
 
-			cols, err := store.GetAllP2PCollections(cmd.Context())
+			cols, err := p2p.GetAllP2PCollections(cmd.Context())
 			if err != nil {
 				return err
 			}

cli/p2p_collection_remove.go

@@ -22,8 +22,8 @@ func MakeP2PCollectionRemoveCommand() *cobra.Command {
 The removed collections will no longer be synchronized between nodes.`,
 		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
-			return store.RemoveP2PCollection(cmd.Context(), args[0])
+			p2p := mustGetP2PContext(cmd)
+			return p2p.RemoveP2PCollection(cmd.Context(), args[0])
 		},
 	}
 	return cmd

cli/p2p_info.go

@@ -23,12 +23,7 @@ func MakeP2PInfoCommand() *cobra.Command {
 		Long:  `Get peer info from a DefraDB node`,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			db := cmd.Context().Value(dbContextKey).(*http.Client)
-
-			res, err := db.PeerInfo(cmd.Context())
-			if err != nil {
-				return err
-			}
-			return writeJSON(cmd, res)
+			return writeJSON(cmd, db.PeerInfo())
 		},
 	}
 	return cmd

cli/p2p_replicator_delete.go

@@ -11,27 +11,36 @@
 package cli
 
 import (
+	"encoding/json"
+
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/spf13/cobra"
 
 	"github.com/sourcenetwork/defradb/client"
 )
 
 func MakeP2PReplicatorDeleteCommand() *cobra.Command {
+	var collections []string
 	var cmd = &cobra.Command{
-		Use:   "delete <peer>",
+		Use:   "delete [-c, --collection] <peer>",
 		Short: "Delete a replicator. It will stop synchronizing",
 		Long:  `Delete a replicator. It will stop synchronizing.`,
 		Args:  cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
+			p2p := mustGetP2PContext(cmd)
 
-			addr, err := peer.AddrInfoFromString(args[0])
-			if err != nil {
+			var info peer.AddrInfo
+			if err := json.Unmarshal([]byte(args[0]), &info); err != nil {
 				return err
 			}
-			return store.DeleteReplicator(cmd.Context(), client.Replicator{Info: *addr})
+			rep := client.Replicator{
+				Info:    info,
+				Schemas: collections,
+			}
+			return p2p.DeleteReplicator(cmd.Context(), rep)
 		},
 	}
+	cmd.Flags().StringSliceVarP(&collections, "collection", "c",
+		[]string{}, "Collection(s) to stop replicating")
 	return cmd
 }

cli/p2p_replicator_getall.go

@@ -21,9 +21,9 @@ func MakeP2PReplicatorGetAllCommand() *cobra.Command {
 		Long: `Get all the replicators active in the P2P data sync system.
 These are the replicators that are currently replicating data from one node to another.`,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
+			p2p := mustGetP2PContext(cmd)
 
-			reps, err := store.GetAllReplicators(cmd.Context())
+			reps, err := p2p.GetAllReplicators(cmd.Context())
 			if err != nil {
 				return err
 			}

cli/p2p_replicator_set.go

@@ -11,6 +11,8 @@
 package cli
 
 import (
+	"encoding/json"
+
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/spf13/cobra"
 
@@ -27,21 +29,21 @@ A replicator replicates one or all collection(s) from this node to another.
 `,
 		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			store := mustGetStoreContext(cmd)
+			p2p := mustGetP2PContext(cmd)
 
-			addr, err := peer.AddrInfoFromString(args[0])
-			if err != nil {
+			var info peer.AddrInfo
+			if err := json.Unmarshal([]byte(args[0]), &info); err != nil {
 				return err
 			}
 			rep := client.Replicator{
-				Info:    *addr,
+				Info:    info,
 				Schemas: collections,
 			}
-			return store.SetReplicator(cmd.Context(), rep)
+			return p2p.SetReplicator(cmd.Context(), rep)
 		},
 	}
 
 	cmd.Flags().StringSliceVarP(&collections, "collection", "c",
-		[]string{}, "Define the collection for the replicator")
+		[]string{}, "Collection(s) to replicate")
 	return cmd
 }

cli/start.go

@@ -13,20 +13,14 @@ package cli
 import (
 	"context"
 	"fmt"
-	gonet "net"
 	"net/http"
 	"os"
 	"os/signal"
 	"strings"
 	"syscall"
 
 	badger "github.com/dgraph-io/badger/v4"
-	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
-	grpc_recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
-	ma "github.com/multiformats/go-multiaddr"
 	"github.com/spf13/cobra"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/keepalive"
 
 	"github.com/sourcenetwork/defradb/client"
 	"github.com/sourcenetwork/defradb/config"
@@ -37,7 +31,6 @@ import (
 	httpapi "github.com/sourcenetwork/defradb/http"
 	"github.com/sourcenetwork/defradb/logging"
 	"github.com/sourcenetwork/defradb/net"
-	netpb "github.com/sourcenetwork/defradb/net/pb"
 	netutils "github.com/sourcenetwork/defradb/net/utils"
 )
 
@@ -113,15 +106,6 @@ func MakeStartCommand(cfg *config.Config) *cobra.Command {
 		log.FeedbackFatalE(context.Background(), "Could not bind net.p2paddress", err)
 	}
 
-	cmd.Flags().String(
-		"tcpaddr", cfg.Net.TCPAddress,
-		"Listener address for the tcp gRPC server (formatted as a libp2p MultiAddr)",
-	)
-	err = cfg.BindFlag("net.tcpaddress", cmd.Flags().Lookup("tcpaddr"))
-	if err != nil {
-		log.FeedbackFatalE(context.Background(), "Could not bind net.tcpaddress", err)
-	}
-
 	cmd.Flags().Bool(
 		"no-p2p", cfg.Net.P2PDisabled,
 		"Disable the peer-to-peer network synchronization system",
@@ -186,15 +170,10 @@ type defraInstance struct {
 
 func (di *defraInstance) close(ctx context.Context) {
 	if di.node != nil {
-		if err := di.node.Close(); err != nil {
-			log.FeedbackInfo(
-				ctx,
-				"The node could not be closed successfully",
-				logging.NewKV("Error", err.Error()),
-			)
-		}
+		di.node.Close()
+	} else {
+		di.db.Close()
 	}
-	di.db.Close(ctx)
 	if err := di.server.Close(); err != nil {
 		log.FeedbackInfo(
 			ctx,
@@ -237,16 +216,16 @@ func start(ctx context.Context, cfg *config.Config) (*defraInstance, error) {
 	}
 
 	// init the p2p node
-	var n *net.Node
+	var node *net.Node
 	if !cfg.Net.P2PDisabled {
 		log.FeedbackInfo(ctx, "Starting P2P node", logging.NewKV("P2P address", cfg.Net.P2PAddress))
-		n, err = net.NewNode(
+		node, err = net.NewNode(
 			ctx,
 			db,
 			net.WithConfig(cfg),
 		)
 		if err != nil {
-			db.Close(ctx)
+			db.Close()
 			return nil, errors.Wrap("failed to start P2P node", err)
 		}
 
@@ -258,55 +237,13 @@ func start(ctx context.Context, cfg *config.Config) (*defraInstance, error) {
 				return nil, errors.Wrap(fmt.Sprintf("failed to parse bootstrap peers %v", cfg.Net.Peers), err)
 			}
 			log.Debug(ctx, "Bootstrapping with peers", logging.NewKV("Addresses", addrs))
-			n.Boostrap(addrs)
+			node.Bootstrap(addrs)
 		}
 
-		if err := n.Start(); err != nil {
-			if e := n.Close(); e != nil {
-				err = errors.Wrap(fmt.Sprintf("failed to close node: %v", e.Error()), err)
-			}
-			db.Close(ctx)
+		if err := node.Start(); err != nil {
+			node.Close()
 			return nil, errors.Wrap("failed to start P2P listeners", err)
 		}
-
-		MtcpAddr, err := ma.NewMultiaddr(cfg.Net.TCPAddress)
-		if err != nil {
-			return nil, errors.Wrap("failed to parse multiaddress", err)
-		}
-		addr, err := netutils.TCPAddrFromMultiAddr(MtcpAddr)
-		if err != nil {
-			return nil, errors.Wrap("failed to parse TCP address", err)
-		}
-
-		rpcTimeoutDuration, err := cfg.Net.RPCTimeoutDuration()
-		if err != nil {
-			return nil, errors.Wrap("failed to parse RPC timeout duration", err)
-		}
-
-		server := grpc.NewServer(
-			grpc.UnaryInterceptor(
-				grpc_middleware.ChainUnaryServer(
-					grpc_recovery.UnaryServerInterceptor(),
-				),
-			),
-			grpc.KeepaliveParams(
-				keepalive.ServerParameters{
-					MaxConnectionIdle: rpcTimeoutDuration,
-				},
-			),
-		)
-		tcplistener, err := gonet.Listen("tcp", addr)
-		if err != nil {
-			return nil, errors.Wrap(fmt.Sprintf("failed to listen on TCP address %v", addr), err)
-		}
-
-		go func() {
-			log.FeedbackInfo(ctx, "Started RPC server", logging.NewKV("Address", addr))
-			netpb.RegisterCollectionServer(server, n.Peer)
-			if err := server.Serve(tcplistener); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
-				log.FeedbackFatalE(ctx, "Failed to start RPC server", err)
-			}
-		}()
 	}
 
 	sOpt := []func(*httpapi.Server){
@@ -315,10 +252,6 @@ func start(ctx context.Context, cfg *config.Config) (*defraInstance, error) {
 		httpapi.WithAllowedOrigins(cfg.API.AllowedOrigins...),
 	}
 
-	if n != nil {
-		sOpt = append(sOpt, httpapi.WithPeerID(n.PeerID().String()))
-	}
-
 	if cfg.API.TLS {
 		sOpt = append(
 			sOpt,
@@ -328,32 +261,36 @@ func start(ctx context.Context, cfg *config.Config) (*defraInstance, error) {
 		)
 	}
 
-	s := httpapi.NewServer(db, sOpt...)
-	if err := s.Listen(ctx); err != nil {
-		return nil, errors.Wrap(fmt.Sprintf("failed to listen on TCP address %v", s.Addr), err)
+	var server *httpapi.Server
+	if node != nil {
+		server = httpapi.NewServer(node, sOpt...)
+	} else {
+		server = httpapi.NewServer(db, sOpt...)
+	}
+	if err := server.Listen(ctx); err != nil {
+		return nil, errors.Wrap(fmt.Sprintf("failed to listen on TCP address %v", server.Addr), err)
 	}
 	// save the address on the config in case the port number was set to random
-	cfg.API.Address = s.AssignedAddr()
+	cfg.API.Address = server.AssignedAddr()
 
 	// run the server in a separate goroutine
 	go func() {
 		log.FeedbackInfo(ctx, fmt.Sprintf("Providing HTTP API at %s.", cfg.API.AddressToURL()))
-		if err := s.Run(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		if err := server.Run(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) {
 			log.FeedbackErrorE(ctx, "Failed to run the HTTP server", err)
-			if n != nil {
-				if err := n.Close(); err != nil {
-					log.FeedbackErrorE(ctx, "Failed to close node", err)
-				}
+			if node != nil {
+				node.Close()
+			} else {
+				db.Close()
 			}
-			db.Close(ctx)
 			os.Exit(1)
 		}
 	}()
 
 	return &defraInstance{
-		node:   n,
+		node:   node,
 		db:     db,
-		server: s,
+		server: server,
 	}, nil
 }
 

cli/utils.go

@@ -50,6 +50,13 @@ func mustGetStoreContext(cmd *cobra.Command) client.Store {
 	return cmd.Context().Value(storeContextKey).(client.Store)
 }
 
+// mustGetP2PContext returns the p2p implementation for the current command context.
+//
+// If a p2p implementation is not set in the current context this function panics.
+func mustGetP2PContext(cmd *cobra.Command) client.P2P {
+	return cmd.Context().Value(dbContextKey).(client.P2P)
+}
+
 // tryGetCollectionContext returns the collection for the current command context
 // and a boolean indicating if the collection was set.
 func tryGetCollectionContext(cmd *cobra.Command) (client.Collection, bool) {

client/db.go

@@ -60,7 +60,7 @@ type DB interface {
 	// be created after calling this to resume operations on the prior data - this is however dependant on
 	// the behaviour of the rootstore provided on database instance creation, as this function will Close
 	// the provided rootstore.
-	Close(context.Context)
+	Close()
 
 	// Events returns the database event queue.
 	//
@@ -82,9 +82,6 @@ type DB interface {
 
 // Store contains the core DefraDB read-write operations.
 type Store interface {
-	// P2P holds the P2P related methods that must be implemented by the database.
-	P2P
-
 	// Backup holds the backup related methods that must be implemented by the database.
 	Backup
 

client/p2p.go

@@ -12,9 +12,17 @@ package client
 
 import (
 	"context"
+
+	"github.com/libp2p/go-libp2p/core/peer"
 )
 
+// P2P is a peer connected database implementation.
 type P2P interface {
+	DB
+
+	// PeerInfo returns the p2p host id and listening addresses.
+	PeerInfo() peer.AddrInfo
+
 	// SetReplicator adds a replicator to the persisted list or adds
 	// schemas if the replicator already exists.
 	SetReplicator(ctx context.Context, rep Replicator) error