diff --git a/radicle-node/src/clock.rs b/radicle-node/src/clock.rs index 12181975..1912a190 100644 --- a/radicle-node/src/clock.rs +++ b/radicle-node/src/clock.rs @@ -3,6 +3,9 @@ use std::rc::Rc; use crate::{LocalDuration, LocalTime}; +/// Seconds since epoch. +pub type Timestamp = u64; + /// Clock with interior mutability. #[derive(Debug, Clone)] pub struct RefClock(Rc>); @@ -28,6 +31,10 @@ impl RefClock { pub fn set(&mut self, time: LocalTime) { *self.borrow_mut() = time; } + + pub fn timestamp(&self) -> Timestamp { + self.local_time().as_secs() + } } impl From for RefClock { diff --git a/radicle-node/src/lib.rs b/radicle-node/src/lib.rs index 3da1232f..1316e125 100644 --- a/radicle-node/src/lib.rs +++ b/radicle-node/src/lib.rs @@ -1,6 +1,4 @@ #![allow(dead_code)] -pub use nakamoto_net::{Io, Link, LocalDuration, LocalTime}; - pub mod address_book; pub mod address_manager; pub mod client; @@ -14,15 +12,18 @@ pub mod test; pub mod transport; pub mod wire; +pub use nakamoto_net::{Io, Link, LocalDuration, LocalTime}; pub use radicle::{collections, crypto, git, hash, identity, rad, storage}; pub mod prelude { + pub use crate::clock::Timestamp; pub use crate::crypto::{PublicKey, Signature, Signer}; pub use crate::decoder::Decoder; pub use crate::hash::Digest; pub use crate::identity::{Did, Id}; pub use crate::service::filter::Filter; - pub use crate::service::{NodeId, Timestamp}; + pub use crate::service::{DisconnectReason, Envelope, Event, Message, Network, NodeId}; pub use crate::storage::refs::Refs; pub use crate::storage::WriteStorage; + pub use crate::{LocalDuration, LocalTime}; } diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index cb08d5b1..b6a42cf5 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -1,10 +1,10 @@ -#![allow(dead_code)] pub mod config; pub mod filter; pub mod message; pub mod peer; +pub mod reactor; -use std::collections::{BTreeMap, VecDeque}; +use std::collections::BTreeMap; use std::ops::{Deref, DerefMut}; use std::{fmt, net, net::IpAddr}; @@ -15,14 +15,15 @@ use nakamoto::{LocalDuration, LocalTime}; use nakamoto_net as nakamoto; use nakamoto_net::Link; use nonempty::NonEmpty; +use radicle::storage::ReadStorage; use crate::address_book; use crate::address_book::AddressBook; use crate::address_manager::AddressManager; -use crate::clock::RefClock; +use crate::clock::{RefClock, Timestamp}; use crate::collections::{HashMap, HashSet}; use crate::crypto; -use crate::crypto::Verified; +use crate::crypto::{Signer, Verified}; use crate::git; use crate::git::Url; use crate::identity::{Doc, Id}; @@ -35,8 +36,8 @@ use crate::storage::{Inventory, ReadRepository, RefUpdate, WriteRepository, Writ pub use crate::service::config::{Config, Network}; pub use crate::service::message::{Envelope, Message}; -use self::filter::Filter; use self::message::{InventoryAnnouncement, NodeFeatures}; +use self::reactor::Reactor; pub const DEFAULT_PORT: u16 = 8776; pub const PROTOCOL_VERSION: u32 = 1; @@ -52,23 +53,6 @@ pub const MAX_TIME_DELTA: LocalDuration = LocalDuration::from_mins(60); pub type NodeId = crypto::PublicKey; /// Network routing table. Keeps track of where projects are hosted. pub type Routing = HashMap>; -/// Seconds since epoch. -pub type Timestamp = u64; - -/// Output of a state transition. -#[derive(Debug)] -pub enum Io { - /// There are some messages ready to be sent to a peer. - Write(net::SocketAddr, Vec), - /// Connect to a peer. - Connect(net::SocketAddr), - /// Disconnect from a peer. - Disconnect(net::SocketAddr, DisconnectReason), - /// Ask for a wakeup in a specified amount of time. - Wakeup(LocalDuration), - /// Emit an event. - Event(Event), -} /// A service event. #[derive(Debug, Clone)] @@ -138,11 +122,27 @@ pub enum Command { pub enum CommandError {} #[derive(Debug)] -pub struct Service { +pub struct Service { + /// Service configuration. + config: Config, + /// Our cryptographic signer and key. + signer: G, + /// Project storage. + storage: S, + /// Tracks the location of projects. + routing: Routing, /// Peer sessions, currently or recently connected. sessions: Sessions, - /// Service state that isn't peer-specific. - context: Context, + /// Keeps track of peer states. + peers: BTreeMap, + /// Clock. Tells the time. + clock: RefClock, + /// Interface to the I/O reactor. + reactor: Reactor, + /// Peer address manager. + addrmgr: AddressManager, + /// Source of entropy. + rng: Rng, /// Whether our local inventory no long represents what we have announced to the network. out_of_sync: bool, /// Last time the service was idle. @@ -157,20 +157,36 @@ pub struct Service { start_time: LocalTime, } -impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service { +impl Service +where + A: address_book::Store, + S: WriteStorage + 'static, + G: crypto::Signer, +{ pub fn new( config: Config, clock: RefClock, - storage: T, - addresses: S, + storage: S, + addresses: A, signer: G, rng: Rng, ) -> Self { let addrmgr = AddressManager::new(addresses); + let routing = HashMap::with_hasher(rng.clone().into()); + let sessions = Sessions::new(rng.clone()); + let network = config.network; Self { - context: Context::new(config, clock, storage, addrmgr, signer, rng.clone()), - sessions: Sessions::new(rng), + config, + storage, + addrmgr, + signer, + rng, + clock, + routing, + peers: BTreeMap::new(), + reactor: Reactor::new(network), + sessions, out_of_sync: false, last_idle: LocalTime::default(), last_sync: LocalTime::default(), @@ -180,10 +196,8 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service } } - pub fn disconnect(&mut self, remote: &IpAddr, reason: DisconnectReason) { - if let Some(addr) = self.sessions.get(remote).map(|p| p.addr) { - self.context.disconnect(addr, reason); - } + pub fn node_id(&self) -> NodeId { + *self.signer.public_key() } pub fn seeds(&self, id: &Id) -> Box + '_> { @@ -243,17 +257,17 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service /// Get the current inventory. pub fn inventory(&self) -> Result { - self.context.storage.inventory() + self.storage.inventory() } /// Get the storage instance. - pub fn storage(&self) -> &T { - &self.context.storage + pub fn storage(&self) -> &S { + &self.storage } /// Get the mutable storage instance. - pub fn storage_mut(&mut self) -> &mut T { - &mut self.context.storage + pub fn storage_mut(&mut self) -> &mut S { + &mut self.storage } /// Get a project from storage, using the local node's key. @@ -263,34 +277,38 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service /// Get the local signer. pub fn signer(&self) -> &G { - &self.context.signer + &self.signer + } + + /// Get the clock. + pub fn clock(&self) -> &RefClock { + &self.clock } /// Get the local service time. pub fn local_time(&self) -> LocalTime { - self.context.clock.local_time() + self.clock.local_time() } /// Get service configuration. pub fn config(&self) -> &Config { - &self.context.config + &self.config } /// Get reference to routing table. pub fn routing(&self) -> &Routing { - &self.context.routing + &self.routing } - /// Get I/O outbox. - pub fn outbox(&mut self) -> &mut VecDeque { - &mut self.context.io + /// Get I/O reactor. + pub fn reactor(&mut self) -> &mut Reactor { + &mut self.reactor } pub fn lookup(&self, id: &Id) -> Lookup { Lookup { - local: self.context.storage.get(&self.node_id(), id).unwrap(), + local: self.storage.get(&self.node_id(), id).unwrap(), remote: self - .context .routing .get(id) .map_or(vec![], |r| r.iter().cloned().collect()), @@ -303,20 +321,20 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service self.start_time = time; // Connect to configured peers. - let addrs = self.context.config.connect.clone(); + let addrs = self.config.connect.clone(); for addr in addrs { - self.context.connect(addr); + self.reactor.connect(addr); } } pub fn tick(&mut self, now: nakamoto::LocalTime) { trace!("Tick +{}", now - self.start_time); - self.context.clock.set(now); + self.clock.set(now); } pub fn wake(&mut self) { - let now = self.context.clock.local_time(); + let now = self.clock.local_time(); trace!("Wake +{}", now - self.start_time); @@ -324,28 +342,28 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service debug!("Running 'idle' task..."); self.maintain_connections(); - self.context.io.push_back(Io::Wakeup(IDLE_INTERVAL)); + self.reactor.wakeup(IDLE_INTERVAL); self.last_idle = now; } if now - self.last_sync >= SYNC_INTERVAL { debug!("Running 'sync' task..."); // TODO: What do we do here? - self.context.io.push_back(Io::Wakeup(SYNC_INTERVAL)); + self.reactor.wakeup(SYNC_INTERVAL); self.last_sync = now; } if now - self.last_announce >= ANNOUNCE_INTERVAL { if self.out_of_sync { self.announce_inventory().unwrap(); } - self.context.io.push_back(Io::Wakeup(ANNOUNCE_INTERVAL)); + self.reactor.wakeup(ANNOUNCE_INTERVAL); self.last_announce = now; } if now - self.last_prune >= PRUNE_INTERVAL { debug!("Running 'prune' task..."); self.prune_routing_entries(); - self.context.io.push_back(Io::Wakeup(PRUNE_INTERVAL)); + self.reactor.wakeup(PRUNE_INTERVAL); self.last_prune = now; } } @@ -354,7 +372,7 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service debug!("Command {:?}", cmd); match cmd { - Command::Connect(addr) => self.context.connect(addr), + Command::Connect(addr) => self.reactor.connect(addr), Command::Fetch(id, resp) => { if !self.config.is_tracking(&id) { resp.send(FetchLookup::NotTracking).ok(); @@ -433,7 +451,7 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service let message = RefsAnnouncement { id, refs }; let signature = message.sign(&self.signer); - self.context.broadcast( + self.reactor.broadcast( Message::RefsAnnouncement { node, message, @@ -447,7 +465,7 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service pub fn attempted(&mut self, addr: &std::net::SocketAddr) { let ip = addr.ip(); - let persistent = self.context.config.is_persistent(addr); + let persistent = self.config.is_persistent(addr); let peer = self .sessions .entry(ip) @@ -470,21 +488,24 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service // For inbound connections, we wait for the remote to say "Hello" first. // TODO: How should we deal with multiple peers connecting from the same IP address? if link.is_outbound() { - // TODO: Refactor this so that we don't create messages if the peer isn't found. - let messages = self.handshake_messages(); - if let Some(peer) = self.sessions.get_mut(&ip) { - self.context.write_all(peer.addr, messages); - peer.connected(); + if link.is_outbound() { + self.reactor.write_all( + addr, + gossip::handshake( + self.clock.timestamp(), + &self.storage, + &self.signer, + &self.config, + ), + ); + } + peer.connected(link); } } else { self.sessions.insert( ip, - Session::new( - addr, - Link::Inbound, - self.context.config.is_persistent(&addr), - ), + Session::new(addr, Link::Inbound, self.config.is_persistent(&addr)), ); } } @@ -503,8 +524,7 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service peer.state = SessionState::Disconnected { since }; // Attempt to re-connect to persistent peers. - if self.context.config.is_persistent(addr) && peer.attempts() < MAX_CONNECTION_ATTEMPTS - { + if self.config.is_persistent(addr) && peer.attempts() < MAX_CONNECTION_ATTEMPTS { if reason.is_dial_err() { return; } @@ -520,7 +540,7 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service // TODO: Try to reconnect only if the peer was attempted. A disconnect without // even a successful attempt means that we're unlikely to be able to reconnect. - self.context.connect(*addr); + self.reactor.connect(*addr); } else { // TODO: Non-persistent peers should be removed from the // map here or at some later point. @@ -528,36 +548,208 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service } } - pub fn received_message(&mut self, addr: &std::net::SocketAddr, msg: Envelope) { - let peer_ip = addr.ip(); + pub fn received_message(&mut self, addr: &net::SocketAddr, envelope: Envelope) { + match self.handle_message(addr, envelope) { + Ok(relay) => { + if let Some(msg) = relay { + let negotiated = self + .sessions + .negotiated() + .filter(|(ip, _)| **ip != addr.ip()) + .map(|(_, p)| p); + + self.reactor.relay(msg, negotiated.clone()); + } + } + Err(SessionError::NotFound(ip)) => { + error!("Session not found for {}", ip); + } + Err(err) => { + // If there's an error, stop processing messages from this peer. + // However, we still relay messages returned up to this point. + self.reactor.disconnect(*addr, DisconnectReason::Error(err)); + + // FIXME: The peer should be set in a state such that we don'that + // process further messages. + } + } + } + + pub fn handle_message( + &mut self, + remote: &net::SocketAddr, + envelope: Envelope, + ) -> Result, peer::SessionError> { + let peer_ip = remote.ip(); let peer = if let Some(peer) = self.sessions.get_mut(&peer_ip) { peer } else { - return; + return Err(SessionError::NotFound(remote.ip())); }; - let relay = match peer.received(msg, &mut self.context) { - Ok(msg) => msg, - Err(err) => { - self.context - .disconnect(peer.addr, DisconnectReason::Error(err)); - // If there's an error, stop processing messages from this peer. - // However, we still relay messages returned up to this point. - // - // FIXME: The peer should be set in a state such that we don'that - // process further messages. - return; + if envelope.magic != self.config.network.magic() { + return Err(SessionError::WrongMagic(envelope.magic)); + } + debug!("Received {:?} from {}", &envelope.msg, peer.ip()); + + match (&peer.state, envelope.msg) { + ( + SessionState::Initial, + Message::Initialize { + id, + version, + addrs, + git, + }, + ) => { + if version != PROTOCOL_VERSION { + return Err(SessionError::WrongVersion(version)); + } + // Nb. This is a very primitive handshake. Eventually we should have anyhow + // extra "acknowledgment" message sent when the `Initialize` is well received. + if peer.link.is_inbound() { + self.reactor.write_all( + peer.addr, + gossip::handshake( + self.clock.timestamp(), + &self.storage, + &self.signer, + &self.config, + ), + ); + } + // Nb. we don't set the peer timestamp here, since it is going to be + // set after the first message is received only. Setting it here would + // mean that messages received right after the handshake could be ignored. + peer.state = SessionState::Negotiated { + id, + since: self.clock.local_time(), + addrs, + git, + }; } - }; + (SessionState::Initial, _) => { + debug!( + "Disconnecting peer {} for sending us a message before handshake", + peer.ip() + ); + return Err(SessionError::Misbehavior); + } + ( + SessionState::Negotiated { git, .. }, + Message::InventoryAnnouncement { + node, + message, + signature, + }, + ) => { + let now = self.clock.local_time(); + let peer = self.peers.entry(node).or_insert_with(Peer::default); + let relay = self.config.relay; + let git = git.clone(); - if let Some(msg) = relay { - let negotiated = self - .sessions - .negotiated() - .filter(|(ip, _)| **ip != peer_ip) - .map(|(_, p)| p); + // Don't allow messages from too far in the future. + if message.timestamp.saturating_sub(now.as_secs()) > MAX_TIME_DELTA.as_secs() { + return Err(SessionError::InvalidTimestamp(message.timestamp)); + } + // Discard inventory messages we've already seen, otherwise update + // out last seen time. + if message.timestamp > peer.last_message { + peer.last_message = message.timestamp; + } else { + return Ok(None); + } + self.process_inventory(&message.inventory, node, &git); - self.context.relay(msg, negotiated.clone()); + if relay { + return Ok(Some(Message::InventoryAnnouncement { + node, + message, + signature, + })); + } + } + // Process a peer inventory update announcement by (maybe) fetching. + ( + SessionState::Negotiated { git, .. }, + Message::RefsAnnouncement { + node, + message, + signature, + }, + ) => { + // FIXME: Check message timestamp. + + if message.verify(&node, &signature) { + // TODO: Buffer/throttle fetches. + // TODO: Check that we're tracking this user as well. + if self.config.is_tracking(&message.id) { + // TODO: Check refs to see if we should try to fetch or not. + let updated = self.storage.fetch(&message.id, git).unwrap(); + let is_updated = !updated.is_empty(); + + self.reactor.event(Event::RefsFetched { + from: git.clone(), + project: message.id.clone(), + updated, + }); + + if is_updated { + return Ok(Some(Message::RefsAnnouncement { + node, + message, + signature, + })); + } + } + } else { + return Err(SessionError::Misbehavior); + } + } + ( + SessionState::Negotiated { .. }, + Message::NodeAnnouncement { + node, + message, + signature, + }, + ) => { + // FIXME: Check message timestamp. + + if !message.verify(&node, &signature) { + return Err(SessionError::Misbehavior); + } + log::warn!("Node announcement handling is not implemented"); + } + (SessionState::Negotiated { .. }, Message::Subscribe(subscribe)) => { + peer.subscribe = Some(subscribe); + } + (SessionState::Negotiated { .. }, Message::Initialize { .. }) => { + debug!( + "Disconnecting peer {} for sending us a redundant handshake message", + peer.ip() + ); + return Err(SessionError::Misbehavior); + } + (SessionState::Disconnected { .. }, msg) => { + debug!("Ignoring {:?} from disconnected peer {}", msg, peer.ip()); + } + } + Ok(None) + } + + /// Process a peer inventory announcement by updating our routing table. + fn process_inventory(&mut self, inventory: &Inventory, from: NodeId, remote: &Url) { + for proj_id in inventory { + let inventory = self + .routing + .entry(proj_id.clone()) + .or_insert_with(|| HashSet::with_hasher(self.rng.clone().into())); + + // TODO: Fire an event on routing update. + if inventory.insert(from) && self.config.is_tracking(proj_id) { + self.storage.fetch(proj_id, remote).unwrap(); + } } } @@ -567,10 +759,14 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service /// Announce our inventory to all connected peers. fn announce_inventory(&mut self) -> Result<(), storage::Error> { - let inv = Message::inventory(self.context.inventory_announcement()?, &self.context.signer); + let inventory = self.storage().inventory()?; + let inv = Message::inventory( + gossip::inventory(self.clock.timestamp(), inventory), + &self.signer, + ); for addr in self.sessions.negotiated().map(|(_, p)| p.addr) { - self.context.write(addr, inv.clone()); + self.reactor.write(addr, inv.clone()); } Ok(()) } @@ -591,20 +787,6 @@ impl<'r, T: WriteStorage<'r>, S: address_book::Store, G: crypto::Signer> Service } } -impl Deref for Service { - type Target = Context; - - fn deref(&self) -> &Self::Target { - &self.context - } -} - -impl DerefMut for Service { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.context - } -} - #[derive(Debug, Clone)] pub enum DisconnectReason { User, @@ -635,11 +817,11 @@ impl fmt::Display for DisconnectReason { } } -impl Iterator for Service { - type Item = Io; +impl Iterator for Service { + type Item = reactor::Io; fn next(&mut self) -> Option { - self.context.io.pop_front() + self.reactor.next() } } @@ -659,202 +841,6 @@ pub struct Peer { pub last_message: Timestamp, } -/// Global service state used across peers. -#[derive(Debug)] -pub struct Context { - /// Service configuration. - config: Config, - /// Our cryptographic signer and key. - signer: G, - /// Tracks the location of projects. - routing: Routing, - /// Keeps track of peer states. - peers: BTreeMap, - /// Outgoing I/O queue. - io: VecDeque, - /// Clock. Tells the time. - clock: RefClock, - /// Project storage. - storage: T, - /// Peer address manager. - addrmgr: AddressManager, - /// Source of entropy. - rng: Rng, -} - -impl Context -where - T: storage::ReadStorage, - G: crypto::Signer, -{ - pub(crate) fn node_id(&self) -> NodeId { - *self.signer.public_key() - } -} - -impl<'r, S, T, G> Context -where - T: storage::WriteStorage<'r>, - G: crypto::Signer, -{ - pub(crate) fn new( - config: Config, - clock: RefClock, - storage: T, - addrmgr: AddressManager, - signer: G, - rng: Rng, - ) -> Self { - Self { - config, - signer, - clock, - routing: HashMap::with_hasher(rng.clone().into()), - peers: BTreeMap::new(), - io: VecDeque::new(), - storage, - addrmgr, - rng, - } - } - - fn node_announcement(&self) -> NodeAnnouncement { - let timestamp = self.timestamp(); - let features = NodeFeatures::default(); - let alias = self.alias(); - let addresses = vec![]; // TODO - - NodeAnnouncement { - features, - timestamp, - alias, - addresses, - } - } - - fn inventory_announcement(&self) -> Result { - let timestamp = self.timestamp(); - let inventory = self.storage.inventory()?; - - Ok(InventoryAnnouncement { - inventory, - timestamp, - }) - } - - fn filter(&self) -> Filter { - match &self.config.project_tracking { - ProjectTracking::All { .. } => Filter::default(), - ProjectTracking::Allowed(ids) => Filter::new(ids.iter()), - } - } - - fn handshake_messages(&self) -> [Message; 4] { - let git = self.config.git_url.clone(); - [ - Message::init(self.node_id(), self.config.listen.clone(), git), - Message::node(self.node_announcement(), &self.signer), - Message::inventory(self.inventory_announcement().unwrap(), &self.signer), - Message::subscribe(self.filter(), self.timestamp(), Timestamp::MAX), - ] - } - - fn alias(&self) -> [u8; 32] { - let mut alias = [0u8; 32]; - - alias[..9].copy_from_slice("anonymous".as_bytes()); - alias - } - - /// Process a peer inventory announcement by updating our routing table. - fn process_inventory(&mut self, inventory: &Inventory, from: NodeId, remote: &Url) { - for proj_id in inventory { - let inventory = self - .routing - .entry(proj_id.clone()) - .or_insert_with(|| HashSet::with_hasher(self.rng.clone().into())); - - // TODO: Fire an event on routing update. - if inventory.insert(from) && self.config.is_tracking(proj_id) { - self.fetch(proj_id, remote); - } - } - } - - fn fetch(&mut self, proj_id: &Id, remote: &Url) -> Vec { - let mut repo = self.storage.repository(proj_id).unwrap(); - let mut path = remote.path.clone(); - - path.push(b'/'); - path.extend(proj_id.to_string().into_bytes()); - - repo.fetch(&Url { - path, - ..remote.clone() - }) - .unwrap() - } - - /// Disconnect a peer. - fn disconnect(&mut self, addr: net::SocketAddr, reason: DisconnectReason) { - self.io.push_back(Io::Disconnect(addr, reason)); - } -} - -impl Context { - /// Get current local timestamp. - pub(crate) fn timestamp(&self) -> Timestamp { - self.clock.local_time().as_secs() - } - - /// Connect to a peer. - fn connect(&mut self, addr: net::SocketAddr) { - // TODO: Make sure we don't try to connect more than once to the same address. - self.io.push_back(Io::Connect(addr)); - } - - fn write_all(&mut self, remote: net::SocketAddr, msgs: impl IntoIterator) { - let envelopes = msgs - .into_iter() - .map(|msg| self.config.network.envelope(msg)) - .collect(); - self.io.push_back(Io::Write(remote, envelopes)); - } - - fn write(&mut self, remote: net::SocketAddr, msg: Message) { - debug!("Write {:?} to {}", &msg, remote.ip()); - - let envelope = self.config.network.envelope(msg); - self.io.push_back(Io::Write(remote, vec![envelope])); - } - - /// Broadcast a message to a list of peers. - fn broadcast<'a>(&mut self, msg: Message, peers: impl IntoIterator) { - for peer in peers { - self.write(peer.addr, msg.clone()); - } - } - - /// Relay a message to interested peers. - fn relay<'a>(&mut self, msg: Message, peers: impl IntoIterator) { - if let Message::RefsAnnouncement { message, .. } = &msg { - let id = message.id.clone(); - let peers = peers.into_iter().filter(|p| { - if let Some(subscribe) = &p.subscribe { - subscribe.filter.contains(&id) - } else { - // If the peer did not send us a `subscribe` message, we don'the - // relay any messages to them. - false - } - }); - self.broadcast(msg, peers); - } else { - self.broadcast(msg, peers); - } - } -} - #[derive(Debug)] /// Holds currently (or recently) connected peers. pub struct Sessions(AddressBook); @@ -893,3 +879,44 @@ impl DerefMut for Sessions { &mut self.0 } } + +mod gossip { + use super::*; + + pub fn handshake( + timestamp: Timestamp, + storage: &S, + signer: &G, + config: &Config, + ) -> [Message; 4] { + let git = config.git_url.clone(); + let inventory = storage.inventory().unwrap(); + + [ + Message::init(*signer.public_key(), config.listen.clone(), git), + Message::node(gossip::node(timestamp, config), signer), + Message::inventory(gossip::inventory(timestamp, inventory), signer), + Message::subscribe(config.filter(), timestamp, Timestamp::MAX), + ] + } + + pub fn node(timestamp: Timestamp, config: &Config) -> NodeAnnouncement { + let features = NodeFeatures::default(); + let alias = config.alias(); + let addresses = vec![]; // TODO + + NodeAnnouncement { + features, + timestamp, + alias, + addresses, + } + } + + pub fn inventory(timestamp: Timestamp, inventory: Vec) -> InventoryAnnouncement { + InventoryAnnouncement { + inventory, + timestamp, + } + } +} diff --git a/radicle-node/src/service/config.rs b/radicle-node/src/service/config.rs index eb4d7e68..f5b53e94 100644 --- a/radicle-node/src/service/config.rs +++ b/radicle-node/src/service/config.rs @@ -4,6 +4,7 @@ use crate::collections::HashSet; use crate::git; use crate::git::Url; use crate::identity::{Id, PublicKey}; +use crate::service::filter::Filter; use crate::service::message::{Address, Envelope, Message}; /// Peer-to-peer network. @@ -124,4 +125,18 @@ impl Config { ProjectTracking::Allowed(ids) => ids.remove(&id), } } + + pub fn filter(&self) -> Filter { + match &self.project_tracking { + ProjectTracking::All { .. } => Filter::default(), + ProjectTracking::Allowed(ids) => Filter::new(ids.iter()), + } + } + + pub fn alias(&self) -> [u8; 32] { + let mut alias = [0u8; 32]; + + alias[..9].copy_from_slice("anonymous".as_bytes()); + alias + } } diff --git a/radicle-node/src/service/peer.rs b/radicle-node/src/service/peer.rs index 0bf9bb82..fcc79525 100644 --- a/radicle-node/src/service/peer.rs +++ b/radicle-node/src/service/peer.rs @@ -31,6 +31,8 @@ pub enum SessionError { WrongVersion(u32), #[error("invalid announcement timestamp: {0}")] InvalidTimestamp(u64), + #[error("session not found for address `{0}`")] + NotFound(net::IpAddr), #[error("peer misbehaved")] Misbehavior, } @@ -84,157 +86,7 @@ impl Session { self.attempts += 1; } - pub fn connected(&mut self) { + pub fn connected(&mut self, _link: Link) { self.attempts = 0; } - - pub fn received<'r, S, T, G>( - &mut self, - envelope: Envelope, - ctx: &mut Context, - ) -> Result, SessionError> - where - T: storage::WriteStorage<'r>, - G: crypto::Signer, - { - if envelope.magic != ctx.config.network.magic() { - return Err(SessionError::WrongMagic(envelope.magic)); - } - debug!("Received {:?} from {}", &envelope.msg, self.ip()); - - match (&self.state, envelope.msg) { - ( - SessionState::Initial, - Message::Initialize { - id, - version, - addrs, - git, - }, - ) => { - if version != PROTOCOL_VERSION { - return Err(SessionError::WrongVersion(version)); - } - // Nb. This is a very primitive handshake. Eventually we should have anyhow - // extra "acknowledgment" message sent when the `Initialize` is well received. - if self.link.is_inbound() { - ctx.write_all(self.addr, ctx.handshake_messages()); - } - // Nb. we don't set the peer timestamp here, since it is going to be - // set after the first message is received only. Setting it here would - // mean that messages received right after the handshake could be ignored. - self.state = SessionState::Negotiated { - id, - since: ctx.clock.local_time(), - addrs, - git, - }; - } - (SessionState::Initial, _) => { - debug!( - "Disconnecting peer {} for sending us a message before handshake", - self.ip() - ); - return Err(SessionError::Misbehavior); - } - ( - SessionState::Negotiated { git, .. }, - Message::InventoryAnnouncement { - node, - message, - signature, - }, - ) => { - let now = ctx.clock.local_time(); - let peer = ctx.peers.entry(node).or_insert_with(Peer::default); - - // Don't allow messages from too far in the future. - if message.timestamp.saturating_sub(now.as_secs()) > MAX_TIME_DELTA.as_secs() { - return Err(SessionError::InvalidTimestamp(message.timestamp)); - } - // Discard inventory messages we've already seen, otherwise update - // out last seen time. - if message.timestamp > peer.last_message { - peer.last_message = message.timestamp; - } else { - return Ok(None); - } - ctx.process_inventory(&message.inventory, node, git); - - if ctx.config.relay { - return Ok(Some(Message::InventoryAnnouncement { - node, - message, - signature, - })); - } - } - // Process a peer inventory update announcement by (maybe) fetching. - ( - SessionState::Negotiated { git, .. }, - Message::RefsAnnouncement { - node, - message, - signature, - }, - ) => { - // FIXME: Check message timestamp. - - if message.verify(&node, &signature) { - // TODO: Buffer/throttle fetches. - // TODO: Check that we're tracking this user as well. - if ctx.config.is_tracking(&message.id) { - // TODO: Check refs to see if we should try to fetch or not. - let updated_refs = ctx.fetch(&message.id, git); - let is_updated = !updated_refs.is_empty(); - - ctx.io.push_back(Io::Event(Event::RefsFetched { - from: git.clone(), - project: message.id.clone(), - updated: updated_refs, - })); - - if is_updated { - return Ok(Some(Message::RefsAnnouncement { - node, - message, - signature, - })); - } - } - } else { - return Err(SessionError::Misbehavior); - } - } - ( - SessionState::Negotiated { .. }, - Message::NodeAnnouncement { - node, - message, - signature, - }, - ) => { - // FIXME: Check message timestamp. - - if !message.verify(&node, &signature) { - return Err(SessionError::Misbehavior); - } - log::warn!("Node announcement handling is not implemented"); - } - (SessionState::Negotiated { .. }, Message::Subscribe(subscribe)) => { - self.subscribe = Some(subscribe); - } - (SessionState::Negotiated { .. }, Message::Initialize { .. }) => { - debug!( - "Disconnecting peer {} for sending us a redundant handshake message", - self.ip() - ); - return Err(SessionError::Misbehavior); - } - (SessionState::Disconnected { .. }, msg) => { - debug!("Ignoring {:?} from disconnected peer {}", msg, self.ip()); - } - } - Ok(None) - } } diff --git a/radicle-node/src/service/reactor.rs b/radicle-node/src/service/reactor.rs new file mode 100644 index 00000000..6c9c94a4 --- /dev/null +++ b/radicle-node/src/service/reactor.rs @@ -0,0 +1,114 @@ +use std::collections::VecDeque; +use std::net; + +use log::*; + +use crate::prelude::*; +use crate::service::peer::Session; + +/// Output of a state transition. +#[derive(Debug)] +pub enum Io { + /// There are some messages ready to be sent to a peer. + Write(net::SocketAddr, Vec), + /// Connect to a peer. + Connect(net::SocketAddr), + /// Disconnect from a peer. + Disconnect(net::SocketAddr, DisconnectReason), + /// Ask for a wakeup in a specified amount of time. + Wakeup(LocalDuration), + /// Emit an event. + Event(Event), +} + +/// Interface to the network reactor. +#[derive(Debug)] +pub struct Reactor { + /// The network we're on. + network: Network, + /// Outgoing I/O queue. + io: VecDeque, +} + +impl Reactor { + pub fn new(network: Network) -> Self { + Self { + network, + io: VecDeque::new(), + } + } + + /// Emit an event. + pub fn event(&mut self, event: Event) { + self.io.push_back(Io::Event(event)); + } + + /// Connect to a peer. + pub fn connect(&mut self, addr: net::SocketAddr) { + // TODO: Make sure we don't try to connect more than once to the same address. + self.io.push_back(Io::Connect(addr)); + } + + /// Disconnect a peer. + pub fn disconnect(&mut self, addr: net::SocketAddr, reason: DisconnectReason) { + self.io.push_back(Io::Disconnect(addr, reason)); + } + + pub fn write(&mut self, remote: net::SocketAddr, msg: Message) { + debug!("Write {:?} to {}", &msg, remote.ip()); + + let envelope = self.network.envelope(msg); + self.io.push_back(Io::Write(remote, vec![envelope])); + } + + pub fn write_all(&mut self, remote: net::SocketAddr, msgs: impl IntoIterator) { + let envelopes = msgs + .into_iter() + .map(|msg| self.network.envelope(msg)) + .collect(); + self.io.push_back(Io::Write(remote, envelopes)); + } + + pub fn wakeup(&mut self, after: LocalDuration) { + self.io.push_back(Io::Wakeup(after)); + } + + /// Broadcast a message to a list of peers. + pub fn broadcast<'a>(&mut self, msg: Message, peers: impl IntoIterator) { + for peer in peers { + self.write(peer.addr, msg.clone()); + } + } + + /// Relay a message to interested peers. + pub fn relay<'a>(&mut self, msg: Message, peers: impl IntoIterator) { + if let Message::RefsAnnouncement { message, .. } = &msg { + let id = message.id.clone(); + let peers = peers.into_iter().filter(|p| { + if let Some(subscribe) = &p.subscribe { + subscribe.filter.contains(&id) + } else { + // If the peer did not send us a `subscribe` message, we don'the + // relay any messages to them. + false + } + }); + self.broadcast(msg, peers); + } else { + self.broadcast(msg, peers); + } + } + + #[cfg(test)] + pub(crate) fn outbox(&mut self) -> &mut VecDeque { + &mut self.io + } +} + +impl Iterator for Reactor { + type Item = Io; + + fn next(&mut self) -> Option { + self.io.pop_front() + } +} diff --git a/radicle-node/src/test/arbitrary.rs b/radicle-node/src/test/arbitrary.rs index 316ed107..eebae12f 100644 --- a/radicle-node/src/test/arbitrary.rs +++ b/radicle-node/src/test/arbitrary.rs @@ -4,14 +4,12 @@ use bloomy::BloomFilter; use quickcheck::Arbitrary; use crate::crypto; -use crate::identity::Id; +use crate::prelude::{Id, NodeId, Refs, Timestamp}; use crate::service::filter::{Filter, FILTER_SIZE}; use crate::service::message::{ Address, Envelope, InventoryAnnouncement, Message, NodeAnnouncement, RefsAnnouncement, Subscribe, }; -use crate::service::{NodeId, Timestamp}; -use crate::storage::refs::Refs; use crate::wire::message::MessageType; pub use radicle::test::arbitrary::*; diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index c61e77df..bb04d78f 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -1,16 +1,18 @@ +use std::iter; use std::net; use std::ops::{Deref, DerefMut}; use log::*; use crate::address_book::{KnownAddress, Source}; -use crate::clock::RefClock; +use crate::clock::{RefClock, Timestamp}; use crate::collections::HashMap; use crate::git; use crate::git::Url; use crate::service; use crate::service::config::*; use crate::service::message::*; +use crate::service::reactor::Io; use crate::service::*; use crate::storage::WriteStorage; use crate::test::signer::MockSigner; @@ -32,9 +34,9 @@ pub struct Peer { initialized: bool, } -impl<'r, S> simulator::Peer for Peer +impl simulator::Peer for Peer where - S: WriteStorage<'r> + 'static, + S: WriteStorage + 'static, { fn init(&mut self) { self.initialize() @@ -59,9 +61,9 @@ impl DerefMut for Peer { } } -impl<'r, S> Peer +impl Peer where - S: WriteStorage<'r> + 'static, + S: WriteStorage + 'static, { pub fn new(name: &'static str, ip: impl Into, storage: S) -> Self { let git_url = Url { @@ -124,7 +126,7 @@ where } pub fn timestamp(&self) -> Timestamp { - self.service.timestamp() + self.service.clock().timestamp() } pub fn git_url(&self) -> Url { @@ -185,8 +187,8 @@ where pub fn messages(&mut self, remote: &net::SocketAddr) -> impl Iterator { let mut msgs = Vec::new(); - self.service.outbox().retain(|o| match o { - service::Io::Write(a, envelopes) if a == remote => { + self.service.reactor().outbox().retain(|o| match o { + Io::Write(a, envelopes) if a == remote => { msgs.extend(envelopes.iter().map(|e| e.msg.clone())); false } @@ -204,6 +206,6 @@ where /// Get a draining iterator over the peer's I/O outbox. pub fn outbox(&mut self) -> impl Iterator + '_ { - self.service.outbox().drain(..) + iter::from_fn(|| self.service.reactor().next()) } } diff --git a/radicle-node/src/test/simulator.rs b/radicle-node/src/test/simulator.rs index da559925..810ab3c3 100644 --- a/radicle-node/src/test/simulator.rs +++ b/radicle-node/src/test/simulator.rs @@ -13,7 +13,8 @@ use log::*; use nakamoto_net as nakamoto; use nakamoto_net::{Link, LocalDuration, LocalTime}; -use crate::service::{DisconnectReason, Envelope, Event, Io}; +use crate::service::reactor::Io; +use crate::service::{DisconnectReason, Envelope, Event}; use crate::storage::WriteStorage; use crate::test::peer::Service; @@ -191,7 +192,7 @@ pub struct Simulation { storage: PhantomData, } -impl<'r, S: WriteStorage<'r>> Simulation { +impl Simulation { /// Create a new simulation. pub fn new(time: LocalTime, rng: fastrand::Rng, opts: Options) -> Self { Self { diff --git a/radicle-node/src/test/tests.rs b/radicle-node/src/test/tests.rs index 789ecac5..7ef70a8d 100644 --- a/radicle-node/src/test/tests.rs +++ b/radicle-node/src/test/tests.rs @@ -8,6 +8,7 @@ use crate::collections::{HashMap, HashSet}; use crate::service::config::*; use crate::service::message::*; use crate::service::peer::*; +use crate::service::reactor::Io; use crate::service::*; use crate::storage::git::Storage; use crate::storage::ReadStorage; diff --git a/radicle-node/src/transport.rs b/radicle-node/src/transport.rs index fbb7b8d7..12934162 100644 --- a/radicle-node/src/transport.rs +++ b/radicle-node/src/transport.rs @@ -32,9 +32,9 @@ impl Transport { } } -impl<'r, S, T, G> nakamoto::Protocol for Transport +impl nakamoto::Protocol for Transport where - T: WriteStorage<'r> + 'static, + T: WriteStorage + 'static, S: address_book::Store, G: crypto::Signer, { diff --git a/radicle-node/src/wire.rs b/radicle-node/src/wire.rs index 313fddfb..f3886d1e 100644 --- a/radicle-node/src/wire.rs +++ b/radicle-node/src/wire.rs @@ -20,6 +20,7 @@ use crate::hash::Digest; use crate::identity::Id; use crate::service; use crate::service::filter; +use crate::service::reactor::Io; use crate::storage::refs::Refs; use crate::storage::refs::SignedRefs; use crate::storage::WriteStorage; @@ -450,10 +451,10 @@ impl Wire { } } -impl<'r, S, T, G> Wire +impl Wire where S: address_book::Store, - T: WriteStorage<'r> + 'static, + T: WriteStorage + 'static, G: Signer, { pub fn connected( @@ -505,7 +506,7 @@ impl Iterator for Wire { fn next(&mut self) -> Option { match self.inner.next() { - Some(service::Io::Write(addr, msgs)) => { + Some(Io::Write(addr, msgs)) => { let mut buf = Vec::new(); for msg in msgs { log::debug!("Write {:?} to {}", &msg, addr.ip()); @@ -515,10 +516,10 @@ impl Iterator for Wire { } Some(nakamoto::Io::Write(addr, buf)) } - Some(service::Io::Event(e)) => Some(nakamoto::Io::Event(e)), - Some(service::Io::Connect(a)) => Some(nakamoto::Io::Connect(a)), - Some(service::Io::Disconnect(a, r)) => Some(nakamoto::Io::Disconnect(a, r)), - Some(service::Io::Wakeup(d)) => Some(nakamoto::Io::Wakeup(d)), + Some(Io::Event(e)) => Some(nakamoto::Io::Event(e)), + Some(Io::Connect(a)) => Some(nakamoto::Io::Connect(a)), + Some(Io::Disconnect(a, r)) => Some(nakamoto::Io::Disconnect(a, r)), + Some(Io::Wakeup(d)) => Some(nakamoto::Io::Wakeup(d)), None => None, } diff --git a/radicle/src/identity/project.rs b/radicle/src/identity/project.rs index 153b7175..7bd4da3d 100644 --- a/radicle/src/identity/project.rs +++ b/radicle/src/identity/project.rs @@ -125,11 +125,11 @@ impl Doc { Ok((oid, sig)) } - pub fn create<'r, S: WriteStorage<'r>>( + pub fn create( &self, remote: &RemoteId, msg: &str, - storage: &'r S, + storage: &S, ) -> Result<(Id, git::Oid, S::Repository), Error> { // You can checkout this branch in your working copy with: // @@ -147,7 +147,7 @@ impl Doc { Ok((id, oid, repo)) } - pub fn update<'r, R: WriteRepository<'r>>( + pub fn update( &self, remote: &RemoteId, msg: &str, @@ -320,7 +320,7 @@ impl Doc { }) } - pub fn blob_at<'r, R: ReadRepository<'r>>( + pub fn blob_at( commit: Oid, repo: &R, ) -> Result, git::Error> { @@ -331,7 +331,7 @@ impl Doc { } } - pub fn load_at<'r, R: ReadRepository<'r>>( + pub fn load_at( commit: Oid, repo: &R, ) -> Result, git::Error> { @@ -342,7 +342,7 @@ impl Doc { Ok(None) } - pub fn load<'r, R: ReadRepository<'r>>( + pub fn load( remote: &RemoteId, repo: &R, ) -> Result, git::Error> { @@ -355,10 +355,7 @@ impl Doc { } impl Doc { - pub fn head<'r, R: ReadRepository<'r>>( - remote: &RemoteId, - repo: &R, - ) -> Result, git::Error> { + pub fn head(remote: &RemoteId, repo: &R) -> Result, git::Error> { let head = &git::refname!("heads").join(&*git::refs::IDENTITY_BRANCH); if let Some(oid) = repo.reference_oid(remote, head)? { Ok(Some(oid)) @@ -421,7 +418,7 @@ impl Identity { } impl Identity { - pub fn load<'r, R: ReadRepository<'r>>( + pub fn load( remote: &RemoteId, repo: &R, ) -> Result>, IdentityError> { diff --git a/radicle/src/rad.rs b/radicle/src/rad.rs index 7238b9c5..47b34dc2 100644 --- a/radicle/src/rad.rs +++ b/radicle/src/rad.rs @@ -35,13 +35,13 @@ pub enum InitError { } /// Initialize a new radicle project from a git repository. -pub fn init<'r, G: Signer, S: storage::WriteStorage<'r>>( +pub fn init( repo: &git2::Repository, name: &str, description: &str, default_branch: BranchName, signer: G, - storage: &'r S, + storage: &S, ) -> Result<(Id, SignedRefs), InitError> { let pk = signer.public_key(); let delegate = identity::Delegate { @@ -98,7 +98,7 @@ pub enum ForkError { } /// Create a local tree for an existing project, from an existing remote. -pub fn fork_remote<'r, G: Signer, S: storage::WriteStorage<'r>>( +pub fn fork_remote( proj: &Id, remote: &RemoteId, signer: G, @@ -147,7 +147,7 @@ pub fn fork_remote<'r, G: Signer, S: storage::WriteStorage<'r>>( Ok(()) } -pub fn fork<'r, G: Signer, S: storage::WriteStorage<'r>>( +pub fn fork( proj: &Id, signer: &G, storage: &S, @@ -206,7 +206,7 @@ pub enum CloneError { Checkout(#[from] CheckoutError), } -pub fn clone<'r, P: AsRef, G: Signer, S: storage::WriteStorage<'r>, H: node::Handle>( +pub fn clone, G: Signer, S: storage::WriteStorage, H: node::Handle>( proj: &Id, path: P, signer: &G, @@ -232,7 +232,7 @@ pub enum CloneUrlError { Checkout(#[from] CheckoutError), } -pub fn clone_url<'r, P: AsRef, G: Signer, S: storage::WriteStorage<'r>>( +pub fn clone_url, G: Signer, S: storage::WriteStorage>( proj: &Id, url: &git::Url, path: P, diff --git a/radicle/src/storage.rs b/radicle/src/storage.rs index d5655ada..fc84b8fc 100644 --- a/radicle/src/storage.rs +++ b/radicle/src/storage.rs @@ -222,8 +222,8 @@ pub trait ReadStorage { fn inventory(&self) -> Result; } -pub trait WriteStorage<'r>: ReadStorage { - type Repository: WriteRepository<'r>; +pub trait WriteStorage: ReadStorage { + type Repository: WriteRepository; fn repository(&self, proj: &Id) -> Result; fn sign_refs( @@ -231,11 +231,10 @@ pub trait WriteStorage<'r>: ReadStorage { repository: &Self::Repository, signer: G, ) -> Result, Error>; + fn fetch(&self, proj_id: &Id, remote: &Url) -> Result, FetchError>; } -pub trait ReadRepository<'r> { - type Remotes: Iterator), refs::Error>> + 'r; - +pub trait ReadRepository { fn is_empty(&self) -> Result; fn path(&self) -> &Path; fn blob_at<'a>(&'a self, oid: Oid, path: &'a Path) -> Result, git_ext::Error>; @@ -253,13 +252,13 @@ pub trait ReadRepository<'r> { ) -> Result, git2::Error>; fn references(&self, remote: &RemoteId) -> Result; fn remote(&self, remote: &RemoteId) -> Result, refs::Error>; - fn remotes(&'r self) -> Result; + fn remotes(&self) -> Result, refs::Error>; /// Return the project associated with this repository. fn project(&self) -> Result, Error>; fn project_identity(&self) -> Result<(Oid, identity::Doc), ProjectError>; } -pub trait WriteRepository<'r>: ReadRepository<'r> { +pub trait WriteRepository: ReadRepository { fn fetch(&mut self, url: &Url) -> Result, FetchError>; fn raw(&self) -> &git2::Repository; } @@ -286,10 +285,10 @@ where } } -impl<'r, T, S> WriteStorage<'r> for T +impl WriteStorage for T where T: Deref, - S: WriteStorage<'r> + 'static, + S: WriteStorage + 'static, { type Repository = S::Repository; @@ -304,6 +303,10 @@ where ) -> Result, Error> { self.deref().sign_refs(repository, signer) } + + fn fetch(&self, proj_id: &Id, remote: &Url) -> Result, FetchError> { + self.deref().fetch(proj_id, remote) + } } #[cfg(test)] diff --git a/radicle/src/storage/git.rs b/radicle/src/storage/git.rs index 6e530f58..093a93c6 100644 --- a/radicle/src/storage/git.rs +++ b/radicle/src/storage/git.rs @@ -13,7 +13,7 @@ use crate::identity::{Doc, Id}; use crate::storage::refs; use crate::storage::refs::{Refs, SignedRefs}; use crate::storage::{ - Error, FetchError, Inventory, ReadRepository, ReadStorage, Remote, WriteRepository, + Error, FetchError, Inventory, ReadRepository, ReadStorage, Remote, Remotes, WriteRepository, WriteStorage, }; @@ -79,7 +79,7 @@ impl ReadStorage for Storage { } } -impl<'r> WriteStorage<'r> for Storage { +impl WriteStorage for Storage { type Repository = Repository; fn repository(&self, proj: &Id) -> Result { @@ -99,6 +99,19 @@ impl<'r> WriteStorage<'r> for Storage { Ok(signed) } + + fn fetch(&self, proj_id: &Id, remote: &Url) -> Result, FetchError> { + let mut repo = self.repository(proj_id).unwrap(); + let mut path = remote.path.clone(); + + path.push(b'/'); + path.extend(proj_id.to_string().into_bytes()); + + repo.fetch(&Url { + path, + ..remote.clone() + }) + } } impl Storage { @@ -338,11 +351,28 @@ impl Repository { ); Ok(iter) } + + pub fn remotes( + &self, + ) -> Result< + impl Iterator), refs::Error>> + '_, + git2::Error, + > { + let remotes = self.backend.references_glob(SIGNATURES_GLOB.as_str())?.map( + |reference| -> Result<_, _> { + let r = reference?; + let name = r.name().ok_or(refs::Error::InvalidRef)?; + let (id, _) = git::parse_ref::(name)?; + let remote = self.remote(&id)?; + + Ok((id, remote)) + }, + ); + Ok(remotes) + } } -impl<'r> ReadRepository<'r> for Repository { - type Remotes = Box), refs::Error>> + 'r>; - +impl ReadRepository for Repository { fn is_empty(&self) -> Result { let some = self.remotes()?.next().is_some(); Ok(!some) @@ -425,19 +455,12 @@ impl<'r> ReadRepository<'r> for Repository { Ok(refs.into()) } - fn remotes(&'r self) -> Result { - let iter = self.backend.references_glob(SIGNATURES_GLOB.as_str())?.map( - |reference| -> Result<(RemoteId, Remote), refs::Error> { - let r = reference?; - let name = r.name().ok_or(refs::Error::InvalidRef)?; - let (id, _) = git::parse_ref::(name)?; - let remote = self.remote(&id)?; - - Ok((id, remote)) - }, - ); - - Ok(Box::new(iter)) + fn remotes(&self) -> Result, refs::Error> { + let mut remotes = Vec::new(); + for remote in Repository::remotes(self)? { + remotes.push(remote?); + } + Ok(Remotes::from_iter(remotes)) } fn project(&self) -> Result, Error> { @@ -449,7 +472,7 @@ impl<'r> ReadRepository<'r> for Repository { } } -impl<'r> WriteRepository<'r> for Repository { +impl WriteRepository for Repository { /// Fetch all remotes of a project from the given URL. fn fetch(&mut self, url: &git::Url) -> Result, FetchError> { // TODO: Have function to fetch specific remotes. diff --git a/radicle/src/storage/refs.rs b/radicle/src/storage/refs.rs index 515f1842..8c8841fc 100644 --- a/radicle/src/storage/refs.rs +++ b/radicle/src/storage/refs.rs @@ -213,9 +213,9 @@ impl SignedRefs { } impl SignedRefs { - pub fn load<'r, S>(remote: &RemoteId, repo: &S) -> Result + pub fn load(remote: &RemoteId, repo: &S) -> Result where - S: ReadRepository<'r>, + S: ReadRepository, { if let Some(oid) = repo.reference_oid(remote, &SIGNATURE_REF)? { Self::load_at(oid, remote, repo) @@ -224,9 +224,9 @@ impl SignedRefs { } } - pub fn load_at<'r, S>(oid: Oid, remote: &RemoteId, repo: &S) -> Result + pub fn load_at(oid: Oid, remote: &RemoteId, repo: &S) -> Result where - S: storage::ReadRepository<'r>, + S: storage::ReadRepository, { let refs = repo.blob_at(oid, Path::new(REFS_BLOB_PATH))?; let signature = repo.blob_at(oid, Path::new(SIGNATURE_BLOB_PATH))?; @@ -248,7 +248,7 @@ impl SignedRefs { /// Save the signed refs to disk. /// This creates a new commit on the signed refs branch, and updates the branch pointer. - pub fn save<'r, S: WriteRepository<'r>>( + pub fn save( &self, // TODO: This should be part of the signed refs. remote: &RemoteId, diff --git a/radicle/src/test/fixtures.rs b/radicle/src/test/fixtures.rs index 1e9ff0e0..2ef83275 100644 --- a/radicle/src/test/fixtures.rs +++ b/radicle/src/test/fixtures.rs @@ -33,9 +33,9 @@ pub fn storage, G: Signer>(path: P, signer: G) -> Result, S: WriteStorage<'r>, G: Signer>( +pub fn project, S: WriteStorage, G: Signer>( path: P, - storage: &'r S, + storage: &S, signer: G, ) -> Result<(Id, SignedRefs, git2::Repository, git2::Oid), rad::InitError> { let (repo, head) = repository(path); diff --git a/radicle/src/test/storage.rs b/radicle/src/test/storage.rs index ef2452fe..6c223a78 100644 --- a/radicle/src/test/storage.rs +++ b/radicle/src/test/storage.rs @@ -53,7 +53,7 @@ impl ReadStorage for MockStorage { } } -impl WriteStorage<'_> for MockStorage { +impl WriteStorage for MockStorage { type Repository = MockRepository; fn repository(&self, _proj: &Id) -> Result { @@ -67,13 +67,15 @@ impl WriteStorage<'_> for MockStorage { ) -> Result, Error> { todo!() } + + fn fetch(&self, _proj_id: &Id, _remote: &Url) -> Result, FetchError> { + Ok(vec![]) + } } pub struct MockRepository {} -impl ReadRepository<'_> for MockRepository { - type Remotes = std::iter::Empty), refs::Error>>; - +impl ReadRepository for MockRepository { fn is_empty(&self) -> Result { Ok(true) } @@ -86,7 +88,7 @@ impl ReadRepository<'_> for MockRepository { todo!() } - fn remotes(&self) -> Result { + fn remotes(&self) -> Result, refs::Error> { todo!() } @@ -137,7 +139,7 @@ impl ReadRepository<'_> for MockRepository { } } -impl WriteRepository<'_> for MockRepository { +impl WriteRepository for MockRepository { fn fetch(&mut self, _url: &Url) -> Result, FetchError> { Ok(vec![]) }