feature: 'ipfs swarm peering' command (#8147)

* feat: added swarm peering command supporting add, ls and rm

Co-authored-by: Steven Allen <[email protected]>

core/commands/commands_test.go

@@ -237,6 +237,10 @@ func TestCommands(t *testing.T) {
 		"/swarm/filters/add",
 		"/swarm/filters/rm",
 		"/swarm/peers",
+		"/swarm/peering",
+		"/swarm/peering/add",
+		"/swarm/peering/ls",
+		"/swarm/peering/rm",
 		"/tar",
 		"/tar/add",
 		"/tar/cat",

core/commands/swarm.go

@@ -51,6 +51,7 @@ ipfs peers in the internet.
 		"disconnect": swarmDisconnectCmd,
 		"filters":    swarmFiltersCmd,
 		"peers":      swarmPeersCmd,
+		"peering":    swarmPeeringCmd,
 	},
 }
 
@@ -61,6 +62,149 @@ const (
 	swarmDirectionOptionName = "direction"
 )
 
+type peeringResult struct {
+	ID     peer.ID
+	Status string
+}
+
+var swarmPeeringCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Modify the peering subsystem.",
+		ShortDescription: `
+'ipfs swarm peering' manages the peering subsystem. 
+Peers in the peering subsystem is maintained to be connected, reconnected 
+on disconnect with a back-off.
+The changes are not saved to the config.
+`,
+	},
+	Subcommands: map[string]*cmds.Command{
+		"add": swarmPeeringAddCmd,
+		"ls":  swarmPeeringLsCmd,
+		"rm":  swarmPeeringRmCmd,
+	},
+}
+
+var swarmPeeringAddCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Add peers into the peering subsystem.",
+		ShortDescription: `
+'ipfs swarm peering add' will add the new address to the peering subsystem as one that should always be connected to.
+`,
+	},
+	Arguments: []cmds.Argument{
+		cmds.StringArg("address", true, true, "address of peer to add into the peering subsystem"),
+	},
+	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
+		addrs := make([]ma.Multiaddr, len(req.Arguments))
+
+		for i, arg := range req.Arguments {
+			addr, err := ma.NewMultiaddr(arg)
+			if err != nil {
+				return err
+			}
+
+			addrs[i] = addr
+		}
+
+		addInfos, err := peer.AddrInfosFromP2pAddrs(addrs...)
+		if err != nil {
+			return err
+		}
+
+		node, err := cmdenv.GetNode(env)
+		if err != nil {
+			return err
+		}
+
+		for _, addrinfo := range addInfos {
+			node.Peering.AddPeer(addrinfo)
+			err = res.Emit(peeringResult{addrinfo.ID, "success"})
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	},
+	Encoders: cmds.EncoderMap{
+		cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, pr *peeringResult) error {
+			fmt.Fprintf(w, "add %s %s\n", pr.ID.String(), pr.Status)
+			return nil
+		}),
+	},
+	Type: peeringResult{},
+}
+
+var swarmPeeringLsCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "List peers registered in the peering subsystem.",
+		ShortDescription: `
+'ipfs swarm peering ls' lists the peers that are registered in the peering subsystem and to which the daemon is always connected.
+`,
+	},
+	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
+		node, err := cmdenv.GetNode(env)
+		if err != nil {
+			return err
+		}
+		peers := node.Peering.ListPeers()
+		return cmds.EmitOnce(res, addrInfos{Peers: peers})
+	},
+	Type: addrInfos{},
+	Encoders: cmds.EncoderMap{
+		cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, ai *addrInfos) error {
+			for _, info := range ai.Peers {
+				fmt.Fprintf(w, "%s\n", info.ID)
+				for _, addr := range info.Addrs {
+					fmt.Fprintf(w, "\t%s\n", addr)
+				}
+			}
+			return nil
+		}),
+	},
+}
+
+type addrInfos struct {
+	Peers []peer.AddrInfo
+}
+
+var swarmPeeringRmCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Remove a peer from the peering subsystem.",
+		ShortDescription: `
+'ipfs swarm peering rm' will remove the given ID from the peering subsystem and remove it from the always-on connection.
+`,
+	},
+	Arguments: []cmds.Argument{
+		cmds.StringArg("ID", true, true, "ID of peer to remove from the peering subsystem"),
+	},
+	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
+		node, err := cmdenv.GetNode(env)
+		if err != nil {
+			return err
+		}
+
+		for _, arg := range req.Arguments {
+			id, err := peer.Decode(arg)
+			if err != nil {
+				return err
+			}
+
+			node.Peering.RemovePeer(id)
+			if err = res.Emit(peeringResult{id, "success"}); err != nil {
+				return err
+			}
+		}
+		return nil
+	},
+	Type: peeringResult{},
+	Encoders: cmds.EncoderMap{
+		cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, pr *peeringResult) error {
+			fmt.Fprintf(w, "add %s %s\n", pr.ID.String(), pr.Status)
+			return nil
+		}),
+	},
+}
+
 var swarmPeersCmd = &cmds.Command{
 	Helptext: cmds.HelpText{
 		Tagline: "List peers with open connections.",

peering/peering.go

@@ -242,6 +242,17 @@ func (ps *PeeringService) AddPeer(info peer.AddrInfo) {
 	}
 }
 
+// ListPeers lists peers in the peering service.
+func (ps *PeeringService) ListPeers() []peer.AddrInfo {
+	out := make([]peer.AddrInfo, 0, len(ps.peers))
+	for id, addrs := range ps.peers {
+		ai := peer.AddrInfo{ID: id}
+		ai.Addrs = append(ai.Addrs, addrs.addrs...)
+		out = append(out, ai)
+	}
+	return out
+}
+
 // RemovePeer removes a peer from the peering service. This function may be
 // safely called at any time: before the service is started, while running, or
 // after it stops.

peering/peering_test.go

@@ -39,6 +39,7 @@ func TestPeeringService(t *testing.T) {
 
 	// peer 1 -> 2
 	ps1.AddPeer(peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
+	require.Contains(t, ps1.ListPeers(), peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
 
 	// We haven't started so we shouldn't have any peers.
 	require.Never(t, func() bool {
@@ -109,6 +110,7 @@ func TestPeeringService(t *testing.T) {
 
 	// Unprotect 2 from 1.
 	ps1.RemovePeer(h2.ID())
+	require.NotContains(t, ps1.ListPeers(), peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
 
 	// Trim connections.
 	h1.ConnManager().TrimOpenConns(ctx)
@@ -127,7 +129,9 @@ func TestPeeringService(t *testing.T) {
 
 	// Until added back
 	ps1.AddPeer(peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
+	require.Contains(t, ps1.ListPeers(), peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
 	ps1.AddPeer(peer.AddrInfo{ID: h3.ID(), Addrs: h3.Addrs()})
+	require.Contains(t, ps1.ListPeers(), peer.AddrInfo{ID: h3.ID(), Addrs: h3.Addrs()})
 	t.Logf("wait for h1 to connect to h2 and h3 again")
 	require.Eventually(t, func() bool {
 		return h1.Network().Connectedness(h2.ID()) == network.Connected
@@ -142,7 +146,9 @@ func TestPeeringService(t *testing.T) {
 
 	// Adding and removing should work after stopping.
 	ps1.AddPeer(peer.AddrInfo{ID: h4.ID(), Addrs: h4.Addrs()})
+	require.Contains(t, ps1.ListPeers(), peer.AddrInfo{ID: h4.ID(), Addrs: h4.Addrs()})
 	ps1.RemovePeer(h2.ID())
+	require.NotContains(t, ps1.ListPeers(), peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
 }
 
 func TestNextBackoff(t *testing.T) {

test/sharness/t0140-swarm.sh

@@ -97,6 +97,40 @@ test_expect_success "Addresses.NoAnnounce with /ipcidr affects addresses" '
 
 test_kill_ipfs_daemon
 
+test_launch_ipfs_daemon
+
+test_expect_success "'ipfs swarm peering ls' lists peerings" '
+  ipfs swarm peering ls
+'
+
+peeringID='QmYyQSo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjhx5N'
+peeringID2='QmYyQSo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjhx5K'
+peeringAddr='/ip4/1.2.3.4/tcp/1234/p2p/QmYyQSo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjhx5N'
+peeringAddr2='/ip4/1.2.3.4/tcp/1234/p2p/QmYyQSo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjhx5K'
+test_expect_success "'ipfs swarm peering add' adds a peering" '
+  ipfs swarm peering ls > peeringls &&
+  ! test_should_contain ${peeringID} peeringls &&
+  ! test_should_contain ${peeringID2} peeringls &&
+  ipfs swarm peering add ${peeringAddr} ${peeringAddr2}
+'
+
+test_expect_success 'a peering is added' '
+  ipfs swarm peering ls > peeringadd &&
+  test_should_contain ${peeringID} peeringadd &&
+  test_should_contain ${peeringID2} peeringadd
+'
+
+test_expect_success "'swarm peering rm' removes a peering" '
+  ipfs swarm peering rm ${peeringID}
+'
+
+test_expect_success 'peering is removed' '
+  ipfs swarm peering ls > peeringrm &&
+  ! test_should_contain ${peeringID} peeringrm
+'
+
+test_kill_ipfs_daemon
+
 test_expect_success "set up tcp testbed" '
   iptb testbed create -type localipfs -count 2 -force -init
 '