gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix incorrectly signed CrdsValues (#6696) (#6699)

Browse Source 
automerge
This commit is contained in:
mergify[bot]
2019-11-03 10:57:59 -08:00
committed by Grimes
parent c86bf60a40
commit d9a9d6547f
8 changed files with 320 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
core/src/cluster_info.rs
View File

@ -12,17 +12,20 @@
//! * layer 2 - Everyone else, if layer 1 is `2^10`, layer 2 should be able to fit `2^20` number of nodes. //! * layer 2 - Everyone else, if layer 1 is `2^10`, layer 2 should be able to fit `2^20` number of nodes.
//! //!
//! Bank needs to provide an interface for us to query the stake weight //! Bank needs to provide an interface for us to query the stake weight
use crate::contact_info::ContactInfo; use crate::crds_value::CrdsValue;
use crate::crds_gossip::CrdsGossip; use crate::{
use crate::crds_gossip_error::CrdsGossipError; contact_info::ContactInfo,
use crate::crds_gossip_pull::{CrdsFilter, CRDS_GOSSIP_PULL_CRDS_TIMEOUT_MS}; crds_gossip::CrdsGossip,
use crate::crds_value::{CrdsValue, CrdsValueLabel, EpochSlots, Vote}; crds_gossip_error::CrdsGossipError,
use crate::packet::{to_shared_blob, Blob, Packet, SharedBlob}; crds_gossip_pull::{CrdsFilter, CRDS_GOSSIP_PULL_CRDS_TIMEOUT_MS},
use crate::repair_service::RepairType; crds_value::{CrdsData, CrdsValueLabel, EpochSlots, Vote},
use crate::result::{Error, Result}; packet::{to_shared_blob, Blob, Packet, SharedBlob},
use crate::sendmmsg::{multicast, send_mmsg}; repair_service::RepairType,
use crate::streamer::{BlobReceiver, BlobSender}; result::{Error, Result},
use crate::weighted_shuffle::{weighted_best, weighted_shuffle}; sendmmsg::{multicast, send_mmsg},
streamer::{BlobReceiver, BlobSender},
weighted_shuffle::{weighted_best, weighted_shuffle},
};
use bincode::{deserialize, serialize, serialized_size}; use bincode::{deserialize, serialize, serialized_size};
use core::cmp; use core::cmp;
use itertools::Itertools; use itertools::Itertools;
@ -195,8 +198,8 @@ impl ClusterInfo {
pub fn insert_self(&mut self, contact_info: ContactInfo) { pub fn insert_self(&mut self, contact_info: ContactInfo) {
if self.id() == contact_info.id { if self.id() == contact_info.id {
let mut value = CrdsValue::ContactInfo(contact_info.clone()); let value =
value.sign(&self.keypair); CrdsValue::new_signed(CrdsData::ContactInfo(contact_info.clone()), &self.keypair);
let _ = self.gossip.crds.insert(value, timestamp()); let _ = self.gossip.crds.insert(value, timestamp());
} }
} }
@ -205,8 +208,7 @@ impl ClusterInfo {
let mut my_data = self.my_data(); let mut my_data = self.my_data();
let now = timestamp(); let now = timestamp();
my_data.wallclock = now; my_data.wallclock = now;
let mut entry = CrdsValue::ContactInfo(my_data); let entry = CrdsValue::new_signed(CrdsData::ContactInfo(my_data), &self.keypair);
entry.sign(&self.keypair);
self.gossip.refresh_push_active_set(stakes); self.gossip.refresh_push_active_set(stakes);
self.gossip self.gossip
.process_push_message(&self.id(), vec![entry], now); .process_push_message(&self.id(), vec![entry], now);
@ -214,8 +216,7 @@ impl ClusterInfo {
// TODO kill insert_info, only used by tests // TODO kill insert_info, only used by tests
pub fn insert_info(&mut self, contact_info: ContactInfo) { pub fn insert_info(&mut self, contact_info: ContactInfo) {
let mut value = CrdsValue::ContactInfo(contact_info); let value = CrdsValue::new_signed(CrdsData::ContactInfo(contact_info), &self.keypair);
value.sign(&self.keypair);
let _ = self.gossip.crds.insert(value, timestamp()); let _ = self.gossip.crds.insert(value, timestamp());
} }
@ -297,8 +298,10 @@ impl ClusterInfo {
pub fn push_epoch_slots(&mut self, id: Pubkey, root: u64, slots: BTreeSet<u64>) { pub fn push_epoch_slots(&mut self, id: Pubkey, root: u64, slots: BTreeSet<u64>) {
let now = timestamp(); let now = timestamp();
let mut entry = CrdsValue::EpochSlots(EpochSlots::new(id, root, slots, now)); let entry = CrdsValue::new_signed(
entry.sign(&self.keypair); CrdsData::EpochSlots(EpochSlots::new(id, root, slots, now)),
&self.keypair,
);
self.gossip self.gossip
.process_push_message(&self.id(), vec![entry], now); .process_push_message(&self.id(), vec![entry], now);
} }
@ -306,8 +309,7 @@ impl ClusterInfo {
pub fn push_vote(&mut self, vote: Transaction) { pub fn push_vote(&mut self, vote: Transaction) {
let now = timestamp(); let now = timestamp();
let vote = Vote::new(&self.id(), vote, now); let vote = Vote::new(&self.id(), vote, now);
let mut entry = CrdsValue::Vote(vote); let entry = CrdsValue::new_signed(CrdsData::Vote(vote), &self.keypair);
entry.sign(&self.keypair);
self.gossip self.gossip
.process_push_message(&self.id(), vec![entry], now); .process_push_message(&self.id(), vec![entry], now);
} }
@ -915,7 +917,7 @@ impl ClusterInfo {
.expect("unable to serialize default filter") as usize; .expect("unable to serialize default filter") as usize;
let protocol = Protocol::PullRequest( let protocol = Protocol::PullRequest(
CrdsFilter::default(), CrdsFilter::default(),
CrdsValue::ContactInfo(ContactInfo::default()), CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default())),
); );
let protocol_size = let protocol_size =
serialized_size(&protocol).expect("unable to serialize gossip protocol") as usize; serialized_size(&protocol).expect("unable to serialize gossip protocol") as usize;
@ -1161,9 +1163,7 @@ impl ClusterInfo {
1 1
); );
} else if caller.contact_info().is_some() { } else if caller.contact_info().is_some() {
if caller.contact_info().unwrap().pubkey() if caller.contact_info().unwrap().id == me.read().unwrap().gossip.id {
== me.read().unwrap().gossip.id
{
warn!("PullRequest ignored, I'm talking to myself"); warn!("PullRequest ignored, I'm talking to myself");
inc_new_counter_debug!("cluster_info-window-request-loopback", 1); inc_new_counter_debug!("cluster_info-window-request-loopback", 1);
} else { } else {
@ -2384,7 +2384,8 @@ mod tests {
} }
// now add this message back to the table and make sure after the next pull, the entrypoint is unset // now add this message back to the table and make sure after the next pull, the entrypoint is unset
let entrypoint_crdsvalue = CrdsValue::ContactInfo(entrypoint.clone()); let entrypoint_crdsvalue =
CrdsValue::new_unsigned(CrdsData::ContactInfo(entrypoint.clone()));
let cluster_info = Arc::new(RwLock::new(cluster_info)); let cluster_info = Arc::new(RwLock::new(cluster_info));
ClusterInfo::handle_pull_response( ClusterInfo::handle_pull_response(
&cluster_info, &cluster_info,
@ -2401,7 +2402,7 @@ mod tests {
#[test] #[test]
fn test_split_messages_small() { fn test_split_messages_small() {
let value = CrdsValue::ContactInfo(ContactInfo::default()); let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
test_split_messages(value); test_split_messages(value);
} }
@ -2411,13 +2412,12 @@ mod tests {
for i in 0..128 { for i in 0..128 {
btree_slots.insert(i); btree_slots.insert(i);
} }
let value = CrdsValue::EpochSlots(EpochSlots { let value = CrdsValue::new_unsigned(CrdsData::EpochSlots(EpochSlots {
from: Pubkey::default(), from: Pubkey::default(),
root: 0, root: 0,
slots: btree_slots, slots: btree_slots,
signature: Signature::default(),
wallclock: 0, wallclock: 0,
}); }));
test_split_messages(value); test_split_messages(value);
} }
@ -2441,7 +2441,7 @@ mod tests {
} }
fn check_pull_request_size(filter: CrdsFilter) { fn check_pull_request_size(filter: CrdsFilter) {
let value = CrdsValue::ContactInfo(ContactInfo::default()); let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
let protocol = Protocol::PullRequest(filter, value.clone()); let protocol = Protocol::PullRequest(filter, value.clone());
assert!(serialized_size(&protocol).unwrap() <= PACKET_DATA_SIZE as u64); assert!(serialized_size(&protocol).unwrap() <= PACKET_DATA_SIZE as u64);
} }

52
core/src/contact_info.rs
View File

@ -1,12 +1,9 @@
use bincode::serialize;
use solana_sdk::pubkey::Pubkey; use solana_sdk::pubkey::Pubkey;
#[cfg(test)] #[cfg(test)]
use solana_sdk::rpc_port; use solana_sdk::rpc_port;
#[cfg(test)] #[cfg(test)]
use solana_sdk::signature::{Keypair, KeypairUtil}; use solana_sdk::signature::{Keypair, KeypairUtil};
use solana_sdk::signature::{Signable, Signature};
use solana_sdk::timing::timestamp; use solana_sdk::timing::timestamp;
use std::borrow::Cow;
use std::cmp::{Ord, Ordering, PartialEq, PartialOrd}; use std::cmp::{Ord, Ordering, PartialEq, PartialOrd};
use std::net::{IpAddr, SocketAddr}; use std::net::{IpAddr, SocketAddr};
@ -14,8 +11,6 @@ use std::net::{IpAddr, SocketAddr};
#[derive(Serialize, Deserialize, Clone, Debug)] #[derive(Serialize, Deserialize, Clone, Debug)]
pub struct ContactInfo { pub struct ContactInfo {
pub id: Pubkey, pub id: Pubkey,
/// signature of this ContactInfo
pub signature: Signature,
/// gossip address /// gossip address
pub gossip: SocketAddr, pub gossip: SocketAddr,
/// address to connect to for replication /// address to connect to for replication
@ -89,7 +84,6 @@ impl Default for ContactInfo {
rpc: socketaddr_any!(), rpc: socketaddr_any!(),
rpc_pubsub: socketaddr_any!(), rpc_pubsub: socketaddr_any!(),
wallclock: 0, wallclock: 0,
signature: Signature::default(),
} }
} }
} }
@ -111,7 +105,6 @@ impl ContactInfo {
) -> Self { ) -> Self {
Self { Self {
id: *id, id: *id,
signature: Signature::default(),
gossip, gossip,
tvu, tvu,
tvu_forwards, tvu_forwards,
@ -242,51 +235,6 @@ impl ContactInfo {
} }
} }
impl Signable for ContactInfo {
fn pubkey(&self) -> Pubkey {
self.id
}
fn signable_data(&self) -> Cow<[u8]> {
#[derive(Serialize)]
struct SignData {
id: Pubkey,
gossip: SocketAddr,
tvu: SocketAddr,
tpu: SocketAddr,
tpu_forwards: SocketAddr,
repair: SocketAddr,
storage_addr: SocketAddr,
rpc: SocketAddr,
rpc_pubsub: SocketAddr,
wallclock: u64,
}
let me = self;
let data = SignData {
id: me.id,
gossip: me.gossip,
tvu: me.tvu,
tpu: me.tpu,
storage_addr: me.storage_addr,
tpu_forwards: me.tpu_forwards,
repair: me.repair,
rpc: me.rpc,
rpc_pubsub: me.rpc_pubsub,
wallclock: me.wallclock,
};
Cow::Owned(serialize(&data).expect("failed to serialize ContactInfo"))
}
fn get_signature(&self) -> Signature {
self.signature
}
fn set_signature(&mut self, signature: Signature) {
self.signature = signature
}
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;

57
core/src/crds.rs
View File

@ -165,11 +165,12 @@ impl Crds {
mod test { mod test {
use super::*; use super::*;
use crate::contact_info::ContactInfo; use crate::contact_info::ContactInfo;
use crate::crds_value::CrdsData;
#[test] #[test]
fn test_insert() { fn test_insert() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let val = CrdsValue::ContactInfo(ContactInfo::default()); let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
assert_eq!(crds.insert(val.clone(), 0).ok(), Some(None)); assert_eq!(crds.insert(val.clone(), 0).ok(), Some(None));
assert_eq!(crds.table.len(), 1); assert_eq!(crds.table.len(), 1);
assert!(crds.table.contains_key(&val.label())); assert!(crds.table.contains_key(&val.label()));
@ -178,7 +179,7 @@ mod test {
#[test] #[test]
fn test_update_old() { fn test_update_old() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let val = CrdsValue::ContactInfo(ContactInfo::default()); let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
assert_eq!(crds.insert(val.clone(), 0), Ok(None)); assert_eq!(crds.insert(val.clone(), 0), Ok(None));
assert_eq!(crds.insert(val.clone(), 1), Err(CrdsError::InsertFailed)); assert_eq!(crds.insert(val.clone(), 1), Err(CrdsError::InsertFailed));
assert_eq!(crds.table[&val.label()].local_timestamp, 0); assert_eq!(crds.table[&val.label()].local_timestamp, 0);
@ -186,9 +187,15 @@ mod test {
#[test] #[test]
fn test_update_new() { fn test_update_new() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let original = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::default(), 0)); let original = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::default(),
0,
)));
assert_matches!(crds.insert(original.clone(), 0), Ok(_)); assert_matches!(crds.insert(original.clone(), 0), Ok(_));
let val = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::default(), 1)); let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::default(),
1,
)));
assert_eq!( assert_eq!(
crds.insert(val.clone(), 1).unwrap().unwrap().value, crds.insert(val.clone(), 1).unwrap().unwrap().value,
original original
@ -198,14 +205,17 @@ mod test {
#[test] #[test]
fn test_update_timestamp() { fn test_update_timestamp() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let val = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::default(), 0)); let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::default(),
0,
)));
assert_eq!(crds.insert(val.clone(), 0), Ok(None)); assert_eq!(crds.insert(val.clone(), 0), Ok(None));
crds.update_label_timestamp(&val.label(), 1); crds.update_label_timestamp(&val.label(), 1);
assert_eq!(crds.table[&val.label()].local_timestamp, 1); assert_eq!(crds.table[&val.label()].local_timestamp, 1);
assert_eq!(crds.table[&val.label()].insert_timestamp, 0); assert_eq!(crds.table[&val.label()].insert_timestamp, 0);
let val2 = CrdsValue::ContactInfo(ContactInfo::default()); let val2 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
assert_eq!(val2.label().pubkey(), val.label().pubkey()); assert_eq!(val2.label().pubkey(), val.label().pubkey());
assert_matches!(crds.insert(val2.clone(), 0), Ok(Some(_))); assert_matches!(crds.insert(val2.clone(), 0), Ok(Some(_)));
@ -221,7 +231,7 @@ mod test {
let mut ci = ContactInfo::default(); let mut ci = ContactInfo::default();
ci.wallclock += 1; ci.wallclock += 1;
let val3 = CrdsValue::ContactInfo(ci); let val3 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci));
assert_matches!(crds.insert(val3.clone(), 3), Ok(Some(_))); assert_matches!(crds.insert(val3.clone(), 3), Ok(Some(_)));
assert_eq!(crds.table[&val2.label()].local_timestamp, 3); assert_eq!(crds.table[&val2.label()].local_timestamp, 3);
assert_eq!(crds.table[&val2.label()].insert_timestamp, 3); assert_eq!(crds.table[&val2.label()].insert_timestamp, 3);
@ -229,7 +239,7 @@ mod test {
#[test] #[test]
fn test_find_old_records() { fn test_find_old_records() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let val = CrdsValue::ContactInfo(ContactInfo::default()); let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
assert_eq!(crds.insert(val.clone(), 1), Ok(None)); assert_eq!(crds.insert(val.clone(), 1), Ok(None));
assert!(crds.find_old_labels(0).is_empty()); assert!(crds.find_old_labels(0).is_empty());
@ -239,7 +249,7 @@ mod test {
#[test] #[test]
fn test_remove() { fn test_remove() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let val = CrdsValue::ContactInfo(ContactInfo::default()); let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
assert_matches!(crds.insert(val.clone(), 1), Ok(_)); assert_matches!(crds.insert(val.clone(), 1), Ok(_));
assert_eq!(crds.find_old_labels(1), vec![val.label()]); assert_eq!(crds.find_old_labels(1), vec![val.label()]);
@ -248,7 +258,7 @@ mod test {
} }
#[test] #[test]
fn test_equal() { fn test_equal() {
let val = CrdsValue::ContactInfo(ContactInfo::default()); let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
let v1 = VersionedCrdsValue::new(1, val.clone()); let v1 = VersionedCrdsValue::new(1, val.clone());
let v2 = VersionedCrdsValue::new(1, val); let v2 = VersionedCrdsValue::new(1, val);
assert_eq!(v1, v2); assert_eq!(v1, v2);
@ -258,12 +268,15 @@ mod test {
fn test_hash_order() { fn test_hash_order() {
let v1 = VersionedCrdsValue::new( let v1 = VersionedCrdsValue::new(
1, 1,
CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::default(), 0)), CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::default(),
0,
))),
); );
let v2 = VersionedCrdsValue::new(1, { let v2 = VersionedCrdsValue::new(1, {
let mut contact_info = ContactInfo::new_localhost(&Pubkey::default(), 0); let mut contact_info = ContactInfo::new_localhost(&Pubkey::default(), 0);
contact_info.rpc = socketaddr!("0.0.0.0:0"); contact_info.rpc = socketaddr!("0.0.0.0:0");
CrdsValue::ContactInfo(contact_info) CrdsValue::new_unsigned(CrdsData::ContactInfo(contact_info))
}); });
assert_eq!(v1.value.label(), v2.value.label()); assert_eq!(v1.value.label(), v2.value.label());
@ -285,11 +298,17 @@ mod test {
fn test_wallclock_order() { fn test_wallclock_order() {
let v1 = VersionedCrdsValue::new( let v1 = VersionedCrdsValue::new(
1, 1,
CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::default(), 1)), CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::default(),
1,
))),
); );
let v2 = VersionedCrdsValue::new( let v2 = VersionedCrdsValue::new(
1, 1,
CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::default(), 0)), CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::default(),
0,
))),
); );
assert_eq!(v1.value.label(), v2.value.label()); assert_eq!(v1.value.label(), v2.value.label());
assert!(v1 > v2); assert!(v1 > v2);
@ -301,11 +320,17 @@ mod test {
fn test_label_order() { fn test_label_order() {
let v1 = VersionedCrdsValue::new( let v1 = VersionedCrdsValue::new(
1, 1,
CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)), CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
))),
); );
let v2 = VersionedCrdsValue::new( let v2 = VersionedCrdsValue::new(
1, 1,
CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)), CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
))),
); );
assert_ne!(v1, v2); assert_ne!(v1, v2);
assert!(!(v1 == v2)); assert!(!(v1 == v2));

7
core/src/crds_gossip.rs
View File

@ -9,7 +9,6 @@ use crate::crds_gossip_pull::{CrdsFilter, CrdsGossipPull};
use crate::crds_gossip_push::{CrdsGossipPush, CRDS_GOSSIP_NUM_ACTIVE}; use crate::crds_gossip_push::{CrdsGossipPush, CRDS_GOSSIP_NUM_ACTIVE};
use crate::crds_value::{CrdsValue, CrdsValueLabel}; use crate::crds_value::{CrdsValue, CrdsValueLabel};
use solana_sdk::pubkey::Pubkey; use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::Signable;
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
///The min size for bloom filters ///The min size for bloom filters
@ -204,6 +203,7 @@ pub fn get_weight(max_weight: f32, time_since_last_selected: u32, stake: f32) ->
mod test { mod test {
use super::*; use super::*;
use crate::contact_info::ContactInfo; use crate::contact_info::ContactInfo;
use crate::crds_value::CrdsData;
use solana_sdk::hash::hash; use solana_sdk::hash::hash;
use solana_sdk::timing::timestamp; use solana_sdk::timing::timestamp;
@ -216,7 +216,10 @@ mod test {
let prune_pubkey = Pubkey::new(&[2; 32]); let prune_pubkey = Pubkey::new(&[2; 32]);
crds_gossip crds_gossip
.crds .crds
.insert(CrdsValue::ContactInfo(ci.clone()), 0) .insert(
CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone())),
0,
)
.unwrap(); .unwrap();
crds_gossip.refresh_push_active_set(&HashMap::new()); crds_gossip.refresh_push_active_set(&HashMap::new());
let now = timestamp(); let now = timestamp();

74
core/src/crds_gossip_pull.rs
View File

@ -294,6 +294,7 @@ impl CrdsGossipPull {
mod test { mod test {
use super::*; use super::*;
use crate::contact_info::ContactInfo; use crate::contact_info::ContactInfo;
use crate::crds_value::CrdsData;
use itertools::Itertools; use itertools::Itertools;
use solana_sdk::hash::hash; use solana_sdk::hash::hash;
use solana_sdk::packet::PACKET_DATA_SIZE; use solana_sdk::packet::PACKET_DATA_SIZE;
@ -303,10 +304,16 @@ mod test {
let mut crds = Crds::default(); let mut crds = Crds::default();
let mut stakes = HashMap::new(); let mut stakes = HashMap::new();
let node = CrdsGossipPull::default(); let node = CrdsGossipPull::default();
let me = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let me = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
crds.insert(me.clone(), 0).unwrap(); crds.insert(me.clone(), 0).unwrap();
for i in 1..=30 { for i in 1..=30 {
let entry = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let entry = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let id = entry.label().pubkey(); let id = entry.label().pubkey();
crds.insert(entry.clone(), 0).unwrap(); crds.insert(entry.clone(), 0).unwrap();
stakes.insert(id, i * 100); stakes.insert(id, i * 100);
@ -325,7 +332,10 @@ mod test {
#[test] #[test]
fn test_new_pull_request() { fn test_new_pull_request() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let entry = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let entry = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let id = entry.label().pubkey(); let id = entry.label().pubkey();
let node = CrdsGossipPull::default(); let node = CrdsGossipPull::default();
assert_eq!( assert_eq!(
@ -339,7 +349,10 @@ mod test {
Err(CrdsGossipError::NoPeers) Err(CrdsGossipError::NoPeers)
); );
let new = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let new = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
crds.insert(new.clone(), 0).unwrap(); crds.insert(new.clone(), 0).unwrap();
let req = node.new_pull_request(&crds, &id, 0, &HashMap::new(), PACKET_DATA_SIZE); let req = node.new_pull_request(&crds, &id, 0, &HashMap::new(), PACKET_DATA_SIZE);
let (to, _, self_info) = req.unwrap(); let (to, _, self_info) = req.unwrap();
@ -350,13 +363,22 @@ mod test {
#[test] #[test]
fn test_new_mark_creation_time() { fn test_new_mark_creation_time() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let entry = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let entry = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let node_pubkey = entry.label().pubkey(); let node_pubkey = entry.label().pubkey();
let mut node = CrdsGossipPull::default(); let mut node = CrdsGossipPull::default();
crds.insert(entry.clone(), 0).unwrap(); crds.insert(entry.clone(), 0).unwrap();
let old = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let old = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
crds.insert(old.clone(), 0).unwrap(); crds.insert(old.clone(), 0).unwrap();
let new = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let new = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
crds.insert(new.clone(), 0).unwrap(); crds.insert(new.clone(), 0).unwrap();
// set request creation time to max_value // set request creation time to max_value
@ -380,11 +402,17 @@ mod test {
#[test] #[test]
fn test_process_pull_request() { fn test_process_pull_request() {
let mut node_crds = Crds::default(); let mut node_crds = Crds::default();
let entry = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let entry = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let node_pubkey = entry.label().pubkey(); let node_pubkey = entry.label().pubkey();
let node = CrdsGossipPull::default(); let node = CrdsGossipPull::default();
node_crds.insert(entry.clone(), 0).unwrap(); node_crds.insert(entry.clone(), 0).unwrap();
let new = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let new = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
node_crds.insert(new.clone(), 0).unwrap(); node_crds.insert(new.clone(), 0).unwrap();
let req = node.new_pull_request( let req = node.new_pull_request(
&node_crds, &node_crds,
@ -419,22 +447,32 @@ mod test {
#[test] #[test]
fn test_process_pull_request_response() { fn test_process_pull_request_response() {
let mut node_crds = Crds::default(); let mut node_crds = Crds::default();
let entry = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let entry = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let node_pubkey = entry.label().pubkey(); let node_pubkey = entry.label().pubkey();
let mut node = CrdsGossipPull::default(); let mut node = CrdsGossipPull::default();
node_crds.insert(entry.clone(), 0).unwrap(); node_crds.insert(entry.clone(), 0).unwrap();
let new = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let new = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
node_crds.insert(new.clone(), 0).unwrap(); node_crds.insert(new.clone(), 0).unwrap();
let mut dest = CrdsGossipPull::default(); let mut dest = CrdsGossipPull::default();
let mut dest_crds = Crds::default(); let mut dest_crds = Crds::default();
let new_id = Pubkey::new_rand(); let new_id = Pubkey::new_rand();
let new = CrdsValue::ContactInfo(ContactInfo::new_localhost(&new_id, 1)); let new = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&new_id, 1,
)));
dest_crds.insert(new.clone(), 0).unwrap(); dest_crds.insert(new.clone(), 0).unwrap();
// node contains a key from the dest node, but at an older local timestamp // node contains a key from the dest node, but at an older local timestamp
let same_key = CrdsValue::ContactInfo(ContactInfo::new_localhost(&new_id, 0)); let same_key = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&new_id, 0,
)));
assert_eq!(same_key.label(), new.label()); assert_eq!(same_key.label(), new.label());
assert!(same_key.wallclock() < new.wallclock()); assert!(same_key.wallclock() < new.wallclock());
node_crds.insert(same_key.clone(), 0).unwrap(); node_crds.insert(same_key.clone(), 0).unwrap();
@ -494,12 +532,18 @@ mod test {
#[test] #[test]
fn test_gossip_purge() { fn test_gossip_purge() {
let mut node_crds = Crds::default(); let mut node_crds = Crds::default();
let entry = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let entry = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let node_label = entry.label(); let node_label = entry.label();
let node_pubkey = node_label.pubkey(); let node_pubkey = node_label.pubkey();
let mut node = CrdsGossipPull::default(); let mut node = CrdsGossipPull::default();
node_crds.insert(entry.clone(), 0).unwrap(); node_crds.insert(entry.clone(), 0).unwrap();
let old = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let old = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
node_crds.insert(old.clone(), 0).unwrap(); node_crds.insert(old.clone(), 0).unwrap();
let value_hash = node_crds.lookup_versioned(&old.label()).unwrap().value_hash; let value_hash = node_crds.lookup_versioned(&old.label()).unwrap().value_hash;

92
core/src/crds_gossip_push.rs
View File

@ -340,7 +340,7 @@ impl CrdsGossipPush {
mod test { mod test {
use super::*; use super::*;
use crate::contact_info::ContactInfo; use crate::contact_info::ContactInfo;
use solana_sdk::signature::Signable; use crate::crds_value::CrdsData;
#[test] #[test]
fn test_prune() { fn test_prune() {
@ -353,7 +353,9 @@ mod test {
stakes.insert(self_id, 100); stakes.insert(self_id, 100);
stakes.insert(origin, 100); stakes.insert(origin, 100);
let value = CrdsValue::ContactInfo(ContactInfo::new_localhost(&origin, 0)); let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&origin, 0,
)));
let label = value.label(); let label = value.label();
let low_staked_peers = (0..10).map(|_| Pubkey::new_rand()); let low_staked_peers = (0..10).map(|_| Pubkey::new_rand());
let mut low_staked_set = HashSet::new(); let mut low_staked_set = HashSet::new();
@ -395,7 +397,10 @@ mod test {
fn test_process_push() { fn test_process_push() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let mut push = CrdsGossipPush::default(); let mut push = CrdsGossipPush::default();
let value = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let label = value.label(); let label = value.label();
// push a new message // push a new message
assert_eq!( assert_eq!(
@ -416,7 +421,7 @@ mod test {
let mut push = CrdsGossipPush::default(); let mut push = CrdsGossipPush::default();
let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0); let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0);
ci.wallclock = 1; ci.wallclock = 1;
let value = CrdsValue::ContactInfo(ci.clone()); let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
// push a new message // push a new message
assert_eq!( assert_eq!(
@ -426,7 +431,7 @@ mod test {
// push an old version // push an old version
ci.wallclock = 0; ci.wallclock = 0;
let value = CrdsValue::ContactInfo(ci.clone()); let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
assert_eq!( assert_eq!(
push.process_push_message(&mut crds, &Pubkey::default(), value, 0), push.process_push_message(&mut crds, &Pubkey::default(), value, 0),
Err(CrdsGossipError::PushMessageOldVersion) Err(CrdsGossipError::PushMessageOldVersion)
@ -441,7 +446,7 @@ mod test {
// push a version to far in the future // push a version to far in the future
ci.wallclock = timeout + 1; ci.wallclock = timeout + 1;
let value = CrdsValue::ContactInfo(ci.clone()); let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
assert_eq!( assert_eq!(
push.process_push_message(&mut crds, &Pubkey::default(), value, 0), push.process_push_message(&mut crds, &Pubkey::default(), value, 0),
Err(CrdsGossipError::PushMessageTimeout) Err(CrdsGossipError::PushMessageTimeout)
@ -449,7 +454,7 @@ mod test {
// push a version to far in the past // push a version to far in the past
ci.wallclock = 0; ci.wallclock = 0;
let value = CrdsValue::ContactInfo(ci.clone()); let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
assert_eq!( assert_eq!(
push.process_push_message(&mut crds, &Pubkey::default(), value, timeout + 1), push.process_push_message(&mut crds, &Pubkey::default(), value, timeout + 1),
Err(CrdsGossipError::PushMessageTimeout) Err(CrdsGossipError::PushMessageTimeout)
@ -461,7 +466,7 @@ mod test {
let mut push = CrdsGossipPush::default(); let mut push = CrdsGossipPush::default();
let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0); let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0);
ci.wallclock = 0; ci.wallclock = 0;
let value_old = CrdsValue::ContactInfo(ci.clone()); let value_old = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
// push a new message // push a new message
assert_eq!( assert_eq!(
@ -471,7 +476,7 @@ mod test {
// push an old version // push an old version
ci.wallclock = 1; ci.wallclock = 1;
let value = CrdsValue::ContactInfo(ci.clone()); let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
assert_eq!( assert_eq!(
push.process_push_message(&mut crds, &Pubkey::default(), value, 0) push.process_push_message(&mut crds, &Pubkey::default(), value, 0)
.unwrap() .unwrap()
@ -492,13 +497,19 @@ mod test {
solana_logger::setup(); solana_logger::setup();
let mut crds = Crds::default(); let mut crds = Crds::default();
let mut push = CrdsGossipPush::default(); let mut push = CrdsGossipPush::default();
let value1 = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let value1 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
assert_eq!(crds.insert(value1.clone(), 0), Ok(None)); assert_eq!(crds.insert(value1.clone(), 0), Ok(None));
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1); push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
assert!(push.active_set.get(&value1.label().pubkey()).is_some()); assert!(push.active_set.get(&value1.label().pubkey()).is_some());
let value2 = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let value2 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
assert!(push.active_set.get(&value2.label().pubkey()).is_none()); assert!(push.active_set.get(&value2.label().pubkey()).is_none());
assert_eq!(crds.insert(value2.clone(), 0), Ok(None)); assert_eq!(crds.insert(value2.clone(), 0), Ok(None));
for _ in 0..30 { for _ in 0..30 {
@ -510,7 +521,9 @@ mod test {
assert!(push.active_set.get(&value2.label().pubkey()).is_some()); assert!(push.active_set.get(&value2.label().pubkey()).is_some());
for _ in 0..push.num_active { for _ in 0..push.num_active {
let value2 = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let value2 = CrdsValue::new_unsigned(CrdsData::ContactInfo(
ContactInfo::new_localhost(&Pubkey::new_rand(), 0),
));
assert_eq!(crds.insert(value2.clone(), 0), Ok(None)); assert_eq!(crds.insert(value2.clone(), 0), Ok(None));
} }
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1); push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
@ -523,8 +536,10 @@ mod test {
let push = CrdsGossipPush::default(); let push = CrdsGossipPush::default();
let mut stakes = HashMap::new(); let mut stakes = HashMap::new();
for i in 1..=100 { for i in 1..=100 {
let peer = let peer = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), time)); &Pubkey::new_rand(),
time,
)));
let id = peer.label().pubkey(); let id = peer.label().pubkey();
crds.insert(peer.clone(), time).unwrap(); crds.insert(peer.clone(), time).unwrap();
stakes.insert(id, i * 100); stakes.insert(id, i * 100);
@ -542,11 +557,17 @@ mod test {
fn test_new_push_messages() { fn test_new_push_messages() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let mut push = CrdsGossipPush::default(); let mut push = CrdsGossipPush::default();
let peer = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let peer = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
assert_eq!(crds.insert(peer.clone(), 0), Ok(None)); assert_eq!(crds.insert(peer.clone(), 0), Ok(None));
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1); push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
let new_msg = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let new_msg = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let mut expected = HashMap::new(); let mut expected = HashMap::new();
expected.insert(peer.label().pubkey(), vec![new_msg.clone()]); expected.insert(peer.label().pubkey(), vec![new_msg.clone()]);
assert_eq!( assert_eq!(
@ -560,11 +581,20 @@ mod test {
fn test_personalized_push_messages() { fn test_personalized_push_messages() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let mut push = CrdsGossipPush::default(); let mut push = CrdsGossipPush::default();
let peer_1 = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let peer_1 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
assert_eq!(crds.insert(peer_1.clone(), 0), Ok(None)); assert_eq!(crds.insert(peer_1.clone(), 0), Ok(None));
let peer_2 = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let peer_2 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
assert_eq!(crds.insert(peer_2.clone(), 0), Ok(None)); assert_eq!(crds.insert(peer_2.clone(), 0), Ok(None));
let peer_3 = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let peer_3 = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
assert_eq!( assert_eq!(
push.process_push_message(&mut crds, &Pubkey::default(), peer_3.clone(), 0), push.process_push_message(&mut crds, &Pubkey::default(), peer_3.clone(), 0),
Ok(None) Ok(None)
@ -572,7 +602,10 @@ mod test {
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1); push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
// push 3's contact info to 1 and 2 and 3 // push 3's contact info to 1 and 2 and 3
let new_msg = CrdsValue::ContactInfo(ContactInfo::new_localhost(&peer_3.pubkey(), 0)); let new_msg = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&peer_3.pubkey(),
0,
)));
let mut expected = HashMap::new(); let mut expected = HashMap::new();
expected.insert(peer_1.pubkey(), vec![new_msg.clone()]); expected.insert(peer_1.pubkey(), vec![new_msg.clone()]);
expected.insert(peer_2.pubkey(), vec![new_msg.clone()]); expected.insert(peer_2.pubkey(), vec![new_msg.clone()]);
@ -583,11 +616,17 @@ mod test {
fn test_process_prune() { fn test_process_prune() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let mut push = CrdsGossipPush::default(); let mut push = CrdsGossipPush::default();
let peer = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let peer = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
assert_eq!(crds.insert(peer.clone(), 0), Ok(None)); assert_eq!(crds.insert(peer.clone(), 0), Ok(None));
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1); push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
let new_msg = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let new_msg = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let expected = HashMap::new(); let expected = HashMap::new();
assert_eq!( assert_eq!(
push.process_push_message(&mut crds, &Pubkey::default(), new_msg.clone(), 0), push.process_push_message(&mut crds, &Pubkey::default(), new_msg.clone(), 0),
@ -600,13 +639,16 @@ mod test {
fn test_purge_old_pending_push_messages() { fn test_purge_old_pending_push_messages() {
let mut crds = Crds::default(); let mut crds = Crds::default();
let mut push = CrdsGossipPush::default(); let mut push = CrdsGossipPush::default();
let peer = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let peer = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
assert_eq!(crds.insert(peer.clone(), 0), Ok(None)); assert_eq!(crds.insert(peer.clone(), 0), Ok(None));
push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1); push.refresh_push_active_set(&crds, &HashMap::new(), &Pubkey::default(), 1, 1);
let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0); let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0);
ci.wallclock = 1; ci.wallclock = 1;
let new_msg = CrdsValue::ContactInfo(ci.clone()); let new_msg = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
let expected = HashMap::new(); let expected = HashMap::new();
assert_eq!( assert_eq!(
push.process_push_message(&mut crds, &Pubkey::default(), new_msg.clone(), 1), push.process_push_message(&mut crds, &Pubkey::default(), new_msg.clone(), 1),
@ -622,7 +664,7 @@ mod test {
let mut push = CrdsGossipPush::default(); let mut push = CrdsGossipPush::default();
let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0); let mut ci = ContactInfo::new_localhost(&Pubkey::new_rand(), 0);
ci.wallclock = 0; ci.wallclock = 0;
let value = CrdsValue::ContactInfo(ci.clone()); let value = CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone()));
let label = value.label(); let label = value.label();
// push a new message // push a new message
assert_eq!( assert_eq!(

209
core/src/crds_value.rs
View File

@ -8,9 +8,34 @@ use std::collections::BTreeSet;
use std::fmt; use std::fmt;
/// CrdsValue that is replicated across the cluster /// CrdsValue that is replicated across the cluster
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct CrdsValue {
pub signature: Signature,
pub data: CrdsData,
}
impl Signable for CrdsValue {
fn pubkey(&self) -> Pubkey {
self.pubkey()
}
fn signable_data(&self) -> Cow<[u8]> {
Cow::Owned(serialize(&self.data).expect("failed to serialize CrdsData"))
}
fn get_signature(&self) -> Signature {
self.signature
}
fn set_signature(&mut self, signature: Signature) {
self.signature = signature
}
}
/// CrdsData that defines the different types of items CrdsValues can hold
#[allow(clippy::large_enum_variant)] #[allow(clippy::large_enum_variant)]
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub enum CrdsValue { pub enum CrdsData {
/// * Merge Strategy - Latest wallclock is picked /// * Merge Strategy - Latest wallclock is picked
ContactInfo(ContactInfo), ContactInfo(ContactInfo),
/// * Merge Strategy - Latest wallclock is picked /// * Merge Strategy - Latest wallclock is picked
@ -24,7 +49,6 @@ pub struct EpochSlots {
pub from: Pubkey, pub from: Pubkey,
pub root: u64, pub root: u64,
pub slots: BTreeSet<u64>, pub slots: BTreeSet<u64>,
pub signature: Signature,
pub wallclock: u64, pub wallclock: u64,
} }
@ -34,46 +58,15 @@ impl EpochSlots {
from, from,
root, root,
slots, slots,
signature: Signature::default(),
wallclock, wallclock,
} }
} }
} }
impl Signable for EpochSlots {
fn pubkey(&self) -> Pubkey {
self.from
}
fn signable_data(&self) -> Cow<[u8]> {
#[derive(Serialize)]
struct SignData<'a> {
root: u64,
slots: &'a BTreeSet<u64>,
wallclock: u64,
}
let data = SignData {
root: self.root,
slots: &self.slots,
wallclock: self.wallclock,
};
Cow::Owned(serialize(&data).expect("unable to serialize EpochSlots"))
}
fn get_signature(&self) -> Signature {
self.signature
}
fn set_signature(&mut self, signature: Signature) {
self.signature = signature;
}
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)] #[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct Vote { pub struct Vote {
pub from: Pubkey, pub from: Pubkey,
pub transaction: Transaction, pub transaction: Transaction,
pub signature: Signature,
pub wallclock: u64, pub wallclock: u64,
} }
@ -82,39 +75,11 @@ impl Vote {
Self { Self {
from: *from, from: *from,
transaction, transaction,
signature: Signature::default(),
wallclock, wallclock,
} }
} }
} }
impl Signable for Vote {
fn pubkey(&self) -> Pubkey {
self.from
}
fn signable_data(&self) -> Cow<[u8]> {
#[derive(Serialize)]
struct SignData<'a> {
transaction: &'a Transaction,
wallclock: u64,
}
let data = SignData {
transaction: &self.transaction,
wallclock: self.wallclock,
};
Cow::Owned(serialize(&data).expect("unable to serialize Vote"))
}
fn get_signature(&self) -> Signature {
self.signature
}
fn set_signature(&mut self, signature: Signature) {
self.signature = signature
}
}
/// Type of the replicated value /// Type of the replicated value
/// These are labels for values in a record that is associated with `Pubkey` /// These are labels for values in a record that is associated with `Pubkey`
#[derive(PartialEq, Hash, Eq, Clone, Debug)] #[derive(PartialEq, Hash, Eq, Clone, Debug)]
@ -145,40 +110,57 @@ impl CrdsValueLabel {
} }
impl CrdsValue { impl CrdsValue {
pub fn new_unsigned(data: CrdsData) -> Self {
Self {
signature: Signature::default(),
data,
}
}
pub fn new_signed(data: CrdsData, keypair: &Keypair) -> Self {
let mut value = Self::new_unsigned(data);
value.sign(keypair);
value
}
/// Totally unsecure unverfiable wallclock of the node that generated this message /// Totally unsecure unverfiable wallclock of the node that generated this message
/// Latest wallclock is always picked. /// Latest wallclock is always picked.
/// This is used to time out push messages. /// This is used to time out push messages.
pub fn wallclock(&self) -> u64 { pub fn wallclock(&self) -> u64 {
match self { match &self.data {
CrdsValue::ContactInfo(contact_info) => contact_info.wallclock, CrdsData::ContactInfo(contact_info) => contact_info.wallclock,
CrdsValue::Vote(vote) => vote.wallclock, CrdsData::Vote(vote) => vote.wallclock,
CrdsValue::EpochSlots(vote) => vote.wallclock, CrdsData::EpochSlots(vote) => vote.wallclock,
}
}
pub fn pubkey(&self) -> Pubkey {
match &self.data {
CrdsData::ContactInfo(contact_info) => contact_info.id,
CrdsData::Vote(vote) => vote.from,
CrdsData::EpochSlots(slots) => slots.from,
} }
} }
pub fn label(&self) -> CrdsValueLabel { pub fn label(&self) -> CrdsValueLabel {
match self { match &self.data {
CrdsValue::ContactInfo(contact_info) => { CrdsData::ContactInfo(_) => CrdsValueLabel::ContactInfo(self.pubkey()),
CrdsValueLabel::ContactInfo(contact_info.pubkey()) CrdsData::Vote(_) => CrdsValueLabel::Vote(self.pubkey()),
} CrdsData::EpochSlots(_) => CrdsValueLabel::EpochSlots(self.pubkey()),
CrdsValue::Vote(vote) => CrdsValueLabel::Vote(vote.pubkey()),
CrdsValue::EpochSlots(slots) => CrdsValueLabel::EpochSlots(slots.pubkey()),
} }
} }
pub fn contact_info(&self) -> Option<&ContactInfo> { pub fn contact_info(&self) -> Option<&ContactInfo> {
match self { match &self.data {
CrdsValue::ContactInfo(contact_info) => Some(contact_info), CrdsData::ContactInfo(contact_info) => Some(contact_info),
_ => None, _ => None,
} }
} }
pub fn vote(&self) -> Option<&Vote> { pub fn vote(&self) -> Option<&Vote> {
match self { match &self.data {
CrdsValue::Vote(vote) => Some(vote), CrdsData::Vote(vote) => Some(vote),
_ => None, _ => None,
} }
} }
pub fn epoch_slots(&self) -> Option<&EpochSlots> { pub fn epoch_slots(&self) -> Option<&EpochSlots> {
match self { match &self.data {
CrdsValue::EpochSlots(slots) => Some(slots), CrdsData::EpochSlots(slots) => Some(slots),
_ => None, _ => None,
} }
} }
@ -197,48 +179,6 @@ impl CrdsValue {
} }
} }
impl Signable for CrdsValue {
fn sign(&mut self, keypair: &Keypair) {
match self {
CrdsValue::ContactInfo(contact_info) => contact_info.sign(keypair),
CrdsValue::Vote(vote) => vote.sign(keypair),
CrdsValue::EpochSlots(epoch_slots) => epoch_slots.sign(keypair),
};
}
fn verify(&self) -> bool {
match self {
CrdsValue::ContactInfo(contact_info) => contact_info.verify(),
CrdsValue::Vote(vote) => vote.verify(),
CrdsValue::EpochSlots(epoch_slots) => epoch_slots.verify(),
}
}
fn pubkey(&self) -> Pubkey {
match self {
CrdsValue::ContactInfo(contact_info) => contact_info.pubkey(),
CrdsValue::Vote(vote) => vote.pubkey(),
CrdsValue::EpochSlots(epoch_slots) => epoch_slots.pubkey(),
}
}
fn signable_data(&self) -> Cow<[u8]> {
unimplemented!()
}
fn get_signature(&self) -> Signature {
match self {
CrdsValue::ContactInfo(contact_info) => contact_info.get_signature(),
CrdsValue::Vote(vote) => vote.get_signature(),
CrdsValue::EpochSlots(epoch_slots) => epoch_slots.get_signature(),
}
}
fn set_signature(&mut self, _: Signature) {
unimplemented!()
}
}
#[cfg(test)] #[cfg(test)]
mod test { mod test {
use super::*; use super::*;
@ -263,17 +203,23 @@ mod test {
} }
#[test] #[test]
fn test_keys_and_values() { fn test_keys_and_values() {
let v = CrdsValue::ContactInfo(ContactInfo::default()); let v = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::default()));
assert_eq!(v.wallclock(), 0); assert_eq!(v.wallclock(), 0);
let key = v.clone().contact_info().unwrap().id; let key = v.clone().contact_info().unwrap().id;
assert_eq!(v.label(), CrdsValueLabel::ContactInfo(key)); assert_eq!(v.label(), CrdsValueLabel::ContactInfo(key));
let v = CrdsValue::Vote(Vote::new(&Pubkey::default(), test_tx(), 0)); let v =
CrdsValue::new_unsigned(CrdsData::Vote(Vote::new(&Pubkey::default(), test_tx(), 0)));
assert_eq!(v.wallclock(), 0); assert_eq!(v.wallclock(), 0);
let key = v.clone().vote().unwrap().from; let key = v.clone().vote().unwrap().from;
assert_eq!(v.label(), CrdsValueLabel::Vote(key)); assert_eq!(v.label(), CrdsValueLabel::Vote(key));
let v = CrdsValue::EpochSlots(EpochSlots::new(Pubkey::default(), 0, BTreeSet::new(), 0)); let v = CrdsValue::new_unsigned(CrdsData::EpochSlots(EpochSlots::new(
Pubkey::default(),
0,
BTreeSet::new(),
0,
)));
assert_eq!(v.wallclock(), 0); assert_eq!(v.wallclock(), 0);
let key = v.clone().epoch_slots().unwrap().from; let key = v.clone().epoch_slots().unwrap().from;
assert_eq!(v.label(), CrdsValueLabel::EpochSlots(key)); assert_eq!(v.label(), CrdsValueLabel::EpochSlots(key));
@ -282,13 +228,24 @@ mod test {
fn test_signature() { fn test_signature() {
let keypair = Keypair::new(); let keypair = Keypair::new();
let wrong_keypair = Keypair::new(); let wrong_keypair = Keypair::new();
let mut v = let mut v = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
CrdsValue::ContactInfo(ContactInfo::new_localhost(&keypair.pubkey(), timestamp())); &keypair.pubkey(),
timestamp(),
)));
verify_signatures(&mut v, &keypair, &wrong_keypair); verify_signatures(&mut v, &keypair, &wrong_keypair);
v = CrdsValue::Vote(Vote::new(&keypair.pubkey(), test_tx(), timestamp())); v = CrdsValue::new_unsigned(CrdsData::Vote(Vote::new(
&keypair.pubkey(),
test_tx(),
timestamp(),
)));
verify_signatures(&mut v, &keypair, &wrong_keypair); verify_signatures(&mut v, &keypair, &wrong_keypair);
let btreeset: BTreeSet<u64> = vec![1, 2, 3, 6, 8].into_iter().collect(); let btreeset: BTreeSet<u64> = vec![1, 2, 3, 6, 8].into_iter().collect();
v = CrdsValue::EpochSlots(EpochSlots::new(keypair.pubkey(), 0, btreeset, timestamp())); v = CrdsValue::new_unsigned(CrdsData::EpochSlots(EpochSlots::new(
keypair.pubkey(),
0,
btreeset,
timestamp(),
)));
verify_signatures(&mut v, &keypair, &wrong_keypair); verify_signatures(&mut v, &keypair, &wrong_keypair);
} }

43
core/tests/crds_gossip.rs
View File

@ -6,8 +6,8 @@ use solana_core::contact_info::ContactInfo;
use solana_core::crds_gossip::*; use solana_core::crds_gossip::*;
use solana_core::crds_gossip_error::CrdsGossipError; use solana_core::crds_gossip_error::CrdsGossipError;
use solana_core::crds_gossip_push::CRDS_GOSSIP_PUSH_MSG_TIMEOUT_MS; use solana_core::crds_gossip_push::CRDS_GOSSIP_PUSH_MSG_TIMEOUT_MS;
use solana_core::crds_value::CrdsValue;
use solana_core::crds_value::CrdsValueLabel; use solana_core::crds_value::CrdsValueLabel;
use solana_core::crds_value::{CrdsData, CrdsValue};
use solana_sdk::hash::hash; use solana_sdk::hash::hash;
use solana_sdk::pubkey::Pubkey; use solana_sdk::pubkey::Pubkey;
use solana_sdk::timing::timestamp; use solana_sdk::timing::timestamp;
@ -72,10 +72,16 @@ fn stakes(network: &Network) -> HashMap<Pubkey, u64> {
} }
fn star_network_create(num: usize) -> Network { fn star_network_create(num: usize) -> Network {
let entry = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let entry = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let mut network: HashMap<_, _> = (1..num) let mut network: HashMap<_, _> = (1..num)
.map(|_| { .map(|_| {
let new = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let new = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let id = new.label().pubkey(); let id = new.label().pubkey();
let mut node = CrdsGossip::default(); let mut node = CrdsGossip::default();
node.crds.insert(new.clone(), 0).unwrap(); node.crds.insert(new.clone(), 0).unwrap();
@ -93,14 +99,20 @@ fn star_network_create(num: usize) -> Network {
} }
fn rstar_network_create(num: usize) -> Network { fn rstar_network_create(num: usize) -> Network {
let entry = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let entry = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let mut origin = CrdsGossip::default(); let mut origin = CrdsGossip::default();
let id = entry.label().pubkey(); let id = entry.label().pubkey();
origin.crds.insert(entry.clone(), 0).unwrap(); origin.crds.insert(entry.clone(), 0).unwrap();
origin.set_self(&id); origin.set_self(&id);
let mut network: HashMap<_, _> = (1..num) let mut network: HashMap<_, _> = (1..num)
.map(|_| { .map(|_| {
let new = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let new = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let id = new.label().pubkey(); let id = new.label().pubkey();
let mut node = CrdsGossip::default(); let mut node = CrdsGossip::default();
node.crds.insert(new.clone(), 0).unwrap(); node.crds.insert(new.clone(), 0).unwrap();
@ -116,7 +128,10 @@ fn rstar_network_create(num: usize) -> Network {
fn ring_network_create(num: usize) -> Network { fn ring_network_create(num: usize) -> Network {
let mut network: HashMap<_, _> = (0..num) let mut network: HashMap<_, _> = (0..num)
.map(|_| { .map(|_| {
let new = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let new = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let id = new.label().pubkey(); let id = new.label().pubkey();
let mut node = CrdsGossip::default(); let mut node = CrdsGossip::default();
node.crds.insert(new.clone(), 0).unwrap(); node.crds.insert(new.clone(), 0).unwrap();
@ -147,7 +162,10 @@ fn connected_staked_network_create(stakes: &[u64]) -> Network {
let num = stakes.len(); let num = stakes.len();
let mut network: HashMap<_, _> = (0..num) let mut network: HashMap<_, _> = (0..num)
.map(|n| { .map(|n| {
let new = CrdsValue::ContactInfo(ContactInfo::new_localhost(&Pubkey::new_rand(), 0)); let new = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
&Pubkey::new_rand(),
0,
)));
let id = new.label().pubkey(); let id = new.label().pubkey();
let mut node = CrdsGossip::default(); let mut node = CrdsGossip::default();
node.crds.insert(new.clone(), 0).unwrap(); node.crds.insert(new.clone(), 0).unwrap();
@ -219,7 +237,11 @@ fn network_simulator(network: &mut Network, max_convergance: f64) {
.and_then(|v| v.contact_info().cloned()) .and_then(|v| v.contact_info().cloned())
.unwrap(); .unwrap();
m.wallclock = now; m.wallclock = now;
node.process_push_message(&Pubkey::default(), vec![CrdsValue::ContactInfo(m)], now); node.process_push_message(
&Pubkey::default(),
vec![CrdsValue::new_unsigned(CrdsData::ContactInfo(m))],
now,
);
}); });
// push for a bit // push for a bit
let (queue_size, bytes_tx) = network_run_push(network, start, end); let (queue_size, bytes_tx) = network_run_push(network, start, end);
@ -547,7 +569,10 @@ fn test_prune_errors() {
let prune_pubkey = Pubkey::new(&[2; 32]); let prune_pubkey = Pubkey::new(&[2; 32]);
crds_gossip crds_gossip
.crds .crds
.insert(CrdsValue::ContactInfo(ci.clone()), 0) .insert(
CrdsValue::new_unsigned(CrdsData::ContactInfo(ci.clone())),
0,
)
.unwrap(); .unwrap();
crds_gossip.refresh_push_active_set(&HashMap::new()); crds_gossip.refresh_push_active_set(&HashMap::new());
let now = timestamp(); let now = timestamp();