From fa81e4aa4c8cd75bdc5c96d3568047cc66f50ae0 Mon Sep 17 00:00:00 2001 From: behzad nouri Date: Tue, 16 Aug 2022 16:30:08 -0400 Subject: [PATCH] adds hash domain to ping-pong protocol In order to maintain backward compatibility, for now the responding node will hash the token both with and without domain so that the other node will accept the response regardless of its upgrade status. Once the cluster has upgraded to the new code, we will remove the legacy domain = false case. --- core/src/ancestor_hashes_service.rs | 15 +++++++---- core/src/serve_repair.rs | 15 +++++++---- gossip/src/cluster_info.rs | 30 +++++++++++++--------- gossip/src/ping_pong.rs | 40 ++++++++++++++++++++++++----- 4 files changed, 72 insertions(+), 28 deletions(-) diff --git a/core/src/ancestor_hashes_service.rs b/core/src/ancestor_hashes_service.rs index 330ebb072abc10..4813ed11685069 100644 --- a/core/src/ancestor_hashes_service.rs +++ b/core/src/ancestor_hashes_service.rs @@ -425,16 +425,21 @@ impl AncestorHashesService { stats.invalid_packets += 1; return None; } - if ping.verify() { - stats.ping_count += 1; - if let Ok(pong) = Pong::new(&ping, keypair) { + if !ping.verify() { + stats.ping_err_verify_count += 1; + return None; + } + stats.ping_count += 1; + // Respond both with and without domain so that the other node + // will accept the response regardless of its upgrade status. + // TODO: remove domain = false once cluster is upgraded. + for domain in [false, true] { + if let Ok(pong) = Pong::new(domain, &ping, keypair) { let pong = RepairProtocol::Pong(pong); if let Ok(pong_bytes) = serialize(&pong) { let _ignore = ancestor_socket.send_to(&pong_bytes[..], from_addr); } } - } else { - stats.ping_err_verify_count += 1; } None } diff --git a/core/src/serve_repair.rs b/core/src/serve_repair.rs index 2f755ebb17f4bb..47443bcd9acc2d 100644 --- a/core/src/serve_repair.rs +++ b/core/src/serve_repair.rs @@ -1044,11 +1044,16 @@ impl ServeRepair { } packet.meta.set_discard(true); stats.ping_count += 1; - if let Ok(pong) = Pong::new(&ping, keypair) { - let pong = RepairProtocol::Pong(pong); - if let Ok(pong_bytes) = serialize(&pong) { - let from_addr = packet.meta.socket_addr(); - pending_pongs.push((pong_bytes, from_addr)); + // Respond both with and without domain so that the other node + // will accept the response regardless of its upgrade status. + // TODO: remove domain = false once cluster is upgraded. + for domain in [false, true] { + if let Ok(pong) = Pong::new(domain, &ping, keypair) { + let pong = RepairProtocol::Pong(pong); + if let Ok(pong_bytes) = serialize(&pong) { + let from_addr = packet.meta.socket_addr(); + pending_pongs.push((pong_bytes, from_addr)); + } } } } diff --git a/gossip/src/cluster_info.rs b/gossip/src/cluster_info.rs index 40142f70e3d336..9d692b8a08aba6 100644 --- a/gossip/src/cluster_info.rs +++ b/gossip/src/cluster_info.rs @@ -2170,14 +2170,18 @@ impl ClusterInfo { I: IntoIterator, { let keypair = self.keypair(); - let pongs_and_dests: Vec<_> = pings - .into_iter() - .filter_map(|(addr, ping)| { - let pong = Pong::new(&ping, &keypair).ok()?; - let pong = Protocol::PongMessage(pong); - Some((addr, pong)) - }) - .collect(); + let mut pongs_and_dests = Vec::new(); + for (addr, ping) in pings { + // Respond both with and without domain so that the other node will + // accept the response regardless of its upgrade status. + // TODO: remove domain = false once cluster is upgraded. + for domain in [false, true] { + if let Ok(pong) = Pong::new(domain, &ping, &keypair) { + let pong = Protocol::PongMessage(pong); + pongs_and_dests.push((addr, pong)); + } + } + } if pongs_and_dests.is_empty() { None } else { @@ -3287,7 +3291,9 @@ RPC Enabled Nodes: 1"#; let pongs: Vec<(SocketAddr, Pong)> = pings .iter() .zip(&remote_nodes) - .map(|(ping, (keypair, socket))| (*socket, Pong::new(ping, keypair).unwrap())) + .map(|(ping, (keypair, socket))| { + (*socket, Pong::new(/*domain:*/ true, ping, keypair).unwrap()) + }) .collect(); let now = now + Duration::from_millis(1); cluster_info.handle_batch_pong_messages(pongs, now); @@ -3330,7 +3336,7 @@ RPC Enabled Nodes: 1"#; .collect(); let pongs: Vec<_> = pings .iter() - .map(|ping| Pong::new(ping, &this_node).unwrap()) + .map(|ping| Pong::new(/*domain:*/ false, ping, &this_node).unwrap()) .collect(); let recycler = PacketBatchRecycler::default(); let packets = cluster_info @@ -3342,9 +3348,9 @@ RPC Enabled Nodes: 1"#; &recycler, ) .unwrap(); - assert_eq!(remote_nodes.len(), packets.len()); + assert_eq!(remote_nodes.len() * 2, packets.len()); for (packet, (_, socket), pong) in izip!( - packets.into_iter(), + packets.into_iter().step_by(2), remote_nodes.into_iter(), pongs.into_iter() ) { diff --git a/gossip/src/ping_pong.rs b/gossip/src/ping_pong.rs index 6c3a219cfdb81b..16961f26f18388 100644 --- a/gossip/src/ping_pong.rs +++ b/gossip/src/ping_pong.rs @@ -16,6 +16,8 @@ use { }, }; +const PING_PONG_HASH_PREFIX: &[u8] = "SOLANA_PING_PONG".as_bytes(); + #[derive(AbiExample, Debug, Deserialize, Serialize)] pub struct Ping { from: Pubkey, @@ -100,8 +102,17 @@ impl Signable for Ping { } impl Pong { - pub fn new(ping: &Ping, keypair: &Keypair) -> Result { - let hash = hash::hash(&serialize(&ping.token)?); + pub fn new( + domain: bool, + ping: &Ping, + keypair: &Keypair, + ) -> Result { + let token = serialize(&ping.token)?; + let hash = if domain { + hash::hashv(&[PING_PONG_HASH_PREFIX, &token]) + } else { + hash::hash(&token) + }; let pong = Pong { from: keypair.pubkey(), hash, @@ -187,9 +198,15 @@ impl PingCache { Some(t) if now.saturating_duration_since(*t) < delay => None, _ => { let ping = pingf()?; - let hash = hash::hash(&serialize(&ping.token).ok()?); - self.pings.put(node, now); + let token = serialize(&ping.token).ok()?; + // For backward compatibility, for now responses both with and + // without domain are accepted. + // TODO: remove no domain case once cluster is upgraded. + let hash = hash::hash(&token); + self.pending_cache.put(hash, node); + let hash = hash::hashv(&[PING_PONG_HASH_PREFIX, &token]); self.pending_cache.put(hash, node); + self.pings.put(node, now); Some(ping) } } @@ -281,10 +298,18 @@ mod tests { assert!(ping.verify()); assert!(ping.sanitize().is_ok()); - let pong = Pong::new(&ping, &keypair).unwrap(); + let pong = Pong::new(/*domain:*/ false, &ping, &keypair).unwrap(); assert!(pong.verify()); assert!(pong.sanitize().is_ok()); assert_eq!(hash::hash(&ping.token), pong.hash); + + let pong = Pong::new(/*domian:*/ true, &ping, &keypair).unwrap(); + assert!(pong.verify()); + assert!(pong.sanitize().is_ok()); + assert_eq!( + hash::hashv(&[PING_PONG_HASH_PREFIX, &ping.token]), + pong.hash + ); } #[test] @@ -339,7 +364,10 @@ mod tests { assert!(ping.is_none()); } Some(ping) => { - let pong = Pong::new(ping, keypair).unwrap(); + let domain = rng.gen_ratio(1, 2); + let pong = Pong::new(domain, ping, keypair).unwrap(); + assert!(cache.add(&pong, *socket, now)); + let pong = Pong::new(!domain, ping, keypair).unwrap(); assert!(cache.add(&pong, *socket, now)); } }