mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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.
This commit is contained in:
Lorenz Leutgeb 2025-06-20 10:01:58 +02:00
parent 61c468778a
commit 1c20f64a26
36 changed files with 458 additions and 449 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
Cargo.lock generated
View File

@ -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",
]

4
crates/radicle-cli/tests/commands.rs
View File

@ -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;

2
crates/radicle-node/Cargo.toml
View File

@ -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 }

2
crates/radicle-node/src/control.rs
View File

@ -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]

22
crates/radicle-node/src/lib.rs
View File

@ -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<UserAgent> = 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};

6
crates/radicle-node/src/runtime.rs
View File

@ -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<G>(
home: Home,
config: service::Config,
config: radicle::node::Config,
listen: Vec<net::SocketAddr>,
signals: chan::Receiver<Signal>,
signer: Device<G>,

10
crates/radicle-node/src/runtime/handle.rs
View File

@ -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;

1
crates/radicle-node/src/test.rs
View File

@ -1,4 +1,3 @@
pub mod arbitrary;
pub mod gossip;
pub mod handle;
pub mod node;

142
crates/radicle-node/src/test/arbitrary.rs
View File

@ -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<T, const N: usize> Arbitrary for BoundedVec<T, N>
where
T: Arbitrary + Eq,
{
fn arbitrary(g: &mut qcheck::Gen) -> Self {
let mut v: Vec<T> = Arbitrary::arbitrary(g);
v.truncate(N);
v.try_into().expect("size within bounds")
}
}

5
crates/radicle-node/src/test/environment.rs
View File

@ -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,

4
crates/radicle-node/src/test/handle.rs
View File

@ -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 {

2
crates/radicle-node/src/test/node.rs
View File

@ -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};

9
crates/radicle-node/src/test/peer.rs
View File

@ -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<S, G> = service::Service<Database, S, G>;
@ -100,7 +103,7 @@ where
}
pub struct Config<G: crypto::signature::Signer<crypto::Signature> + 'static> {
pub config: service::Config,
pub config: radicle::node::Config,
pub local_time: LocalTime,
pub policy: SeedingPolicy,
pub signer: Device<G>,
@ -113,7 +116,7 @@ impl Default for Config<MockSigner> {
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,

7
crates/radicle-node/src/test/simulator.rs
View File

@ -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.

9
crates/radicle-node/src/tests.rs
View File

@ -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()
)
})

19
crates/radicle-node/src/tests/e2e.rs
View File

@ -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();

30
crates/radicle-protocol/src/wire/protocol.rs → crates/radicle-node/src/wire.rs
View File

@ -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::<WireSession<G>>::with_session(session, Link::Outbound)
NetTransport::<WireSession<G>>::with_session(
session,
netservices::Direction::Outbound,
)
}) {
Ok(transport) => {
self.outbound.insert(
@ -1134,7 +1142,7 @@ pub fn dial<G: Ecdh<Pk = NodeId>>(
remote_addr: NetAddr<HostName>,
remote_id: <G as EcSk>::Pk,
signer: G,
config: &service::Config,
config: &radicle::node::Config,
) -> io::Result<WireSession<G>> {
// 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 {

117
crates/radicle-node/src/worker.rs
View File

@ -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<Vec<RefsAt>>,
},
/// 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<Event>,
},
}
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<fetch::FetchResult, FetchError>,
},
Responder {
/// Repo requested.
rid: Option<RepoId>,
/// 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 {

28
crates/radicle-node/src/worker/fetch.rs
View File

@ -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<RefUpdate>,
/// The set of remote namespaces that were updated.
pub namespaces: HashSet<PublicKey>,
/// 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<ChannelsFlush>,

16
crates/radicle-protocol/Cargo.toml
View File

@ -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 }

12
crates/radicle-protocol/src/bounded.rs
View File

@ -239,3 +239,15 @@ impl<T: std::fmt::Debug, const N: usize> std::fmt::Debug for BoundedVec<T, N> {
self.v.fmt(f)
}
}
#[cfg(any(test, feature = "test"))]
impl<T, const N: usize> qcheck::Arbitrary for BoundedVec<T, N>
where
T: qcheck::Arbitrary + Eq,
{
fn arbitrary(g: &mut qcheck::Gen) -> Self {
let mut v: Vec<T> = qcheck::Arbitrary::arbitrary(g);
v.truncate(N);
v.try_into().expect("size within bounds")
}
}

8
crates/radicle-node/src/deserializer.rs → crates/radicle-protocol/src/deserializer.rs
View File

@ -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());

8
crates/radicle-protocol/src/lib.rs Normal file
View File

@ -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;

44
crates/radicle-protocol/src/service.rs
View File

@ -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<fetch::FetchResult, FetchError>,
result: Result<crate::worker::fetch::FetchResult, crate::worker::FetchError>,
) {
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,

20
crates/radicle-protocol/src/service/filter.rs
View File

@ -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<BloomFilter<RepoId>> 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() {

9
crates/radicle-protocol/src/service/gossip.rs
View File

@ -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<UserAgent> = LazyLock::new(|| {
FromStr::from_str(format!("/radicle:{}/", VERSION.version).as_str())
pub static PROTOCOL_VERSION_STRING: LazyLock<UserAgent> = 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 {

11
crates/radicle-protocol/src/service/gossip/store.rs
View File

@ -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() {

9
crates/radicle-protocol/src/service/io.rs
View File

@ -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<Io> {
pub fn queue(&mut self) -> &mut VecDeque<Io> {
&mut self.io
}
}

99
crates/radicle-protocol/src/service/message.rs
View File

@ -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() {

11
crates/radicle-protocol/src/service/session.rs
View File

@ -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);

40
crates/radicle-protocol/src/wire.rs
View File

@ -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) {

4
crates/radicle-protocol/src/wire/frame.rs
View File

@ -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.

26
crates/radicle-protocol/src/wire/message.rs
View File

@ -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<Option<Self>, 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<usize, Self::Error> {
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() {

119
crates/radicle-protocol/src/worker.rs Normal file
View File

@ -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<Vec<RefsAt>>,
},
/// 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<Event>,
},
}
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<fetch::FetchResult, FetchError>,
},
Responder {
/// Repo requested.
rid: Option<RepoId>,
/// Upload result.
result: Result<(), UploadError>,
},
}

41
crates/radicle-protocol/src/worker/fetch.rs Normal file
View File

@ -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<RefUpdate>,
/// The set of remote namespaces that were updated.
pub namespaces: HashSet<PublicKey>,
/// 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),
}
}
}

0
crates/radicle-node/src/worker/fetch/error.rs → crates/radicle-protocol/src/worker/fetch/error.rs
View File