feat(rpc): impl Filecoin.NetProtectAdd and NetProtectRemove (#4743)

scripts/tests/api_compare/filter-list-offline

@@ -9,6 +9,9 @@
 !Filecoin.NetAutoNatStatus
 !Filecoin.NetPeers
 !Filecoin.NetFindPeer
+!Filecoin.NetProtectAdd
+!Filecoin.NetProtectRemove
+!Filecoin.NetProtectList
 # CustomCheckFailed in Forest: https://github.com/ChainSafe/forest/actions/runs/9593268587/job/26453560366
 !Filecoin.StateReplay
 # CustomCheckFailed in Forest: https://github.com/ChainSafe/forest/actions/runs/9593268587/job/26453560366

src/libp2p/peer_manager.rs

@@ -349,7 +349,6 @@ impl PeerManager {
         self.protected_peers.write().insert(peer_id);
     }
 
-    #[allow(dead_code)]
     pub fn unprotect_peer(&self, peer_id: &PeerId) {
         self.protected_peers.write().remove(peer_id);
     }

src/libp2p/service.rs

@@ -152,7 +152,8 @@ pub enum NetRPCMethods {
     AddrsListen(flume::Sender<(PeerId, HashSet<Multiaddr>)>),
     Peer(flume::Sender<Option<HashSet<Multiaddr>>>, PeerId),
     Peers(flume::Sender<HashMap<PeerId, HashSet<Multiaddr>>>),
-    ProtectPeer(flume::Sender<()>, PeerId),
+    ProtectPeer(flume::Sender<()>, HashSet<PeerId>),
+    UnprotectPeer(flume::Sender<()>, HashSet<PeerId>),
     Info(flume::Sender<NetInfoResult>),
     Connect(flume::Sender<bool>, PeerId, HashSet<Multiaddr>),
     Disconnect(flume::Sender<()>, PeerId),
@@ -499,8 +500,16 @@ async fn handle_network_message(
                     let peer_addresses = swarm.behaviour().peer_addresses();
                     response_channel.send_or_warn(peer_addresses);
                 }
-                NetRPCMethods::ProtectPeer(tx, peer_id) => {
-                    peer_manager.protect_peer(peer_id);
+                NetRPCMethods::ProtectPeer(tx, peer_ids) => {
+                    peer_ids.into_iter().for_each(|peer_id| {
+                        peer_manager.protect_peer(peer_id);
+                    });
+                    tx.send_or_warn(());
+                }
+                NetRPCMethods::UnprotectPeer(tx, peer_ids) => {
+                    peer_ids.iter().for_each(|peer_id| {
+                        peer_manager.unprotect_peer(peer_id);
+                    });
                     tx.send_or_warn(());
                 }
                 NetRPCMethods::Info(response_channel) => {

src/rpc/methods/f3.rs

@@ -416,7 +416,7 @@ impl RpcMethod<1> for ProtectPeer {
         let (tx, rx) = flume::bounded(1);
         ctx.network_send
             .send_async(NetworkMessage::JSONRPCRequest {
-                method: NetRPCMethods::ProtectPeer(tx, peer_id),
+                method: NetRPCMethods::ProtectPeer(tx, std::iter::once(peer_id).collect()),
             })
             .await?;
         rx.recv_async().await?;

src/rpc/methods/net.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0, MIT
 
 mod types;
+use itertools::Itertools;
 pub use types::*;
 
 use std::any::Any;
@@ -254,21 +255,33 @@ impl RpcMethod<0> for NetVersion {
 pub enum NetProtectAdd {}
 impl RpcMethod<1> for NetProtectAdd {
     const NAME: &'static str = "Filecoin.NetProtectAdd";
-    const PARAM_NAMES: [&'static str; 1] = ["peer_id"];
+    const PARAM_NAMES: [&'static str; 1] = ["peer_ids"];
     const API_PATHS: ApiPaths = ApiPaths::V0;
     const PERMISSION: Permission = Permission::Admin;
 
-    type Params = (String,);
+    type Params = (Vec<String>,);
     type Ok = ();
 
-    // This is a no-op due to the fact that `rust-libp2p` implementation is very different to that
+    // This whitelists a peer in forest peer manager but has no impact on libp2p swarm
+    // due to the fact that `rust-libp2p` implementation is very different to that
     // in go. However it would be nice to investigate connection limiting options in Rust.
     // See: <https://github.com/ChainSafe/forest/issues/4355>.
     async fn handle(
-        _: Ctx<impl Blockstore>,
-        (peer_id,): Self::Params,
+        ctx: Ctx<impl Blockstore>,
+        (peer_ids,): Self::Params,
     ) -> Result<Self::Ok, ServerError> {
-        let _ = PeerId::from_str(&peer_id)?;
+        let peer_ids = peer_ids
+            .iter()
+            .map(String::as_str)
+            .map(PeerId::from_str)
+            .try_collect()?;
+        let (tx, rx) = flume::bounded(1);
+        ctx.network_send
+            .send_async(NetworkMessage::JSONRPCRequest {
+                method: NetRPCMethods::ProtectPeer(tx, peer_ids),
+            })
+            .await?;
+        rx.recv_async().await?;
         Ok(())
     }
 }
@@ -285,16 +298,34 @@ impl RpcMethod<0> for NetProtectList {
         Err(ServerError::stubbed_for_openrpc())
     }
 }
+
 pub enum NetProtectRemove {}
 impl RpcMethod<1> for NetProtectRemove {
     const NAME: &'static str = "Filecoin.NetProtectRemove";
-    const PARAM_NAMES: [&'static str; 1] = ["peer_id"];
+    const PARAM_NAMES: [&'static str; 1] = ["peer_ids"];
     const API_PATHS: ApiPaths = ApiPaths::Both;
     const PERMISSION: Permission = Permission::Admin;
 
-    type Params = (String,);
+    type Params = (Vec<String>,);
     type Ok = ();
-    async fn handle(_: Ctx<impl Blockstore>, (_,): Self::Params) -> Result<Self::Ok, ServerError> {
-        Err(ServerError::stubbed_for_openrpc())
+
+    // Similar to NetProtectAdd
+    async fn handle(
+        ctx: Ctx<impl Blockstore>,
+        (peer_ids,): Self::Params,
+    ) -> Result<Self::Ok, ServerError> {
+        let peer_ids = peer_ids
+            .iter()
+            .map(String::as_str)
+            .map(PeerId::from_str)
+            .try_collect()?;
+        let (tx, rx) = flume::bounded(1);
+        ctx.network_send
+            .send_async(NetworkMessage::JSONRPCRequest {
+                method: NetRPCMethods::UnprotectPeer(tx, peer_ids),
+            })
+            .await?;
+        rx.recv_async().await?;
+        Ok(())
     }
 }

src/tool/subcommands/api_cmd.rs

@@ -40,6 +40,7 @@ use futures::{stream::FuturesUnordered, StreamExt};
 use fvm_ipld_blockstore::Blockstore;
 use itertools::Itertools as _;
 use jsonrpsee::types::ErrorCode;
+use libp2p::PeerId;
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
@@ -656,24 +657,26 @@ fn mpool_tests_with_tipset(tipset: &Tipset) -> Vec<RpcTest> {
 }
 
 fn net_tests() -> Vec<RpcTest> {
-    // Tests with a known peer id tend to be flaky, use a random peer id to test the unhappy path only
-    let random_peer_id = libp2p::PeerId::random().to_string();
-
     // More net commands should be tested. Tracking issue:
     // https://github.com/ChainSafe/forest/issues/3639
     vec![
         RpcTest::basic(NetAddrsListen::request(()).unwrap()),
         RpcTest::basic(NetPeers::request(()).unwrap()),
         RpcTest::identity(NetListening::request(()).unwrap()),
-        RpcTest::basic(NetAgentVersion::request((random_peer_id.clone(),)).unwrap())
+        // Tests with a known peer id tend to be flaky, use a random peer id to test the unhappy path only
+        RpcTest::basic(NetAgentVersion::request((PeerId::random().to_string(),)).unwrap())
             .policy_on_rejected(PolicyOnRejected::PassWithIdenticalError),
-        RpcTest::basic(NetFindPeer::request((random_peer_id,)).unwrap())
+        RpcTest::basic(NetFindPeer::request((PeerId::random().to_string(),)).unwrap())
             .policy_on_rejected(PolicyOnRejected::Pass)
             .ignore("It times out in lotus when peer not found"),
         RpcTest::basic(NetInfo::request(()).unwrap())
             .ignore("Not implemented in Lotus. Why do we even have this method?"),
         RpcTest::basic(NetAutoNatStatus::request(()).unwrap()),
         RpcTest::identity(NetVersion::request(()).unwrap()),
+        RpcTest::identity(NetProtectAdd::request((vec![PeerId::random().to_string()],)).unwrap()),
+        RpcTest::identity(
+            NetProtectRemove::request((vec![PeerId::random().to_string()],)).unwrap(),
+        ),
     ]
 }