Skip to content
This repository has been archived by the owner on Nov 15, 2023. It is now read-only.

Add a JSON-RPC request for reserved nodes #8704

Merged
6 commits merged into from
May 3, 2021
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions client/network/src/protocol.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1014,6 +1014,12 @@ impl<B: BlockT> Protocol<B> {
self.peerset_handle.remove_reserved_peer(HARDCODED_PEERSETS_SYNC, peer.clone());
}

/// Returns the list of reserved peers.
pub fn reserved_peers(&self, pending_response: oneshot::Sender<Vec<PeerId>>) {
self.peerset_handle
.get_reserved_peers(HARDCODED_PEERSETS_SYNC, pending_response)
}

/// Adds a `PeerId` to the list of reserved peers for syncing purposes.
pub fn add_reserved_peer(&self, peer: PeerId) {
self.peerset_handle.add_reserved_peer(HARDCODED_PEERSETS_SYNC, peer.clone());
Expand Down
5 changes: 5 additions & 0 deletions client/network/src/service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -619,6 +619,11 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
pub fn add_reserved_peer(&self, peer: String) -> Result<(), String> {
self.service.add_reserved_peer(peer)
}

/// Returns the list of reserved peers.
pub fn reserved_peers(&self, pending_response: oneshot::Sender<Vec<PeerId>>) {
self.network_service.behaviour().user_protocol().reserved_peers(pending_response)
}
}

impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
Expand Down
15 changes: 14 additions & 1 deletion client/peerset/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@
mod peersstate;

use std::{collections::HashSet, collections::VecDeque};
use futures::prelude::*;
use futures::{channel::oneshot, prelude::*};
use log::{debug, error, trace};
use serde_json::json;
use std::{collections::HashMap, pin::Pin, task::{Context, Poll}, time::Duration};
Expand All @@ -58,6 +58,7 @@ enum Action {
RemoveReservedPeer(SetId, PeerId),
SetReservedPeers(SetId, HashSet<PeerId>),
SetReservedOnly(SetId, bool),
GetReservedPeers(SetId, oneshot::Sender<Vec<PeerId>>),
ReportPeer(PeerId, ReputationChange),
AddToPeersSet(SetId, PeerId),
RemoveFromPeersSet(SetId, PeerId),
Expand Down Expand Up @@ -147,6 +148,11 @@ impl PeersetHandle {
let _ = self.tx.unbounded_send(Action::SetReservedPeers(set_id, peer_ids));
}

/// Get the list of reserved peers for the given set.
pub fn get_reserved_peers(&self, set_id: SetId, pending_response: oneshot::Sender<Vec<PeerId>>) {
let _ = self.tx.unbounded_send(Action::GetReservedPeers(set_id, pending_response));
}

/// Reports an adjustment to the reputation of the given peer.
pub fn report_peer(&self, peer_id: PeerId, score_diff: ReputationChange) {
let _ = self.tx.unbounded_send(Action::ReportPeer(peer_id, score_diff));
Expand Down Expand Up @@ -379,6 +385,11 @@ impl Peerset {
}
}

fn on_get_reserved_peers(&mut self, set_id: SetId, pending_response: oneshot::Sender<Vec<PeerId>>) {
let peers = self.reserved_nodes[set_id.0].0.iter().cloned().collect::<Vec<_>>();
let _ = pending_response.send(peers);
}

/// Adds a node to the given set. The peerset will, if possible and not already the case,
/// try to connect to it.
///
Expand Down Expand Up @@ -709,6 +720,8 @@ impl Stream for Peerset {
self.on_set_reserved_peers(set_id, peer_ids),
Action::SetReservedOnly(set_id, reserved) =>
self.on_set_reserved_only(set_id, reserved),
Action::GetReservedPeers(set_id, pending_response) =>
self.on_get_reserved_peers(set_id, pending_response),
Action::ReportPeer(peer_id, score_diff) =>
self.on_report_peer(peer_id, score_diff),
Action::AddToPeersSet(sets_name, peer_id) =>
Expand Down
4 changes: 4 additions & 0 deletions client/rpc-api/src/system/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,10 @@ pub trait SystemApi<Hash, Number> {
fn system_remove_reserved_peer(&self, peer_id: String)
-> Compat<BoxFuture<'static, Result<(), jsonrpc_core::Error>>>;

/// Returns the list of reserved peers
#[rpc(name = "system_reservedPeers", returns = "Vec<String>")]
fn system_reserved_peers(&self) -> Receiver<Vec<String>>;

/// Returns the roles the node is running as.
#[rpc(name = "system_nodeRoles", returns = "Vec<NodeRole>")]
fn system_node_roles(&self) -> Receiver<Vec<NodeRole>>;
Expand Down
8 changes: 8 additions & 0 deletions client/rpc/src/system/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,8 @@ pub enum Request<B: traits::Block> {
NetworkAddReservedPeer(String, oneshot::Sender<Result<()>>),
/// Must return any potential parse error.
NetworkRemoveReservedPeer(String, oneshot::Sender<Result<()>>),
/// Must return the list of reserved peers
NetworkReservedPeers(oneshot::Sender<Vec<String>>),
/// Must return the node role.
NodeRoles(oneshot::Sender<Vec<NodeRole>>),
/// Must return the state of the node syncing.
Expand Down Expand Up @@ -187,6 +189,12 @@ impl<B: traits::Block> SystemApi<B::Hash, <B::Header as HeaderT>::Number> for Sy
}.boxed().compat()
}

fn system_reserved_peers(&self) -> Receiver<Vec<String>> {
let (tx, rx) = oneshot::channel();
let _ = self.send_back.unbounded_send(Request::NetworkReservedPeers(tx));
Receiver(Compat::new(rx))
}

fn system_node_roles(&self) -> Receiver<Vec<NodeRole>> {
let (tx, rx) = oneshot::channel();
let _ = self.send_back.unbounded_send(Request::NodeRoles(tx));
Expand Down
11 changes: 11 additions & 0 deletions client/rpc/src/system/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,9 @@ fn api<T: Into<Option<Status>>>(sync: T) -> System<Block> {
Err(s) => sender.send(Err(error::Error::MalformattedPeerArg(s.to_string()))),
};
}
Request::NetworkReservedPeers(sender) => {
let _ = sender.send(vec!["peer_A".to_string()]);
kpp marked this conversation as resolved.
Show resolved Hide resolved
}
Request::NodeRoles(sender) => {
let _ = sender.send(vec![NodeRole::Authority]);
}
Expand Down Expand Up @@ -337,6 +340,14 @@ fn system_network_remove_reserved() {
assert!(runtime.block_on(bad_fut).is_err());
}

#[test]
fn system_network_reserved_peers() {
assert_eq!(
wait_receiver(api(None).system_reserved_peers()),
vec!["peer_A".to_string()]
);
}

#[test]
fn test_add_reset_log_filter() {
const EXPECTED_BEFORE_ADD: &'static str = "EXPECTED_BEFORE_ADD";
Expand Down
11 changes: 10 additions & 1 deletion client/service/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ use std::collections::HashMap;
use std::time::Duration;
use std::task::Poll;

use futures::{Future, FutureExt, Stream, StreamExt, stream, compat::*};
use futures::{Future, FutureExt, Stream, StreamExt, stream, compat::*, channel::oneshot};
use sc_network::{NetworkStatus, network_state::NetworkState, PeerId};
use log::{warn, debug, error};
use codec::{Encode, Decode};
Expand Down Expand Up @@ -302,6 +302,15 @@ async fn build_network_future<
))),
};
}
sc_rpc::system::Request::NetworkReservedPeers(sender) => {
let (tx, rx) = oneshot::channel();
let _ = network.reserved_peers(tx);
let reserved_peers = rx.await.unwrap();
let reserved_peers = reserved_peers.into_iter()
kpp marked this conversation as resolved.
Show resolved Hide resolved
.map(|peer_id| peer_id.to_base58())
.collect();
let _ = sender.send(reserved_peers);
}
sc_rpc::system::Request::NodeRoles(sender) => {
use sc_rpc::system::NodeRole;

Expand Down