From fdfe1d508b44f6fa5687b4bf82ac1bce25f553c8 Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Thu, 16 Mar 2023 15:04:03 +0100 Subject: [PATCH] node: Use deserializer when decoding messages For whatever reason, we weren't using the deserializer for message decoding. This meant that if a message came in two pieces, it would cause decoding problems, since the first piece would be discarded. --- radicle-node/src/deserializer.rs | 45 +++++++++++++++++++++++++++++++ radicle-node/src/wire/message.rs | 2 +- radicle-node/src/wire/protocol.rs | 27 ++++++++++--------- 3 files changed, 60 insertions(+), 14 deletions(-) diff --git a/radicle-node/src/deserializer.rs b/radicle-node/src/deserializer.rs index 8805d1b1..51875beb 100644 --- a/radicle-node/src/deserializer.rs +++ b/radicle-node/src/deserializer.rs @@ -13,6 +13,12 @@ pub struct Deserializer { item: PhantomData, } +impl Default for Deserializer { + fn default() -> Self { + Self::new(wire::Size::MAX as usize + 1) + } +} + impl From> for Deserializer { fn from(unparsed: Vec) -> Self { Self { @@ -51,6 +57,16 @@ impl Deserializer { Err(err) => Err(err), } } + + /// Drain the unparsed buffer. + pub fn unparsed(&mut self) -> impl ExactSizeIterator + '_ { + self.unparsed.drain(..) + } + + /// Return whether there are unparsed bytes. + pub fn is_empty(&self) -> bool { + self.unparsed.is_empty() + } } impl io::Write for Deserializer { @@ -78,9 +94,38 @@ mod test { use super::*; use qcheck_macros::quickcheck; + use crate::test::assert_matches; + const MSG_HELLO: &[u8] = &[5, b'h', b'e', b'l', b'l', b'o']; const MSG_BYE: &[u8] = &[3, b'b', b'y', b'e']; + #[test] + fn test_decode_next() { + let mut decoder = Deserializer::::new(8); + + decoder.input(&[3, b'b']); + assert_matches!(decoder.deserialize_next(), Ok(None)); + assert_eq!(decoder.unparsed.len(), 2); + + decoder.input(&[b'y']); + assert_matches!(decoder.deserialize_next(), Ok(None)); + assert_eq!(decoder.unparsed.len(), 3); + + decoder.input(&[b'e']); + assert_matches!(decoder.deserialize_next(), Ok(Some(s)) if s.as_str() == "bye"); + assert_eq!(decoder.unparsed.len(), 0); + assert!(decoder.is_empty()); + } + + #[test] + fn test_unparsed() { + let mut decoder = Deserializer::::new(8); + + decoder.input(&[3, b'b', b'y']); + assert_eq!(decoder.unparsed().collect::>(), vec![3, b'b', b'y']); + assert!(decoder.is_empty()); + } + #[quickcheck] fn prop_decode_next(chunk_size: usize) { let mut bytes = vec![]; diff --git a/radicle-node/src/wire/message.rs b/radicle-node/src/wire/message.rs index 766ab130..59fabf3a 100644 --- a/radicle-node/src/wire/message.rs +++ b/radicle-node/src/wire/message.rs @@ -365,7 +365,7 @@ impl wire::Encode for ZeroBytes { fn encode(&self, writer: &mut W) -> Result { let mut n = (self.len() as u16).encode(writer)?; for _ in 0..self.len() { - n += 0u8.encode(writer)? + n += 0u8.encode(writer)?; } Ok(n) } diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index 29cb6048..e0568bf9 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -26,9 +26,10 @@ use radicle::node::{routing, NodeId}; use radicle::storage::WriteStorage; use crate::crypto::Signer; +use crate::prelude::Deserializer; use crate::service::reactor::{Fetch, Io}; use crate::service::{session, DisconnectReason, Message, Service}; -use crate::wire::{Decode, Encode, Error}; +use crate::wire::{Encode, Error}; use crate::worker; use crate::worker::{Task, TaskResult}; use crate::Link; @@ -79,7 +80,7 @@ enum Peer { Connected { link: Link, id: NodeId, - inbox: VecDeque, + inbox: Deserializer, }, /// The peer was scheduled for disconnection. Once the transport is handed over /// by the reactor, we can consider it disconnected. @@ -93,7 +94,7 @@ enum Peer { fetch: Fetch, link: Link, id: NodeId, - inbox: VecDeque, + inbox: Vec, }, /// The peer is now upgraded and we are in control of the socket. Upgraded { link: Link, id: NodeId }, @@ -151,7 +152,7 @@ impl Peer { *self = Self::Connected { link, id, - inbox: VecDeque::new(), + inbox: Deserializer::default(), }; link } else if let Self::Outbound { id: expected } = self { @@ -161,7 +162,7 @@ impl Peer { *self = Self::Connected { link, id, - inbox: VecDeque::new(), + inbox: Deserializer::default(), }; link } else { @@ -195,7 +196,7 @@ impl Peer { fetch, id: *id, link: *link, - inbox: inbox.clone(), + inbox: inbox.unparsed().collect(), }; } else { panic!("Peer::upgrading: session is not fully connected"); @@ -232,7 +233,7 @@ impl Peer { *self = Self::Connected { id: *id, link: *link, - inbox: VecDeque::new(), + inbox: Deserializer::default(), }; } else { panic!("Peer::downgrade: can't downgrade if not in upgraded state"); @@ -508,12 +509,12 @@ where } SessionEvent::Data(data) => { if let Some(Peer::Connected { id, inbox, .. }) = self.peers.get_mut(&fd) { - inbox.extend(data); + inbox.input(&data); loop { - match Message::decode(inbox) { - Ok(msg) => self.service.received_message(*id, msg), - Err(err) if err.is_eof() => { + match inbox.deserialize_next() { + Ok(Some(msg)) => self.service.received_message(*id, msg), + Ok(None) => { // Buffer is empty, or message isn't complete. break; } @@ -522,7 +523,7 @@ where if let Error::UnknownMessageType(t) = e { let mut leftover = t.to_be_bytes().to_vec(); - leftover.extend(inbox.drain(..)); + leftover.extend(inbox.unparsed()); if let Ok(header) = str::from_utf8(&leftover[..worker::pktline::HEADER_LEN]) @@ -538,7 +539,7 @@ where } if !inbox.is_empty() { - log::debug!(target: "wire", "Dropping read buffer with {} bytes", inbox.len()); + log::debug!(target: "wire", "Dropping read buffer for {id} with {} bytes", inbox.unparsed().count()); } self.disconnect( fd,