From 5a4c2cb37ef3db97d6b8de76059dc3e99c61d647 Mon Sep 17 00:00:00 2001 From: John Smith Date: Mon, 27 Feb 2023 21:29:51 -0500 Subject: [PATCH] refactor --- veilid-core/src/routing_table/privacy.rs | 6 +- .../rpc_processor/coders/operations/answer.rs | 8 +-- .../coders/operations/operation_find_block.rs | 4 +- .../coders/operations/operation_get_value.rs | 14 +++-- .../coders/operations/operation_set_value.rs | 14 +++-- .../operations/operation_supply_block.rs | 9 +-- .../operations/operation_value_changed.rs | 53 +++++++++++++++-- .../operations/operation_watch_value.rs | 59 +++++++++++++++++-- 8 files changed, 135 insertions(+), 32 deletions(-) diff --git a/veilid-core/src/routing_table/privacy.rs b/veilid-core/src/routing_table/privacy.rs index bec7cc91..e4aa4c34 100644 --- a/veilid-core/src/routing_table/privacy.rs +++ b/veilid-core/src/routing_table/privacy.rs @@ -30,7 +30,7 @@ impl RouteNode { match self { RouteNode::NodeId(id) => { // - routing_table.lookup_node_ref(TypedKey::new(crypto_kind, id)) + routing_table.lookup_node_ref(TypedKey::new(crypto_kind, *id)) } RouteNode::PeerInfo(pi) => { // @@ -46,10 +46,10 @@ impl RouteNode { pub fn describe(&self, crypto_kind: CryptoKind) -> String { match self { RouteNode::NodeId(id) => { - format!("{}", TypedKey::new(crypto_kind, id)) + format!("{}", TypedKey::new(crypto_kind, *id)) } RouteNode::PeerInfo(pi) => match pi.node_ids.get(crypto_kind) { - Some(id) => format!("{}", TypedKey::new(crypto_kind, id)), + Some(id) => format!("{}", id), None => { format!("({})?{}", crypto_kind, pi.node_ids) } diff --git a/veilid-core/src/rpc_processor/coders/operations/answer.rs b/veilid-core/src/rpc_processor/coders/operations/answer.rs index 7d068adf..d940c736 100644 --- a/veilid-core/src/rpc_processor/coders/operations/answer.rs +++ b/veilid-core/src/rpc_processor/coders/operations/answer.rs @@ -84,22 +84,22 @@ impl RPCAnswerDetail { } veilid_capnp::answer::detail::GetValueA(r) => { let op_reader = r.map_err(RPCError::protocol)?; - let out = RPCOperationGetValueA::decode(&op_reader)?; + let out = RPCOperationGetValueA::decode(&op_reader, crypto)?; RPCAnswerDetail::GetValueA(out) } veilid_capnp::answer::detail::SetValueA(r) => { let op_reader = r.map_err(RPCError::protocol)?; - let out = RPCOperationSetValueA::decode(&op_reader)?; + let out = RPCOperationSetValueA::decode(&op_reader, crypto)?; RPCAnswerDetail::SetValueA(out) } veilid_capnp::answer::detail::WatchValueA(r) => { let op_reader = r.map_err(RPCError::protocol)?; - let out = RPCOperationWatchValueA::decode(&op_reader)?; + let out = RPCOperationWatchValueA::decode(&op_reader, crypto)?; RPCAnswerDetail::WatchValueA(out) } veilid_capnp::answer::detail::SupplyBlockA(r) => { let op_reader = r.map_err(RPCError::protocol)?; - let out = RPCOperationSupplyBlockA::decode(&op_reader)?; + let out = RPCOperationSupplyBlockA::decode(&op_reader, crypto)?; RPCAnswerDetail::SupplyBlockA(out) } veilid_capnp::answer::detail::FindBlockA(r) => { diff --git a/veilid-core/src/rpc_processor/coders/operations/operation_find_block.rs b/veilid-core/src/rpc_processor/coders/operations/operation_find_block.rs index 4f15e6f0..b5ecab45 100644 --- a/veilid-core/src/rpc_processor/coders/operations/operation_find_block.rs +++ b/veilid-core/src/rpc_processor/coders/operations/operation_find_block.rs @@ -10,7 +10,7 @@ impl RPCOperationFindBlockQ { reader: &veilid_capnp::operation_find_block_q::Reader, ) -> Result { let bi_reader = reader.get_block_id().map_err(RPCError::protocol)?; - let block_id = decode_key256(&bi_reader); + let block_id = decode_typed_key(&bi_reader)?; Ok(RPCOperationFindBlockQ { block_id }) } @@ -19,7 +19,7 @@ impl RPCOperationFindBlockQ { builder: &mut veilid_capnp::operation_find_block_q::Builder, ) -> Result<(), RPCError> { let mut bi_builder = builder.reborrow().init_block_id(); - encode_key256(&self.block_id, &mut bi_builder)?; + encode_typed_key(&self.block_id, &mut bi_builder); Ok(()) } diff --git a/veilid-core/src/rpc_processor/coders/operations/operation_get_value.rs b/veilid-core/src/rpc_processor/coders/operations/operation_get_value.rs index f9fc9959..5db1f993 100644 --- a/veilid-core/src/rpc_processor/coders/operations/operation_get_value.rs +++ b/veilid-core/src/rpc_processor/coders/operations/operation_get_value.rs @@ -2,7 +2,8 @@ use super::*; #[derive(Debug, Clone)] pub struct RPCOperationGetValueQ { - pub key: ValueKey, + pub key: TypedKey, + pub subkey: ValueSubkey, } impl RPCOperationGetValueQ { @@ -10,15 +11,17 @@ impl RPCOperationGetValueQ { reader: &veilid_capnp::operation_get_value_q::Reader, ) -> Result { let k_reader = reader.get_key().map_err(RPCError::protocol)?; - let key = decode_value_key(&k_reader)?; - Ok(RPCOperationGetValueQ { key }) + let key = decode_typed_key(&k_reader)?; + let subkey = reader.get_subkey(); + Ok(RPCOperationGetValueQ { key, subkey }) } pub fn encode( &self, builder: &mut veilid_capnp::operation_get_value_q::Builder, ) -> Result<(), RPCError> { let mut k_builder = builder.reborrow().init_key(); - encode_value_key(&self.key, &mut k_builder)?; + encode_typed_key(&self.key, &mut k_builder); + builder.set_subkey(self.subkey); Ok(()) } } @@ -32,6 +35,7 @@ pub enum RPCOperationGetValueA { impl RPCOperationGetValueA { pub fn decode( reader: &veilid_capnp::operation_get_value_a::Reader, + crypto: Crypto, ) -> Result { match reader.which().map_err(RPCError::protocol)? { veilid_capnp::operation_get_value_a::Which::Data(r) => { @@ -47,7 +51,7 @@ impl RPCOperationGetValueA { .map_err(RPCError::map_internal("too many peers"))?, ); for p in peers_reader.iter() { - let peer_info = decode_peer_info(&p)?; + let peer_info = decode_peer_info(&p, crypto.clone())?; peers.push(peer_info); } diff --git a/veilid-core/src/rpc_processor/coders/operations/operation_set_value.rs b/veilid-core/src/rpc_processor/coders/operations/operation_set_value.rs index 23a34421..4f1c5763 100644 --- a/veilid-core/src/rpc_processor/coders/operations/operation_set_value.rs +++ b/veilid-core/src/rpc_processor/coders/operations/operation_set_value.rs @@ -2,7 +2,8 @@ use super::*; #[derive(Debug, Clone)] pub struct RPCOperationSetValueQ { - pub key: ValueKey, + pub key: TypedKey, + pub subkey: ValueSubkey, pub value: ValueData, } @@ -11,17 +12,19 @@ impl RPCOperationSetValueQ { reader: &veilid_capnp::operation_set_value_q::Reader, ) -> Result { let k_reader = reader.get_key().map_err(RPCError::protocol)?; - let key = decode_value_key(&k_reader)?; + let key = decode_typed_key(&k_reader)?; + let subkey = reader.get_subkey(); let v_reader = reader.get_value().map_err(RPCError::protocol)?; let value = decode_value_data(&v_reader)?; - Ok(RPCOperationSetValueQ { key, value }) + Ok(RPCOperationSetValueQ { key, subkey, value }) } pub fn encode( &self, builder: &mut veilid_capnp::operation_set_value_q::Builder, ) -> Result<(), RPCError> { let mut k_builder = builder.reborrow().init_key(); - encode_value_key(&self.key, &mut k_builder)?; + encode_typed_key(&self.key, &mut k_builder); + builder.set_subkey(self.subkey); let mut v_builder = builder.reborrow().init_value(); encode_value_data(&self.value, &mut v_builder)?; Ok(()) @@ -37,6 +40,7 @@ pub enum RPCOperationSetValueA { impl RPCOperationSetValueA { pub fn decode( reader: &veilid_capnp::operation_set_value_a::Reader, + crypto: Crypto, ) -> Result { match reader.which().map_err(RPCError::protocol)? { veilid_capnp::operation_set_value_a::Which::Data(r) => { @@ -52,7 +56,7 @@ impl RPCOperationSetValueA { .map_err(RPCError::map_internal("too many peers"))?, ); for p in peers_reader.iter() { - let peer_info = decode_peer_info(&p)?; + let peer_info = decode_peer_info(&p, crypto.clone())?; peers.push(peer_info); } diff --git a/veilid-core/src/rpc_processor/coders/operations/operation_supply_block.rs b/veilid-core/src/rpc_processor/coders/operations/operation_supply_block.rs index 2e606661..f68de596 100644 --- a/veilid-core/src/rpc_processor/coders/operations/operation_supply_block.rs +++ b/veilid-core/src/rpc_processor/coders/operations/operation_supply_block.rs @@ -2,7 +2,7 @@ use super::*; #[derive(Debug, Clone)] pub struct RPCOperationSupplyBlockQ { - pub block_id: PublicKey, + pub block_id: TypedKey, } impl RPCOperationSupplyBlockQ { @@ -10,7 +10,7 @@ impl RPCOperationSupplyBlockQ { reader: &veilid_capnp::operation_supply_block_q::Reader, ) -> Result { let bi_reader = reader.get_block_id().map_err(RPCError::protocol)?; - let block_id = decode_key256(&bi_reader); + let block_id = decode_typed_key(&bi_reader)?; Ok(RPCOperationSupplyBlockQ { block_id }) } @@ -19,7 +19,7 @@ impl RPCOperationSupplyBlockQ { builder: &mut veilid_capnp::operation_supply_block_q::Builder, ) -> Result<(), RPCError> { let mut bi_builder = builder.reborrow().init_block_id(); - encode_key256(&self.block_id, &mut bi_builder)?; + encode_typed_key(&self.block_id, &mut bi_builder); Ok(()) } @@ -34,6 +34,7 @@ pub enum RPCOperationSupplyBlockA { impl RPCOperationSupplyBlockA { pub fn decode( reader: &veilid_capnp::operation_supply_block_a::Reader, + crypto: Crypto, ) -> Result { match reader.which().map_err(RPCError::protocol)? { veilid_capnp::operation_supply_block_a::Which::Expiration(r) => { @@ -48,7 +49,7 @@ impl RPCOperationSupplyBlockA { .map_err(RPCError::map_internal("too many peers"))?, ); for p in peers_reader.iter() { - let peer_info = decode_peer_info(&p)?; + let peer_info = decode_peer_info(&p, crypto.clone())?; peers.push(peer_info); } diff --git a/veilid-core/src/rpc_processor/coders/operations/operation_value_changed.rs b/veilid-core/src/rpc_processor/coders/operations/operation_value_changed.rs index 3d1c08cf..86bc9c69 100644 --- a/veilid-core/src/rpc_processor/coders/operations/operation_value_changed.rs +++ b/veilid-core/src/rpc_processor/coders/operations/operation_value_changed.rs @@ -2,7 +2,9 @@ use super::*; #[derive(Debug, Clone)] pub struct RPCOperationValueChanged { - pub key: ValueKey, + pub key: TypedKey, + pub subkeys: Vec, + pub count: u32, pub value: ValueData, } @@ -11,17 +13,60 @@ impl RPCOperationValueChanged { reader: &veilid_capnp::operation_value_changed::Reader, ) -> Result { let k_reader = reader.get_key().map_err(RPCError::protocol)?; - let key = decode_value_key(&k_reader)?; + let key = decode_typed_key(&k_reader)?; + + let sk_reader = reader.get_subkeys().map_err(RPCError::protocol)?; + let mut subkeys = Vec::::with_capacity( + sk_reader + .len() + .try_into() + .map_err(RPCError::map_protocol("too many subkey ranges"))?, + ); + for skr in sk_reader.iter() { + let vskr = (skr.get_start(), skr.get_end()); + if vskr.0 > vskr.1 { + return Err(RPCError::protocol("invalid subkey range")); + } + if let Some(lvskr) = subkeys.last() { + if lvskr.1 >= vskr.0 { + return Err(RPCError::protocol( + "subkey range out of order or not merged", + )); + } + } + subkeys.push(vskr); + } + let count = reader.get_count(); let v_reader = reader.get_value().map_err(RPCError::protocol)?; let value = decode_value_data(&v_reader)?; - Ok(RPCOperationValueChanged { key, value }) + Ok(RPCOperationValueChanged { + key, + subkeys, + count, + value, + }) } pub fn encode( &self, builder: &mut veilid_capnp::operation_value_changed::Builder, ) -> Result<(), RPCError> { let mut k_builder = builder.reborrow().init_key(); - encode_value_key(&self.key, &mut k_builder)?; + encode_typed_key(&self.key, &mut k_builder); + + let mut sk_builder = builder.reborrow().init_subkeys( + self.subkeys + .len() + .try_into() + .map_err(RPCError::map_internal("invalid subkey range list length"))?, + ); + for (i, skr) in self.subkeys.iter().enumerate() { + let mut skr_builder = sk_builder.reborrow().get(i as u32); + skr_builder.set_start(skr.0); + skr_builder.set_end(skr.1); + } + + builder.set_count(self.count); + let mut v_builder = builder.reborrow().init_value(); encode_value_data(&self.value, &mut v_builder)?; Ok(()) diff --git a/veilid-core/src/rpc_processor/coders/operations/operation_watch_value.rs b/veilid-core/src/rpc_processor/coders/operations/operation_watch_value.rs index 00c0199c..7812b98b 100644 --- a/veilid-core/src/rpc_processor/coders/operations/operation_watch_value.rs +++ b/veilid-core/src/rpc_processor/coders/operations/operation_watch_value.rs @@ -2,7 +2,10 @@ use super::*; #[derive(Debug, Clone)] pub struct RPCOperationWatchValueQ { - pub key: ValueKey, + pub key: TypedKey, + pub subkeys: Vec, + pub expiration: u64, + pub count: u32, } impl RPCOperationWatchValueQ { @@ -10,15 +13,60 @@ impl RPCOperationWatchValueQ { reader: &veilid_capnp::operation_watch_value_q::Reader, ) -> Result { let k_reader = reader.get_key().map_err(RPCError::protocol)?; - let key = decode_value_key(&k_reader)?; - Ok(RPCOperationWatchValueQ { key }) + let key = decode_typed_key(&k_reader)?; + + let sk_reader = reader.get_subkeys().map_err(RPCError::protocol)?; + let mut subkeys = Vec::::with_capacity( + sk_reader + .len() + .try_into() + .map_err(RPCError::map_protocol("too many subkey ranges"))?, + ); + for skr in sk_reader.iter() { + let vskr = (skr.get_start(), skr.get_end()); + if vskr.0 > vskr.1 { + return Err(RPCError::protocol("invalid subkey range")); + } + if let Some(lvskr) = subkeys.last() { + if lvskr.1 >= vskr.0 { + return Err(RPCError::protocol( + "subkey range out of order or not merged", + )); + } + } + subkeys.push(vskr); + } + + let expiration = reader.get_expiration(); + let count = reader.get_count(); + + Ok(RPCOperationWatchValueQ { + key, + subkeys, + expiration, + count, + }) } pub fn encode( &self, builder: &mut veilid_capnp::operation_watch_value_q::Builder, ) -> Result<(), RPCError> { let mut k_builder = builder.reborrow().init_key(); - encode_value_key(&self.key, &mut k_builder)?; + encode_typed_key(&self.key, &mut k_builder); + + let mut sk_builder = builder.reborrow().init_subkeys( + self.subkeys + .len() + .try_into() + .map_err(RPCError::map_internal("invalid subkey range list length"))?, + ); + for (i, skr) in self.subkeys.iter().enumerate() { + let mut skr_builder = sk_builder.reborrow().get(i as u32); + skr_builder.set_start(skr.0); + skr_builder.set_end(skr.1); + } + builder.set_expiration(self.expiration); + builder.set_count(self.count); Ok(()) } } @@ -32,6 +80,7 @@ pub struct RPCOperationWatchValueA { impl RPCOperationWatchValueA { pub fn decode( reader: &veilid_capnp::operation_watch_value_a::Reader, + crypto: Crypto, ) -> Result { let expiration = reader.get_expiration(); let peers_reader = reader.get_peers().map_err(RPCError::protocol)?; @@ -42,7 +91,7 @@ impl RPCOperationWatchValueA { .map_err(RPCError::map_internal("too many peers"))?, ); for p in peers_reader.iter() { - let peer_info = decode_peer_info(&p)?; + let peer_info = decode_peer_info(&p, crypto.clone())?; peers.push(peer_info); }