[P2P] [Tooling] Peer discovery peer list subcommand

app/client/cli/cmd.go

@@ -8,7 +8,7 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/pokt-network/pocket/app/client/cli/flags"
-	"github.com/pokt-network/pocket/runtime/configs"
+	"github.com/pokt-network/pocket/app/client/cli/peer"
 	"github.com/pokt-network/pocket/runtime/defaults"
 )
 
@@ -17,8 +17,6 @@ const (
 	flagBindErrFormat = "could not bind flag %q: %v"
 )
 
-var cfg *configs.Config
-
 func init() {
 	rootCmd.PersistentFlags().StringVar(&flags.RemoteCLIURL, "remote_cli_url", defaults.DefaultRemoteCLIURL, "takes a remote endpoint in the form of <protocol>://<host>:<port> (uses RPC Port)")
 	// ensure that this flag can be overridden by the respective viper-conventional environment variable (i.e. `POCKET_REMOTE_CLI_URL`)
@@ -39,19 +37,15 @@ func init() {
 	if err := viper.BindPFlag("verbose", rootCmd.PersistentFlags().Lookup("verbose")); err != nil {
 		log.Fatalf(flagBindErrFormat, "verbose", err)
 	}
+
+	rootCmd.AddCommand(peer.PeerCmd)
 }
 
 var rootCmd = &cobra.Command{
-	Use:   cliExecutableName,
-	Short: "Pocket Network Command Line Interface (CLI)",
-	Long:  "The CLI is meant to be an user but also a machine friendly way for interacting with Pocket Network.",
-	PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
-		// by this time, the config path should be set
-		cfg = configs.ParseConfig(flags.ConfigPath)
-		// set final `remote_cli_url` value; order of precedence: flag > env var > config > default
-		flags.RemoteCLIURL = viper.GetString("remote_cli_url")
-		return nil
-	},
+	Use:               cliExecutableName,
+	Short:             "Pocket Network Command Line Interface (CLI)",
+	Long:              "The CLI is meant to be an user but also a machine friendly way for interacting with Pocket Network.",
+	PersistentPreRunE: flags.ParseConfigAndFlags,
 }
 
 func ExecuteContext(ctx context.Context) error {

app/client/cli/flags/config.go

@@ -0,0 +1,19 @@
+package flags
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+
+	"github.com/pokt-network/pocket/runtime/configs"
+)
+
+var Cfg *configs.Config
+
+func ParseConfigAndFlags(_ *cobra.Command, _ []string) error {
+	// by this time, the config path should be set
+	Cfg = configs.ParseConfig(ConfigPath)
+
+	// set final `remote_cli_url` value; order of precedence: flag > env var > config > default
+	RemoteCLIURL = viper.GetString("remote_cli_url")
+	return nil
+}

app/client/cli/helpers/common.go

@@ -23,15 +23,14 @@ func FetchPeerstore(cmd *cobra.Command) (types.Peerstore, error) {
 	if err != nil {
 		return nil, err
 	}
-	// TECHDEBT(#811): use `bus.GetPeerstoreProvider()` after peerstore provider
-	// is retrievable as a proper submodule
-	pstoreProvider, err := bus.GetModulesRegistry().GetModule(peerstore_provider.PeerstoreProviderSubmoduleName)
+	// TECHDEBT(#811): Remove type casting once Peerstore is available as a submodule
+	pstoreProvider, err := peerstore_provider.GetPeerstoreProvider(bus)
 	if err != nil {
-		return nil, errors.New("retrieving peerstore provider")
+		return nil, err
 	}
 	currentHeightProvider := bus.GetCurrentHeightProvider()
 	height := currentHeightProvider.CurrentHeight()
-	pstore, err := pstoreProvider.(peerstore_provider.PeerstoreProvider).GetStakedPeerstoreAtHeight(height)
+	pstore, err := pstoreProvider.GetStakedPeerstoreAtHeight(height)
 	if err != nil {
 		return nil, fmt.Errorf("retrieving peerstore at height %d", height)
 	}

app/client/cli/peer/list.go

@@ -0,0 +1,101 @@
+package peer
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	"github.com/pokt-network/pocket/app/client/cli/helpers"
+	"github.com/pokt-network/pocket/logger"
+	"github.com/pokt-network/pocket/p2p/debug"
+	"github.com/pokt-network/pocket/shared/messaging"
+)
+
+var (
+	listCmd = &cobra.Command{
+		Use:     "List",
+		Short:   "List the known peers",
+		Long:    "Prints a table of the Peer ID, Pokt Address and Service URL of the known peers",
+		Aliases: []string{"list", "ls"},
+		RunE:    listRunE,
+	}
+
+	ErrRouterType = fmt.Errorf("must specify one of --staked, --unstaked, or --all")
+)
+
+func init() {
+	PeerCmd.AddCommand(listCmd)
+}
+
+func listRunE(cmd *cobra.Command, _ []string) error {
+	var routerType debug.RouterType
+
+	bus, err := helpers.GetBusFromCmd(cmd)
+	if err != nil {
+		return err
+	}
+
+	switch {
+	case stakedFlag && !unstakedFlag && !allFlag:
+		routerType = debug.StakedRouterType
+	case unstakedFlag && !stakedFlag && !allFlag:
+		routerType = debug.UnstakedRouterType
+	case stakedFlag || unstakedFlag:
+		return ErrRouterType
+	// even if `allFlag` is false, we still want to print all connections
+	default:
+		routerType = debug.AllRouterTypes
+	}
+
+	debugMsg := &messaging.DebugMessage{
+		Action: messaging.DebugMessageAction_DEBUG_P2P_PRINT_PEER_LIST,
+		Type:   messaging.DebugMessageRoutingType_DEBUG_MESSAGE_TYPE_BROADCAST,
+		Message: &anypb.Any{
+			Value: []byte(routerType),
+		},
+	}
+	debugMsgAny, err := anypb.New(debugMsg)
+	if err != nil {
+		return fmt.Errorf("error creating anypb from debug message: %w", err)
+	}
+
+	if localFlag {
+		if err := debug.PrintPeerList(bus, routerType); err != nil {
+			return fmt.Errorf("error printing peer list: %w", err)
+		}
+		return nil
+	}
+
+	// TECHDEBT(#811): will need to wait for DHT bootstrapping to complete before
+	// p2p broadcast can be used with to reach unstaked actors.
+	// CONSIDERATION: add the peer commands to the interactive CLI as the P2P module
+	// instance could persist between commands. Other interactive CLI commands which
+	// rely on unstaked actor router broadcast are working as expected.
+
+	// TECHDEBT(#811): use broadcast instead to reach all peers.
+	return sendToStakedPeers(cmd, debugMsgAny)
+}
+
+func sendToStakedPeers(cmd *cobra.Command, debugMsgAny *anypb.Any) error {
+	bus, err := helpers.GetBusFromCmd(cmd)
+	if err != nil {
+		return err
+	}
+
+	pstore, err := helpers.FetchPeerstore(cmd)
+	if err != nil {
+		logger.Global.Fatal().Err(err).Msg("unable to retrieve the pstore")
+	}
+
+	if pstore.Size() == 0 {
+		logger.Global.Fatal().Msg("no validators found")
+	}
+
+	for _, peer := range pstore.GetPeerList() {
+		if err := bus.GetP2PModule().Send(peer.GetAddress(), debugMsgAny); err != nil {
+			logger.Global.Error().Err(err).Msg("failed to send debug message")
+		}
+	}
+	return nil
+}

app/client/cli/peer/peer.go

@@ -0,0 +1,27 @@
+package peer
+
+import (
+	"github.com/spf13/cobra"
+
+	"github.com/pokt-network/pocket/app/client/cli/helpers"
+)
+
+var (
+	allFlag,
+	stakedFlag,
+	unstakedFlag,
+	localFlag bool
+
+	PeerCmd = &cobra.Command{
+		Use:               "peer",
+		Short:             "Manage peers",
+		PersistentPreRunE: helpers.P2PDependenciesPreRunE,
+	}
+)
+
+func init() {
+	PeerCmd.PersistentFlags().BoolVarP(&allFlag, "all", "a", false, "operations apply to both staked & unstaked router peerstores (default)")
+	PeerCmd.PersistentFlags().BoolVarP(&stakedFlag, "staked", "s", false, "operations only apply to staked router peerstore (i.e. raintree)")
+	PeerCmd.PersistentFlags().BoolVarP(&unstakedFlag, "unstaked", "u", false, "operations only apply to unstaked (including staked as a subset) router peerstore (i.e. gossipsub)")
+	PeerCmd.PersistentFlags().BoolVarP(&localFlag, "local", "l", false, "operations apply to the local (CLI binary's) P2P module instead of being broadcast")
+}

app/client/cli/utils.go

@@ -311,7 +311,7 @@ func attachKeybaseFlagsToSubcommands() []cmdOption {
 }
 
 func keybaseForCLI() (keybase.Keybase, error) {
-	return keybase.NewKeybase(cfg.Keybase)
+	return keybase.NewKeybase(flags.Cfg.Keybase)
 }
 
 func unableToConnectToRpc(err error) error {

charts/pocket/README.md

@@ -46,6 +46,7 @@ privateKeySecretKeyRef:
 | config.fisherman.enabled | bool | `false` |  |
 | config.logger.format | string | `"json"` |  |
 | config.logger.level | string | `"debug"` |  |
+| config.p2p.enable_peer_discovery_debug_rpc | bool | `false` |  |
 | config.p2p.hostname | string | `""` |  |
 | config.p2p.is_empty_connection_type | bool | `false` |  |
 | config.p2p.max_mempool_count | int | `100000` |  |

charts/pocket/templates/statefulset.yaml

@@ -100,6 +100,8 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: status.podIP
+            - name: POCKET_P2P_ENABLE_PEER_DISCOVERY_DEBUG_RPC
+              value: "true"
           livenessProbe:
             httpGet:
               path: /v1/health

charts/pocket/values.yaml

@@ -101,6 +101,7 @@ config:
     is_empty_connection_type: false
     private_key: "" # @ignored This value is needed but ignored - use privateKeySecretKeyRef instead
     max_mempool_count: 100000
+    enable_peer_discovery_debug_rpc: false
   telemetry:
     enabled: true
     address: 0.0.0.0:9000

p2p/background/router.go

@@ -17,7 +17,6 @@ import (
 	"github.com/pokt-network/pocket/logger"
 	"github.com/pokt-network/pocket/p2p/config"
 	"github.com/pokt-network/pocket/p2p/protocol"
-	"github.com/pokt-network/pocket/p2p/providers"
 	"github.com/pokt-network/pocket/p2p/providers/peerstore_provider"
 	typesP2P "github.com/pokt-network/pocket/p2p/types"
 	"github.com/pokt-network/pocket/p2p/unicast"
@@ -36,8 +35,8 @@ var (
 // TECHDEBT: Make these values configurable
 // TECHDEBT: Consider using an exponential backoff instead
 const (
-	connectMaxRetries   = 5
-	connectRetryTimeout = time.Second * 2
+	connectMaxRetries   = 7
+	connectRetryTimeout = time.Second * 3
 )
 
 type backgroundRouterFactory = modules.FactoryWithConfig[typesP2P.Router, *config.BackgroundConfig]
@@ -73,7 +72,7 @@ type backgroundRouter struct {
 	subscription *pubsub.Subscription
 	// kadDHT is a kademlia distributed hash table used for routing and peer discovery.
 	kadDHT *dht.IpfsDHT
-	// TECHDEBT: `pstore` will likely be removed in future refactoring / simplification
+	// TECHDEBT(#747, #749): `pstore` will likely be removed in future refactoring / simplification
 	// of the `Router` interface.
 	// pstore is the background router's peerstore. Assigned in `backgroundRouter#setupPeerstore()`.
 	pstore typesP2P.Peerstore
@@ -258,18 +257,11 @@ func (rtr *backgroundRouter) setupDependencies(ctx context.Context, _ *config.Ba
 }
 
 func (rtr *backgroundRouter) setupPeerstore(ctx context.Context) (err error) {
-	rtr.logger.Warn().Msg("setting up peerstore...")
-
-	// TECHDEBT(#810, #811): use `bus.GetPeerstoreProvider()` after peerstore provider
+	// TECHDEBT(#811): use `bus.GetPeerstoreProvider()` after peerstore provider
 	// is retrievable as a proper submodule
-	pstoreProviderModule, err := rtr.GetBus().GetModulesRegistry().
-		GetModule(peerstore_provider.PeerstoreProviderSubmoduleName)
+	pstoreProvider, err := peerstore_provider.GetPeerstoreProvider(rtr.GetBus())
 	if err != nil {
-		return fmt.Errorf("retrieving peerstore provider: %w", err)
-	}
-	pstoreProvider, ok := pstoreProviderModule.(providers.PeerstoreProvider)
-	if !ok {
-		return fmt.Errorf("unexpected peerstore provider type: %T", pstoreProviderModule)
+		return err
 	}
 
 	rtr.logger.Debug().Msg("setupCurrentHeightProvider")
@@ -284,10 +276,7 @@ func (rtr *backgroundRouter) setupPeerstore(ctx context.Context) (err error) {
 	}
 
 	// TECHDEBT(#859): integrate with `p2pModule#bootstrap()`.
-	if err := rtr.bootstrap(ctx); err != nil {
-		return fmt.Errorf("bootstrapping peerstore: %w", err)
-	}
-
+	rtr.bootstrap(ctx)
 	return nil
 }
 
@@ -343,33 +332,38 @@ func (rtr *backgroundRouter) setupSubscription() (err error) {
 }
 
 // TECHDEBT(#859): integrate with `p2pModule#bootstrap()`.
-func (rtr *backgroundRouter) bootstrap(ctx context.Context) error {
+func (rtr *backgroundRouter) bootstrap(ctx context.Context) {
 	// CONSIDERATION: add `GetPeers` method, which returns a map,
 	// to the `PeerstoreProvider` interface to simplify this loop.
-	for _, peer := range rtr.pstore.GetPeerList() {
+	peerList := rtr.pstore.GetPeerList()
+	for _, peer := range peerList {
 		if err := utils.AddPeerToLibp2pHost(rtr.host, peer); err != nil {
-			return err
+			rtr.logger.Error().Err(err).Msg("error adding peer to libp2p host")
+			continue
 		}
 
 		libp2pAddrInfo, err := utils.Libp2pAddrInfoFromPeer(peer)
 		if err != nil {
-			return fmt.Errorf(
-				"converting peer info, pokt address: %s: %w",
-				peer.GetAddress(),
-				err,
-			)
+			rtr.logger.Error().Err(err).Msg("error converting peer info")
+			continue
 		}
 
 		// don't attempt to connect to self
 		if rtr.host.ID() == libp2pAddrInfo.ID {
-			return nil
+			rtr.logger.Debug().Msg("not bootstrapping against self")
+			continue
 		}
 
+		rtr.logger.Debug().Fields(map[string]any{
+			"peer_id":   libp2pAddrInfo.ID.String(),
+			"peer_addr": libp2pAddrInfo.Addrs[0].String(),
+			"num_peers": len(peerList) - 1, // -1 as includes self
+		}).Msg("connecting to peer")
 		if err := rtr.connectWithRetry(ctx, libp2pAddrInfo); err != nil {
-			return fmt.Errorf("connecting to peer: %w", err)
+			rtr.logger.Error().Err(err).Msg("error connecting to bootstrap peer")
+			continue
 		}
 	}
-	return nil
 }
 
 // connectWithRetry attempts to connect to the given peer, retrying up to connectMaxRetries times
@@ -382,11 +376,11 @@ func (rtr *backgroundRouter) connectWithRetry(ctx context.Context, libp2pAddrInf
 			return nil
 		}
 
-		fmt.Printf("Failed to connect (attempt %d), retrying in %v...\n", i+1, connectRetryTimeout)
+		rtr.logger.Error().Msgf("failed to connect (attempt %d), retrying in %v...", i+1, connectRetryTimeout)
 		time.Sleep(connectRetryTimeout)
 	}
 
-	return fmt.Errorf("failed to connect after %d attempts, last error: %w", 5, err)
+	return fmt.Errorf("failed to connect after %d attempts, last error: %w", connectMaxRetries, err)
 }
 
 // topicValidator is used in conjunction with libp2p-pubsub's notion of "topic
@@ -430,7 +424,6 @@ func (rtr *backgroundRouter) readSubscription(ctx context.Context) {
 			return
 		}
 		msg, err := rtr.subscription.Next(ctx)
-
 		if err != nil {
 			rtr.logger.Error().Err(err).
 				Msg("error reading from background topic subscription")