Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: Use handshake version 2 #3157

Merged
merged 25 commits into from
Sep 4, 2020
Merged

fix: Use handshake version 2 #3157

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
25 commits
Select commit Hold shift + click to select a range
ed4a565
fix: Use both version and oldest_supported version
mfornet Jul 20, 2020
6ebcf07
Fix failing test
mfornet Aug 13, 2020
8873ddd
fix: Use handshake version 2
mfornet Aug 13, 2020
0947747
Update handshake.py
mfornet Aug 13, 2020
5480df9
Use previous version as default version for incoming handshakes
mfornet Aug 13, 2020
8d71ec8
Use +1 predicted version
mfornet Sep 1, 2020
40bc724
Respond with current version to other nodes
mfornet Sep 2, 2020
fa35a0f
Reaction to ProtocolVersionMismatch and better API
mfornet Sep 4, 2020
b153dc6
Update backward compatible test
mfornet Sep 4, 2020
985ae66
Remove Option
mfornet Sep 4, 2020
002d8f0
Fix codec test
mfornet Sep 4, 2020
77e8d8a
Backward compatible with version bump
mfornet Sep 4, 2020
259dd43
hotfix upgradable tests
bowenwang1996 Sep 4, 2020
e830271
Address bowen comments
mfornet Sep 4, 2020
ea96331
Merge remote-tracking branch 'origin/latest_supported_version_2' into…
mfornet Sep 4, 2020
3886464
test fix
bowenwang1996 Sep 4, 2020
0362242
test another attempt
bowenwang1996 Sep 4, 2020
36f7c7d
compile current
bowenwang1996 Sep 4, 2020
1acef3e
add debug info
bowenwang1996 Sep 4, 2020
9b155e9
try again
bowenwang1996 Sep 4, 2020
fd6e767
remove debug info
bowenwang1996 Sep 4, 2020
51ee7b2
Remove unnecessary version checks
mfornet Sep 4, 2020
465a4ac
Add responses to Handshake with future version
mfornet Sep 4, 2020
99b576d
Fix test
mfornet Sep 4, 2020
f5aafd8
Don't use match
mfornet Sep 4, 2020
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
71 changes: 65 additions & 6 deletions chain/network/src/codec.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,16 +76,19 @@ pub fn bytes_to_peer_message(bytes: &[u8]) -> Result<PeerMessage, std::io::Error

#[cfg(test)]
mod test {
use near_crypto::{KeyType, SecretKey};
use near_crypto::{KeyType, PublicKey, SecretKey};
use near_primitives::block::{Approval, ApprovalInner};
use near_primitives::hash::CryptoHash;
use near_primitives::network::AnnounceAccount;
use near_primitives::types::EpochId;
use near_primitives::network::{AnnounceAccount, PeerId};
use near_primitives::{
types::EpochId,
version::{OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION, PROTOCOL_VERSION},
};

use crate::routing::EdgeInfo;
use crate::types::{
Handshake, PeerChainInfo, PeerIdOrHash, PeerInfo, RoutedMessage, RoutedMessageBody,
SyncData,
Handshake, HandshakeFailureReason, HandshakeV2, PeerChainInfo, PeerIdOrHash, PeerInfo,
RoutedMessage, RoutedMessageBody, SyncData,
};

use super::*;
Expand All @@ -102,7 +105,7 @@ mod test {
fn test_peer_message_handshake() {
let peer_info = PeerInfo::random();
let fake_handshake = Handshake {
version: 1,
version: PROTOCOL_VERSION,
peer_id: peer_info.id.clone(),
target_peer_id: peer_info.id,
listen_port: None,
Expand All @@ -117,6 +120,62 @@ mod test {
test_codec(msg);
}

#[test]
fn test_peer_message_handshake_v2() {
let peer_info = PeerInfo::random();
let fake_handshake = HandshakeV2 {
version: PROTOCOL_VERSION,
oldest_supported_version: OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION,
peer_id: peer_info.id.clone(),
target_peer_id: peer_info.id,
listen_port: None,
chain_info: PeerChainInfo {
genesis_id: Default::default(),
height: 0,
tracked_shards: vec![],
},
edge_info: EdgeInfo::default(),
};
let msg = PeerMessage::HandshakeV2(fake_handshake);
test_codec(msg);
}

#[test]
fn test_peer_message_handshake_v2_00() {
let fake_handshake = HandshakeV2 {
version: 0,
oldest_supported_version: 0,
peer_id: PeerId::new(PublicKey::empty(KeyType::ED25519)),
target_peer_id: PeerId::new(PublicKey::empty(KeyType::ED25519)),
listen_port: None,
chain_info: PeerChainInfo {
genesis_id: Default::default(),
height: 0,
tracked_shards: vec![],
},
edge_info: EdgeInfo::default(),
};
let msg = PeerMessage::HandshakeV2(fake_handshake);

let mut codec = Codec::new();
let mut buffer = BytesMut::new();
codec.encode(peer_message_to_bytes(msg.clone()).unwrap(), &mut buffer).unwrap();
let decoded = codec.decode(&mut buffer).unwrap().unwrap().unwrap();

let err = bytes_to_peer_message(&decoded).unwrap_err();

assert_eq!(
*err.get_ref()
.map(|inner| inner.downcast_ref::<HandshakeFailureReason>())
.unwrap()
.unwrap(),
HandshakeFailureReason::ProtocolVersionMismatch {
version: 0,
oldest_supported_version: 0,
}
);
}

#[test]
fn test_peer_message_info_gossip() {
let peer_info1 = PeerInfo::random();
Expand Down
86 changes: 58 additions & 28 deletions chain/network/src/peer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,17 @@ use near_primitives::hash::CryptoHash;
use near_primitives::network::PeerId;
use near_primitives::unwrap_option_or_return;
use near_primitives::utils::DisplayOption;
use near_primitives::version::{OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION, PROTOCOL_VERSION};
use near_primitives::version::{
ProtocolVersion, NETWORK_PROTOCOL_VERSION, OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION,
};

use crate::codec::{bytes_to_peer_message, peer_message_to_bytes, Codec};
use crate::rate_counter::RateCounter;
#[cfg(feature = "metric_recorder")]
use crate::recorder::{PeerMessageMetadata, Status};
use crate::routing::{Edge, EdgeInfo};
use crate::types::{
Ban, Consolidate, ConsolidateResponse, Handshake, HandshakeFailureReason,
Ban, Consolidate, ConsolidateResponse, HandshakeFailureReason, HandshakeV2,
NetworkClientMessages, NetworkClientResponses, NetworkRequests, NetworkViewClientMessages,
NetworkViewClientResponses, PeerChainInfo, PeerInfo, PeerManagerRequest, PeerMessage,
PeerRequest, PeerResponse, PeerStatsResult, PeerStatus, PeerType, PeersRequest, PeersResponse,
Expand Down Expand Up @@ -139,6 +141,8 @@ pub struct Peer {
pub peer_type: PeerType,
/// Peer status.
pub peer_status: PeerStatus,
/// Protocol version to communicate with this peer.
pub protocol_version: ProtocolVersion,
/// Framed wrapper to send messages through the TCP connection.
framed: FramedWrite<WriteHalf, Codec>,
/// Handshake timeout.
Expand Down Expand Up @@ -183,6 +187,7 @@ impl Peer {
peer_info: peer_info.into(),
peer_type,
peer_status: PeerStatus::Connecting,
protocol_version: NETWORK_PROTOCOL_VERSION,
framed,
handshake_timeout,
peer_manager_addr,
Expand Down Expand Up @@ -269,15 +274,15 @@ impl Peer {
height,
tracked_shards,
}) => {
// TODO(HandshakeV2): Use handshake v2
let handshake = Handshake::new(
let handshake = HandshakeV2::new(
act.protocol_version,
act.node_id(),
act.peer_id().unwrap(),
act.node_info.addr_port(),
PeerChainInfo { genesis_id, height, tracked_shards },
act.edge_info.as_ref().unwrap().clone(),
);
act.send_message(PeerMessage::Handshake(handshake));
act.send_message(PeerMessage::HandshakeV2(handshake));
actix::fut::ready(())
}
Err(err) => {
Expand Down Expand Up @@ -605,10 +610,34 @@ impl StreamHandler<Result<Vec<u8>, ReasonForBan>> for Peer {
let msg_size = msg.len();

self.tracker.increment_received(msg.len() as u64);

let mut peer_msg = match bytes_to_peer_message(&msg) {
Ok(peer_msg) => peer_msg,
Err(err) => {
info!(target: "network", "Received invalid data {:?} from {}: {}", msg, self.peer_info, err);
if let Some(version) = err
.get_ref()
.and_then(|err| err.downcast_ref::<HandshakeFailureReason>())
.and_then(|inner| {
if let HandshakeFailureReason::ProtocolVersionMismatch { version, .. } =
*inner
{
Some(version)
} else {
None
}
})
{
debug!(target: "network", "Received connection from node with unsupported version: {}", version);
self.send_message(PeerMessage::HandshakeFailure(
self.node_info.clone(),
HandshakeFailureReason::ProtocolVersionMismatch {
version: NETWORK_PROTOCOL_VERSION,
oldest_supported_version: OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION,
},
));
} else {
info!(target: "network", "Received invalid data {:?} from {}: {}", msg, self.peer_info, err);
}
return;
}
};
Expand Down Expand Up @@ -638,9 +667,8 @@ impl StreamHandler<Result<Vec<u8>, ReasonForBan>> for Peer {
msg.len() as i64,
);

// TODO(HandshakeV2): Remove this, and deprecate PeerMessage::Handshake
if let PeerMessage::HandshakeV2(handshake) = peer_msg {
peer_msg = PeerMessage::Handshake(handshake.into());
if let PeerMessage::Handshake(handshake) = peer_msg {
peer_msg = PeerMessage::HandshakeV2(handshake.into());
}

match (self.peer_type, self.peer_status, peer_msg) {
Expand All @@ -653,7 +681,21 @@ impl StreamHandler<Result<Vec<u8>, ReasonForBan>> for Peer {
version,
oldest_supported_version,
} => {
warn!(target: "network", "Unable to connect to a node ({}) due to a network protocol version mismatch. Our version: {:?}, their: {:?}", peer_info, (PROTOCOL_VERSION, OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION), (version, oldest_supported_version));
let target_version = std::cmp::min(version, NETWORK_PROTOCOL_VERSION);

if target_version
>= std::cmp::max(
oldest_supported_version,
OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION,
)
{
// Use target_version as protocol_version to talk with this peer
self.protocol_version = target_version;
self.send_handshake(ctx);
return;
} else {
warn!(target: "network", "Unable to connect to a node ({}) due to a network protocol version mismatch. Our version: {:?}, their: {:?}", peer_info, (NETWORK_PROTOCOL_VERSION, OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION), (version, oldest_supported_version));
}
}
HandshakeFailureReason::InvalidTarget => {
debug!(target: "network", "Peer found was not what expected. Updating peer info with {:?}", peer_info);
Expand All @@ -662,9 +704,14 @@ impl StreamHandler<Result<Vec<u8>, ReasonForBan>> for Peer {
}
ctx.stop();
}
(_, PeerStatus::Connecting, PeerMessage::Handshake(handshake)) => {
(_, PeerStatus::Connecting, PeerMessage::HandshakeV2(handshake)) => {
debug!(target: "network", "{:?}: Received handshake {:?}", self.node_info.id, handshake);

debug_assert!(
OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION <= handshake.version
&& handshake.version <= NETWORK_PROTOCOL_VERSION
);

if handshake.chain_info.genesis_id != self.genesis_id {
debug!(target: "network", "Received connection from node with different genesis.");
ctx.address().do_send(SendMessage {
Expand All @@ -677,23 +724,6 @@ impl StreamHandler<Result<Vec<u8>, ReasonForBan>> for Peer {
// Connection will be closed by a handshake timeout
}

if handshake.version < OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION
// TODO(HandshakeV2): Check that our current version is supported by other party.
// PROTOCOL_VERSION < handshake.oldest_supported_version
|| PROTOCOL_VERSION < handshake.version
{
debug!(target: "network", "Received connection from node with incompatible network protocol version.");
self.send_message(PeerMessage::HandshakeFailure(
self.node_info.clone(),
HandshakeFailureReason::ProtocolVersionMismatch {
version: PROTOCOL_VERSION,
oldest_supported_version: OLDEST_BACKWARD_COMPATIBLE_PROTOCOL_VERSION,
},
));
return;
// Connection will be closed by a handshake timeout
}

if handshake.peer_id == self.node_info.id {
near_metrics::inc_counter(&metrics::RECEIVED_INFO_ABOUT_ITSELF);
debug!(target: "network", "Received info about itself. Disconnecting this peer.");
Expand Down
Loading