From 89b9eb53b76adf9e38e0da5a15bf810f8d72a3c8 Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Fri, 7 Apr 2023 11:36:28 +0200 Subject: [PATCH] node: Implement protocol multiplexing To improve the reliability and flexibility of the protocol, we introduce multiplexing over peer connections. This involves a new `Frame` type that carries a stream-id and payload. Three stream types are made available: 1. Control 2. Gossip 3. Git This change brings the following improvements: * Removed need to queue fetch requests * Removed need to queue gossip messages * Removed need for `Fetch` and `FetchOk` messages * Service doesn't need to know about inbound fetches * Removed one round-trip for fetch negotiation * Removed special `done` git packet * Removed session logic and state around Git/Fetch protocols * Removed code around upgrading/downgrading transport * Worker in responder mode is able to process any number of fetches * Connections support any number of concurrent fetches We had to introduce a few extra things however to make it all work: * A `VarInt` type for variable-length integers, since we want the frames to be lightweight. * A custom "tunnel" implementation, since we couldn't use the existing one anymore. Overall the change removes more complexity than it adds, while improving the protocol along the way. --- radicle-cli/tests/commands.rs | 6 +- radicle-node/src/deserializer.rs | 2 +- radicle-node/src/runtime.rs | 14 +- radicle-node/src/runtime/handle.rs | 51 +-- radicle-node/src/service.rs | 278 ++++----------- radicle-node/src/service/message.rs | 8 - radicle-node/src/service/reactor.rs | 124 ++----- radicle-node/src/service/session.rs | 163 ++------- radicle-node/src/test/arbitrary.rs | 1 - radicle-node/src/test/environment.rs | 2 +- radicle-node/src/test/simulator.rs | 72 ++-- radicle-node/src/tests.rs | 184 +--------- radicle-node/src/wire.rs | 9 + radicle-node/src/wire/frame.rs | 359 +++++++++++++++++++ radicle-node/src/wire/message.rs | 20 -- radicle-node/src/wire/protocol.rs | 494 +++++++++++++++----------- radicle-node/src/wire/varint.rs | 225 ++++++++++++ radicle-node/src/worker.rs | 496 ++++++++++++--------------- radicle-node/src/worker/channels.rs | 136 ++++++++ radicle-node/src/worker/tunnel.rs | 80 +++++ 20 files changed, 1522 insertions(+), 1202 deletions(-) create mode 100644 radicle-node/src/wire/frame.rs create mode 100644 radicle-node/src/wire/varint.rs create mode 100644 radicle-node/src/worker/channels.rs create mode 100644 radicle-node/src/worker/tunnel.rs diff --git a/radicle-cli/tests/commands.rs b/radicle-cli/tests/commands.rs index 6977755d..d69c8a60 100644 --- a/radicle-cli/tests/commands.rs +++ b/radicle-cli/tests/commands.rs @@ -312,10 +312,12 @@ fn rad_clone() { let working = environment.tmp().join("working"); // Setup a test project. - let _ = alice.project("heartwood", "Radicle Heartwood Protocol & Stack"); + let acme = alice.project("heartwood", "Radicle Heartwood Protocol & Stack"); - let alice = alice.spawn(Config::default()); + let mut alice = alice.spawn(Config::default()); let mut bob = bob.spawn(Config::default()); + // Prevent Alice from fetching Bob's fork, as we're not testing that and it may cause errors. + alice.handle.track_repo(acme, Scope::Trusted).unwrap(); bob.connect(&alice).converge([&alice]); diff --git a/radicle-node/src/deserializer.rs b/radicle-node/src/deserializer.rs index 51875beb..7acbfbbb 100644 --- a/radicle-node/src/deserializer.rs +++ b/radicle-node/src/deserializer.rs @@ -13,7 +13,7 @@ pub struct Deserializer { item: PhantomData, } -impl Default for Deserializer { +impl Default for Deserializer { fn default() -> Self { Self::new(wire::Size::MAX as usize + 1) } diff --git a/radicle-node/src/runtime.rs b/radicle-node/src/runtime.rs index 33f8641c..ecead7ec 100644 --- a/radicle-node/src/runtime.rs +++ b/radicle-node/src/runtime.rs @@ -96,24 +96,24 @@ impl Emitter { } /// Holds join handles to the client threads, as well as a client handle. -pub struct Runtime { +pub struct Runtime { pub id: NodeId, pub home: Home, pub control: UnixListener, - pub handle: Handle, + pub handle: Handle, pub storage: Storage, - pub reactor: Reactor>, + pub reactor: Reactor, pub daemon: net::SocketAddr, pub pool: worker::Pool, pub local_addrs: Vec, pub signals: chan::Receiver<()>, } -impl Runtime { +impl Runtime { /// Initialize the runtime. /// /// This function spawns threads. - pub fn init( + pub fn init( home: Home, config: service::Config, listen: Vec, @@ -121,7 +121,7 @@ impl Runtime { daemon: net::SocketAddr, signals: chan::Receiver<()>, signer: G, - ) -> Result, Error> + ) -> Result where G: Ecdh + Clone, { @@ -160,7 +160,7 @@ impl Runtime { emitter.clone(), ); - let (worker_send, worker_recv) = chan::unbounded::>(); + let (worker_send, worker_recv) = chan::unbounded::(); let mut wire = Wire::new(service, worker_send, signer, proxy, clock); let mut local_addrs = Vec::new(); diff --git a/radicle-node/src/runtime/handle.rs b/radicle-node/src/runtime/handle.rs index 0a6ace63..a64fb101 100644 --- a/radicle-node/src/runtime/handle.rs +++ b/radicle-node/src/runtime/handle.rs @@ -5,11 +5,9 @@ use std::sync::Arc; use std::{fmt, io, time}; use crossbeam_channel as chan; -use cyphernet::Ecdh; use radicle::node::Seeds; use thiserror::Error; -use crate::crypto::Signer; use crate::identity::Id; use crate::node::{Command, FetchResult}; use crate::profile::Home; @@ -20,6 +18,7 @@ use crate::service::Event; use crate::service::{CommandError, QueryState}; use crate::service::{NodeId, Sessions}; use crate::wire; +use crate::wire::StreamId; use crate::worker::TaskResult; /// An error resulting from a handle method. @@ -27,7 +26,7 @@ use crate::worker::TaskResult; pub enum Error { /// The command channel is no longer connected. #[error("command channel is not connected")] - NotConnected, + ChannelDisconnected, /// The command returned an error. #[error("command failed: {0}")] Command(#[from] CommandError), @@ -41,7 +40,7 @@ pub enum Error { impl From for Error { fn from(_: chan::RecvError) -> Self { - Self::NotConnected + Self::ChannelDisconnected } } @@ -49,20 +48,20 @@ impl From for Error { fn from(err: chan::RecvTimeoutError) -> Self { match err { chan::RecvTimeoutError::Timeout => Self::Timeout, - chan::RecvTimeoutError::Disconnected => Self::NotConnected, + chan::RecvTimeoutError::Disconnected => Self::ChannelDisconnected, } } } impl From> for Error { fn from(_: chan::SendError) -> Self { - Self::NotConnected + Self::ChannelDisconnected } } -pub struct Handle { +pub struct Handle { pub(crate) home: Home, - pub(crate) controller: reactor::Controller>, + pub(crate) controller: reactor::Controller, /// Whether a shutdown was initiated or not. Prevents attempting to shutdown twice. shutdown: Arc, @@ -117,20 +116,20 @@ impl Events { } } -impl Handle { +impl Handle { /// Subscribe to events stream. pub fn events(&self) -> Events { Events(self.emitter.subscribe()) } } -impl fmt::Debug for Handle { +impl fmt::Debug for Handle { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Handle").field("home", &self.home).finish() } } -impl Clone for Handle { +impl Clone for Handle { fn clone(&self) -> Self { Self { home: self.home.clone(), @@ -141,10 +140,10 @@ impl Clone for Handle { } } -impl Handle { +impl Handle { pub fn new( home: Home, - controller: reactor::Controller>, + controller: reactor::Controller, emitter: Emitter, ) -> Self { Self { @@ -155,22 +154,20 @@ impl Handle { } } - pub fn worker_result(&mut self, resp: TaskResult) -> Result<(), Error> { - match self.controller.cmd(wire::Control::Worker(resp)) { - Ok(()) => {} - Err(err) if err.kind() == io::ErrorKind::BrokenPipe => return Err(Error::NotConnected), - Err(err) => return Err(err.into()), - } - Ok(()) + pub fn worker_result(&mut self, result: TaskResult) -> Result<(), io::Error> { + self.controller.cmd(wire::Control::Worker(result)) } - fn command(&self, cmd: service::Command) -> Result<(), Error> { - self.controller.cmd(wire::Control::User(cmd))?; - Ok(()) + pub fn flush(&mut self, remote: NodeId, stream: StreamId) -> Result<(), io::Error> { + self.controller.cmd(wire::Control::Flush { remote, stream }) + } + + fn command(&self, cmd: service::Command) -> Result<(), io::Error> { + self.controller.cmd(wire::Control::User(cmd)) } } -impl radicle::node::Handle for Handle { +impl radicle::node::Handle for Handle { type Sessions = Sessions; type Error = Error; @@ -222,10 +219,12 @@ impl radicle::node::Handle for Handle { fn announce_refs(&mut self, id: Id) -> Result<(), Error> { self.command(service::Command::AnnounceRefs(id)) + .map_err(Error::from) } fn announce_inventory(&mut self) -> Result<(), Error> { self.command(service::Command::AnnounceInventory) + .map_err(Error::from) } fn sync_inventory(&mut self) -> Result { @@ -265,6 +264,8 @@ impl radicle::node::Handle for Handle { .and_then(|sock| Command::SHUTDOWN.to_writer(sock)) .ok(); - self.controller.shutdown().map_err(|_| Error::NotConnected) + self.controller + .shutdown() + .map_err(|_| Error::ChannelDisconnected) } } diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index b9856a53..ba1a704a 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -32,8 +32,6 @@ use crate::prelude::*; use crate::runtime::Emitter; use crate::service::message::{Announcement, AnnouncementMessage, Ping}; use crate::service::message::{NodeAnnouncement, RefsAnnouncement}; -use crate::service::reactor::FetchDirection; -use crate::service::session::GossipState; use crate::service::tracking::Scope; use crate::storage; use crate::storage::{Namespaces, ReadStorage}; @@ -44,7 +42,6 @@ use crate::Link; pub use crate::node::NodeId; pub use crate::service::config::{Config, Network}; pub use crate::service::message::{Message, ZeroBytes}; -pub use crate::service::reactor::Fetch; pub use crate::service::session::Session; use self::gossip::Gossip; @@ -551,13 +548,12 @@ where let seed = session.id; match session.fetch(rid) { - session::FetchResult::Ready(fetch) => { + session::FetchResult::Ready => { debug!(target: "service", "Fetch initiated for {rid} with {seed}.."); match self.tracking.namespaces_for(&self.storage, &rid) { - Ok(ns) => { - self.reactor.write(session, fetch); - session.to_requesting(rid, ns); + Ok(namespaces) => { + self.reactor.fetch(session, rid, namespaces); } Err(err) => { error!(target: "service", "Error getting namespaces for {rid}: {err}"); @@ -571,19 +567,8 @@ where } }; } - session::FetchResult::AlreadyFetching(other) => { - if other == rid { - debug!(target: "service", "Ignoring redundant attempt to fetch {rid} from {from}"); - } else { - // If we can't fetch, it's because we're already fetching from - // this peer. So we need to queue the request. - // TODO: consider to find another peer. - debug!( - target: "service", - "Queueing fetch for {rid} from {from}: another fetch is ongoing" - ); - session.queue_fetch(rid); - } + session::FetchResult::AlreadyFetching => { + debug!(target: "service", "Ignoring redundant attempt to fetch {rid} from {from}"); } session::FetchResult::NotConnected => { error!(target: "service", "Unable to fetch {rid} from peer {seed}: peer is not connected"); @@ -591,75 +576,74 @@ where } } - pub fn fetched(&mut self, fetch: Fetch, result: Result, FetchError>) { - let remote = fetch.remote; - let rid = fetch.rid; + pub fn fetched( + &mut self, + rid: Id, + namespaces: Namespaces, + remote: NodeId, + result: Result, FetchError>, + ) { + let result = match result { + Ok(updated) => { + log::debug!(target: "service", "Fetched {rid} from {remote}"); - match fetch.direction { - FetchDirection::Initiator { namespaces } => { - let result = match result { - Ok(updated) => { - log::debug!(target: "service", "Fetched {rid} from {remote}"); + self.emitter.emit(Event::RefsFetched { + remote, + rid, + updated: updated.clone(), + }); - self.emitter.emit(Event::RefsFetched { - remote, - rid, - updated: updated.clone(), - }); - - FetchResult::Success { updated } - } - Err(err) => { - let reason = err.to_string(); - error!(target: "service", "Fetch failed for {rid} from {remote}: {reason}"); - - // For now, we only disconnect the remote in case of timeout. In the future, - // there may be other reasons to disconnect. - if err.is_timeout() { - self.reactor - .disconnect(remote, DisconnectReason::Fetch(err)); - } - FetchResult::Failed { reason } - } - }; - - if let Some(results) = self.fetch_reqs.remove(&(rid, remote)) { - log::debug!(target: "service", "Found existing fetch request, sending result.."); - - if results.send(result).is_err() { - log::error!(target: "service", "Error sending fetch result for {rid}.."); - } else { - log::debug!(target: "service", "Sent fetch result for {rid}.."); - } - } else { - log::debug!(target: "service", "No fetch requests found for {rid}.."); - - // We only announce refs here when the fetch wasn't user-requested. This is - // because the user might want to announce his fork, once he has created one, - // or may choose to not announce anything. - if let Err(e) = self.announce_refs(rid, &namespaces) { - error!(target: "service", "Failed to announce new refs: {e}"); - } - } - - self.switch_to_gossip(remote); - - // TODO: Since this fetch could be either a full clone - // or simply a ref update, we need to either announce - // new inventory, or new refs. Right now, we announce - // both in some cases. - // - // Announce the newly fetched repository to the - // network, if necessary. - // - // Nb. This needs to be run after we've switched back - // to the gossip protocol, otherwise the messages will - // be queued. - self.sync_and_announce(); - self.process_fetch_queue(&remote); + FetchResult::Success { updated } + } + Err(err) => { + let reason = err.to_string(); + error!(target: "service", "Fetch failed for {rid} from {remote}: {reason}"); + + // For now, we only disconnect the remote in case of timeout. In the future, + // there may be other reasons to disconnect. + if err.is_timeout() { + self.reactor + .disconnect(remote, DisconnectReason::Fetch(err)); + } + FetchResult::Failed { reason } + } + }; + + if let Some(results) = self.fetch_reqs.remove(&(rid, remote)) { + log::debug!(target: "service", "Found existing fetch request, sending result.."); + + if results.send(result).is_err() { + log::error!(target: "service", "Error sending fetch result for {rid}.."); + } else { + log::debug!(target: "service", "Sent fetch result for {rid}.."); + } + } else { + log::debug!(target: "service", "No fetch requests found for {rid}.."); + + // We only announce refs here when the fetch wasn't user-requested. This is + // because the user might want to announce his fork, once he has created one, + // or may choose to not announce anything. + if let Err(e) = self.announce_refs(rid, &namespaces) { + error!(target: "service", "Failed to announce new refs: {e}"); } - FetchDirection::Responder => self.switch_to_gossip(remote), } + + // Go back to "idle". + if let Some(s) = self.sessions.get_mut(&remote) { + s.fetched(rid); + } + // TODO: Since this fetch could be either a full clone + // or simply a ref update, we need to either announce + // new inventory, or new refs. Right now, we announce + // both in some cases. + // + // Announce the newly fetched repository to the + // network, if necessary. + // + // Nb. This needs to be run after we've switched back + // to the gossip protocol, otherwise the messages will + // be queued. + self.sync_and_announce(); } pub fn accepted(&mut self, _addr: net::SocketAddr) { @@ -725,8 +709,8 @@ where // If the peer disconnected while we were waiting for a [`Message::FetchOk`], // return a failure to any potential fetcher. - if let Some((requested, _)) = session.requesting() { - if let Some(resp) = self.fetch_reqs.remove(&(*requested, remote)) { + for rid in session.fetching() { + if let Some(resp) = self.fetch_reqs.remove(&(rid, remote)) { resp.send(FetchResult::Failed { reason: format!("disconnected: {reason}"), }) @@ -902,9 +886,7 @@ where if self.sessions.is_connected(announcer) { match self.should_fetch_refs_announcement(message, &repo_entry.scope) { Ok(true) => self.fetch(message.rid, announcer), - Ok(false) => { - debug!(target: "service", "Skipping fetch from {announcer}") - } + Ok(false) => {} Err(e) => { error!(target: "service", "Failed to check refs announcement: {e}"); return Err(session::Error::Misbehavior); @@ -1042,19 +1024,6 @@ where debug!(target: "service", "Received message {:?} from {}", &message, peer.id); match (&mut peer.state, message) { - ( - session::State::Connected { - protocol: session::Protocol::Fetch { .. }, - .. - }, - _, - ) => { - // This should never happen if the service is properly configured, since all - // incoming data is sent directly to the Git worker. - log::error!(target: "service", "Received gossip message from {remote} during git fetch"); - - return Err(session::Error::Misbehavior); - } // Process a peer announcement. (session::State::Connected { .. }, Message::Announcement(ann)) => { let relayer = peer.id; @@ -1107,72 +1076,6 @@ where } } } - ( - session::State::Connected { - protocol: session::Protocol::Gossip { state }, - .. - }, - Message::Fetch { rid }, - ) => { - debug!(target: "service", "Fetch requested for {rid} from {remote}.."); - - // TODO: Check that we have the repo first? - - // We got a fetch request right after sending our own. We have to decide on which - // fetch to run: our own, or the remote's. - if let GossipState::Requesting { rid, .. } = state { - debug!(target: "service", "Received fetch request from {remote} while attempting to fetch {rid}.."); - - // When fetch requests cross, the inbound peer takes precedence. - if peer.link.is_inbound() { - debug!(target: "service", "Cancelling fetch request to {remote}.."); - - // Cancel our own fetch request. This doesn't send anything to the remote, - // it simply updates the local session's state machine. - *state = GossipState::Idle; - - // TODO: Queue the fetch request as if we tried to request twice from - // the same node. - } else { - // In this case, the remote node will cancel its request, so we don't - // want to handover the session to the worker here, we will do it when - // we get the `FetchOk` from the remote. - return Ok(()); - } - } - - // Accept the request and instruct the transport to handover the socket to the worker. - self.reactor.write(peer, Message::FetchOk { rid }); - self.reactor.fetch(peer, rid, FetchDirection::Responder); - } - (session::State::Connected { protocol, .. }, Message::FetchOk { rid }) => { - let session::Protocol::Gossip { - state: GossipState::Requesting { rid: requested, namespaces } - } = protocol else { - // As long as we disconnect peers who don't respond to our fetch requests within - // the alloted time, this shouldn't happen by mistake. - error!( - "Received unexpected message `fetch-ok` from peer {}", - peer.id - ); - return Err(session::Error::Misbehavior); - }; - - if *requested != rid { - error!( - "Received `fetch-ok` from {} for incorrect repository {rid}", - peer.id - ); - return Err(session::Error::Misbehavior); - } - let namespaces = namespaces.clone(); - - debug!(target: "service", "Fetch accepted for {rid} from {remote}.."); - - // Instruct the transport to handover the socket to the worker. - self.reactor - .fetch(peer, rid, FetchDirection::Initiator { namespaces }); - } (session::State::Attempted { .. } | session::State::Initial, msg) => { error!(target: "service", "Received {:?} from connecting peer {}", msg, peer.id); } @@ -1299,19 +1202,6 @@ where Ok(()) } - fn switch_to_gossip(&mut self, remote: NodeId) { - if let Some(session) = self.sessions.get_mut(&remote) { - // Transition session back to gossip protocol. - session.to_gossip(); - // Drain any messages in the session's outbox, which might - // have accumulated during a fetch, and send them to the - // peer. - self.reactor.drain(session); - } else { - log::debug!(target: "service", "Session not found for {remote}"); - } - } - fn sync_and_announce(&mut self) { match self.sync_inventory() { Ok(updated) => { @@ -1331,16 +1221,6 @@ where } } - /// Execute the next pending fetch with `remote`, if any. - fn process_fetch_queue(&mut self, remote: &NodeId) { - if let Some(session) = self.sessions.get_mut(remote) { - if let Some(rid) = session.dequeue_fetch() { - debug!(target: "service", "Dequeued a pending fetch {rid} with {remote}"); - self.fetch(rid, remote); - } - } - } - fn reconnect(&mut self, nid: NodeId, addr: Address) -> bool { if let Some(sess) = self.sessions.get_mut(&nid) { sess.to_initial(); @@ -1370,7 +1250,6 @@ where pub struct Stats { connected: usize, disconnected: usize, - fetching: usize, } let (stats, seeds) = match self.routing.get(rid) { @@ -1378,10 +1257,7 @@ where (Stats::default(), Seeds::default()), |(mut stats, mut seeds), node| { if node != self.node_id() { - if self.sessions.is_fetching(&node) { - seeds.insert(Seed::Fetching(node)); - stats.fetching += 1; - } else if self.sessions.is_connected(&node) { + if self.sessions.is_connected(&node) { seeds.insert(Seed::Connected(node)); stats.connected += 1; } else if self.sessions.is_disconnected(&node) { @@ -1399,8 +1275,8 @@ where }; debug!( target: "service", - "Found {} connected seed(s), {} disconnected seed(s), and {} fetching seed(s) for {}", - stats.connected, stats.disconnected, stats.fetching, rid + "Found {} connected seed(s) and {} disconnected seed(s) for {}", + stats.connected, stats.disconnected, rid ); Ok(seeds) @@ -1788,10 +1664,6 @@ impl Sessions { self.0.get(id).map(|s| s.is_connected()).unwrap_or(false) } - pub fn is_fetching(&self, id: &NodeId) -> bool { - self.0.get(id).map(|s| s.is_fetching()).unwrap_or(false) - } - /// Return whether this node can be connected to. pub fn is_disconnected(&self, id: &NodeId) -> bool { self.0.get(id).map(|s| s.is_disconnected()).unwrap_or(true) diff --git a/radicle-node/src/service/message.rs b/radicle-node/src/service/message.rs index d355de2b..b1d3bb3c 100644 --- a/radicle-node/src/service/message.rs +++ b/radicle-node/src/service/message.rs @@ -341,12 +341,6 @@ pub enum Message { /// The pong payload. zeroes: ZeroBytes, }, - - /// Request a session upgrade to the Git protocol and fetch the given repository. - Fetch { rid: Id }, - - /// Accept a fetch request. - FetchOk { rid: Id }, } impl PartialOrd for Message { @@ -442,8 +436,6 @@ impl fmt::Debug for Message { } Self::Ping(Ping { ponglen, zeroes }) => write!(f, "Ping({ponglen}, {zeroes:?})"), Self::Pong { zeroes } => write!(f, "Pong({zeroes:?})"), - Self::Fetch { rid } => write!(f, "Fetch({rid})"), - Self::FetchOk { rid } => write!(f, "FetchOk({rid})"), } } } diff --git a/radicle-node/src/service/reactor.rs b/radicle-node/src/service/reactor.rs index 3d152040..ac32d8cf 100644 --- a/radicle-node/src/service/reactor.rs +++ b/radicle-node/src/service/reactor.rs @@ -1,5 +1,4 @@ -use std::collections::{HashMap, VecDeque}; -use std::mem; +use std::collections::VecDeque; use log::*; @@ -19,52 +18,16 @@ pub enum Io { /// Disconnect from a peer. Disconnect(NodeId, DisconnectReason), /// Fetch repository data from a peer. - Fetch(Fetch), - /// Ask for a wakeup in a specified amount of time. - Wakeup(LocalDuration), -} - -/// Fetch job sent to worker thread. -#[derive(Debug, Clone)] -pub struct Fetch { - /// Repo to fetch. - pub rid: Id, - /// Indicates whether the fetch request was initiated or is a response. - pub direction: FetchDirection, - /// Remote peer we are interacting with. - pub remote: NodeId, -} - -impl Fetch { - pub fn is_initiator(&self) -> bool { - self.direction.is_initiator() - } - - pub fn initiated(&self) -> Option<&Namespaces> { - match &self.direction { - FetchDirection::Initiator { namespaces } => Some(namespaces), - FetchDirection::Responder => None, - } - } -} - -#[derive(Debug, Clone)] -pub enum FetchDirection { - /// Client is initiating a fetch in order to receive the specified - /// `refspecs` determined by [`Namespaces`]. - Initiator { - /// Namespaces to fetch. + Fetch { + /// Repo being fetched. + rid: Id, + /// Remote node being fetched from. + remote: NodeId, + /// Namespaces being fetched. namespaces: Namespaces, }, - /// Server is responding to a fetch request by uploading the - /// specified `refspecs` sent by the client. - Responder, -} - -impl FetchDirection { - pub fn is_initiator(&self) -> bool { - matches!(self, Self::Initiator { .. }) - } + /// Ask for a wakeup in a specified amount of time. + Wakeup(LocalDuration), } /// Interface to the network reactor. @@ -72,10 +35,6 @@ impl FetchDirection { pub struct Reactor { /// Outgoing I/O queue. io: VecDeque, - /// Message outbox for each node. - /// If messages can't be sent to a node immediately, they are stored in the outbox. - /// This can happen if for eg. a fetch is ongoing with that node. - outbox: HashMap>, } impl Reactor { @@ -90,71 +49,36 @@ impl Reactor { } pub fn write(&mut self, remote: &Session, msg: Message) { - // If we've requested a fetch or are currently fetching, any message to be written - // to the remote peer should be queued. - if remote.is_requesting() || remote.is_fetching() { - debug!(target: "service", "Queue {:?} for {}", &msg, remote); - self.outbox.entry(remote.id).or_default().push(msg); - } else { - debug!(target: "service", "Write {:?} to {}", &msg, remote); - self.io.push_back(Io::Write(remote.id, vec![msg])); - } + debug!(target: "service", "Write {:?} to {}", &msg, remote); + self.io.push_back(Io::Write(remote.id, vec![msg])); } pub fn write_all(&mut self, remote: &Session, msgs: impl IntoIterator) { let msgs = msgs.into_iter().collect::>(); - let queue = remote.is_fetching() || remote.is_requesting(); for (ix, msg) in msgs.iter().enumerate() { - if queue { - debug!( - target: "service", - "Queue {:?} for {} ({}/{})", - msg, - remote, - ix + 1, - msgs.len() - ); - } else { - debug!( - target: "service", - "Write {:?} to {} ({}/{})", - msg, - remote, - ix + 1, - msgs.len() - ); - } - } - if queue { - self.outbox.entry(remote.id).or_default().extend(msgs); - } else { - self.io.push_back(Io::Write(remote.id, msgs)); - } - } - - pub fn drain(&mut self, remote: &Session) { - if let Some(outbox) = self.outbox.get_mut(&remote.id) { - debug!(target: "service", "Draining outbox for session {} ({} message(s))", remote.id, outbox.len()); - - let msgs = mem::take(outbox); - self.write_all(remote, msgs); + debug!( + target: "service", + "Write {:?} to {} ({}/{})", + msg, + remote, + ix + 1, + msgs.len() + ); } + self.io.push_back(Io::Write(remote.id, msgs)); } pub fn wakeup(&mut self, after: LocalDuration) { self.io.push_back(Io::Wakeup(after)); } - pub fn fetch(&mut self, remote: &mut Session, rid: Id, direction: FetchDirection) { - // Transition the session state machine to "fetching". - remote.to_fetching(rid); - - self.io.push_back(Io::Fetch(Fetch { + pub fn fetch(&mut self, remote: &mut Session, rid: Id, namespaces: Namespaces) { + self.io.push_back(Io::Fetch { rid, - direction, + namespaces, remote: remote.id, - })); + }); } /// Broadcast a message to a list of peers. diff --git a/radicle-node/src/service/session.rs b/radicle-node/src/service/session.rs index a11fb9d7..3eeae834 100644 --- a/radicle-node/src/service/session.rs +++ b/radicle-node/src/service/session.rs @@ -1,8 +1,6 @@ -use std::collections::VecDeque; +use std::collections::HashSet; use std::fmt; -use radicle::storage::Namespaces; - use crate::service::message; use crate::service::message::Message; use crate::service::{Id, LocalTime, NodeId, Reactor, Rng}; @@ -19,35 +17,6 @@ pub enum PingState { Ok, } -/// Sub-state of the gossip protocol. -#[derive(Debug, Default, PartialEq, Eq, Clone)] -pub enum GossipState { - /// Regular gossip, no pending fetch requests. - #[default] - Idle, - /// Requesting a fetch for the given RID. Waiting for a [`Message::FetchOk`]. - Requesting { rid: Id, namespaces: Namespaces }, -} - -/// Session protocol. -#[derive(Debug, PartialEq, Eq, Clone)] -pub enum Protocol { - /// The default message-based gossip protocol. - Gossip { state: GossipState }, - /// Git smart protocol. Used for fetching repository data. - /// This protocol is used after a connection upgrade via the - /// [`Message::Fetch`] message. - Fetch { rid: Id }, -} - -impl Default for Protocol { - fn default() -> Self { - Self::Gossip { - state: GossipState::default(), - } - } -} - #[derive(Debug, Clone)] #[allow(clippy::large_enum_variant)] pub enum State { @@ -61,8 +30,8 @@ pub enum State { since: LocalTime, /// Ping state. ping: PingState, - /// Session protocol. - protocol: Protocol, + /// Ongoing fetches. + fetching: HashSet, }, /// When a peer is disconnected. Disconnected { @@ -82,23 +51,9 @@ impl fmt::Display for State { Self::Attempted => { write!(f, "attempted") } - Self::Connected { protocol, .. } => match protocol { - Protocol::Gossip { - state: GossipState::Idle, - .. - } => { - write!(f, "connected ") - } - Protocol::Gossip { - state: GossipState::Requesting { rid, .. }, - .. - } => { - write!(f, "connected requested={rid}") - } - Protocol::Fetch { .. } => { - write!(f, "connected ") - } - }, + Self::Connected { .. } => { + write!(f, "connected") + } Self::Disconnected { .. } => { write!(f, "disconnected") } @@ -109,10 +64,10 @@ impl fmt::Display for State { /// Return value of [`Session::fetch`]. #[derive(Debug)] pub enum FetchResult { - /// We are already fetching from this peer. - AlreadyFetching(Id), + /// We are already fetching the given repo from this peer. + AlreadyFetching, /// Ok, ready to fetch. - Ready(Message), + Ready, /// This peer is not ready to fetch. NotConnected, } @@ -164,9 +119,6 @@ pub struct Session { /// Last time a message was received from the peer. pub last_active: LocalTime, - /// Fetches queued due to another ongoing fetch. - pending_fetches: VecDeque, - /// Connection attempts. For persistent peers, Tracks /// how many times we've attempted to connect. We reset this to zero /// upon successful connection. @@ -205,7 +157,6 @@ impl Session { persistent, last_active: LocalTime::default(), attempts: 1, - pending_fetches: VecDeque::new(), rng, } } @@ -216,14 +167,13 @@ impl Session { state: State::Connected { since: time, ping: PingState::default(), - protocol: Protocol::default(), + fetching: HashSet::default(), }, link: Link::Inbound, subscribe: None, persistent, last_active: LocalTime::default(), attempts: 0, - pending_fetches: VecDeque::new(), rng, } } @@ -236,16 +186,6 @@ impl Session { matches!(self.state, State::Connected { .. }) } - pub fn is_fetching(&self) -> bool { - matches!( - self.state, - State::Connected { - protocol: Protocol::Fetch { .. }, - .. - } - ) - } - pub fn is_disconnected(&self) -> bool { matches!(self.state, State::Disconnected { .. }) } @@ -254,64 +194,26 @@ impl Session { matches!(self.state, State::Initial) } - pub fn is_requesting(&self) -> bool { - matches!( - self.state, - State::Connected { - protocol: Protocol::Gossip { - state: GossipState::Requesting { .. } - }, - .. - } - ) - } - pub fn attempts(&self) -> usize { self.attempts } - pub fn fetch(&self, rid: Id) -> FetchResult { - if let State::Connected { protocol, .. } = &self.state { - match protocol { - Protocol::Gossip { state } => { - if let GossipState::Requesting { rid, .. } = state { - FetchResult::AlreadyFetching(*rid) - } else { - FetchResult::Ready(Message::Fetch { rid }) - } - } - Protocol::Fetch { rid } => FetchResult::AlreadyFetching(*rid), + pub fn fetch(&mut self, rid: Id) -> FetchResult { + if let State::Connected { fetching, .. } = &mut self.state { + if fetching.insert(rid) { + FetchResult::Ready + } else { + FetchResult::AlreadyFetching } } else { FetchResult::NotConnected } } - pub fn to_requesting(&mut self, rid: Id, namespaces: Namespaces) { - let State::Connected { protocol, .. } = &mut self.state else { - panic!("Session::to_requesting: cannot transition to 'requesting': session is not connected"); - }; - *protocol = Protocol::Gossip { - state: GossipState::Requesting { rid, namespaces }, - }; - } - - pub fn to_fetching(&mut self, rid: Id) { - let State::Connected { protocol, .. } = &mut self.state else { - panic!("Session::to_fetching: cannot transition to 'fetching': session is not connected"); - }; - *protocol = Protocol::Fetch { rid }; - } - - pub fn to_gossip(&mut self) { - if let State::Connected { protocol, .. } = &mut self.state { - if let Protocol::Fetch { .. } = protocol { - *protocol = Protocol::default(); - } else { - panic!( - "Unexpected session state for {}: expected 'fetch' protocol, got 'gossip'", - self.id - ); + pub fn fetched(&mut self, rid: Id) { + if let State::Connected { fetching, .. } = &mut self.state { + if !fetching.remove(&rid) { + log::error!(target: "service", "Fetched unknown repository {rid}"); } } } @@ -334,7 +236,7 @@ impl Session { self.state = State::Connected { since, ping: PingState::default(), - protocol: Protocol::default(), + fetching: HashSet::default(), }; } @@ -354,18 +256,11 @@ impl Session { self.state = State::Initial; } - pub fn requesting(&self) -> Option<(&Id, &Namespaces)> { - if let State::Connected { - protocol: - Protocol::Gossip { - state: GossipState::Requesting { rid, namespaces }, - }, - .. - } = &self.state - { - Some((rid, namespaces)) + pub fn fetching(&self) -> HashSet { + if let State::Connected { fetching, .. } = &self.state { + fetching.clone() } else { - None + HashSet::default() } } @@ -378,12 +273,4 @@ impl Session { } Ok(()) } - - pub(crate) fn queue_fetch(&mut self, rid: Id) { - self.pending_fetches.push_back(rid); - } - - pub(crate) fn dequeue_fetch(&mut self) -> Option { - self.pending_fetches.pop_front() - } } diff --git a/radicle-node/src/test/arbitrary.rs b/radicle-node/src/test/arbitrary.rs index 642dca35..1481a820 100644 --- a/radicle-node/src/test/arbitrary.rs +++ b/radicle-node/src/test/arbitrary.rs @@ -93,7 +93,6 @@ impl Arbitrary for Message { MessageType::Pong => Self::Pong { zeroes: ZeroBytes::new(u16::arbitrary(g).min(Ping::MAX_PONG_ZEROES)), }, - _ => unreachable!(), } } } diff --git a/radicle-node/src/test/environment.rs b/radicle-node/src/test/environment.rs index 8221e8ee..fece4817 100644 --- a/radicle-node/src/test/environment.rs +++ b/radicle-node/src/test/environment.rs @@ -114,7 +114,7 @@ pub struct NodeHandle { pub home: Home, pub addr: net::SocketAddr, pub thread: ManuallyDrop>>, - pub handle: ManuallyDrop>, + pub handle: ManuallyDrop, } impl Drop for NodeHandle { diff --git a/radicle-node/src/test/simulator.rs b/radicle-node/src/test/simulator.rs index e761614b..60b4166a 100644 --- a/radicle-node/src/test/simulator.rs +++ b/radicle-node/src/test/simulator.rs @@ -14,9 +14,9 @@ use log::*; use crate::crypto::Signer; use crate::git::raw as git; -use crate::prelude::Address; +use crate::prelude::{Address, Id}; use crate::service::reactor::Io; -use crate::service::{DisconnectReason, Event, Fetch, Message, NodeId}; +use crate::service::{DisconnectReason, Event, Message, NodeId}; use crate::storage::{Namespaces, RefUpdate}; use crate::storage::{WriteRepository, WriteStorage}; use crate::test::peer::Service; @@ -63,7 +63,12 @@ pub enum Input { /// Received a message from a remote peer. Received(NodeId, Vec), /// Fetch completed for a node. - Fetched(Fetch, Rc, FetchError>>), + Fetched( + Id, + Namespaces, + NodeId, + Rc, FetchError>>, + ), /// Used to advance the state machine after some wall time has passed. Wake, } @@ -105,15 +110,8 @@ impl fmt::Display for Scheduled { Input::Wake => { write!(f, "{}: Tock", self.node) } - Input::Fetched(fetch, _) => { - write!( - f, - "{} <<~ {} ({}): Fetched (initiated={})", - self.node, - fetch.remote, - fetch.rid, - fetch.is_initiator() - ) + Input::Fetched(rid, _, nid, _) => { + write!(f, "{} <<~ {} ({}): Fetched", self.node, nid, rid) } } } @@ -411,17 +409,15 @@ impl Simulation { p.received_message(id, msg); } } - Input::Fetched(f, result) => { + Input::Fetched(rid, ns, nid, result) => { let result = Rc::try_unwrap(result).unwrap(); - if let Some(namespaces) = f.initiated() { - let mut repo = match p.storage().repository_mut(f.rid) { - Ok(repo) => repo, - Err(e) if e.is_not_found() => p.storage().create(f.rid).unwrap(), - Err(e) => panic!("Failed to open repository: {e}"), - }; - fetch(&mut repo, &f.remote, namespaces.clone()).unwrap(); - } - p.fetched(f, result); + let mut repo = match p.storage().repository_mut(rid) { + Ok(repo) => repo, + Err(e) if e.is_not_found() => p.storage().create(rid).unwrap(), + Err(e) => panic!("Failed to open repository: {e}"), + }; + fetch(&mut repo, &nid, ns.clone()).unwrap(); + p.fetched(rid, ns, nid, result); } } while let Some(o) = p.next() { @@ -606,22 +602,16 @@ impl Simulation { ); } } - Io::Fetch(fetch) => { - let remote = fetch.remote; - - if fetch.is_initiator() { - log::info!( - target: "sim", - "{:05} {} ~> {} ({}): Fetch outgoing", - self.elapsed().as_millis(), node, remote, fetch.rid - ); - } else { - log::info!( - target: "sim", - "{:05} {} <~ {} ({}): Fetch incoming", - self.elapsed().as_millis(), node, remote, fetch.rid - ); - } + Io::Fetch { + rid, + remote, + namespaces, + } => { + log::info!( + target: "sim", + "{:05} {} ~> {} ({}): Fetch outgoing", + self.elapsed().as_millis(), node, remote, rid + ); if self.is_fallible() { self.inbox.insert( @@ -630,7 +620,9 @@ impl Simulation { node, remote, input: Input::Fetched( - fetch, + rid, + namespaces, + remote, Rc::new(Err(FetchError::Io(io::ErrorKind::Other.into()))), ), }, @@ -641,7 +633,7 @@ impl Simulation { Scheduled { node, remote, - input: Input::Fetched(fetch, Rc::new(Ok(vec![]))), + input: Input::Fetched(rid, namespaces, remote, Rc::new(Ok(vec![]))), }, ); } diff --git a/radicle-node/src/tests.rs b/radicle-node/src/tests.rs index 38700166..12ca1d88 100644 --- a/radicle-node/src/tests.rs +++ b/radicle-node/src/tests.rs @@ -17,13 +17,12 @@ use crate::prelude::{LocalDuration, Timestamp}; use crate::service::config::*; use crate::service::filter::Filter; use crate::service::message::*; -use crate::service::reactor::FetchDirection; use crate::service::reactor::Io; use crate::service::ServiceState as _; use crate::service::*; use crate::storage::git::transport::{local, remote}; use crate::storage::git::Storage; -use crate::storage::{Namespaces, ReadStorage}; +use crate::storage::ReadStorage; use crate::test::arbitrary; use crate::test::assert_matches; use crate::test::fixtures; @@ -707,10 +706,7 @@ fn test_refs_announcement_fetch_trusted_no_inventory() { alice.receive(bob.id(), bob.refs_announcement(rid)); // Alice fetches Bob's refs as this is a new repo. - assert_eq!( - alice.messages(bob.id()).next(), - Some(Message::Fetch { rid }) - ); + assert_matches!(alice.outbox().next(), Some(Io::Fetch { .. })); } /// Alice and Bob both have the same repo. @@ -758,7 +754,7 @@ fn test_refs_announcement_trusted() { // Bob announces refs again. bob.elapse(LocalDuration::from_mins(1)); // Make sure our announcement is fresh. alice.receive(bob.id(), bob.refs_announcement(rid)); - assert_matches!(alice.messages(bob.id()).next(), Some(Message::Fetch { .. })); + assert_matches!(alice.outbox().next(), Some(Io::Fetch { .. })); } #[test] @@ -778,86 +774,6 @@ fn test_refs_announcement_no_subscribe() { assert!(alice.messages(eve.id()).next().is_none()); } -#[test] -fn test_gossip_during_fetch() { - let storage = arbitrary::nonempty_storage(1); - let rid = *storage.inventory.keys().next().unwrap(); - let mut alice = Peer::with_storage("alice", [7, 7, 7, 7], storage); - let bob = Peer::new("bob", [8, 8, 8, 8]); - let eve = Peer::new("eve", [9, 9, 9, 9]); - let now = LocalTime::now().as_millis(); - let (send, _recv) = chan::bounded::(1); - let inventory1 = BoundedVec::try_from(arbitrary::vec(1)).unwrap(); - let inventory2 = BoundedVec::try_from(arbitrary::vec(1)).unwrap(); - - alice.connect_to(&bob); - alice.connect_to(&eve); - alice.command(Command::Fetch(rid, bob.id, send)); - - assert_matches!(alice.messages(bob.id).next(), Some(Message::Fetch { .. })); - - logger::init(log::Level::Debug); - - alice.receive( - eve.id(), - Message::inventory( - InventoryAnnouncement { - inventory: inventory1.clone(), - timestamp: now + 1, - }, - eve.signer(), - ), - ); - // We shouldn't relay to Bob while we're fetching from him. - assert_matches!(alice.messages(bob.id).next(), None); - - alice.receive(bob.id(), Message::FetchOk { rid }); - alice.receive( - eve.id(), - Message::inventory( - InventoryAnnouncement { - inventory: inventory2.clone(), - timestamp: now + 2, - }, - eve.signer(), - ), - ); - // We shouldn't relay to Bob while we're fetching from him. - assert_matches!(alice.messages(bob.id).next(), None); - - // Have enough time pass that Alice sends a "ping" to Bob. - alice.elapse(KEEP_ALIVE_DELTA); - - // Now that the fetch is done, the messages Bob missed should be relayed to him. - alice.fetched( - Fetch { - rid, - direction: FetchDirection::Initiator { - namespaces: Namespaces::All, - }, - remote: bob.id, - }, - Ok(vec![]), - ); - let mut messages = alice.messages(bob.id); - - assert_matches!( - messages.next(), - Some(Message::Announcement(Announcement { - message: AnnouncementMessage::Inventory(InventoryAnnouncement { inventory, .. }), - .. - })) if inventory == inventory1 - ); - assert_matches!( - messages.next(), - Some(Message::Announcement(Announcement { - message: AnnouncementMessage::Inventory(InventoryAnnouncement { inventory, .. }), - .. - })) if inventory == inventory2 - ); - assert_matches!(messages.next(), Some(Message::Ping { .. })); -} - #[test] fn test_inventory_relay() { // Topology is eve <-> alice <-> bob @@ -1155,12 +1071,12 @@ fn test_fetch_missing_inventory() { alice.elapse(service::SYNC_INTERVAL); alice - .messages(bob.id) - .find(|m| matches!(m, Message::Fetch { .. })) + .outbox() + .find(|m| matches!(m, Io::Fetch { .. })) .unwrap(); alice - .messages(eve.id) - .find(|m| matches!(m, Message::Fetch { .. })) + .outbox() + .find(|m| matches!(m, Io::Fetch { .. })) .unwrap(); } @@ -1356,89 +1272,3 @@ fn prop_inventory_exchange_dense() { .tests(20) .quickcheck(property as fn(MockStorage, MockStorage, MockStorage)); } - -#[test] -fn test_queued_fetch() { - let storage = arbitrary::nonempty_storage(3); - let mut repo_keys = storage.inventory.keys(); - let rid = *repo_keys.next().unwrap(); - let rid2 = *repo_keys.next().unwrap(); - let rid3 = *repo_keys.next().unwrap(); - let mut alice = Peer::with_storage("alice", [7, 7, 7, 7], storage); - let bob = Peer::new("bob", [8, 8, 8, 8]); - let (send, _recv) = chan::bounded::(1); - - logger::init(log::Level::Debug); - - // Send the first fetch. - alice.connect_to(&bob); - alice.command(Command::Fetch(rid, bob.id, send)); - - assert_matches!(alice.messages(bob.id).next(), Some(Message::Fetch { .. })); - - // Send the 2nd fetch that will be queued. - let (send2, _recv2) = chan::bounded::(1); - alice.command(Command::Fetch(rid2, bob.id, send2)); - - // Send the 3rd fetch that will be queued. - let (send3, _recv3) = chan::bounded::(1); - alice.command(Command::Fetch(rid3, bob.id, send3)); - - // We shouldn't send out the 2nd, 3rd fetch while we're doing the 1st fetch. - assert_matches!(alice.messages(bob.id).next(), None); - - alice.receive(bob.id(), Message::FetchOk { rid }); - assert_matches!(alice.messages(bob.id).next(), None); - - // Have enough time pass that Alice sends a "ping" to Bob. - alice.elapse(KEEP_ALIVE_DELTA); - - // Finish the 1st fetch. - alice.fetched( - Fetch { - rid, - direction: FetchDirection::Initiator { - namespaces: Namespaces::All, - }, - remote: bob.id, - }, - Ok(vec![]), - ); - - // Now the 1st fetch is done, the gossip messages are drained. - let mut messages = alice.messages(bob.id); - assert_matches!(messages.next(), Some(Message::Ping(_))); - - // The message after all queued gossip messages is Fetch. - assert_eq!(messages.last(), Some(Message::Fetch { rid: rid2 })); - - // `FetchOk` for the 2nd fetch. - alice.receive(bob.id(), Message::FetchOk { rid: rid2 }); - - // The 2nd fetch should be in `Io` now. Not the 3rd fetch yet. - let last_io = alice.outbox().last().unwrap(); - assert_matches!(last_io, Io::Fetch(fetch) if fetch.rid == rid2); - - // Finish the 2nd fetch. - alice.fetched( - Fetch { - rid: rid2, - direction: FetchDirection::Initiator { - namespaces: Namespaces::All, - }, - remote: bob.id, - }, - Ok(vec![]), - ); - - // Now the 2nd fetch is done, the 3rd fetch is drained. - let mut messages = alice.messages(bob.id); - assert_eq!(messages.next(), Some(Message::Fetch { rid: rid3 })); - - // `FetchOk` for the 3rd fetch. - alice.receive(bob.id(), Message::FetchOk { rid: rid3 }); - - // The 3rd fetch should be in `Io` now. - let last_io = alice.outbox().last().unwrap(); - assert_matches!(last_io, Io::Fetch(fetch) if fetch.rid == rid3); -} diff --git a/radicle-node/src/wire.rs b/radicle-node/src/wire.rs index b5f6e835..3c907beb 100644 --- a/radicle-node/src/wire.rs +++ b/radicle-node/src/wire.rs @@ -1,6 +1,9 @@ +mod frame; mod message; mod protocol; +mod varint; +pub use frame::StreamId; pub use message::{AddressType, MessageType}; pub use protocol::{Control, Wire, WireReader, WireSession, WireWriter}; @@ -40,8 +43,14 @@ pub enum Error { InvalidSize { expected: usize, actual: usize }, #[error("invalid filter size: {0}")] InvalidFilterSize(usize), + #[error("invalid channel type {0:x}")] + InvalidStreamKind(u8), #[error(transparent)] InvalidRefName(#[from] fmt::Error), + #[error("invalid control message with type `{0}`")] + InvalidControlMessage(u8), + #[error("invalid protocol version header `{0:x?}`")] + InvalidProtocolVersion([u8; 4]), #[error("unknown address type `{0}`")] UnknownAddressType(u8), #[error("unknown message type `{0}`")] diff --git a/radicle-node/src/wire/frame.rs b/radicle-node/src/wire/frame.rs new file mode 100644 index 00000000..8a4b2ca6 --- /dev/null +++ b/radicle-node/src/wire/frame.rs @@ -0,0 +1,359 @@ +//! Framing protocol. +#![warn(clippy::missing_docs_in_private_items)] +use std::{fmt, io}; + +use crate::{wire, wire::varint, wire::varint::VarInt, wire::Message, Link}; + +/// Protocol version strings all start with the magic sequence `rad`, followed +/// by a version number. +pub const PROTOCOL_VERSION: Version = Version([b'r', b'a', b'd', 0x1]); + +/// Control open byte. +const CONTROL_OPEN: u8 = 0; +/// Control close byte. +const CONTROL_CLOSE: u8 = 1; +/// Control EOF byte. +const CONTROL_EOF: u8 = 2; + +/// Protocol version. +pub struct Version([u8; 4]); + +impl wire::Encode for Version { + fn encode(&self, writer: &mut W) -> Result { + writer.write_all(&PROTOCOL_VERSION.0)?; + + Ok(PROTOCOL_VERSION.0.len()) + } +} + +impl wire::Decode for Version { + fn decode(reader: &mut R) -> Result { + let mut version = [0u8; 4]; + reader.read_exact(&mut version[..])?; + + if version != PROTOCOL_VERSION.0 { + return Err(wire::Error::InvalidProtocolVersion(version)); + } + Ok(Self(version)) + } +} + +/// Identifies a (multiplexed) stream. +/// +/// Stream IDs are variable-length integers with the least significant 3 bits +/// denoting the stream type and initiator. +/// +/// The first bit denotes the initiator (outbound or inbound), while the second +/// and third bit denote the stream type. See `StreamKind`. +/// +/// In a situation where Alice connects to Bob, Alice will have the initiator +/// bit set to `1` for all streams she creates, while Bob will have it set to `0`. +/// +/// This ensures that Stream IDs never collide. +/// Additionally, Stream IDs must never be re-used within a connection. +/// +/// +=======+==================================+ +/// | Bits | Stream Type | +/// +=======+==================================+ +/// | 0b000 | Outbound Control stream | +/// +-------+----------------------------------+ +/// | 0b001 | Inbound Control stream | +/// +-------+----------------------------------+ +/// | 0b010 | Outbound Gossip stream | +/// +-------+----------------------------------+ +/// | 0b011 | Inbound Gossip stream | +/// +-------+----------------------------------+ +/// | 0b100 | Outbound Git stream | +/// +-------+----------------------------------+ +/// | 0b101 | Inbound Git stream | +/// +-------+----------------------------------+ +/// +#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)] +pub struct StreamId(VarInt); + +impl StreamId { + /// Get the initiator of this stream. + pub fn link(&self) -> Link { + let n = *self.0; + if 0b1 & n == 0 { + Link::Outbound + } else { + Link::Inbound + } + } + + /// Get the kind of stream this is. + pub fn kind(&self) -> Result { + let id = *self.0; + match (id >> 1) & 0b11 { + 0 => Ok(StreamKind::Control), + 1 => Ok(StreamKind::Gossip), + 2 => Ok(StreamKind::Git), + n => Err(n as u8), + } + } + + /// Create a control identifier. + pub fn control(link: Link) -> Self { + match link { + Link::Outbound => Self(VarInt::from(0b000u8)), + Link::Inbound => Self(VarInt::from(0b001u8)), + } + } + + /// Create a gossip identifier. + pub fn gossip(link: Link) -> Self { + match link { + Link::Outbound => Self(VarInt::from(0b010u8)), + Link::Inbound => Self(VarInt::from(0b011u8)), + } + } + + /// Create a git identifier. + pub fn git(link: Link) -> Self { + match link { + Link::Outbound => Self(VarInt::from(0b100u8)), + Link::Inbound => Self(VarInt::from(0b101u8)), + } + } + + /// Get the nth identifier while preserving the stream type and initiator. + pub fn nth(self, n: u64) -> Result { + let id = *self.0 + (n << 3); + VarInt::new(id).map(Self) + } +} + +impl From for u64 { + fn from(value: StreamId) -> Self { + *value.0 + } +} + +impl From for VarInt { + fn from(value: StreamId) -> Self { + value.0 + } +} + +impl fmt::Display for StreamId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", *self.0) + } +} + +impl wire::Decode for StreamId { + fn decode(reader: &mut R) -> Result { + let id = VarInt::decode(reader)?; + Ok(Self(id)) + } +} + +impl wire::Encode for StreamId { + fn encode(&self, writer: &mut W) -> Result { + self.0.encode(writer) + } +} + +/// Type of stream. +#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)] +pub enum StreamKind { + /// Control stream, used to open and close streams. + Control, + /// Gossip stream, used to exchange messages. + Gossip, + /// Git stream, used for replication. + Git, +} + +/// Protocol frame. +/// +/// 0 1 2 3 +/// 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 +/// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +/// | 'r' | 'a' | 'd' | 0x1 | Version +/// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +/// | Stream ID |TTT|I| Stream ID with Stream [T]ype and [I]nitiator bits +/// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +/// | Data ...| Data (variable size) +/// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +pub struct Frame { + /// The protocol version. + pub version: Version, + /// The stream identifier. + pub stream: StreamId, + /// The frame payload. + pub data: FrameData, +} + +impl Frame { + /// Create a 'git' protocol frame. + pub fn git(stream: StreamId, data: Vec) -> Self { + Self { + version: PROTOCOL_VERSION, + stream, + data: FrameData::Git(data), + } + } + + /// Create a 'control' protocol frame. + pub fn control(link: Link, ctrl: Control) -> Self { + Self { + version: PROTOCOL_VERSION, + stream: StreamId::control(link), + data: FrameData::Control(ctrl), + } + } + + /// Create a 'gossip' protocol frame. + pub fn gossip(link: Link, msg: Message) -> Self { + Self { + version: PROTOCOL_VERSION, + stream: StreamId::gossip(link), + data: FrameData::Gossip(msg), + } + } + + /// Serialize frame to bytes. + pub fn to_bytes(&self) -> Vec { + wire::serialize(self) + } +} + +/// Frame payload. +pub enum FrameData { + /// Control frame payload. + Control(Control), + /// Gossip frame payload. + Gossip(Message), + /// Git frame payload. May contain packet-lines as well as packfile data. + Git(Vec), +} + +/// A control message sent over a control stream. +pub enum Control { + /// Open a new stream. + Open { + /// The stream to open. + stream: StreamId, + }, + /// Close an existing stream. + Close { + /// The stream to close. + stream: StreamId, + }, + /// Signal an end-of-file. This can be used to simulate connections terminating + /// without having to close the connection. These control messages are turned into + /// [`io::ErrorKind::UnexpectedEof`] errors on read. + Eof { + /// The stream to send an EOF on. + stream: StreamId, + }, +} + +impl wire::Decode for Control { + fn decode(reader: &mut R) -> Result { + let command = u8::decode(reader)?; + match command { + CONTROL_OPEN => { + let stream = StreamId::decode(reader)?; + Ok(Control::Open { stream }) + } + CONTROL_CLOSE => { + let stream = StreamId::decode(reader)?; + Ok(Control::Close { stream }) + } + CONTROL_EOF => { + let stream = StreamId::decode(reader)?; + Ok(Control::Eof { stream }) + } + other => Err(wire::Error::InvalidControlMessage(other)), + } + } +} + +impl wire::Encode for Control { + fn encode(&self, writer: &mut W) -> Result { + let mut n = 0; + + match self { + Self::Open { stream: id } => { + n += CONTROL_OPEN.encode(writer)?; + n += id.encode(writer)?; + } + Self::Eof { stream: id } => { + n += CONTROL_EOF.encode(writer)?; + n += id.encode(writer)?; + } + Self::Close { stream: id } => { + n += CONTROL_CLOSE.encode(writer)?; + n += id.encode(writer)?; + } + } + Ok(n) + } +} + +impl wire::Decode for Frame { + fn decode(reader: &mut R) -> Result { + let version = Version::decode(reader)?; + let stream = StreamId::decode(reader)?; + + match stream.kind() { + Ok(StreamKind::Control) => { + let ctrl = Control::decode(reader)?; + let frame = Frame { + version, + stream, + data: FrameData::Control(ctrl), + }; + Ok(frame) + } + Ok(StreamKind::Gossip) => { + let msg = Message::decode(reader)?; + let frame = Frame { + version, + stream, + data: FrameData::Gossip(msg), + }; + Ok(frame) + } + Ok(StreamKind::Git { .. }) => { + let size = VarInt::decode(reader)?; + let mut data = vec![0; *size as usize]; + reader.read_exact(&mut data[..])?; + + Ok(Frame::git(stream, data)) + } + Err(n) => Err(wire::Error::InvalidStreamKind(n)), + } + } +} + +impl wire::Encode for Frame { + fn encode(&self, writer: &mut W) -> Result { + let mut n = 0; + + n += self.version.encode(writer)?; + n += self.stream.encode(writer)?; + + match &self.data { + FrameData::Control(ctrl) => { + n += ctrl.encode(writer)?; + } + FrameData::Gossip(msg) => { + n += msg.encode(writer)?; + } + FrameData::Git(data) => { + let len = data.len(); + let size = VarInt::new(len as u64) + .map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?; + n += size.encode(writer)?; + + writer.write_all(data.as_slice())?; + n += len; + } + } + Ok(n) + } +} diff --git a/radicle-node/src/wire/message.rs b/radicle-node/src/wire/message.rs index 59fabf3a..5c1d9efd 100644 --- a/radicle-node/src/wire/message.rs +++ b/radicle-node/src/wire/message.rs @@ -19,8 +19,6 @@ pub enum MessageType { Subscribe = 8, Ping = 10, Pong = 12, - Fetch = 14, - FetchOk = 16, } impl From for u16 { @@ -40,8 +38,6 @@ impl TryFrom for MessageType { 8 => Ok(MessageType::Subscribe), 10 => Ok(MessageType::Ping), 12 => Ok(MessageType::Pong), - 14 => Ok(MessageType::Fetch), - 16 => Ok(MessageType::FetchOk), _ => Err(other), } } @@ -62,8 +58,6 @@ impl Message { }, Self::Ping { .. } => MessageType::Ping, Self::Pong { .. } => MessageType::Pong, - Self::Fetch { .. } => MessageType::Fetch, - Self::FetchOk { .. } => MessageType::FetchOk, } .into() } @@ -216,12 +210,6 @@ impl wire::Encode for Message { Self::Pong { zeroes } => { n += zeroes.encode(writer)?; } - Self::Fetch { rid } => { - n += rid.encode(writer)?; - } - Self::FetchOk { rid } => { - n += rid.encode(writer)?; - } } if n > wire::Size::MAX as usize { @@ -295,14 +283,6 @@ impl wire::Decode for Message { let zeroes = ZeroBytes::decode(reader)?; Ok(Self::Pong { zeroes }) } - Ok(MessageType::Fetch) => { - let rid = Id::decode(reader)?; - Ok(Self::Fetch { rid }) - } - Ok(MessageType::FetchOk) => { - let rid = Id::decode(reader)?; - Ok(Self::FetchOk { rid }) - } Err(other) => Err(wire::Error::UnknownMessageType(other)), } } diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index d915e687..46cf3061 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -7,7 +7,7 @@ use std::collections::VecDeque; use std::os::unix::io::AsRawFd; use std::os::unix::prelude::RawFd; use std::sync::Arc; -use std::{fmt, io, net, str}; +use std::{io, net}; use amplify::Wrapper as _; use crossbeam_channel as chan; @@ -18,7 +18,7 @@ use cyphernet::{Digest, EcSk, Ecdh, Sha256}; use localtime::LocalTime; use netservices::resource::{ListenerEvent, NetAccept, NetTransport, SessionEvent}; use netservices::session::{ProtocolArtifact, Socks5Session}; -use netservices::{NetConnection, NetProtocol, NetReader, NetSession, NetWriter}; +use netservices::{NetConnection, NetProtocol, NetReader, NetWriter}; use reactor::Timestamp; use radicle::collections::HashMap; @@ -27,11 +27,12 @@ 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::{Encode, Error}; -use crate::worker; -use crate::worker::{Task, TaskResult}; +use crate::service::reactor::Io; +use crate::service::{session, DisconnectReason, Service}; +use crate::wire::frame; +use crate::wire::frame::{Frame, FrameData, StreamId}; +use crate::wire::Encode; +use crate::worker::{ChannelEvent, Fetch, Task, TaskResult}; use crate::Link; use crate::{address, service}; @@ -41,22 +42,16 @@ pub const NOISE_XK: HandshakePattern = HandshakePattern { responder: cyphernet::encrypt::noise::OneWayPattern::Known, }; -#[allow(clippy::large_enum_variant)] /// Control message used internally between workers, users, and the service. -pub enum Control { +#[allow(clippy::large_enum_variant)] +#[derive(Debug)] +pub enum Control { /// Message from the user to the service. User(service::Command), /// Message from a worker to the service. - Worker(TaskResult), -} - -impl fmt::Debug for Control { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - Self::User(cmd) => cmd.fmt(f), - Self::Worker(resp) => resp.result.fmt(f), - } - } + Worker(TaskResult), + /// Flush data in the given stream to the remote. + Flush { remote: NodeId, stream: StreamId }, } /// Peer session type. @@ -69,6 +64,81 @@ pub type WireWriter = NetWriter, Socks5Session = reactor::Action>, NetTransport>>; +/// Worker channels used to send Git frames back and forth. +struct WorkerChannels { + /// Send data to the git worker. + sender: chan::Sender, + /// Receive data from the git worker. + receiver: chan::Receiver, +} + +/// Streams associated with a connected peer. +struct Streams { + /// Active streams and their associated worker channels. + /// Note that the gossip and control streams are not included here as they are always + /// implied to exist. + streams: HashMap, + /// Connection direction. + link: Link, + /// Sequence number used to compute the next stream id. + seq: u64, +} + +impl Streams { + /// Create a new [`Streams`] object, passing the connection link. + fn new(link: Link) -> Self { + Self { + streams: HashMap::default(), + link, + seq: 0, + } + } + + /// Get a known stream. + fn get(&self, stream: &StreamId) -> Option<&WorkerChannels> { + self.streams.get(stream) + } + + /// Open a new stream. + fn open(&mut self) -> (StreamId, WorkerChannels) { + self.seq += 1; + + let id = StreamId::git(self.link) + .nth(self.seq) + .expect("Streams::open: too many streams"); + let channels = self + .register(id) + .expect("Streams::open: stream was already open"); + + (id, channels) + } + + /// Register an open stream. + fn register(&mut self, stream: StreamId) -> Option { + let (wire_send, wire_recv) = chan::unbounded::(); + let (work_send, work_recv) = chan::unbounded::(); + + match self.streams.entry(stream) { + Entry::Vacant(e) => { + e.insert(WorkerChannels { + sender: wire_send, + receiver: work_recv, + }); + Some(WorkerChannels { + sender: work_send, + receiver: wire_recv, + }) + } + Entry::Occupied(_) => None, + } + } + + /// Unregister an open stream. + fn unregister(&mut self, stream: &StreamId) -> Option { + self.streams.remove(stream) + } +} + /// Peer connection state machine. enum Peer { /// The initial state of an inbound peer before handshake is completed. @@ -79,8 +149,9 @@ enum Peer { /// Peers in this state are handled by the underlying service. Connected { link: Link, - id: NodeId, - inbox: Deserializer, + nid: NodeId, + inbox: Deserializer, + streams: Streams, }, /// The peer was scheduled for disconnection. Once the transport is handed over /// by the reactor, we can consider it disconnected. @@ -88,16 +159,6 @@ enum Peer { id: Option, reason: DisconnectReason, }, - /// The state after we've started the process of upgraded the peer for a fetch. - /// The request to handover the socket was made to the reactor. - Upgrading { - fetch: Fetch, - link: Link, - id: NodeId, - inbox: Vec, - }, - /// The peer is now upgraded and we are in control of the socket. - Upgraded { link: Link, id: NodeId }, } impl std::fmt::Debug for Peer { @@ -105,16 +166,8 @@ impl std::fmt::Debug for Peer { match self { Self::Inbound {} => write!(f, "Inbound"), Self::Outbound { id } => write!(f, "Outbound({id})"), - Self::Connected { link, id, .. } => write!(f, "Connected({link:?}, {id})"), + Self::Connected { link, nid, .. } => write!(f, "Connected({link:?}, {nid})"), Self::Disconnecting { .. } => write!(f, "Disconnecting"), - Self::Upgrading { - fetch, link, id, .. - } => write!( - f, - "Upgrading(initiated={}, {link:?}, {id})", - fetch.is_initiator(), - ), - Self::Upgraded { link, id, .. } => write!(f, "Upgraded({link:?}, {id})"), } } } @@ -124,11 +177,8 @@ impl Peer { fn id(&self) -> Option<&NodeId> { match self { Peer::Outbound { id } - | Peer::Connected { id, .. } - | Peer::Disconnecting { id: Some(id), .. } - | Peer::Upgrading { id, .. } - | Peer::Upgraded { id, .. } => Some(id), - + | Peer::Connected { nid: id, .. } + | Peer::Disconnecting { id: Some(id), .. } => Some(id), Peer::Inbound {} => None, Peer::Disconnecting { id: None, .. } => None, } @@ -151,8 +201,9 @@ impl Peer { *self = Self::Connected { link, - id, + nid: id, inbox: Deserializer::default(), + streams: Streams::new(link), }; link } else if let Self::Outbound { id: expected } = self { @@ -161,8 +212,9 @@ impl Peer { *self = Self::Connected { link, - id, + nid: id, inbox: Deserializer::default(), + streams: Streams::new(link), }; link } else { @@ -172,7 +224,7 @@ impl Peer { /// Switch to disconnecting state. fn disconnecting(&mut self, reason: DisconnectReason) { - if let Self::Connected { id, .. } = self { + if let Self::Connected { nid: id, .. } = self { *self = Self::Disconnecting { id: Some(*id), reason, @@ -188,57 +240,6 @@ impl Peer { panic!("Peer::disconnected: session is not connected ({self:?})"); } } - - /// Switch to upgrading state. - fn upgrading(&mut self, fetch: Fetch) { - if let Self::Connected { id, link, inbox } = self { - *self = Self::Upgrading { - fetch, - id: *id, - link: *link, - inbox: inbox.unparsed().collect(), - }; - } else { - panic!("Peer::upgrading: session is not fully connected"); - } - } - - /// Switch to upgraded state. Returns the unread bytes from the peer. - #[must_use] - fn upgraded(&mut self) -> (Fetch, Vec) { - if let Self::Upgrading { - fetch, - id, - link, - inbox, - } = self - { - let fetch = fetch.clone(); - let inbox = inbox.drain(..).collect(); - log::debug!(target: "wire", "Peer {id} upgraded for fetch {}", fetch.rid); - - *self = Self::Upgraded { - id: *id, - link: *link, - }; - (fetch, inbox) - } else { - panic!("Peer::upgraded: can't upgrade before handover"); - } - } - - /// Switch back from upgraded to connected state. - fn downgrade(&mut self) { - if let Self::Upgraded { id, link, .. } = self { - *self = Self::Connected { - id: *id, - link: *link, - inbox: Deserializer::default(), - }; - } else { - panic!("Peer::downgrade: can't downgrade if not in upgraded state"); - } - } } /// Wire protocol implementation for a set of peers. @@ -246,7 +247,7 @@ pub struct Wire { /// Backing service instance. service: Service, /// Worker pool interface. - worker: chan::Sender>, + worker: chan::Sender, /// Used for authentication. signer: G, /// Internal queue of actions to send to the reactor. @@ -266,7 +267,7 @@ where { pub fn new( mut service: Service, - worker: chan::Sender>, + worker: chan::Sender, signer: G, proxy: net::SocketAddr, clock: LocalTime, @@ -289,13 +290,6 @@ where self.actions.push_back(Action::RegisterListener(socket)); } - fn peer_mut_by_fd(&mut self, fd: RawFd) -> &mut Peer { - self.peers.get_mut(&fd).unwrap_or_else(|| { - log::error!(target: "wire", "Peer with fd {fd} was not found"); - panic!("Peer with fd {fd} is not known"); - }) - } - fn fd_by_id(&self, node_id: &NodeId) -> (RawFd, &Peer) { self.peers .iter() @@ -304,6 +298,14 @@ where .unwrap_or_else(|| panic!("Peer {node_id} was expected to be known to the transport")) } + fn fd_by_id_mut(&mut self, node_id: &NodeId) -> (RawFd, &mut Peer) { + self.peers + .iter_mut() + .find(|(_, peer)| peer.id() == Some(node_id)) + .map(|(fd, peer)| (*fd, peer)) + .unwrap_or_else(|| panic!("Peer {node_id} was expected to be known to the transport")) + } + fn connected_fd_by_id(&self, node_id: &NodeId) -> RawFd { match self.fd_by_id(node_id) { (fd, Peer::Connected { .. }) => fd, @@ -319,16 +321,14 @@ where self.peers.iter().filter_map(|(fd, peer)| match peer { Peer::Inbound {} => None, Peer::Outbound { id } => Some((*fd, id)), - Peer::Connected { id, .. } => Some((*fd, id)), - Peer::Upgrading { id, .. } => Some((*fd, id)), - Peer::Upgraded { id, .. } => Some((*fd, id)), + Peer::Connected { nid: id, .. } => Some((*fd, id)), Peer::Disconnecting { .. } => None, }) } fn connected(&self) -> impl Iterator { self.peers.iter().filter_map(|(fd, peer)| { - if let Peer::Connected { id, .. } = peer { + if let Peer::Connected { nid: id, .. } = peer { Some((*fd, id)) } else { None @@ -353,65 +353,74 @@ where } } - fn upgrade(&mut self, fd: RawFd, fetch: Fetch) { - let peer = self.peer_mut_by_fd(fd); - if let Peer::Disconnecting { .. } = peer { - log::error!(target: "wire", "Peer (fd={fd}) is disconnecting"); + fn worker_result(&mut self, task: TaskResult) { + log::debug!(target: "wire", "Received fetch result from worker: {:?}", task.result); + + let nid = task.fetch.remote(); + let Some((fd, peer)) = self + .peers + .iter_mut() + .find(|(_, peer)| peer.id() == Some(&nid)) + .map(|(fd, peer)| (*fd, peer)) else { + log::warn!(target: "wire", "Peer {nid} not found; ignoring fetch result"); + return; + }; + + let Peer::Connected { nid, link, streams, .. } = peer else { + log::warn!(target: "wire", "Peer {nid} is not connected; ignoring fetch result"); return; }; - log::debug!(target: "wire", "Requesting transport handover from reactor for peer (fd={fd})"); - peer.upgrading(fetch); + let remote = *nid; - self.actions.push_back(Action::UnregisterTransport(fd)); - } + // Only call into the service if we initiated this fetch. + if let Some((rid, namespaces)) = task.fetch.initiated() { + self.service.fetched(rid, namespaces, remote, task.result); + } - fn upgraded(&mut self, transport: NetTransport>) { - let fd = transport.as_raw_fd(); - let peer = self.peer_mut_by_fd(fd); - let (fetch, drain) = peer.upgraded(); - let session = match transport.into_session() { - Ok(session) => session, - Err(_) => panic!("Wire::upgraded: peer write buffer not empty on upgrade"), - }; - - if self - .worker - .send(Task { - fetch, - session, - drain, - }) - .is_err() - { - log::error!(target: "wire", "Worker pool is disconnected; cannot send fetch request"); + // Nb. It's possible that the stream would already be unregistered if we received an early + // "close" from the remote. Otherwise, we unregister it here and send the "close" ourselves. + if streams.unregister(&task.stream).is_some() { + let frame = Frame::control( + *link, + frame::Control::Close { + stream: task.stream, + }, + ); + self.actions.push_back(Action::Send(fd, frame.to_bytes())); } } - fn worker_result(&mut self, task: TaskResult) { - log::debug!(target: "wire", "Fetch completed: {:?}", task.result); + fn flush(&mut self, remote: NodeId, stream: StreamId) { + let (fd, peer) = self + .peers + .iter() + .find(|(_, peer)| peer.id() == Some(&remote)) + .map(|(fd, peer)| (*fd, peer)) + .unwrap_or_else(|| panic!("Peer {remote} was expected to be known to the transport")); - let session = task.session; - let fd = session.as_connection().as_raw_fd(); - let peer = self.peer_mut_by_fd(fd); - - let session = if let Peer::Disconnecting { .. } = peer { - log::error!(target: "wire", "Peer with fd {fd} is disconnecting"); + let Peer::Connected { streams, link, .. } = peer else { + log::warn!(target: "wire", "Peer {remote} is not connected; ignoring flush"); + return; + }; + let Some(c) = streams.get(&stream) else { + log::debug!(target: "wire", "Stream {stream} cannot be found; ignoring flush"); return; - } else if let Peer::Upgraded { link, .. } = peer { - match NetTransport::with_session(session, *link) { - Ok(session) => session, - Err(err) => { - log::error!(target: "wire", "Session downgrade failed: {err}"); - return; - } - } - } else { - todo!(); }; - peer.downgrade(); - self.actions.push_back(Action::RegisterTransport(session)); - self.service.fetched(task.fetch, task.result); + #[cfg(test)] + if c.receiver.is_empty() { + panic!("Wire:flush: redundant flush"); + } + + for data in c.receiver.try_iter() { + let frame = match data { + ChannelEvent::Data(data) => Frame::git(stream, data), + ChannelEvent::Close => Frame::control(*link, frame::Control::Close { stream }), + ChannelEvent::Eof => Frame::control(*link, frame::Control::Eof { stream }), + }; + self.actions + .push_back(reactor::Action::Send(fd, frame.to_bytes())); + } } } @@ -424,7 +433,7 @@ where { type Listener = NetAccept>; type Transport = NetTransport>; - type Command = Control; + type Command = Control; fn tick(&mut self, time: Timestamp) { self.service @@ -508,57 +517,98 @@ where self.service.connected(id, link); } SessionEvent::Data(data) => { - if let Some(Peer::Connected { id, inbox, .. }) = self.peers.get_mut(&fd) { + if let Some(Peer::Connected { + nid, + inbox, + streams, + .. + }) = self.peers.get_mut(&fd) + { inbox.input(&data); loop { match inbox.deserialize_next() { - Ok(Some(msg)) => self.service.received_message(*id, msg), + Ok(Some(Frame { + data: FrameData::Control(frame::Control::Open { stream }), + .. + })) => { + log::debug!(target: "wire", "Received stream open for id={stream}"); + + let Some(WorkerChannels { + sender: work_send, + receiver: wire_recv, + }) = streams.register(stream) else { + log::warn!(target: "wire", "Peer attempted to open already-open stream id={stream}"); + continue; + }; + + let task = Task { + fetch: Fetch::Responder { remote: *nid }, + stream, + send: work_send, + recv: wire_recv, + }; + if self.worker.send(task).is_err() { + log::error!(target: "wire", "Worker pool is disconnected; cannot send task"); + } + } + Ok(Some(Frame { + data: FrameData::Control(frame::Control::Eof { stream }), + .. + })) => { + if let Some(channels) = streams.get(&stream) { + if channels.sender.send(ChannelEvent::Eof).is_err() { + log::error!(target: "wire", "Worker is disconnected; cannot send `EOF`"); + } + } else { + log::debug!(target: "wire", "Ignoring frame on closed or unknown stream id={stream}"); + } + } + Ok(Some(Frame { + data: FrameData::Control(frame::Control::Close { stream }), + .. + })) => { + log::debug!(target: "wire", "Received stream close command for id={stream}"); + + streams.unregister(&stream); + } + Ok(Some(Frame { + data: FrameData::Gossip(msg), + .. + })) => { + self.service.received_message(*nid, msg); + } + Ok(Some(Frame { + stream, + data: FrameData::Git(data), + .. + })) => { + if let Some(channels) = streams.get(&stream) { + if channels.sender.send(ChannelEvent::Data(data)).is_err() { + log::error!(target: "wire", "Worker is disconnected; cannot send data"); + } + } else { + log::debug!(target: "wire", "Ignoring frame on closed or unknown stream id={stream}"); + } + } Ok(None) => { // Buffer is empty, or message isn't complete. break; } Err(e) => { - log::error!(target: "wire", "Invalid gossip message from {id}: {e}"); - - let mut reason = - DisconnectReason::Session(session::Error::Misbehavior); - if let Error::UnknownMessageType(t) = e { - let leftover = - inbox.unparsed().chain(t.to_be_bytes()).collect::>(); - - if let Ok(header) = - str::from_utf8(&leftover[..worker::pktline::HEADER_LEN]) - { - if header.is_ascii() && !header.is_empty() { - log::error!( - target: "wire", - "Received possible Git packet-line header `{}` from {id} (protocol mismatch)", - header - ); - // In case of protocol mismatch, don't penalize the peer. - reason = DisconnectReason::Session( - session::Error::ProtocolMismatch, - ); - } - } - } + log::error!(target: "wire", "Invalid gossip message from {nid}: {e}"); if !inbox.is_empty() { - log::debug!(target: "wire", "Dropping read buffer for {id} with {} bytes", inbox.unparsed().count()); + log::debug!(target: "wire", "Dropping read buffer for {nid} with {} bytes", inbox.unparsed().count()); } self.disconnect( - fd, // TODO(cloudhead): Include error in reason. - reason, + fd, + DisconnectReason::Session(session::Error::Misbehavior), ); break; } } } - } else if let Some(Peer::Upgrading { inbox, .. }) = self.peers.get_mut(&fd) { - // If somehow the remote peer managed to send git data before the reactor - // unregistered our session, we'll hit this branch. - inbox.extend(data); } else { log::warn!(target: "wire", "Dropping message from unconnected peer (fd={fd})"); } @@ -573,6 +623,7 @@ where match cmd { Control::User(cmd) => self.service.command(cmd), Control::Worker(result) => self.worker_result(result), + Control::Flush { remote, stream } => self.flush(remote, stream), } } @@ -604,7 +655,11 @@ where } reactor::Error::TransportPollError(fd, _) => { log::error!(target: "wire", "Received error: peer (fd={fd}) poll error"); - self.actions.push_back(Action::UnregisterTransport(*fd)); + + self.disconnect( + *fd, + DisconnectReason::Connection(Arc::new(io::Error::from(io::ErrorKind::Other))), + ) } reactor::Error::TransportDisconnect(fd, _, _) => { log::error!(target: "wire", "Received error: peer (fd={fd}) disconnected"); @@ -661,9 +716,6 @@ where } e.remove(); } - Peer::Upgrading { .. } => { - self.upgraded(transport); - } _ => { panic!("Wire::handover_transport: Unexpected peer with fd {fd} handed over from the reactor"); } @@ -689,8 +741,8 @@ where while let Some(ev) = self.service.next() { match ev { Io::Write(node_id, msgs) => { - let fd = match self.fd_by_id(&node_id) { - (fd, Peer::Connected { .. }) => fd, + let (fd, link) = match self.fd_by_id(&node_id) { + (fd, Peer::Connected { link, .. }) => (fd, *link), (_, peer) => { // If the peer is disconnected by the wire protocol, the service may // not be aware of this yet, and may continue to write messages to it. @@ -704,7 +756,9 @@ where let mut data = Vec::new(); for msg in msgs { - msg.encode(&mut data).expect("in-memory writes never fail"); + Frame::gossip(link, msg) + .encode(&mut data) + .expect("in-memory writes never fail"); } self.actions.push_back(reactor::Action::Send(fd, data)); } @@ -753,10 +807,40 @@ where Io::Wakeup(d) => { self.actions.push_back(reactor::Action::SetTimer(d.into())); } - Io::Fetch(fetch) => { - // TODO: Check that the node_id is connected, queue request otherwise. - let fd = self.connected_fd_by_id(&fetch.remote); - self.upgrade(fd, fetch); + Io::Fetch { + rid, + remote, + namespaces, + } => { + log::debug!(target: "wire", "Processing fetch.."); + + let (fd, Peer::Connected { link, streams, .. }) = + self.fd_by_id_mut(&remote) else { + panic!("Wire::next: peer {remote} is not connected"); + }; + let (stream, channels) = streams.open(); + + log::debug!(target: "wire", "Opened new stream with id={stream} for rid={rid}"); + + let link = *link; + let task = Task { + fetch: Fetch::Initiator { + rid, + namespaces, + remote, + }, + stream, + send: channels.sender, + recv: channels.receiver, + }; + + if self.worker.send(task).is_err() { + log::error!(target: "wire", "Worker pool is disconnected; cannot send fetch request"); + } + self.actions.push_back(Action::Send( + fd, + Frame::control(link, frame::Control::Open { stream }).to_bytes(), + )); } } } diff --git a/radicle-node/src/wire/varint.rs b/radicle-node/src/wire/varint.rs new file mode 100644 index 00000000..48bc74e7 --- /dev/null +++ b/radicle-node/src/wire/varint.rs @@ -0,0 +1,225 @@ +//! Variable-length integer implementation based on QUIC. +#![warn(clippy::missing_docs_in_private_items)] + +// This implementation is largely based on the `quinn` crate. +// Copyright (c) 2018 The quinn developers. +use std::{fmt, io, ops}; + +use byteorder::ReadBytesExt; +use thiserror::Error; + +use crate::wire; +use crate::wire::{Decode, Encode}; + +/// An integer less than 2^62 +/// +/// Based on QUIC variable-length integers (RFC 9000). +/// +/// > The QUIC variable-length integer encoding reserves the two most significant bits of the first +/// > byte to encode the base-2 logarithm of the integer encoding length in bytes. The integer value is +/// > encoded on the remaining bits, in network byte order. This means that integers are encoded on 1, +/// > 2, 4, or 8 bytes and can encode 6-, 14-, 30-, or 62-bit values, respectively. Table 4 summarizes +/// > the encoding properties. +/// +/// ```text +/// MSB Length Usable Bits Range +/// ---------------------------------------------------- +/// 00 1 6 0 - 63 +/// 01 2 14 0 - 16383 +/// 10 4 30 0 - 1073741823 +/// 11 8 62 0 - 4611686018427387903 +/// ``` +#[derive(Default, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)] +pub struct VarInt(pub(crate) u64); + +impl VarInt { + /// The largest representable value. + pub const MAX: VarInt = VarInt((1 << 62) - 1); + + /// Succeeds iff `x` < 2^62. + pub fn new(x: u64) -> Result { + if x <= Self::MAX.0 { + Ok(Self(x)) + } else { + Err(BoundsExceeded) + } + } +} + +impl ops::Deref for VarInt { + type Target = u64; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl From for VarInt { + fn from(x: u8) -> Self { + VarInt(x.into()) + } +} + +impl From for VarInt { + fn from(x: u16) -> Self { + VarInt(x.into()) + } +} + +impl From for VarInt { + fn from(x: u32) -> Self { + VarInt(x.into()) + } +} + +impl std::convert::TryFrom for VarInt { + type Error = BoundsExceeded; + /// Succeeds iff `x` < 2^62. + fn try_from(x: u64) -> Result { + VarInt::new(x) + } +} + +impl fmt::Debug for VarInt { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(f) + } +} + +impl fmt::Display for VarInt { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.0.fmt(f) + } +} + +/// Error returned when constructing a `VarInt` from a value >= 2^62. +#[derive(Debug, Copy, Clone, Eq, PartialEq, Error)] +#[error("value too large for varint encoding")] +pub struct BoundsExceeded; + +impl Decode for VarInt { + fn decode(r: &mut R) -> Result { + let mut buf = [0; 8]; + buf[0] = r.read_u8()?; + + // Integer length. + let tag = buf[0] >> 6; + buf[0] &= 0b0011_1111; + + let x = match tag { + 0b00 => u64::from(buf[0]), + 0b01 => { + r.read_exact(&mut buf[1..2])?; + u64::from(u16::from_be_bytes([buf[0], buf[1]])) + } + 0b10 => { + r.read_exact(&mut buf[1..4])?; + u64::from(u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]])) + } + 0b11 => { + r.read_exact(&mut buf[1..8])?; + u64::from_be_bytes(buf) + } + // SAFETY: It should be obvious that we can't have any other bit pattern + // than the above, since all other bits are zeroed. + _ => unreachable! {}, + }; + Ok(Self(x)) + } +} + +impl Encode for VarInt { + fn encode(&self, w: &mut W) -> io::Result { + let x: u64 = self.0; + + if x < 2u64.pow(6) { + (x as u8).encode(w) + } else if x < 2u64.pow(14) { + (0b01 << 14 | x as u16).encode(w) + } else if x < 2u64.pow(30) { + (0b10 << 30 | x as u32).encode(w) + } else if x < 2u64.pow(62) { + (0b11 << 62 | x).encode(w) + } else { + panic!("VarInt::encode: integer overflow"); + } + } +} + +#[cfg(test)] +mod test { + use super::*; + use qcheck_macros::quickcheck; + + impl qcheck::Arbitrary for VarInt { + fn arbitrary(g: &mut qcheck::Gen) -> Self { + let a = u16::arbitrary(g) as u64; + let b = u32::arbitrary(g) as u64; + let n = g + .choose(&[ + 0, + 1, + 3, + 7, + 13, + 37, + 255, + 4931, + 54019, + 69149, + 151288809941952652, + u8::MAX as u64, + u16::MAX as u64, + u16::MAX as u64 - 1, + u32::MAX as u64, + u32::MAX as u64 - 1, + *Self::MAX, + a, + b, + ]) + .copied() + .unwrap(); + + Self(n) + } + } + + #[quickcheck] + fn prop_encode_decode(input: VarInt) { + let encoded = wire::serialize(&input); + let decoded: VarInt = wire::deserialize(&encoded).unwrap(); + + assert_eq!(decoded, input); + } + + #[test] + #[should_panic] + fn test_encode_overflow() { + wire::serialize(&VarInt(u64::MAX)); + } + + #[test] + fn test_encoding() { + assert_eq!(wire::serialize(&VarInt(0)), vec![0x0]); + assert_eq!(wire::serialize(&VarInt(1)), vec![0x01]); + assert_eq!(wire::serialize(&VarInt(10)), vec![0x0a]); + assert_eq!(wire::serialize(&VarInt(37)), vec![0x25]); + assert_eq!( + wire::deserialize::(&[0x40, 0x25]).unwrap(), + VarInt(37) + ); + assert_eq!(wire::serialize(&VarInt(15293)), vec![0x7b, 0xbd]); + assert_eq!( + wire::serialize(&VarInt(494878333)), + vec![0x9d, 0x7f, 0x3e, 0x7d], + ); + assert_eq!( + wire::serialize(&VarInt(151288809941952652)), + vec![0xc2, 0x19, 0x7c, 0x5e, 0xff, 0x14, 0xe8, 0x8c] + ); + assert_eq!( + wire::serialize(&VarInt(10000000000)), + vec![0xc0, 0x00, 0x00, 0x02, 0x54, 0x0b, 0xe4, 0x00], + ); + } +} diff --git a/radicle-node/src/worker.rs b/radicle-node/src/worker.rs index 50da6994..c79a89d0 100644 --- a/radicle-node/src/worker.rs +++ b/radicle-node/src/worker.rs @@ -1,4 +1,6 @@ +mod channels; mod fetch; +mod tunnel; use std::io::{prelude::*, BufReader}; use std::ops::ControlFlow; @@ -6,20 +8,19 @@ use std::thread::JoinHandle; use std::{env, io, net, process, thread, time}; use crossbeam_channel as chan; -use cyphernet::Ecdh; -use netservices::tunnel::Tunnel; -use netservices::{AsConnection, NetSession, SplitIo}; -use radicle::crypto::Signer; use radicle::identity::{Id, IdentityError}; +use radicle::prelude::NodeId; use radicle::storage::{Namespaces, ReadRepository, RefUpdate}; use radicle::{git, Storage}; -use reactor::poller::popol; use crate::runtime::Handle; -use crate::service::reactor::{Fetch, FetchDirection}; use crate::storage; -use crate::wire::{WireReader, WireSession, WireWriter}; +use crate::wire::StreamId; +use channels::{ChannelReader, ChannelWriter, Channels}; +use tunnel::Tunnel; + +pub use channels::ChannelEvent; /// Worker pool configuration. pub struct Config { @@ -54,8 +55,6 @@ pub enum FetchError { Identity(#[from] IdentityError), #[error("upload failed: {0}")] Upload(#[from] UploadError), - #[error("remote aborted fetch")] - RemoteAbortedFetch, #[error(transparent)] StagingInit(#[from] fetch::error::Init), #[error(transparent)] @@ -76,8 +75,6 @@ impl FetchError { pub enum UploadError { #[error("worker failed to connect to git daemon: {0}")] DaemonConnectionFailed(io::Error), - #[error("git pkt-line command does not match fetch request")] - CommandMismatch, #[error("error parsing git command packet-line: {0}")] InvalidPacketLine(io::Error), #[error(transparent)] @@ -91,33 +88,73 @@ impl UploadError { } } +/// Fetch job sent to worker thread. +#[derive(Debug, Clone)] +pub enum Fetch { + /// Client is initiating a fetch in order to receive the specified + /// `refspecs` determined by [`Namespaces`]. + Initiator { + /// Repo to fetch. + rid: Id, + /// Namespaces to fetch. + namespaces: Namespaces, + /// Remote peer we are interacting with. + remote: NodeId, + }, + /// Server is responding to a fetch request by uploading the + /// specified `refspecs` sent by the client. + Responder { + /// Remote peer we are interacting with. + remote: NodeId, + }, +} + +impl Fetch { + pub fn remote(&self) -> NodeId { + match self { + Self::Initiator { remote, .. } | Self::Responder { remote } => *remote, + } + } + + pub fn initiated(self) -> Option<(Id, Namespaces)> { + match self { + Self::Initiator { + rid, namespaces, .. + } => Some((rid, namespaces)), + Self::Responder { .. } => None, + } + } +} + /// Task to be accomplished on a worker thread. /// This is either going to be an outgoing or incoming fetch. -pub struct Task { +pub struct Task { pub fetch: Fetch, - pub session: WireSession, - pub drain: Vec, + pub stream: StreamId, + pub send: chan::Sender, + pub recv: chan::Receiver, } /// Worker response. -pub struct TaskResult { +#[derive(Debug)] +pub struct TaskResult { pub fetch: Fetch, + pub stream: StreamId, pub result: Result, FetchError>, - pub session: WireSession, } /// A worker that replicates git objects. -struct Worker { +struct Worker { storage: Storage, - tasks: chan::Receiver>, + tasks: chan::Receiver, daemon: net::SocketAddr, timeout: time::Duration, - handle: Handle, + handle: Handle, atomic: bool, name: String, } -impl Worker { +impl Worker { /// Waits for tasks and runs them. Blocks indefinitely unless there is an error receiving /// the next task. fn run(mut self) -> Result<(), chan::RecvError> { @@ -127,17 +164,15 @@ impl Worker { } } - fn process(&mut self, task: Task) { + fn process(&mut self, task: Task) { let Task { fetch, - session, - drain, + recv, + send, + stream, } = task; - - let timeout = session.as_connection().read_timeout().unwrap_or_default(); - let (session, result) = self._process(&fetch, drain, session); - // In case the timeout is changed during the fetch, we reset it here. - session.as_connection().set_read_timeout(timeout).ok(); + let channels = Channels::new(send, recv); + let result = self._process(&fetch, stream, channels); log::debug!(target: "worker", "Sending response back to service.."); @@ -145,8 +180,8 @@ impl Worker { .handle .worker_result(TaskResult { fetch, + stream, result, - session, }) .is_err() { @@ -155,176 +190,135 @@ impl Worker { } fn _process( - &self, + &mut self, fetch: &Fetch, - drain: Vec, - mut session: WireSession, - ) -> (WireSession, Result, FetchError>) { - match &fetch.direction { - FetchDirection::Initiator { namespaces } => { - log::debug!(target: "worker", "Worker processing outgoing fetch for {}", fetch.rid); + stream: StreamId, + mut channels: Channels, + ) -> Result, FetchError> { + match &fetch { + Fetch::Initiator { + rid, + namespaces, + remote, + } => { + log::debug!(target: "worker", "Worker processing outgoing fetch for {}", rid); - let (session, result) = self.fetch(fetch.rid, namespaces, session); + let result = self.fetch(*rid, *remote, stream, namespaces, channels); if let Err(err) = &result { log::error!(target: "worker", "Fetch error: {err}"); } - - (session, result) + result } - FetchDirection::Responder => { - log::debug!(target: "worker", "Worker processing incoming fetch for {}", fetch.rid); + Fetch::Responder { .. } => { + log::debug!(target: "worker", "Worker processing incoming fetch.."); - if let Err(err) = session.as_connection_mut().set_nonblocking(false) { - return (session, Err(err.into())); - } - - let (mut stream_r, mut stream_w) = match session.split_io() { - Ok((r, w)) => (r, w), - Err(err) => { - return (err.original, Err(err.error.into())); + let (stream_w, mut stream_r) = channels.split(); + let mut pktline_r = pktline::Reader::new(&mut stream_r); + // Nb. two fetches are usually expected: one for the *special* refs, + // followed by another for the signed refs. + loop { + match self.upload_pack(fetch, stream, &mut pktline_r, stream_w) { + Ok(ControlFlow::Continue(())) => continue, + Ok(ControlFlow::Break(())) => break, + Err(e) => return Err(e.into()), } - }; - let mut pktline_r = pktline::Reader::new(drain, &mut stream_r); - // Nb. two fetches are expected to happen, one for - // the `rad` refs, followed by the refs listed in - // signed refs. - let result = self.upload_pack(fetch, &mut pktline_r, &mut stream_w); - let result = - result.and_then(|_| self.upload_pack(fetch, &mut pktline_r, &mut stream_w)); - (WireSession::from_split_io(stream_r, stream_w), result) + } + Ok(vec![]) } } } fn fetch( - &self, + &mut self, rid: Id, + remote: NodeId, + stream: StreamId, namespaces: &Namespaces, - session: WireSession, - ) -> (WireSession, Result, FetchError>) { - let staging = match fetch::StagingPhaseInitial::new(&self.storage, rid, namespaces.clone()) - { - Ok(staging) => staging, - Err(err) => return (session, Err(err.into())), - }; + mut channels: Channels, + ) -> Result, FetchError> { + let staging = fetch::StagingPhaseInitial::new(&self.storage, rid, namespaces.clone())?; - let session = match self.tunnel_fetch(&staging.repo, staging.refspecs(), session) { - (session, Ok(())) => session, - (session, Err(err)) => return (session, Err(err)), - }; + self._fetch( + &staging.repo, + remote, + staging.refspecs(), + stream, + &mut channels, + )?; + if let Err(e) = self.handle.flush(remote, stream) { + log::error!(target: "worker", "Error flushing worker stream: {e}"); + } let staging = match staging.into_final().map_err(FetchError::from) { Ok(staging) => staging, - Err(e) => return (session, Err(e)), + Err(e) => return Err(e), }; - - let (session, res) = self.tunnel_fetch(&staging.repo, staging.refspecs(), session); - - if let Err(e) = res { - return (session, Err(e)); + self._fetch( + &staging.repo, + remote, + staging.refspecs(), + stream, + &mut channels, + )?; + if let Err(e) = self.handle.flush(remote, stream) { + log::error!(target: "worker", "Error flushing worker stream: {e}"); } - - (session, staging.transfer().map_err(FetchError::from)) + staging.transfer().map_err(FetchError::from) } fn upload_pack( - &self, + &mut self, fetch: &Fetch, - pktline_r: &mut pktline::Reader, - stream_w: &mut WireWriter, - ) -> Result, FetchError> { - match self._upload_pack(fetch, pktline_r, stream_w) { - Ok(()) => { - log::debug!(target: "worker", "Upload of {} to {} exited successfully", fetch.rid, fetch.remote); + stream: StreamId, + pktline_r: &mut pktline::Reader<&mut ChannelReader>, + stream_w: &mut ChannelWriter, + ) -> Result, UploadError> { + log::debug!(target: "worker", "Waiting for Git request pktline for.."); - Ok(vec![]) - } - Err(err) => { - log::error!(target: "worker", "Upload error for {}: {err}", fetch.rid); - - // If we exited without receiving a `done` packet, wait for it here. - // It's possible that the daemon exited first, or the remote crashed. - log::debug!(target: "worker", "Waiting for `done` packet from remote.."); - let mut header = [0; pktline::HEADER_LEN]; - - // Set the read timeout for the `done` packet to twice the configured - // value that is used for the fetching (initiator) side. - // - // This is because the uploader always waits for the `done` packet; - // so in case the fetch is aborted by the uploader, eg. if - // it can't connect with the daemon, it will wait long enough for the - // fetcher to timeout before timing out itself, and will thus receive - // the `done` packet. - pktline_r - .stream() - .as_connection() - .set_read_timeout(Some(self.timeout * 2)) - .ok(); - - loop { - match pktline_r.read_done_pktline(&mut header) { - Ok(()) => { - log::debug!(target: "worker", "Received `done` packet from remote"); - - // If we get the `done` packet, we exit with the original - // error. - return Err(err.into()); - } - Err(e) if e.kind() == io::ErrorKind::InvalidInput => { - // If we get some other packet, because the fetch request - // is still sending stuff, we simply keep reading until we - // get a `done` packet. - continue; - } - Err(_) => { - // If we get any other error, eg. a timeout, we abort. - log::error!( - target: "worker", - "Upload of {} to {} aborted: missing `done` packet from remote", - fetch.rid, - fetch.remote - ); - return Err(FetchError::RemoteAbortedFetch); - } - } - } - } - } - } - - fn _upload_pack( - &self, - fetch: &Fetch, - stream_r: &mut pktline::Reader, - stream_w: &mut WireWriter, - ) -> Result<(), UploadError> { // Read the request packet line to make sure the repository being requested matches what // we expect, and that the service requested is valid. - let request = match stream_r.read_request_pktline() { - Ok((req, pktline)) => { - log::debug!( - target: "worker", - "Parsed git command packet-line for {}: {:?}", fetch.rid, req - ); - if req.repo != fetch.rid { - return Err(UploadError::CommandMismatch); - } - pktline + let (rid, request) = match pktline_r.read_request_pktline() { + Ok((req, pktline)) => (req.repo, pktline), + Err(err) if err.kind() == io::ErrorKind::ConnectionReset => { + return Ok(ControlFlow::Break(())); } Err(err) => { return Err(UploadError::InvalidPacketLine(err)); } }; + log::debug!(target: "worker", "Received Git request pktline for {rid}.."); + + match self._upload_pack(rid, fetch.remote(), request, stream, pktline_r, stream_w) { + Ok(()) => { + log::debug!(target: "worker", "Upload of {rid} to {} exited successfully", fetch.remote()); + + Ok(ControlFlow::Continue(())) + } + Err(e) => Err(e), + } + } + + fn _upload_pack( + &mut self, + rid: Id, + remote: NodeId, + request: Vec, + stream: StreamId, + stream_r: &mut pktline::Reader<&mut ChannelReader>, + stream_w: &mut ChannelWriter, + ) -> Result<(), UploadError> { + log::debug!(target: "worker", "Connecting to daemon.."); // Connect to our local git daemon, running as a child process. let daemon = net::TcpStream::connect_timeout(&self.daemon, self.timeout) .map_err(UploadError::DaemonConnectionFailed)?; let (mut daemon_r, mut daemon_w) = (daemon.try_clone()?, daemon); - let mut daemon_r = pktline::Reader::new(vec![], &mut daemon_r); + let mut daemon_r = pktline::Reader::new(&mut daemon_r); - // Write the raw request to the daemon, once we've verified it. + // Write the raw request to the daemon, once we've parsed it. daemon_w.write_all(&request)?; + log::debug!(target: "worker", "Entering Git protocol loop for {rid}.."); // We now loop, alternating between reading requests from the client, and writing responses // back from the daemon.. Requests are delimited with a flush packet (`flush-pkt`). let mut buffer = [0; u16::MAX as usize + 1]; @@ -333,23 +327,28 @@ impl Worker { if let Err(e) = daemon_r.pipe(stream_w, &mut buffer) { // This is the expected error when the daemon disconnects. if e.kind() == io::ErrorKind::UnexpectedEof { - log::debug!(target: "worker", "Daemon closed the git connection for {}", fetch.rid); + log::debug!(target: "worker", "Daemon closed the git connection for {rid}"); + log::debug!(target: "worker", "Waiting for EOF from remote.."); + + stream_r.wait_for_eof()?; + + return Ok(()); } return Err(e.into()); } + + if let Err(e) = self.handle.flush(remote, stream) { + log::error!(target: "worker", "Worker channel disconnected; aborting"); + return Err(e.into()); + } + // Read from the stream and write to the daemon. match stream_r.pipe(&mut daemon_w, &mut buffer) { - // Triggered by a [`pktline::DONE_PKT`] packet. - Ok(ControlFlow::Break(())) => { - log::debug!(target: "worker", "Received `done` packet from remote for {}", fetch.rid); - return Ok(()); - } - Ok(ControlFlow::Continue(())) => { - continue; - } + Ok(()) => continue, + Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(()), Err(e) => { - if e.kind() == io::ErrorKind::UnexpectedEof { - log::debug!(target: "worker", "Remote closed the git connection for {}", fetch.rid); + if e.kind() == io::ErrorKind::ConnectionReset { + log::debug!(target: "worker", "Remote closed the git connection for {rid}"); } return Err(e.into()); } @@ -357,41 +356,20 @@ impl Worker { } } - fn tunnel_fetch( - &self, - repo: &fetch::StagedRepository, - specs: Specs, - session: WireSession, - ) -> (WireSession, Result<(), FetchError>) - where - Specs: fetch::AsRefspecs, - { - let mut tunnel = match Tunnel::with(session, net::SocketAddr::from(([0, 0, 0, 0], 0))) { - Ok(tunnel) => tunnel, - Err((session, err)) => return (session, Err(err.into())), - }; - let res = self._fetch(repo, specs, &mut tunnel); - - let mut session = tunnel.into_session(); - - log::debug!(target: "worker", "Sending `done` packet to remote.."); - if let Err(err) = pktline::done(&mut session) { - log::error!(target: "worker", "Fetch error: error sending `done` packet: {err}"); - } - (session, res) - } - fn _fetch( &self, repo: &storage::git::Repository, + remote: NodeId, specs: S, - tunnel: &mut Tunnel>, + stream: StreamId, + channels: &mut Channels, ) -> Result<(), FetchError> where S: fetch::AsRefspecs, { + let mut tunnel = Tunnel::with(channels, stream, remote, self.handle.clone())?; let rid = repo.id; - let tunnel_addr = tunnel.local_addr()?; + let tunnel_addr = tunnel.local_addr(); let mut cmd = process::Command::new("git"); cmd.current_dir(repo.path()) .env_clear() @@ -429,10 +407,10 @@ impl Worker { } })?; - let _ = tunnel.tunnel_once(popol::Poller::new(), self.timeout)?; + tunnel.run(self.timeout)?; let result = child.wait()?; - if result.success() { + let result = if result.success() { log::debug!(target: "worker", "Fetch for {} exited successfully", rid); Ok(()) } else { @@ -440,7 +418,14 @@ impl Worker { Err(FetchError::CommandFailed { code: result.code().unwrap_or(1), }) + }; + + log::debug!(target: "worker", "Sending `EOF` to remote.."); + + if let Err(e) = channels.sender.eof() { + log::error!(target: "worker", "Fetch error: error sending `EOF` message: {e}"); } + result } } @@ -451,11 +436,7 @@ pub struct Pool { impl Pool { /// Create a new worker pool with the given parameters. - pub fn with( - tasks: chan::Receiver>, - handle: Handle, - config: Config, - ) -> Self { + pub fn with(tasks: chan::Receiver, handle: Handle, config: Config) -> Self { let mut pool = Vec::with_capacity(config.capacity); for _ in 0..config.capacity { let worker = Worker { @@ -495,7 +476,7 @@ impl Pool { pub mod pktline { use std::io; use std::io::Read; - use std::ops::ControlFlow; + use std::net::TcpStream; use std::str; use super::Id; @@ -504,84 +485,69 @@ pub mod pktline { pub const FLUSH_PKT: &[u8; HEADER_LEN] = b"0000"; pub const DELIM_PKT: &[u8; HEADER_LEN] = b"0001"; pub const RESPONSE_END_PKT: &[u8; HEADER_LEN] = b"0002"; - /// When the remote `fetch` exits, it sends a special `done` packet which triggers - /// an EOF. This `done` packet is not part of the git protocol, and so is - /// not sent to the deamon. - pub const DONE_PKT: &[u8; HEADER_LEN] = b"done"; - - /// Packetline read result. - #[derive(Debug, PartialEq, Eq)] - pub enum Packetline { - /// Received a `done` control packet. - Done, - /// Received a git packet with the given length. - Git(usize), - } - - /// Send a special `done` packet. Since the git protocol is tunneled over an existing - /// connection, we can't signal the end of the protocol via the usual means, which is - /// to close the connection and trigger an EOF on the other side. Git also doesn't have - /// any special message we can send to signal the end of the protocol. Hence, we there's - /// no other way for the server to know that we're done sending commands than to send a - /// message that is not part of the git protocol. This message can then be processed by - /// the remote worker to end the protocol. - pub fn done(w: &mut W) -> io::Result<()> { - w.write_all(DONE_PKT) - } pub struct Reader<'a, R> { - drain: Vec, stream: &'a mut R, } + impl<'a> Reader<'a, TcpStream> { + /// Check whether the stream ended. + pub fn is_eof(&self) -> io::Result { + // Use non-blocking mode instead of timeouts, as we don't want to mess + // with existing timeouts. + self.stream.set_nonblocking(true)?; + let eof = match self.stream.peek(&mut []) { + Ok(0) => true, + Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => true, + _ => false, + }; + self.stream.set_nonblocking(false)?; + + Ok(eof) + } + } + impl<'a, R: io::Read> Reader<'a, R> { /// Create a new packet-line reader. - pub fn new(drain: Vec, stream: &'a mut R) -> Self { - Self { drain, stream } + pub fn new(stream: &'a mut R) -> Self { + Self { stream } } - /// Return the underlying stream. - pub fn stream(&self) -> &R { + /// Get the underlying stream. + pub fn stream(&mut self) -> &mut R { self.stream } + /// Wait for EOF. + pub fn wait_for_eof(&mut self) -> io::Result<()> { + match self.stream.read_to_end(&mut Vec::new()) { + Ok(_) => Ok(()), + Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => Ok(()), + Err(e) => Err(e), + } + } + /// Parse a Git request packet-line. /// /// Example: `0032git-upload-pack /project.git\0host=myserver.com\0` /// pub fn read_request_pktline(&mut self) -> io::Result<(GitRequest, Vec)> { let mut pktline = [0u8; 1024]; - let Packetline::Git(length) = self.read_pktline(&mut pktline)? else { - return Err(io::ErrorKind::InvalidInput.into()); - }; + let length = self.read_pktline(&mut pktline)?; let Some(cmd) = GitRequest::parse(&pktline[4..length]) else { return Err(io::ErrorKind::InvalidInput.into()); }; Ok((cmd, Vec::from(&pktline[..length]))) } - /// Parse a `done` packet-line. - pub fn read_done_pktline(&mut self, buf: &mut [u8]) -> io::Result<()> { - self.read_exact(&mut buf[..HEADER_LEN])?; - - if &buf[..HEADER_LEN] == DONE_PKT { - return Ok(()); - } - Err(io::ErrorKind::InvalidInput.into()) - } - /// Parse a Git packet-line. - pub fn read_pktline(&mut self, buf: &mut [u8]) -> io::Result { + pub fn read_pktline(&mut self, buf: &mut [u8]) -> io::Result { self.read_exact(&mut buf[..HEADER_LEN])?; - - if &buf[..HEADER_LEN] == DONE_PKT { - return Ok(Packetline::Done); - } if &buf[..HEADER_LEN] == FLUSH_PKT || &buf[..HEADER_LEN] == DELIM_PKT || &buf[..HEADER_LEN] == RESPONSE_END_PKT { - return Ok(Packetline::Git(HEADER_LEN)); + return Ok(HEADER_LEN); } let length = str::from_utf8(&buf[..HEADER_LEN]) .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?; @@ -590,25 +556,14 @@ pub mod pktline { self.read_exact(&mut buf[HEADER_LEN..length])?; - Ok(Packetline::Git(length)) + Ok(length) } /// Read packet-lines from the internal reader into `buf`, - /// and write them to the given writer. - /// - /// Returns [`ControlFlow::Break`] if the fetch should be terminated. - /// Otherwise, returns [`ControlFlow::Continue`] to mean that we're - /// expecting a response from the remote. - pub fn pipe( - &mut self, - w: &mut W, - buf: &mut [u8], - ) -> io::Result> { + /// and write them to the given writer. Exits when a [`FLUSH_PKT`] packet is received. + pub fn pipe(&mut self, w: &mut W, buf: &mut [u8]) -> io::Result<()> { loop { - let n = match self.read_pktline(buf)? { - Packetline::Done => return Ok(ControlFlow::Break(())), - Packetline::Git(n) => n, - }; + let n = self.read_pktline(buf)?; if n == 0 { break; } @@ -618,19 +573,12 @@ pub mod pktline { break; } } - Ok(ControlFlow::Continue(())) + Ok(()) } } impl<'a, R: io::Read> io::Read for Reader<'a, R> { fn read(&mut self, buf: &mut [u8]) -> io::Result { - if !self.drain.is_empty() { - let count = buf.len().min(self.drain.len()); - buf[..count].copy_from_slice(&self.drain[..count]); - self.drain.drain(..count); - - return Ok(count); - } self.stream.read(buf) } } diff --git a/radicle-node/src/worker/channels.rs b/radicle-node/src/worker/channels.rs new file mode 100644 index 00000000..328d0125 --- /dev/null +++ b/radicle-node/src/worker/channels.rs @@ -0,0 +1,136 @@ +use std::io::{Read, Write}; +use std::{fmt, io}; + +use crossbeam_channel as chan; + +/// Data that can be sent and received on worker channels. +pub enum ChannelEvent> { + /// Git protocol data. + Data(T), + /// A request to close the channel. + Close, + /// A signal that the git protocol has ended, eg. when the remote fetch closes the + /// connection. + Eof, +} + +impl fmt::Debug for ChannelEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Data(_) => write!(f, "ChannelEvent::Data(..)"), + Self::Close => write!(f, "ChannelEvent::Close"), + Self::Eof => write!(f, "ChannelEvent::Eof"), + } + } +} + +/// Worker channels for communicating through the git stream with the remote. +pub struct Channels> { + pub sender: ChannelWriter, + pub receiver: ChannelReader, +} + +impl Write for Channels { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.sender.write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + self.sender.flush() + } +} + +impl Read for Channels { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.receiver.read(buf) + } +} + +impl Channels { + pub fn new( + sender: chan::Sender>, + receiver: chan::Receiver>, + ) -> Self { + Channels { + sender: ChannelWriter(sender), + receiver: ChannelReader { + receiver, + buffer: io::Cursor::new(Vec::new()), + }, + } + } + + pub fn split(&mut self) -> (&mut ChannelWriter, &mut ChannelReader) { + (&mut self.sender, &mut self.receiver) + } +} + +/// Wraps a [`chan::Receiver`] and provides it with [`io::Read`]. +#[derive(Clone)] +pub struct ChannelReader> { + buffer: io::Cursor>, + receiver: chan::Receiver>, +} + +impl Read for ChannelReader> { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + let read = self.buffer.read(buf)?; + if read == 0 { + let event = self.receiver.recv().map_err(|_| { + io::Error::new( + io::ErrorKind::BrokenPipe, + "error reading from stream: channel is disconnected", + ) + })?; + + match event { + ChannelEvent::Data(data) => { + self.buffer = io::Cursor::new(data); + self.buffer.read(buf) + } + ChannelEvent::Eof => Err(io::ErrorKind::UnexpectedEof.into()), + ChannelEvent::Close => Err(io::ErrorKind::ConnectionReset.into()), + } + } else { + Ok(read) + } + } +} + +/// Wraps a [`chan::Sender`] and provides it with [`io::Write`]. +#[derive(Clone)] +pub struct ChannelWriter>(chan::Sender>); + +impl ChannelWriter { + /// Since the git protocol is tunneled over an existing connection, we can't signal the end of + /// the protocol via the usual means, which is to close the connection. Git also doesn't have + /// any special message we can send to signal the end of the protocol. + /// + /// Hence, we there's no other way for the server to know that we're done sending requests + /// than to send a special message outside the git protocol. This message can then be processed + /// by the remote worker to end the protocol. We use the special "eof" control message for this. + pub fn eof(&self) -> Result<(), chan::SendError> { + self.0.send(ChannelEvent::Eof) + } +} + +impl Write for ChannelWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + let data = buf.to_vec(); + self.0.send(ChannelEvent::Data(data)).map_err(|m| { + io::Error::new( + io::ErrorKind::BrokenPipe, + format!( + "error writing to stream: channel is disconnected: dropped {:?}", + m.into_inner() + ), + ) + })?; + + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} diff --git a/radicle-node/src/worker/tunnel.rs b/radicle-node/src/worker/tunnel.rs new file mode 100644 index 00000000..b2d178b3 --- /dev/null +++ b/radicle-node/src/worker/tunnel.rs @@ -0,0 +1,80 @@ +use std::{ + io::{self, Write}, + net, time, +}; + +use super::channels::Channels; +use super::{pktline, Handle, NodeId, StreamId}; + +/// Tunnels fetches to a remote peer. +pub struct Tunnel<'a> { + stream: &'a mut Channels, + listener: net::TcpListener, + local_addr: net::SocketAddr, + channel: StreamId, + remote: NodeId, + handle: Handle, +} + +impl<'a> Tunnel<'a> { + pub(super) fn with( + stream: &'a mut Channels, + channel: StreamId, + remote: NodeId, + handle: Handle, + ) -> io::Result { + let listener = net::TcpListener::bind(net::SocketAddr::from(([0, 0, 0, 0], 0)))?; + let local_addr = listener.local_addr()?; + + Ok(Self { + stream, + listener, + local_addr, + channel, + remote, + handle, + }) + } + + pub fn local_addr(&self) -> net::SocketAddr { + self.local_addr + } + + /// Run the tunnel until the connection is closed. + pub fn run(&mut self, timeout: time::Duration) -> io::Result<()> { + // We now loop, alternating between reading requests from the client, and writing responses + // back from the daemon.. Requests are delimited with a flush packet (`flush-pkt`). + let mut buffer = [0; u16::MAX as usize + 1]; + let (mut remote_w, mut remote_r) = self.stream.split(); + let (mut stream, _) = self.listener.accept()?; + + let mut local = pktline::Reader::new(&mut stream); + let mut remote_r = pktline::Reader::new(&mut remote_r); + + local.stream().set_read_timeout(Some(timeout))?; + local.stream().set_write_timeout(Some(timeout))?; + + let (_, buf) = local.read_request_pktline()?; + remote_w.write_all(&buf)?; + + // Nb. Annoyingly, we have to always check if the fetch stream is closed on every + // iteration, otherwise we may get stuck waiting for data from the remote while + // we're actually done. After measurement, this checking for EOF only takes + // between 1µs and 4µs, and is therefore an okay compromise. + while !local.is_eof()? { + if self.handle.flush(self.remote, self.channel).is_err() { + return Err(io::ErrorKind::BrokenPipe.into()); + } + remote_r.pipe(local.stream(), &mut buffer)?; + + if let Err(e) = local.pipe(&mut remote_w, &mut buffer) { + // This is the expected error when the git fetch closes the connection. + if e.kind() == io::ErrorKind::UnexpectedEof { + break; + } + return Err(e); + } + } + Ok(()) + } +}