From 1c20f64a26641ecb7777567b9661599d92757e08 Mon Sep 17 00:00:00 2001 From: Lorenz Leutgeb Date: Fri, 20 Jun 2025 10:01:58 +0200 Subject: [PATCH] node, protocol: Refactor Mostly changes of `use`s and moving components to `radicle-protocol`, with the goal of just getting `radicle-node` to work on top of the new `radicle-protocol` crate. --- Cargo.lock | 11 -- crates/radicle-cli/tests/commands.rs | 4 +- crates/radicle-node/Cargo.toml | 2 +- crates/radicle-node/src/control.rs | 2 +- crates/radicle-node/src/lib.rs | 22 ++- crates/radicle-node/src/runtime.rs | 6 +- crates/radicle-node/src/runtime/handle.rs | 10 +- crates/radicle-node/src/test.rs | 1 - crates/radicle-node/src/test/arbitrary.rs | 142 ------------------ crates/radicle-node/src/test/environment.rs | 5 +- crates/radicle-node/src/test/handle.rs | 4 +- crates/radicle-node/src/test/node.rs | 2 +- crates/radicle-node/src/test/peer.rs | 9 +- crates/radicle-node/src/test/simulator.rs | 7 +- crates/radicle-node/src/tests.rs | 9 +- crates/radicle-node/src/tests/e2e.rs | 19 ++- .../protocol.rs => radicle-node/src/wire.rs} | 30 ++-- crates/radicle-node/src/worker.rs | 117 +-------------- crates/radicle-node/src/worker/fetch.rs | 28 +--- crates/radicle-protocol/Cargo.toml | 16 +- crates/radicle-protocol/src/bounded.rs | 12 ++ .../src/deserializer.rs | 8 +- crates/radicle-protocol/src/lib.rs | 8 + crates/radicle-protocol/src/service.rs | 44 +++--- crates/radicle-protocol/src/service/filter.rs | 20 ++- crates/radicle-protocol/src/service/gossip.rs | 9 +- .../src/service/gossip/store.rs | 11 +- crates/radicle-protocol/src/service/io.rs | 9 +- .../radicle-protocol/src/service/message.rs | 99 ++++++++++-- .../radicle-protocol/src/service/session.rs | 11 +- crates/radicle-protocol/src/wire.rs | 40 ++--- crates/radicle-protocol/src/wire/frame.rs | 4 +- crates/radicle-protocol/src/wire/message.rs | 26 +--- crates/radicle-protocol/src/worker.rs | 119 +++++++++++++++ crates/radicle-protocol/src/worker/fetch.rs | 41 +++++ .../src/worker/fetch/error.rs | 0 36 files changed, 458 insertions(+), 449 deletions(-) delete mode 100644 crates/radicle-node/src/test/arbitrary.rs rename crates/{radicle-protocol/src/wire/protocol.rs => radicle-node/src/wire.rs} (98%) rename crates/{radicle-node => radicle-protocol}/src/deserializer.rs (96%) create mode 100644 crates/radicle-protocol/src/lib.rs create mode 100644 crates/radicle-protocol/src/worker.rs create mode 100644 crates/radicle-protocol/src/worker/fetch.rs rename crates/{radicle-node => radicle-protocol}/src/worker/fetch/error.rs (100%) diff --git a/Cargo.lock b/Cargo.lock index a05294e6..b81bb351 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2679,21 +2679,13 @@ dependencies = [ name = "radicle-protocol" version = "0.1.0" dependencies = [ - "amplify", - "anyhow", "bloomy", "byteorder", - "chrono", - "colored", "crossbeam-channel", "cyphernet", "fastrand", - "io-reactor", - "lexopt", - "libc", "localtime", "log", - "netservices", "nonempty 0.9.0", "qcheck", "qcheck-macros", @@ -2704,10 +2696,7 @@ dependencies = [ "scrypt", "serde", "serde_json", - "snapbox", - "socket2", "sqlite", - "tempfile", "thiserror 1.0.69", ] diff --git a/crates/radicle-cli/tests/commands.rs b/crates/radicle-cli/tests/commands.rs index 60b13a42..b0ff852f 100644 --- a/crates/radicle-cli/tests/commands.rs +++ b/crates/radicle-cli/tests/commands.rs @@ -7,6 +7,8 @@ use radicle::node; use radicle::node::address::Store as _; use radicle::node::config::seeds::RADICLE_NODE_BOOTSTRAP_IRIS; use radicle::node::config::DefaultSeedingPolicy; +use radicle::node::events::Event; +use radicle::node::policy::Scope; use radicle::node::routing::Store as _; use radicle::node::UserAgent; use radicle::node::{Address, Alias, Config, Handle as _, DEFAULT_TIMEOUT}; @@ -16,8 +18,6 @@ use radicle::profile::Home; use radicle::storage::{ReadStorage, RefUpdate, RemoteRepository}; use radicle::test::fixtures; -use radicle_node::service::policy::Scope; -use radicle_node::service::Event; #[allow(unused_imports)] use radicle_node::test::logger; use radicle_node::test::node::Node; diff --git a/crates/radicle-node/Cargo.toml b/crates/radicle-node/Cargo.toml index e2e084bf..ce05ef47 100644 --- a/crates/radicle-node/Cargo.toml +++ b/crates/radicle-node/Cargo.toml @@ -12,7 +12,7 @@ rust-version.workspace = true [features] default = ["systemd"] systemd = ["dep:radicle-systemd"] -test = ["radicle/test", "radicle-crypto/test", "radicle-crypto/cyphernet", "qcheck", "snapbox"] +test = ["radicle/test", "radicle-crypto/test", "radicle-crypto/cyphernet", "radicle-protocol/test", "qcheck", "snapbox"] [dependencies] amplify = { workspace = true } diff --git a/crates/radicle-node/src/control.rs b/crates/radicle-node/src/control.rs index fb789667..6c79eee0 100644 --- a/crates/radicle-node/src/control.rs +++ b/crates/radicle-node/src/control.rs @@ -246,9 +246,9 @@ mod tests { use super::*; use crate::identity::RepoId; + use crate::node::policy::Scope; use crate::node::Handle; use crate::node::{Alias, Node, NodeId}; - use crate::service::policy::Scope; use crate::test; #[test] diff --git a/crates/radicle-node/src/lib.rs b/crates/radicle-node/src/lib.rs index 269a63e7..4e9fbfbe 100644 --- a/crates/radicle-node/src/lib.rs +++ b/crates/radicle-node/src/lib.rs @@ -3,11 +3,12 @@ // suggestions did not make sense. #![allow(clippy::byte_char_slices)] -pub mod bounded; +use std::str::FromStr; +use std::sync::LazyLock; + pub mod control; -pub mod deserializer; pub mod runtime; -pub mod service; +pub(crate) use radicle_protocol::service; #[cfg(any(test, feature = "test"))] pub mod test; #[cfg(test)] @@ -18,7 +19,8 @@ pub mod worker; use radicle::version::Version; pub use localtime::{LocalDuration, LocalTime}; -pub use netservices::Direction as Link; +pub use radicle::node::Link; +pub use radicle::node::UserAgent; pub use radicle::node::PROTOCOL_VERSION; pub use radicle::prelude::Timestamp; pub use radicle::{collections, crypto, git, identity, node, profile, rad, storage}; @@ -32,14 +34,18 @@ pub const VERSION: Version = Version { timestamp: env!("SOURCE_DATE_EPOCH"), }; +/// This node's user agent string. +pub static USER_AGENT: LazyLock = LazyLock::new(|| { + FromStr::from_str(format!("/radicle:{}/", VERSION.version).as_str()) + .expect("user agent is valid") +}); + pub mod prelude { - pub use crate::bounded::BoundedVec; pub use crate::crypto::{PublicKey, Signature}; - pub use crate::deserializer::Deserializer; pub use crate::identity::{Did, RepoId}; - pub use crate::node::Address; + pub use crate::node::{config::Network, Address, Event, NodeId}; pub use crate::service::filter::Filter; - pub use crate::service::{DisconnectReason, Event, Message, Network, NodeId}; + pub use crate::service::{DisconnectReason, Message}; pub use crate::storage::refs::Refs; pub use crate::storage::WriteStorage; pub use crate::{LocalDuration, LocalTime, Timestamp}; diff --git a/crates/radicle-node/src/runtime.rs b/crates/radicle-node/src/runtime.rs index ad2d5cc1..2aa4c598 100644 --- a/crates/radicle-node/src/runtime.rs +++ b/crates/radicle-node/src/runtime.rs @@ -21,6 +21,8 @@ use radicle::node; use radicle::node::address; use radicle::node::address::Store as _; use radicle::node::notifications; +use radicle::node::policy::config as policy; +use radicle::node::Event; use radicle::node::Handle as _; use radicle::node::UserAgent; use radicle::profile::Home; @@ -29,7 +31,7 @@ use radicle::{cob, git, storage, Storage}; use crate::control; use crate::node::{routing, NodeId}; use crate::service::message::NodeAnnouncement; -use crate::service::{gossip, policy, Event, INITIAL_SUBSCRIBE_BACKLOG_DELTA}; +use crate::service::{gossip, INITIAL_SUBSCRIBE_BACKLOG_DELTA}; use crate::wire; use crate::wire::{Decode, Wire}; use crate::worker; @@ -117,7 +119,7 @@ impl Runtime { /// This function spawns threads. pub fn init( home: Home, - config: service::Config, + config: radicle::node::Config, listen: Vec, signals: chan::Receiver, signer: Device, diff --git a/crates/radicle-node/src/runtime/handle.rs b/crates/radicle-node/src/runtime/handle.rs index 5721a5aa..cfc94ec0 100644 --- a/crates/radicle-node/src/runtime/handle.rs +++ b/crates/radicle-node/src/runtime/handle.rs @@ -5,8 +5,10 @@ use std::sync::Arc; use std::{fmt, io, time}; use crossbeam_channel as chan; +use radicle::node::events::{Event, Events}; +use radicle::node::policy; +use radicle::node::{Config, NodeId}; use radicle::node::{ConnectOptions, ConnectResult, Seeds}; -use radicle::storage::refs::RefsAt; use reactor::poller::popol::PopolWaker; use serde_json::json; use thiserror::Error; @@ -16,10 +18,8 @@ use crate::node::{Alias, Command, FetchResult}; use crate::profile::Home; use crate::runtime::Emitter; use crate::service; -use crate::service::policy; -use crate::service::NodeId; -use crate::service::{CommandError, Config, QueryState}; -use crate::service::{Event, Events}; +use crate::service::{CommandError, QueryState}; +use crate::storage::refs::RefsAt; use crate::wire; use crate::wire::StreamId; use crate::worker::TaskResult; diff --git a/crates/radicle-node/src/test.rs b/crates/radicle-node/src/test.rs index e4225831..87500a36 100644 --- a/crates/radicle-node/src/test.rs +++ b/crates/radicle-node/src/test.rs @@ -1,4 +1,3 @@ -pub mod arbitrary; pub mod gossip; pub mod handle; pub mod node; diff --git a/crates/radicle-node/src/test/arbitrary.rs b/crates/radicle-node/src/test/arbitrary.rs deleted file mode 100644 index 19728895..00000000 --- a/crates/radicle-node/src/test/arbitrary.rs +++ /dev/null @@ -1,142 +0,0 @@ -use std::collections::HashSet; - -use bloomy::BloomFilter; -use qcheck::Arbitrary; -use radicle::node::UserAgent; - -use crate::crypto; -use crate::identity::DocAt; -use crate::node::Alias; -use crate::prelude::{BoundedVec, NodeId, RepoId, Timestamp}; -use crate::service::filter::{Filter, FILTER_SIZE_L, FILTER_SIZE_M, FILTER_SIZE_S}; -use crate::service::message::{ - Announcement, Info, InventoryAnnouncement, Message, NodeAnnouncement, Ping, RefsAnnouncement, - Subscribe, ZeroBytes, -}; -use crate::wire::MessageType; -use crate::worker::fetch::FetchResult; - -pub use radicle::test::arbitrary::*; - -impl Arbitrary for Filter { - fn arbitrary(g: &mut qcheck::Gen) -> Self { - let size = *g - .choose(&[FILTER_SIZE_S, FILTER_SIZE_M, FILTER_SIZE_L]) - .unwrap(); - let mut bytes = vec![0; size]; - for _ in 0..64 { - let index = usize::arbitrary(g) % bytes.len(); - bytes[index] = u8::arbitrary(g); - } - Self::from(BloomFilter::from(bytes)) - } -} - -impl Arbitrary for FetchResult { - fn arbitrary(g: &mut qcheck::Gen) -> Self { - FetchResult { - updated: vec![], - namespaces: HashSet::arbitrary(g), - clone: bool::arbitrary(g), - doc: DocAt::arbitrary(g), - } - } -} - -impl Arbitrary for Message { - fn arbitrary(g: &mut qcheck::Gen) -> Self { - let type_id = g - .choose(&[ - MessageType::InventoryAnnouncement, - MessageType::NodeAnnouncement, - MessageType::RefsAnnouncement, - MessageType::Info, - MessageType::Subscribe, - MessageType::Ping, - MessageType::Pong, - ]) - .unwrap(); - - match type_id { - MessageType::InventoryAnnouncement => Announcement { - node: NodeId::arbitrary(g), - message: InventoryAnnouncement { - inventory: BoundedVec::arbitrary(g), - timestamp: Timestamp::arbitrary(g), - } - .into(), - signature: crypto::Signature::from(<[u8; 64]>::arbitrary(g)), - } - .into(), - MessageType::RefsAnnouncement => Announcement { - node: NodeId::arbitrary(g), - message: RefsAnnouncement { - rid: RepoId::arbitrary(g), - refs: BoundedVec::arbitrary(g), - timestamp: Timestamp::arbitrary(g), - } - .into(), - signature: crypto::Signature::from(<[u8; 64]>::arbitrary(g)), - } - .into(), - MessageType::NodeAnnouncement => { - let message = NodeAnnouncement { - version: u8::arbitrary(g), - features: u64::arbitrary(g).into(), - timestamp: Timestamp::arbitrary(g), - alias: Alias::arbitrary(g), - addresses: Arbitrary::arbitrary(g), - nonce: u64::arbitrary(g), - agent: UserAgent::arbitrary(g), - } - .into(); - let bytes: [u8; 64] = Arbitrary::arbitrary(g); - let signature = crypto::Signature::from(bytes); - - Announcement { - node: NodeId::arbitrary(g), - signature, - message, - } - .into() - } - MessageType::Info => { - let message = Info::RefsAlreadySynced { - rid: RepoId::arbitrary(g), - at: oid(), - }; - Self::Info(message) - } - MessageType::Subscribe => Self::Subscribe(Subscribe { - filter: Filter::arbitrary(g), - since: Timestamp::arbitrary(g), - until: Timestamp::arbitrary(g), - }), - MessageType::Ping => { - let mut rng = fastrand::Rng::with_seed(u64::arbitrary(g)); - - Self::Ping(Ping::new(&mut rng)) - } - MessageType::Pong => Self::Pong { - zeroes: ZeroBytes::new(u16::arbitrary(g).min(Ping::MAX_PONG_ZEROES)), - }, - } - } -} - -impl Arbitrary for ZeroBytes { - fn arbitrary(g: &mut qcheck::Gen) -> Self { - ZeroBytes::new(u16::arbitrary(g)) - } -} - -impl Arbitrary for BoundedVec -where - T: Arbitrary + Eq, -{ - fn arbitrary(g: &mut qcheck::Gen) -> Self { - let mut v: Vec = Arbitrary::arbitrary(g); - v.truncate(N); - v.try_into().expect("size within bounds") - } -} diff --git a/crates/radicle-node/src/test/environment.rs b/crates/radicle-node/src/test/environment.rs index a68d11ce..edf98c13 100644 --- a/crates/radicle-node/src/test/environment.rs +++ b/crates/radicle-node/src/test/environment.rs @@ -19,8 +19,10 @@ use radicle::git::refname; use radicle::identity::{RepoId, Visibility}; use radicle::node::config::ConnectAddress; use radicle::node::device::Device; +use radicle::node::events::Event; use radicle::node::policy::store as policy; use radicle::node::seed::Store as _; +pub use radicle::node::Config; use radicle::node::{Alias, Database, UserAgent, POLICIES_DB_FILE}; use radicle::node::{ConnectOptions, Handle as _}; use radicle::profile; @@ -34,12 +36,9 @@ use radicle::{cob, explorer}; use radicle::{git, web}; use crate::node::NodeId; -use crate::service::Event; use crate::storage::git::transport; use crate::{runtime, runtime::Handle, service, Runtime}; -pub use service::Config; - /// Test environment. pub struct Environment { tempdir: tempfile::TempDir, diff --git a/crates/radicle-node/src/test/handle.rs b/crates/radicle-node/src/test/handle.rs index e37290c0..10ff46f8 100644 --- a/crates/radicle-node/src/test/handle.rs +++ b/crates/radicle-node/src/test/handle.rs @@ -9,8 +9,8 @@ use radicle::storage::refs::RefsAt; use crate::identity::RepoId; use crate::node::{Alias, Config, ConnectOptions, ConnectResult, Event, FetchResult, Seeds}; use crate::runtime::HandleError; -use crate::service::policy; -use crate::service::NodeId; +use radicle::node::policy; +use radicle::node::NodeId; #[derive(Default, Clone)] pub struct Handle { diff --git a/crates/radicle-node/src/test/node.rs b/crates/radicle-node/src/test/node.rs index e583db75..5de098d3 100644 --- a/crates/radicle-node/src/test/node.rs +++ b/crates/radicle-node/src/test/node.rs @@ -22,6 +22,7 @@ use radicle::node::config::ConnectAddress; use radicle::node::policy::store as policy; use radicle::node::seed::Store as _; use radicle::node::Config; +use radicle::node::Event; use radicle::node::{self, Alias}; use radicle::node::{ConnectOptions, Handle as _}; use radicle::node::{Database, POLICIES_DB_FILE}; @@ -33,7 +34,6 @@ use radicle::Storage; use crate::node::device::Device; use crate::node::NodeId; -use crate::service::Event; use crate::storage::git::transport; use crate::{runtime, runtime::Handle, service, Runtime}; diff --git a/crates/radicle-node/src/test/peer.rs b/crates/radicle-node/src/test/peer.rs index a1eaf873..59ffe1e8 100644 --- a/crates/radicle-node/src/test/peer.rs +++ b/crates/radicle-node/src/test/peer.rs @@ -28,7 +28,6 @@ use crate::runtime::Emitter; use crate::service; use crate::service::io::Io; use crate::service::message::*; -use crate::service::policy::{Scope, SeedingPolicy}; use crate::service::*; use crate::storage::git::transport::remote; use crate::storage::{RemoteId, WriteStorage}; @@ -36,6 +35,10 @@ use crate::test::storage::MockStorage; use crate::test::{arbitrary, fixtures, simulator}; use crate::wire::MessageType; use crate::{Link, LocalDuration, LocalTime, PROTOCOL_VERSION}; +use radicle::node::events::Events; +use radicle::node::policy::config as policy; +use radicle::node::policy::{Scope, SeedingPolicy}; +use radicle_protocol::bounded::BoundedVec; /// Service instantiation used for testing. pub type Service = service::Service; @@ -100,7 +103,7 @@ where } pub struct Config + 'static> { - pub config: service::Config, + pub config: radicle::node::Config, pub local_time: LocalTime, pub policy: SeedingPolicy, pub signer: Device, @@ -113,7 +116,7 @@ impl Default for Config { let mut rng = fastrand::Rng::new(); let signer = Device::mock_rng(&mut rng); let tmp = tempfile::TempDir::new().unwrap(); - let config = service::Config::test(Alias::from_str("mocky").unwrap()); + let config = radicle::node::Config::test(Alias::from_str("mocky").unwrap()); Config { config, diff --git a/crates/radicle-node/src/test/simulator.rs b/crates/radicle-node/src/test/simulator.rs index 2f1648f9..5ee1dff4 100644 --- a/crates/radicle-node/src/test/simulator.rs +++ b/crates/radicle-node/src/test/simulator.rs @@ -13,16 +13,19 @@ use std::{fmt, io, net}; use localtime::{LocalDuration, LocalTime}; use log::*; +use radicle::node::events::Event; +use radicle::node::NodeId; +use radicle_protocol::worker::FetchError; use crate::crypto; use crate::prelude::{Address, RepoId}; use crate::service::io::Io; -use crate::service::{DisconnectReason, Event, Message, Metrics, NodeId}; +use crate::service::{DisconnectReason, Message, Metrics}; use crate::storage::Namespaces; use crate::storage::{ReadRepository, WriteStorage}; use crate::test::arbitrary; use crate::test::peer::Service; -use crate::worker::{fetch, FetchError}; +use crate::worker::fetch; use crate::Link; /// Minimum latency between peers. diff --git a/crates/radicle-node/src/tests.rs b/crates/radicle-node/src/tests.rs index 1bd4290f..82e5e83f 100644 --- a/crates/radicle-node/src/tests.rs +++ b/crates/radicle-node/src/tests.rs @@ -9,17 +9,19 @@ use std::sync::LazyLock; use std::time; use crossbeam_channel as chan; -use netservices::Direction as Link; use radicle::identity::Visibility; use radicle::node::address::Store as _; use radicle::node::device::Device; +use radicle::node::policy; use radicle::node::refs::Store as _; use radicle::node::routing::Store as _; +use radicle::node::Link; use radicle::node::{ConnectOptions, DEFAULT_TIMEOUT}; use radicle::storage::refs::RefsAt; use radicle::storage::RefUpdate; use radicle::test::arbitrary::gen; use radicle::test::storage::MockRepository; +use radicle_protocol::bounded::BoundedVec; use crate::collections::{RandomMap, RandomSet}; use crate::identity::RepoId; @@ -49,7 +51,6 @@ use crate::test::simulator::{Peer as _, Simulation}; use crate::test::storage::MockStorage; use crate::wire::Decode; use crate::wire::Encode; -use crate::worker; use crate::worker::fetch; use crate::LocalTime; use crate::{git, identity, rad, runtime, service, test}; @@ -1462,7 +1463,7 @@ fn test_fetch_missing_inventory_on_schedule() { alice.fetched( rid, bob.id, - Err(worker::FetchError::Io( + Err(radicle_protocol::worker::FetchError::Io( io::ErrorKind::ConnectionReset.into(), )), ); @@ -1806,7 +1807,7 @@ fn test_init_and_seed() { .find(|e| { matches!( e, - service::Event::RefsFetched { remote, .. } + radicle::node::events::Event::RefsFetched { remote, .. } if *remote == eve.node_id() ) }) diff --git a/crates/radicle-node/src/tests/e2e.rs b/crates/radicle-node/src/tests/e2e.rs index 6b0d17c3..ff25dbb6 100644 --- a/crates/radicle-node/src/tests/e2e.rs +++ b/crates/radicle-node/src/tests/e2e.rs @@ -1,6 +1,8 @@ use std::{collections::HashSet, thread, time}; use radicle::node::device::Device; +use radicle::node::policy::Scope; +use radicle::node::Event; use radicle::node::{Alias, ConnectResult, FetchResult, Handle as _, DEFAULT_TIMEOUT}; use radicle::storage::{ ReadRepository, ReadStorage, RefUpdate, RemoteRepository, SignRepository, ValidateRepository, @@ -13,7 +15,6 @@ use radicle::{git, issue}; use crate::node::config::Limits; use crate::node::{Config, ConnectOptions}; use crate::service; -use crate::service::policy::Scope; use crate::storage::git::transport; use crate::test::logger; use crate::test::node::{converge, Node}; @@ -688,8 +689,7 @@ fn test_large_fetch() { bob_events .wait( |e| { - matches!(e, service::Event::RefsFetched { updated, .. } if !updated.is_empty()) - .then_some(()) + matches!(e, Event::RefsFetched { updated, .. } if !updated.is_empty()).then_some(()) }, time::Duration::from_secs(9 * scale as u64), ) @@ -717,7 +717,7 @@ fn test_concurrent_fetches() { let mut alice_repos = HashSet::new(); let mut alice = Node::init( tmp.path(), - service::Config { + radicle::node::config::Config { limits: limits.clone(), relay: radicle::node::config::Relay::Always, ..config::relay("alice") @@ -725,7 +725,7 @@ fn test_concurrent_fetches() { ); let mut bob = Node::init( tmp.path(), - service::Config { + radicle::node::config::Config { limits, relay: radicle::node::config::Relay::Always, ..config::relay("bob") @@ -766,7 +766,7 @@ fn test_concurrent_fetches() { while !bob_repos.is_empty() { match alice_events.recv().unwrap() { - service::Event::RefsFetched { rid, updated, .. } if !updated.is_empty() => { + Event::RefsFetched { rid, updated, .. } if !updated.is_empty() => { bob_repos.remove(&rid); log::debug!(target: "test", "{} fetched {rid} ({} left)",alice.id, bob_repos.len()); } @@ -776,7 +776,7 @@ fn test_concurrent_fetches() { while !alice_repos.is_empty() { match bob_events.recv().unwrap() { - service::Event::RefsFetched { rid, updated, .. } if !updated.is_empty() => { + Event::RefsFetched { rid, updated, .. } if !updated.is_empty() => { alice_repos.remove(&rid); log::debug!(target: "test", "{} fetched {rid} ({} left)", bob.id, alice_repos.len()); } @@ -1255,8 +1255,7 @@ fn missing_delegate_default_branch() { bob_events .wait( |e| { - matches!(e, service::Event::RefsFetched { updated, .. } if !updated.is_empty()) - .then_some(()) + matches!(e, Event::RefsFetched { updated, .. } if !updated.is_empty()).then_some(()) }, DEFAULT_TIMEOUT, ) @@ -1416,7 +1415,7 @@ fn test_background_foreground_fetch() { bob.handle.announce_refs(rid).unwrap(); alice_events .wait( - |e| matches!(e, service::Event::RefsAnnounced { .. }).then_some(()), + |e| matches!(e, Event::RefsAnnounced { .. }).then_some(()), DEFAULT_TIMEOUT, ) .unwrap(); diff --git a/crates/radicle-protocol/src/wire/protocol.rs b/crates/radicle-node/src/wire.rs similarity index 98% rename from crates/radicle-protocol/src/wire/protocol.rs rename to crates/radicle-node/src/wire.rs index 3b0f9374..4a3e0553 100644 --- a/crates/radicle-protocol/src/wire/protocol.rs +++ b/crates/radicle-node/src/wire.rs @@ -24,20 +24,22 @@ use reactor::{ResourceId, ResourceType, Timestamp}; use radicle::collections::RandomMap; use radicle::crypto; use radicle::node::config::AddressConfig; +use radicle::node::Link; use radicle::node::NodeId; use radicle::storage::WriteStorage; +use radicle_protocol::deserializer::Deserializer; +pub use radicle_protocol::wire::frame; +pub use radicle_protocol::wire::frame::{Frame, FrameData, StreamId}; +pub use radicle_protocol::wire::*; +use radicle_protocol::worker::{FetchRequest, FetchResult}; -use crate::prelude::Deserializer; use crate::service; use crate::service::io::Io; use crate::service::FETCH_TIMEOUT; use crate::service::{session, DisconnectReason, Metrics, Service}; -use crate::wire::frame; -use crate::wire::frame::{Frame, FrameData, StreamId}; -use crate::wire::Encode; use crate::worker; -use crate::worker::{ChannelEvent, ChannelsConfig, FetchRequest, FetchResult, Task, TaskResult}; -use crate::Link; +use crate::worker::{ChannelEvent, ChannelsConfig}; +use crate::worker::{Task, TaskResult}; /// NoiseXK handshake pattern. pub const NOISE_XK: HandshakePattern = HandshakePattern { @@ -573,7 +575,10 @@ where return; } }; - let transport = match NetTransport::with_session(session, Link::Inbound) { + let transport = match NetTransport::with_session( + session, + netservices::Direction::Inbound, + ) { Ok(transport) => transport, Err(err) => { log::error!(target: "wire", "Failed to create transport for accepted connection: {err}"); @@ -1024,7 +1029,10 @@ where self.service.config(), ) .and_then(|session| { - NetTransport::>::with_session(session, Link::Outbound) + NetTransport::>::with_session( + session, + netservices::Direction::Outbound, + ) }) { Ok(transport) => { self.outbound.insert( @@ -1134,7 +1142,7 @@ pub fn dial>( remote_addr: NetAddr, remote_id: ::Pk, signer: G, - config: &service::Config, + config: &radicle::node::Config, ) -> io::Result> { // Determine what address to establish a TCP connection with, given the remote peer // address and our node configuration. @@ -1248,7 +1256,7 @@ mod test { #[test] fn test_pong_message_with_extension() { - use crate::deserializer; + use radicle_protocol::deserializer; let mut stream = Vec::new(); let pong = Message::Pong { @@ -1281,7 +1289,7 @@ mod test { #[test] fn test_inventory_ann_with_extension() { - use crate::deserializer; + use radicle_protocol::deserializer; #[derive(Debug)] struct MessageWithExt { diff --git a/crates/radicle-node/src/worker.rs b/crates/radicle-node/src/worker.rs index 2a24bd52..721cc805 100644 --- a/crates/radicle-node/src/worker.rs +++ b/crates/radicle-node/src/worker.rs @@ -5,22 +5,25 @@ mod upload_pack; pub mod fetch; pub mod garbage; -use std::io; use std::path::PathBuf; use crossbeam_channel as chan; use radicle::identity::RepoId; -use radicle::node::{notifications, Event}; +use radicle::node::notifications; +use radicle::node::policy::config as policy; +use radicle::node::policy::config::SeedingPolicy; use radicle::prelude::NodeId; use radicle::storage::refs::RefsAt; use radicle::storage::{ReadRepository, ReadStorage}; use radicle::{cob, crypto, Storage}; use radicle_fetch::FetchLimit; -use crate::runtime::{thread, Emitter, Handle}; -use crate::service::policy; -use crate::service::policy::SeedingPolicy; +pub use radicle_protocol::worker::{ + AuthorizationError, FetchError, FetchRequest, FetchResult, UploadError, +}; + +use crate::runtime::{thread, Handle}; use crate::wire::StreamId; pub use channels::{ChannelEvent, Channels, ChannelsConfig}; @@ -39,110 +42,6 @@ pub struct Config { pub policies_db: PathBuf, } -/// Error returned by fetch. -#[derive(thiserror::Error, Debug)] -pub enum FetchError { - #[error("the 'git fetch' command failed with exit code '{code}'")] - CommandFailed { code: i32 }, - #[error(transparent)] - Io(#[from] io::Error), - #[error(transparent)] - Fetch(#[from] fetch::error::Fetch), - #[error(transparent)] - Handle(#[from] fetch::error::Handle), - #[error(transparent)] - Storage(#[from] radicle::storage::Error), - #[error(transparent)] - PolicyStore(#[from] radicle::node::policy::store::Error), - #[error(transparent)] - Policy(#[from] radicle_fetch::policy::error::Policy), - #[error(transparent)] - Blocked(#[from] radicle_fetch::policy::error::Blocked), -} - -impl FetchError { - /// Check if it's a timeout error. - pub fn is_timeout(&self) -> bool { - matches!(self, FetchError::Io(e) if e.kind() == io::ErrorKind::TimedOut) - } -} - -/// Error returned by fetch responder. -#[derive(thiserror::Error, Debug)] -pub enum UploadError { - #[error("error parsing git command packet-line: {0}")] - PacketLine(io::Error), - #[error("error while performing git upload-pack: {0}")] - UploadPack(io::Error), - #[error(transparent)] - Authorization(#[from] AuthorizationError), -} - -#[derive(thiserror::Error, Debug)] -pub enum AuthorizationError { - #[error("{0} is not authorized to fetch {1}")] - Unauthorized(NodeId, RepoId), - #[error(transparent)] - PolicyStore(#[from] radicle::node::policy::store::Error), - #[error(transparent)] - Repository(#[from] radicle::storage::RepositoryError), -} - -impl UploadError { - /// Check if it's an end-of-file error. - pub fn is_eof(&self) -> bool { - matches!(self, UploadError::UploadPack(e) if e.kind() == io::ErrorKind::UnexpectedEof) - } -} - -/// Fetch job sent to worker thread. -#[derive(Debug, Clone)] -pub enum FetchRequest { - /// Client is initiating a fetch for the repository identified by - /// `rid` from the peer identified by `remote`. - Initiator { - /// Repo to fetch. - rid: RepoId, - /// Remote peer we are interacting with. - remote: NodeId, - /// If this fetch is for a particular set of `rad/sigrefs`. - refs_at: Option>, - }, - /// 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, - /// Reporter for upload-pack progress. - emitter: Emitter, - }, -} - -impl FetchRequest { - pub fn remote(&self) -> NodeId { - match self { - Self::Initiator { remote, .. } | Self::Responder { remote, .. } => *remote, - } - } -} - -/// Fetch result of an upload or fetch. -#[derive(Debug)] -pub enum FetchResult { - Initiator { - /// Repo fetched. - rid: RepoId, - /// Fetch result, including remotes fetched. - result: Result, - }, - Responder { - /// Repo requested. - rid: Option, - /// Upload result. - result: Result<(), UploadError>, - }, -} - /// Task to be accomplished on a worker thread. /// This is either going to be an outgoing or incoming fetch. pub struct Task { diff --git a/crates/radicle-node/src/worker/fetch.rs b/crates/radicle-node/src/worker/fetch.rs index f3ae1f44..bd870dc6 100644 --- a/crates/radicle-node/src/worker/fetch.rs +++ b/crates/radicle-node/src/worker/fetch.rs @@ -1,6 +1,5 @@ -pub mod error; +pub(crate) use radicle_protocol::worker::fetch::error; -use std::collections::HashSet; use std::str::FromStr; use localtime::LocalTime; @@ -8,7 +7,6 @@ use localtime::LocalTime; use radicle::cob::TypedId; use radicle::crypto::PublicKey; use radicle::identity::crefs::GetCanonicalRefs as _; -use radicle::identity::DocAt; use radicle::prelude::NodeId; use radicle::prelude::RepoId; use radicle::storage::git::Repository; @@ -20,32 +18,10 @@ use radicle::storage::{ use radicle::{cob, git, node, Storage}; use radicle_fetch::git::refs::Applied; use radicle_fetch::{Allowed, BlockList, FetchLimit}; +pub use radicle_protocol::worker::fetch::FetchResult; use super::channels::ChannelsFlush; -#[derive(Debug, Clone)] -pub struct FetchResult { - /// The set of updated references. - pub updated: Vec, - /// The set of remote namespaces that were updated. - pub namespaces: HashSet, - /// The fetch was a full clone. - pub clone: bool, - /// Identity doc of fetched repo. - pub doc: DocAt, -} - -impl FetchResult { - pub fn new(doc: DocAt) -> Self { - Self { - updated: vec![], - namespaces: HashSet::new(), - clone: false, - doc, - } - } -} - pub enum Handle { Clone { handle: radicle_fetch::Handle, diff --git a/crates/radicle-protocol/Cargo.toml b/crates/radicle-protocol/Cargo.toml index 8d0f5ecd..6a301073 100644 --- a/crates/radicle-protocol/Cargo.toml +++ b/crates/radicle-protocol/Cargo.toml @@ -9,24 +9,16 @@ edition.workspace = true rust-version.workspace = true [features] -test = ["radicle/test", "radicle-crypto/test", "radicle-crypto/cyphernet", "qcheck", "snapbox"] +test = ["radicle/test", "radicle-crypto/test", "radicle-crypto/cyphernet", "qcheck"] [dependencies] -amplify = { workspace = true } -anyhow = { workspace = true } bloomy = "1.2" byteorder = { workspace = true } -chrono = { workspace = true, features = ["clock"] } -colored = { workspace = true } crossbeam-channel = { workspace = true } -cyphernet = { workspace = true, features = ["tor", "dns", "ed25519", "p2p-ed25519"] } +cyphernet = { workspace = true, features = ["tor"] } fastrand = { workspace = true } -io-reactor = { version = "0.5.1", features = ["popol"] } -lexopt = { workspace = true } -libc = { workspace = true } log = { workspace = true, features = ["std"] } localtime = { workspace = true } -netservices = { version = "0.8.0", features = ["io-reactor", "socket2"] } nonempty = { workspace = true, features = ["serialize"] } qcheck = { workspace = true, optional = true } radicle = { workspace = true, features = ["logger"] } @@ -36,9 +28,6 @@ sqlite = { workspace = true, features = ["bundled"] } scrypt = { version = "0.11.0", default-features = false } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true, features = ["preserve_order"] } -snapbox = { workspace = true, optional = true } -socket2 = "0.5.7" -tempfile = { workspace = true } thiserror = { workspace = true } [dev-dependencies] @@ -46,4 +35,3 @@ qcheck = { workspace = true } qcheck-macros = { workspace = true } radicle = { workspace = true, features = ["test"] } radicle-crypto = { workspace = true, features = ["test", "cyphernet"] } -snapbox = { workspace = true } diff --git a/crates/radicle-protocol/src/bounded.rs b/crates/radicle-protocol/src/bounded.rs index 6b9b3c25..e136bc31 100644 --- a/crates/radicle-protocol/src/bounded.rs +++ b/crates/radicle-protocol/src/bounded.rs @@ -239,3 +239,15 @@ impl std::fmt::Debug for BoundedVec { self.v.fmt(f) } } + +#[cfg(any(test, feature = "test"))] +impl qcheck::Arbitrary for BoundedVec +where + T: qcheck::Arbitrary + Eq, +{ + fn arbitrary(g: &mut qcheck::Gen) -> Self { + let mut v: Vec = qcheck::Arbitrary::arbitrary(g); + v.truncate(N); + v.try_into().expect("size within bounds") + } +} diff --git a/crates/radicle-node/src/deserializer.rs b/crates/radicle-protocol/src/deserializer.rs similarity index 96% rename from crates/radicle-node/src/deserializer.rs rename to crates/radicle-protocol/src/deserializer.rs index 3879a9c2..e75ddeaa 100644 --- a/crates/radicle-node/src/deserializer.rs +++ b/crates/radicle-protocol/src/deserializer.rs @@ -2,7 +2,7 @@ use std::io; use std::marker::PhantomData; use crate::bounded; -use crate::prelude::BoundedVec; +use crate::bounded::BoundedVec; use crate::service::message::Message; use crate::wire; @@ -104,7 +104,7 @@ mod test { use super::*; use qcheck_macros::quickcheck; - use crate::test::assert_matches; + use radicle::assert_matches; const MSG_HELLO: &[u8] = &[5, b'h', b'e', b'l', b'l', b'o']; const MSG_BYE: &[u8] = &[3, b'b', b'y', b'e']; @@ -117,11 +117,11 @@ mod test { assert_matches!(decoder.deserialize_next(), Ok(None)); assert_eq!(decoder.unparsed.len(), 2); - decoder.input(&[b'y']).unwrap(); + decoder.input(b"y").unwrap(); assert_matches!(decoder.deserialize_next(), Ok(None)); assert_eq!(decoder.unparsed.len(), 3); - decoder.input(&[b'e']).unwrap(); + decoder.input(b"e").unwrap(); assert_matches!(decoder.deserialize_next(), Ok(Some(s)) if s.as_str() == "bye"); assert_eq!(decoder.unparsed.len(), 0); assert!(decoder.is_empty()); diff --git a/crates/radicle-protocol/src/lib.rs b/crates/radicle-protocol/src/lib.rs new file mode 100644 index 00000000..9c20f918 --- /dev/null +++ b/crates/radicle-protocol/src/lib.rs @@ -0,0 +1,8 @@ +pub mod bounded; +pub mod deserializer; +pub mod service; +pub mod wire; +pub mod worker; + +/// Peer-to-peer protocol version. +pub const PROTOCOL_VERSION: u8 = 1; diff --git a/crates/radicle-protocol/src/service.rs b/crates/radicle-protocol/src/service.rs index 663388b5..f2b53c31 100644 --- a/crates/radicle-protocol/src/service.rs +++ b/crates/radicle-protocol/src/service.rs @@ -38,32 +38,36 @@ use radicle::storage::refs::SIGREFS_BRANCH; use radicle::storage::RepositoryError; use radicle_fetch::policy::SeedingPolicy; -use crate::identity::RepoId; -use crate::node::routing; -use crate::node::routing::InsertResult; -use crate::node::{ - Address, Alias, Features, FetchResult, HostName, Seed, Seeds, SyncStatus, SyncedAt, -}; -use crate::prelude::*; -use crate::runtime::Emitter; use crate::service::gossip::Store as _; use crate::service::message::{ Announcement, AnnouncementMessage, Info, NodeAnnouncement, Ping, RefsAnnouncement, RefsStatus, }; use crate::service::policy::{store::Write, Scope}; -use crate::storage; -use crate::storage::{refs::RefsAt, Namespaces, ReadStorage}; -use crate::worker::fetch; -use crate::worker::FetchError; -use crate::Link; -use crate::{crypto, PROTOCOL_VERSION}; +use radicle::identity::RepoId; +use radicle::node::events::Emitter; +use radicle::node::routing; +use radicle::node::routing::InsertResult; +use radicle::node::{ + Address, Alias, Features, FetchResult, HostName, Seed, Seeds, SyncStatus, SyncedAt, +}; +use radicle::prelude::*; +use radicle::storage; +use radicle::storage::{refs::RefsAt, Namespaces, ReadStorage}; +// use radicle::worker::fetch; +// use crate::worker::FetchError; +use radicle::crypto; +use radicle::node::Link; +use radicle::node::PROTOCOL_VERSION; -pub use crate::node::events::{Event, Events}; -pub use crate::node::{config::Network, Config, NodeId}; +use crate::bounded::BoundedVec; +use crate::service::filter::Filter; pub use crate::service::message::{Message, ZeroBytes}; pub use crate::service::session::{QueuedFetch, Session}; +use crate::worker::FetchError; +use radicle::node::events::{Event, Events}; +use radicle::node::{Config, NodeId}; -pub use radicle::node::policy::config as policy; +use radicle::node::policy::config as policy; use self::io::Outbox; use self::limiter::RateLimiter; @@ -221,7 +225,7 @@ pub trait Store: { } -impl Store for node::Database {} +impl Store for radicle::node::Database {} /// Function used to query internal service state. pub type QueryState = dyn Fn(&dyn ServiceState) -> Result<(), CommandError> + Send + Sync; @@ -1133,7 +1137,7 @@ where &mut self, rid: RepoId, remote: NodeId, - result: Result, + result: Result, ) { let Some(fetching) = self.fetching.remove(&rid) else { error!(target: "service", "Received unexpected fetch result for {rid}, from {remote}"); @@ -1169,7 +1173,7 @@ where } match result { - Ok(fetch::FetchResult { + Ok(crate::worker::fetch::FetchResult { updated, namespaces, clone, diff --git a/crates/radicle-protocol/src/service/filter.rs b/crates/radicle-protocol/src/service/filter.rs index eae7c360..c57db7fc 100644 --- a/crates/radicle-protocol/src/service/filter.rs +++ b/crates/radicle-protocol/src/service/filter.rs @@ -3,7 +3,7 @@ use std::ops::{Deref, DerefMut}; pub use bloomy::BloomFilter; -use crate::identity::RepoId; +use radicle::identity::RepoId; /// Size in bytes of *large* bloom filter. /// It can store about 13'675 items with a false positive rate of 1%. @@ -90,10 +90,26 @@ impl From> for Filter { } } +#[allow(clippy::unwrap_used)] +#[cfg(any(test, feature = "test"))] +impl qcheck::Arbitrary for Filter { + fn arbitrary(g: &mut qcheck::Gen) -> Self { + let size = *g + .choose(&[FILTER_SIZE_S, FILTER_SIZE_M, FILTER_SIZE_L]) + .unwrap(); + let mut bytes = vec![0; size]; + for _ in 0..64 { + let index = usize::arbitrary(g) % bytes.len(); + bytes[index] = u8::arbitrary(g); + } + Self::from(BloomFilter::from(bytes)) + } +} + #[cfg(test)] mod test { use super::*; - use crate::test::arbitrary; + use radicle::test::arbitrary; #[test] fn test_parameters() { diff --git a/crates/radicle-protocol/src/service/gossip.rs b/crates/radicle-protocol/src/service/gossip.rs index 2a677868..8ba0f16f 100644 --- a/crates/radicle-protocol/src/service/gossip.rs +++ b/crates/radicle-protocol/src/service/gossip.rs @@ -4,14 +4,15 @@ use std::str::FromStr; use std::sync::LazyLock; use super::*; -use crate::{PROTOCOL_VERSION, VERSION}; +use crate::bounded::BoundedVec; use radicle::node::UserAgent; +use radicle::node::PROTOCOL_VERSION; pub use store::{AnnouncementId, Error, RelayStatus, Store}; /// This node's user agent string. -pub static USER_AGENT: LazyLock = LazyLock::new(|| { - FromStr::from_str(format!("/radicle:{}/", VERSION.version).as_str()) +pub static PROTOCOL_VERSION_STRING: LazyLock = LazyLock::new(|| { + FromStr::from_str(format!("/radicle:{}/", PROTOCOL_VERSION).as_str()) .expect("user agent is valid") }); @@ -23,7 +24,7 @@ pub fn node(config: &Config, timestamp: Timestamp) -> NodeAnnouncement { .clone() .try_into() .expect("external addresses are within the limit"); - let agent = USER_AGENT.clone(); + let agent = PROTOCOL_VERSION_STRING.clone(); let version = PROTOCOL_VERSION; NodeAnnouncement { diff --git a/crates/radicle-protocol/src/service/gossip/store.rs b/crates/radicle-protocol/src/service/gossip/store.rs index 49d9ba4e..a7288164 100644 --- a/crates/radicle-protocol/src/service/gossip/store.rs +++ b/crates/radicle-protocol/src/service/gossip/store.rs @@ -5,13 +5,15 @@ use radicle::crypto::Signature; use sqlite as sql; use thiserror::Error; -use crate::node::{Database, NodeId}; -use crate::prelude::{Filter, Timestamp}; +use crate::service::filter::Filter; use crate::service::message::{ Announcement, AnnouncementMessage, InventoryAnnouncement, NodeAnnouncement, RefsAnnouncement, }; use crate::wire; use crate::wire::Decode; +use radicle::node::Database; +use radicle::node::NodeId; +use radicle::prelude::Timestamp; #[derive(Error, Debug)] pub enum Error { @@ -395,11 +397,12 @@ mod parse { #[allow(clippy::unwrap_used)] mod test { use super::*; - use crate::prelude::{BoundedVec, RepoId}; - use crate::test::arbitrary; + use crate::bounded::BoundedVec; use localtime::LocalTime; use radicle::assert_matches; + use radicle::identity::RepoId; use radicle::node::device::Device; + use radicle::test::arbitrary; #[test] fn test_announced() { diff --git a/crates/radicle-protocol/src/service/io.rs b/crates/radicle-protocol/src/service/io.rs index 09b05fbb..f0f8eb1d 100644 --- a/crates/radicle-protocol/src/service/io.rs +++ b/crates/radicle-protocol/src/service/io.rs @@ -1,12 +1,17 @@ use std::collections::VecDeque; use std::time; +use localtime::LocalDuration; use log::*; +use radicle::identity::RepoId; use radicle::node::config::FetchPackSizeLimit; +use radicle::node::Address; +use radicle::node::NodeId; use radicle::storage::refs::RefsAt; -use crate::prelude::*; +use crate::service::message::Message; use crate::service::session::Session; +use crate::service::DisconnectReason; use crate::service::Link; use super::gossip; @@ -189,7 +194,7 @@ impl Outbox { } #[cfg(any(test, feature = "test"))] - pub(crate) fn queue(&mut self) -> &mut VecDeque { + pub fn queue(&mut self) -> &mut VecDeque { &mut self.io } } diff --git a/crates/radicle-protocol/src/service/message.rs b/crates/radicle-protocol/src/service/message.rs index 2b92e62d..5995b748 100644 --- a/crates/radicle-protocol/src/service/message.rs +++ b/crates/radicle-protocol/src/service/message.rs @@ -1,18 +1,18 @@ use std::{fmt, io, mem}; use nonempty::NonEmpty; +use radicle::crypto; use radicle::git; +use radicle::identity::RepoId; +use radicle::node; use radicle::node::device::Device; +use radicle::node::{Address, Alias, UserAgent}; +use radicle::storage; use radicle::storage::refs::RefsAt; -use crate::crypto; -use crate::identity::RepoId; -use crate::node; -use crate::node::{Address, Alias, UserAgent}; -use crate::prelude::BoundedVec; +use crate::bounded::BoundedVec; use crate::service::filter::Filter; use crate::service::{Link, NodeId, Timestamp}; -use crate::storage; use crate::wire; /// Maximum number of addresses which can be announced to other nodes. @@ -363,7 +363,7 @@ impl Announcement { #[cfg(not(debug_assertions))] pub const POW_PARAMS: (u8, u32, u32) = (15, 8, 1); /// Salt used for generating PoW. - pub const POW_SALT: &'static [u8] = &[b'r', b'a', b'd']; + pub const POW_SALT: &'static [u8] = b"rad"; /// Verify this announcement's signature. pub fn verify(&self) -> bool { @@ -589,6 +589,87 @@ impl ZeroBytes { } } +#[cfg(any(test, feature = "test"))] +#[allow(clippy::unwrap_used)] +impl qcheck::Arbitrary for Message { + fn arbitrary(g: &mut qcheck::Gen) -> Self { + use qcheck::Arbitrary; + + match g.choose(&[1, 2, 3, 4, 5, 6, 7]).unwrap() { + 1 => Announcement { + node: NodeId::arbitrary(g), + message: InventoryAnnouncement { + inventory: BoundedVec::arbitrary(g), + timestamp: Timestamp::arbitrary(g), + } + .into(), + signature: crypto::Signature::from(<[u8; 64]>::arbitrary(g)), + } + .into(), + 2 => Announcement { + node: NodeId::arbitrary(g), + message: RefsAnnouncement { + rid: RepoId::arbitrary(g), + refs: BoundedVec::arbitrary(g), + timestamp: Timestamp::arbitrary(g), + } + .into(), + signature: crypto::Signature::from(<[u8; 64]>::arbitrary(g)), + } + .into(), + 3 => { + let message = NodeAnnouncement { + version: u8::arbitrary(g), + features: u64::arbitrary(g).into(), + timestamp: Timestamp::arbitrary(g), + alias: Alias::arbitrary(g), + addresses: Arbitrary::arbitrary(g), + nonce: u64::arbitrary(g), + agent: UserAgent::arbitrary(g), + } + .into(); + let bytes: [u8; 64] = Arbitrary::arbitrary(g); + let signature = crypto::Signature::from(bytes); + + Announcement { + node: NodeId::arbitrary(g), + signature, + message, + } + .into() + } + 4 => { + let message = Info::RefsAlreadySynced { + rid: RepoId::arbitrary(g), + at: radicle::test::arbitrary::oid(), + }; + Self::Info(message) + } + 5 => Self::Subscribe(Subscribe { + filter: Filter::arbitrary(g), + since: Timestamp::arbitrary(g), + until: Timestamp::arbitrary(g), + }), + 6 => { + let mut rng = fastrand::Rng::with_seed(u64::arbitrary(g)); + + Self::Ping(Ping::new(&mut rng)) + } + 7 => Self::Pong { + zeroes: ZeroBytes::new(u16::arbitrary(g).min(Ping::MAX_PONG_ZEROES)), + }, + _ => panic!("Invalid choice for Message::arbitrary"), + } + } +} + +#[cfg(any(test, feature = "test"))] +impl qcheck::Arbitrary for ZeroBytes { + fn arbitrary(g: &mut qcheck::Gen) -> Self { + ZeroBytes::new(u16::arbitrary(g)) + } +} + #[cfg(test)] #[allow(clippy::unwrap_used)] mod tests { @@ -599,9 +680,9 @@ mod tests { use radicle::git::raw; use super::*; - use crate::prelude::*; - use crate::test::arbitrary; use crate::wire::Encode; + use localtime::LocalTime; + use radicle::test::arbitrary; #[test] fn test_ref_remote_limit() { diff --git a/crates/radicle-protocol/src/service/session.rs b/crates/radicle-protocol/src/service/session.rs index ab8b32b2..0fc9be0c 100644 --- a/crates/radicle-protocol/src/service/session.rs +++ b/crates/radicle-protocol/src/service/session.rs @@ -2,16 +2,15 @@ use std::collections::{HashSet, VecDeque}; use std::{fmt, time}; use crossbeam_channel as chan; +use radicle::node::config::Limits; +use radicle::node::{FetchResult, Severity}; +use radicle::node::{Link, Timestamp}; +pub use radicle::node::{PingState, State}; +use radicle::storage::refs::RefsAt; -use crate::node::config::Limits; -use crate::node::{FetchResult, Severity}; use crate::service::message; use crate::service::message::Message; use crate::service::{Address, LocalDuration, LocalTime, NodeId, Outbox, RepoId, Rng}; -use crate::storage::refs::RefsAt; -use crate::{Link, Timestamp}; - -pub use crate::node::{PingState, State}; /// Time after which a connection is considered stable. pub const CONNECTION_STABLE_THRESHOLD: LocalDuration = LocalDuration::from_mins(1); diff --git a/crates/radicle-protocol/src/wire.rs b/crates/radicle-protocol/src/wire.rs index c400ea0c..fb696f1d 100644 --- a/crates/radicle-protocol/src/wire.rs +++ b/crates/radicle-protocol/src/wire.rs @@ -1,12 +1,9 @@ -mod frame; -mod message; -mod protocol; -mod varint; +pub mod frame; +pub mod message; +pub mod varint; pub use frame::StreamId; pub use message::{AddressType, MessageType}; -pub use protocol::{Control, Wire, WireReader, WireSession, WireWriter}; -use radicle::node::UserAgent; use std::collections::BTreeMap; use std::convert::TryFrom; @@ -18,18 +15,21 @@ use std::{io, mem}; use byteorder::{NetworkEndian, ReadBytesExt, WriteBytesExt}; use cyphernet::addr::tor; -use crate::crypto::{PublicKey, Signature, Unverified}; -use crate::git; -use crate::git::fmt; -use crate::identity::RepoId; -use crate::node; -use crate::node::Alias; -use crate::prelude::*; +use radicle::crypto::{PublicKey, Signature, Unverified}; +use radicle::git; +use radicle::git::fmt; +use radicle::identity::RepoId; +use radicle::node; +use radicle::node::Alias; +use radicle::node::NodeId; +use radicle::node::Timestamp; +use radicle::node::UserAgent; +use radicle::storage::refs::Refs; +use radicle::storage::refs::RefsAt; +use radicle::storage::refs::SignedRefs; + +use crate::bounded::BoundedVec; use crate::service::filter; -use crate::storage::refs::Refs; -use crate::storage::refs::RefsAt; -use crate::storage::refs::SignedRefs; -use crate::Timestamp; /// The default type we use to represent sizes on the wire. /// @@ -569,9 +569,9 @@ mod tests { use qcheck; use qcheck_macros::quickcheck; - use crate::crypto::Unverified; - use crate::storage::refs::SignedRefs; - use crate::test::assert_matches; + use radicle::assert_matches; + use radicle::crypto::Unverified; + use radicle::storage::refs::SignedRefs; #[quickcheck] fn prop_u8(input: u8) { diff --git a/crates/radicle-protocol/src/wire/frame.rs b/crates/radicle-protocol/src/wire/frame.rs index 64c874c1..fcd5c30f 100644 --- a/crates/radicle-protocol/src/wire/frame.rs +++ b/crates/radicle-protocol/src/wire/frame.rs @@ -2,7 +2,9 @@ #![warn(clippy::missing_docs_in_private_items)] use std::{fmt, io}; -use crate::{wire, wire::varint, wire::varint::VarInt, wire::Message, Link, PROTOCOL_VERSION}; +use crate::service::Message; +use crate::{wire, wire::varint, wire::varint::VarInt, PROTOCOL_VERSION}; +use radicle::node::Link; /// Protocol version strings all start with the magic sequence `rad`, followed /// by a version number. diff --git a/crates/radicle-protocol/src/wire/message.rs b/crates/radicle-protocol/src/wire/message.rs index 4e8062af..bcfc2997 100644 --- a/crates/radicle-protocol/src/wire/message.rs +++ b/crates/radicle-protocol/src/wire/message.rs @@ -2,13 +2,17 @@ use std::{io, mem, net}; use byteorder::{NetworkEndian, ReadBytesExt}; use cyphernet::addr::{tor, Addr, HostName, NetAddr}; +use radicle::crypto::Signature; use radicle::git::Oid; +use radicle::identity::RepoId; use radicle::node::Address; +use radicle::node::NodeId; +use radicle::node::Timestamp; -use crate::prelude::*; +use crate::bounded::BoundedVec; +use crate::service::filter::Filter; use crate::service::message::*; use crate::wire; -use crate::wire::{Decode, Encode}; /// Message type. #[repr(u16)] @@ -67,22 +71,6 @@ impl Message { } } -impl netservices::Frame for Message { - type Error = wire::Error; - - fn unmarshall(mut reader: impl io::Read) -> Result, Self::Error> { - match Message::decode(&mut reader) { - Ok(msg) => Ok(Some(msg)), - Err(wire::Error::Io(_)) => Ok(None), - Err(err) => Err(err), - } - } - - fn marshall(&self, mut writer: impl io::Write) -> Result { - self.encode(&mut writer).map_err(wire::Error::from) - } -} - /// Address type. #[repr(u8)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -463,8 +451,8 @@ mod tests { use radicle::storage::refs::RefsAt; use crate::deserializer::Deserializer; - use crate::test::arbitrary; use crate::wire::{self, Encode}; + use radicle::test::arbitrary; #[test] fn test_refs_ann_max_size() { diff --git a/crates/radicle-protocol/src/worker.rs b/crates/radicle-protocol/src/worker.rs new file mode 100644 index 00000000..7886df93 --- /dev/null +++ b/crates/radicle-protocol/src/worker.rs @@ -0,0 +1,119 @@ +#![allow(clippy::too_many_arguments)] +pub mod fetch; + +use std::io; + +use radicle::identity::RepoId; +use radicle::node::Event; +use radicle::prelude::NodeId; +use radicle::storage::refs::RefsAt; + +// use crate::runtime::{thread, Emitter, Handle}; + +use radicle::node::events::Emitter; + +// pub use channels::{ChannelEvent, Channels, ChannelsConfig}; + +/// Error returned by fetch. +#[derive(thiserror::Error, Debug)] +pub enum FetchError { + #[error("the 'git fetch' command failed with exit code '{code}'")] + CommandFailed { code: i32 }, + #[error(transparent)] + Io(#[from] io::Error), + #[error(transparent)] + Fetch(#[from] fetch::error::Fetch), + #[error(transparent)] + Handle(#[from] fetch::error::Handle), + #[error(transparent)] + Storage(#[from] radicle::storage::Error), + #[error(transparent)] + PolicyStore(#[from] radicle::node::policy::store::Error), + #[error(transparent)] + Policy(#[from] radicle_fetch::policy::error::Policy), + #[error(transparent)] + Blocked(#[from] radicle_fetch::policy::error::Blocked), +} + +impl FetchError { + /// Check if it's a timeout error. + pub fn is_timeout(&self) -> bool { + matches!(self, FetchError::Io(e) if e.kind() == io::ErrorKind::TimedOut) + } +} + +/// Error returned by fetch responder. +#[derive(thiserror::Error, Debug)] +pub enum UploadError { + #[error("error parsing git command packet-line: {0}")] + PacketLine(io::Error), + #[error("error while performing git upload-pack: {0}")] + UploadPack(io::Error), + #[error(transparent)] + Authorization(#[from] AuthorizationError), +} + +impl UploadError { + /// Check if it's an end-of-file error. + pub fn is_eof(&self) -> bool { + matches!(self, UploadError::UploadPack(e) if e.kind() == io::ErrorKind::UnexpectedEof) + } +} + +#[derive(thiserror::Error, Debug)] +pub enum AuthorizationError { + #[error("{0} is not authorized to fetch {1}")] + Unauthorized(NodeId, RepoId), + #[error(transparent)] + PolicyStore(#[from] radicle::node::policy::store::Error), + #[error(transparent)] + Repository(#[from] radicle::storage::RepositoryError), +} + +/// Fetch job sent to worker thread. +#[derive(Debug, Clone)] +pub enum FetchRequest { + /// Client is initiating a fetch for the repository identified by + /// `rid` from the peer identified by `remote`. + Initiator { + /// Repo to fetch. + rid: RepoId, + /// Remote peer we are interacting with. + remote: NodeId, + /// If this fetch is for a particular set of `rad/sigrefs`. + refs_at: Option>, + }, + /// 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, + /// Reporter for upload-pack progress. + emitter: Emitter, + }, +} + +impl FetchRequest { + pub fn remote(&self) -> NodeId { + match self { + Self::Initiator { remote, .. } | Self::Responder { remote, .. } => *remote, + } + } +} + +/// Fetch result of an upload or fetch. +#[derive(Debug)] +pub enum FetchResult { + Initiator { + /// Repo fetched. + rid: RepoId, + /// Fetch result, including remotes fetched. + result: Result, + }, + Responder { + /// Repo requested. + rid: Option, + /// Upload result. + result: Result<(), UploadError>, + }, +} diff --git a/crates/radicle-protocol/src/worker/fetch.rs b/crates/radicle-protocol/src/worker/fetch.rs new file mode 100644 index 00000000..1ff587aa --- /dev/null +++ b/crates/radicle-protocol/src/worker/fetch.rs @@ -0,0 +1,41 @@ +pub mod error; + +use std::collections::HashSet; + +use radicle::crypto::PublicKey; +use radicle::{identity::DocAt, storage::RefUpdate}; + +#[derive(Debug, Clone)] +pub struct FetchResult { + /// The set of updated references. + pub updated: Vec, + /// The set of remote namespaces that were updated. + pub namespaces: HashSet, + /// The fetch was a full clone. + pub clone: bool, + /// Identity doc of fetched repo. + pub doc: DocAt, +} + +impl FetchResult { + pub fn new(doc: DocAt) -> Self { + Self { + updated: vec![], + namespaces: HashSet::new(), + clone: false, + doc, + } + } +} + +#[cfg(any(test, feature = "test"))] +impl qcheck::Arbitrary for FetchResult { + fn arbitrary(g: &mut qcheck::Gen) -> Self { + FetchResult { + updated: vec![], + namespaces: HashSet::arbitrary(g), + clone: bool::arbitrary(g), + doc: DocAt::arbitrary(g), + } + } +} diff --git a/crates/radicle-node/src/worker/fetch/error.rs b/crates/radicle-protocol/src/worker/fetch/error.rs similarity index 100% rename from crates/radicle-node/src/worker/fetch/error.rs rename to crates/radicle-protocol/src/worker/fetch/error.rs