fb45803 mega — heartwood

modified .codespellrc

@@ -1,4 +1,4 @@

 [codespell]
 skip = .git*,*.lock,.codespellrc
 skip = .git*,*.lock,.codespellrc,target,.jj
 check-hidden = true
 ignore-words-list = ser,noes
 ignore-words-list = set,noes

added .typos.toml

@@ -0,0 +1,16 @@

 [default]
 extend-ignore-re = [
     "[0-9a-f]{7}\\.\\.\\.?[0-9a-f]{7}", # Git range between two short commit IDs
     "[0-9a-f]{7}\\[\\.\\.\\]", # Shortened commit IDs as written in tests
     "did:key:z6Mk[123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz]{44}",
     "rad://z[123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz]{28}",
     "rad:z[123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz]{28}",
     "z6Mk[123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz]{44}",
 ]
 [default.extend-identifiers]
 "typ" = "typ" # We may write "typ" instead of "type". The latter is a Rust keyword.
 [type.codespell]
 check-file = false
 extend-glob = [".codespellrc"]

modified Cargo.lock

@@ -3129,6 +3129,7 @@ dependencies = [

  "crossbeam-channel",
  "cyphernet",
  "fastrand",
  "gix-packetline",
  "lexopt",
  "log",
  "mio 1.0.4",

modified Cargo.toml

@@ -31,6 +31,7 @@ dunce = "1.0.5"

 fastrand = { version = "2.0.0", default-features = false }
 git2 = { version = "0.19.0", default-features = false, features = ["vendored-libgit2"] }
 gix-hash = { version = "0.22.1", default-features = false, features = ["sha1"] }
 gix-packetline = { version = "0.21.1", default-features = false }
 human-panic = "2.0.6"
 itertools = "0.14"
 lexopt = "0.3.0"

modified build.rs

@@ -31,7 +31,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {

         // x.y.z, with efe10f95be being a unique prefix of the OID of
         // `HEAD`, and the working directory was dirty.
         // If this is a build pointing to a commit that has release tag, this
         // will just return the tag name itelf, e.g. `releases/x.y.z`.
         // will just return the tag name itself, e.g. `releases/x.y.z`.
         // If all fails, we just use `hash`, which, in the worst case is
         // still "unknown" (see above) but in most cases will just be
         // the short OID of `HEAD`.

modified crates/radicle-node/Cargo.toml

@@ -25,6 +25,7 @@ colored = { workspace = true }

 crossbeam-channel = { workspace = true }
 cyphernet = { workspace = true, features = ["dns", "ed25519", "p2p-ed25519", "noise-framework", "noise_sha2"] }
 fastrand = { workspace = true }
 gix-packetline = { workspace = true, features = ["blocking-io"] }
 lexopt = { workspace = true }
 log = { workspace = true, features = ["kv", "std"] }
 mio = { version = "1", features = ["net", "os-poll"] }

modified crates/radicle-node/src/reactor.rs

@@ -34,6 +34,12 @@ const SECONDS_IN_AN_HOUR: u64 = 60 * 60;

 /// Maximum amount of time to wait for I/O.
 const WAIT_TIMEOUT: Duration = Duration::from_secs(SECONDS_IN_AN_HOUR);
 /// Maximum duration we accept the service to spend handling events (and errors,
 /// ticking, etc.) without warning. We set this to be warned whenever the
 /// service becomes so slow that we would not be able to handle at least 100
 /// "requests" per second, i.e. `1s / 100 = 10ms`.
 const LAG_TIMEOUT: Duration = Duration::from_millis(10);
 /// A resource which can be managed by the reactor.
 pub trait EventHandler {
     /// The type of reactions which this resource may generate upon receiving

@@ -371,10 +377,9 @@ impl<H: ReactionHandler> Runtime<H> {

     fn run(mut self) {
         loop {
             let before_poll = Instant::now();
             let timeout = self
                 .timeouts
                 .next_expiring_from(before_poll)
                 .next_expiring_from(Instant::now())
                 .unwrap_or(WAIT_TIMEOUT);
             self.register_interests()

@@ -384,31 +389,38 @@ impl<H: ReactionHandler> Runtime<H> {

             let mut events = Events::with_capacity(1024);
             // Blocking
             // Block and wait for I/O events or timeout.
             let res = self.poll.poll(&mut events, Some(timeout));
             self.service.tick();
             // This instant allows us to measure the time spent by the service
             // to handle the result of polling.
             let tick = Instant::now();
             // The way this is currently used basically ignores which keys have
             // timed out. So as long as *something* timed out, we wake the service.
             let timers_fired = self.timeouts.remove_expired_by(tick);
             if timers_fired > 0 {
                 log::trace!(target: "reactor", "Timer has fired");
                 self.service.timer_reacted();
             }
             // We inform the service that time has advanced.
             self.service.tick();
             // We inform the service about errors during polling.
             if let Err(err) = res {
                 log::warn!(target: "reactor", "Failure during polling: {err}");
                 self.service.handle_error(Error::Poll(err));
             }
             let awoken = self.handle_events(tick, events);
             // We inform the service that some timers have reacted.
             // The way this is currently used basically ignores which
             // timers have expired. As long as *something* timed out,
             // we inform the service.
             let timers_fired = self.timeouts.remove_expired_by(tick);
             if timers_fired > 0 {
                 log::trace!(target: "reactor", "Timer has fired");
                 self.service.timer_reacted();
             }
             log::trace!(target: "reactor", "Duration between tick and events handled: {:?}", Instant::now().duration_since(tick));
             // Process the commands only if we awoken by the waker.
             if awoken {
             if events.is_empty() {
                 log::trace!(target: "reactor", "Woke up from poll without events");
             } else if self.handle_events(tick, events) {
                 loop {
                     match self.receiver.try_recv() {
                         Err(TryRecvError::Empty) => break,

@@ -421,6 +433,11 @@ impl<H: ReactionHandler> Runtime<H> {

                 }
             }
             let duration = Instant::now().duration_since(tick);
             if duration > LAG_TIMEOUT {
                 log::warn!(target: "reactor", "Service took {:?} to tick and handle errors and events, which exceeds the timeout of {:?}", duration, LAG_TIMEOUT);
             }
             self.handle_actions(tick);
         }
     }

modified crates/radicle-node/src/test/node.rs

@@ -7,6 +7,7 @@ use std::{

     collections::{BTreeMap, BTreeSet},
     fs, io, iter, net, process, thread, time,
     time::Duration,
     time::SystemTime,
 };
 use crossbeam_channel as chan;

@@ -107,22 +108,37 @@ impl<G: Signer<Signature> + cyphernet::Ecdh> NodeHandle<G> {

             .connect(remote.id, remote.addr.into(), ConnectOptions::default())
             .ok();
         local_events
             .iter()
             .find(|e| {
                 matches!(
                     e, Event::PeerConnected { nid } if nid == &remote.id
                 )
             })
             .unwrap();
         remote_events
             .iter()
             .find(|e| {
                 matches!(
                     e, Event::PeerConnected { nid } if nid == &self.id
                 )
             })
             .unwrap();
         // This timeout is rather arbitrary. The main point is to avoid waiting
         // indefinitely, but to also give slow CI enough time to settle.
         const TIMEOUT: Duration = Duration::from_secs(60);
         log::debug!(target: "test", "Waiting {:?} for node {} to be connected to peer {} (direction 1/2).", TIMEOUT, self.id, remote.id);
         if let Err(err) = local_events.wait(
             |event| match event {
                 Event::PeerConnected { nid } if *nid == remote.id => Some(()),
                 _ => None,
             },
             TIMEOUT,
         ) {
             panic!(
                 "Failed to connect node {} to peer {} within {:?}: {err}",
                 self.id, remote.id, TIMEOUT
             );
         }
         log::debug!(target: "test", "Waiting {:?} for node {} to be connected to peer {} (direction 2/2).", TIMEOUT, remote.id, self.id);
         if let Err(err) = remote_events.wait(
             |event| match event {
                 Event::PeerConnected { nid } if *nid == self.id => Some(()),
                 _ => None,
             },
             TIMEOUT,
         ) {
             panic!(
                 "Failed to connect node {} to peer {} within {:?}: {err}",
                 remote.id, self.id, TIMEOUT
             );
         }
         self
     }

@@ -578,6 +594,12 @@ pub fn converge<'a, G: Signer<Signature> + cyphernet::Ecdh + 'static>(

         }
     }
     // This timeout is rather arbitrary. The main point is to avoid waiting
     // indefinitely, but to also give slow CI enough time to settle.
     const TIMEOUT_DURATION: Duration = Duration::from_secs(30);
     let timeout = SystemTime::now() + TIMEOUT_DURATION;
     // Then, while there are nodes remaining to converge, check each node to see if
     // its routing table has all routes. If so, remove it from the remaining nodes.
     while !remaining.is_empty() {

@@ -595,6 +617,11 @@ pub fn converge<'a, G: Signer<Signature> + cyphernet::Ecdh + 'static>(

             true
         });
         thread::sleep(Duration::from_millis(100));
         if SystemTime::now() > timeout {
             log::warn!(target: "test", "Nodes did not converge within {:?}. Remaining nodes: {:?}", TIMEOUT_DURATION, remaining.keys());
             break;
         }
     }
     all_routes
 }

modified crates/radicle-node/src/worker.rs

@@ -141,15 +141,51 @@ impl Worker {

                 let timeout = channels.timeout();
                 let (mut stream_r, stream_w) = channels.split();
                 let header = match upload_pack::pktline::git_request(&mut stream_r) {
                     Ok(header) => header,
                     Err(e) => {
                 let mut iter = gix_packetline::blocking_io::StreamingPeekableIter::new(
                     &mut stream_r,
                     &[gix_packetline::PacketLineRef::Flush],
                     false, /* packet tracing */
                 );
                 let header = match iter.read_line() {
                     None => {
                         return FetchResult::Responder {
                             rid: None,
                             result: Err(UploadError::PacketLine(std::io::Error::new(
                                 std::io::ErrorKind::UnexpectedEof,
                                 "unexpected end of stream while reading upload-pack header",
                             ))),
                         }
                     }
                     Some(Err(e)) => {
                         return FetchResult::Responder {
                             rid: None,
                             result: Err(UploadError::PacketLine(e)),
                         }
                     }
                     Some(Ok(Err(e))) => {
                         return FetchResult::Responder {
                             rid: None,
                             result: Err(UploadError::PacketLine(std::io::Error::new(
                                 std::io::ErrorKind::InvalidData,
                                 format!("invalid upload-pack header: {e}"),
                             ))),
                         }
                     }
                     Some(Ok(Ok(header))) => header,
                 };
                 let Some(header) = upload_pack::GitRequest::from_packetline(header) else {
                     return FetchResult::Responder {
                         rid: None,
                         result: Err(UploadError::PacketLine(std::io::Error::new(
                             std::io::ErrorKind::InvalidData,
                             "failed to parse upload-pack header",
                         ))),
                     };
                 };
                 log::debug!(target: "worker", "Spawning upload-pack process for {} on stream {stream}..", header.repo);
                 if let Err(e) = self.is_authorized(remote, header.repo) {

modified crates/radicle-node/src/worker/upload_pack.rs

@@ -25,7 +25,7 @@ pub fn upload_pack<R, W>(

     remote: NodeId,
     storage: &Storage,
     emitter: &Emitter<Event>,
     header: &pktline::GitRequest,
     header: &GitRequest,
     mut recv: R,
     send: W,
     timeout: Duration,

@@ -239,119 +239,63 @@ where

     }
 }
 pub(super) mod pktline {
     use std::io;
     use std::io::Read;
     use std::str;
     use radicle::prelude::RepoId;
     pub const HEADER_LEN: usize = 4;
     /// Read and parse the `GitRequest` data from the client side.
     pub fn git_request<R>(reader: &mut R) -> io::Result<GitRequest>
     where
         R: io::Read,
     {
         let mut reader = Reader::new(reader);
         let (header, _) = reader.read_request_pktline()?;
         Ok(header)
     }
     struct Reader<'a, R> {
         stream: &'a mut R,
     }
     impl<'a, R: io::Read> Reader<'a, R> {
         /// Create a new packet-line reader.
         pub fn new(stream: &'a mut R) -> Self {
             Self { stream }
         }
         /// Parse a Git request packet-line.
         ///
         /// Example: `0032git-upload-pack /project.git\0host=myserver.com\0`
         ///
         fn read_request_pktline(&mut self) -> io::Result<(GitRequest, Vec<u8>)> {
             let mut pktline = [0u8; 1024];
             let length = self.read_pktline(&mut pktline)?;
             let Some(cmd) = GitRequest::parse(&pktline[4..length]) else {
                 return Err(io::ErrorKind::InvalidInput.into());
             };
             Ok((cmd, Vec::from(&pktline[..length])))
         }
         /// Parse a Git packet-line.
         fn read_pktline(&mut self, buf: &mut [u8]) -> io::Result<usize> {
             self.read_exact(&mut buf[..HEADER_LEN])?;
             let length = str::from_utf8(&buf[..HEADER_LEN])
                 .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?;
             let length = usize::from_str_radix(length, 16)
                 .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?;
             self.read_exact(&mut buf[HEADER_LEN..length])?;
             Ok(length)
         }
     }
     impl<R: io::Read> io::Read for Reader<'_, R> {
         fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
             self.stream.read(buf)
         }
     }
 /// The Git request packet-line for a repository.
 ///
 /// See <https://git-scm.com/docs/pack-protocol.html#_git_transport>.
 ///
 /// Example: `0032git-upload-pack /rad:z3gqcJUoA1n9HaHKufZs5FCSGazv5.git\0host=myserver.com\0`
 #[derive(Debug)]
 pub struct GitRequest {
     pub repo: RepoId,
     #[allow(dead_code)]
     pub path: String,
     #[allow(dead_code)]
     pub host: Option<(String, Option<u16>)>,
     pub extra: Vec<(String, Option<String>)>,
 }
     /// The Git request packet-line for a Heartwood repository.
     ///
     /// Example: `0032git-upload-pack /rad:z3gqcJUoA1n9HaHKufZs5FCSGazv5.git\0host=myserver.com\0`
     #[derive(Debug)]
     pub struct GitRequest {
         pub repo: RepoId,
         #[allow(dead_code)]
         pub path: String,
         #[allow(dead_code)]
         pub host: Option<(String, Option<u16>)>,
         pub extra: Vec<(String, Option<String>)>,
 impl GitRequest {
     pub(super) fn from_packetline(
         packet_line: gix_packetline::PacketLineRef<'_>,
     ) -> Option<GitRequest> {
         packet_line.as_slice().and_then(Self::parse)
     }
     impl GitRequest {
         /// Parse a Git command from a packet-line.
         fn parse(input: &[u8]) -> Option<Self> {
             let input = str::from_utf8(input).ok()?;
             let mut parts = input
                 .strip_prefix("git-upload-pack ")?
                 .split_terminator('\0');
             let path = parts.next()?.to_owned();
             let repo = path.strip_prefix('/')?.parse().ok()?;
             let host = match parts.next() {
                 None | Some("") => None,
                 Some(host) => {
                     let host = host.strip_prefix("host=")?;
                     match host.split_once(':') {
                         None => Some((host.to_owned(), None)),
                         Some((host, port)) => {
                             let port = port.parse::<u16>().ok()?;
                             Some((host.to_owned(), Some(port)))
                         }
     /// Parse a Git command from a packet-line.
     fn parse(input: &[u8]) -> Option<Self> {
         let input = std::str::from_utf8(input).ok()?;
         let mut parts = input
             .strip_prefix("git-upload-pack ")?
             .split_terminator('\0');
         let path = parts.next()?.to_owned();
         let repo = path.strip_prefix('/')?.parse().ok()?;
         let host = match parts.next() {
             None | Some("") => None,
             Some(host) => {
                 let host = host.strip_prefix("host=")?;
                 match host.split_once(':') {
                     None => Some((host.to_owned(), None)),
                     Some((host, port)) => {
                         let port = port.parse::<u16>().ok()?;
                         Some((host.to_owned(), Some(port)))
                     }
                 }
             };
             let extra = parts
                 .skip_while(|part| part.is_empty())
                 .map(|part| match part.split_once('=') {
                     None => (part.to_owned(), None),
                     Some((k, v)) => (k.to_owned(), Some(v.to_owned())),
                 })
                 .collect();
             Some(Self {
                 repo,
                 path,
                 host,
                 extra,
             }
         };
         let extra = parts
             .skip_while(|part| part.is_empty())
             .map(|part| match part.split_once('=') {
                 None => (part.to_owned(), None),
                 Some((k, v)) => (k.to_owned(), Some(v.to_owned())),
             })
         }
             .collect();
         Some(Self {
             repo,
             path,
             host,
             extra,
         })
     }
 }

modified crates/radicle-protocol/Cargo.toml

@@ -9,7 +9,7 @@ edition.workspace = true

 rust-version.workspace = true
 [features]
 i2p = ["cyphernet/i2p"]
 i2p = ["cypheraddr/i2p"]
 test = ["radicle/test", "radicle-crypto/test", "radicle-crypto/cyphernet", "qcheck"]
 tor = ["cypheraddr/tor"]

@@ -37,4 +37,4 @@ paste = "1.0.15"

 qcheck = { workspace = true }
 qcheck-macros = { workspace = true }
 radicle = { workspace = true, features = ["test"] }
 radicle-crypto = { workspace = true, features = ["test", "cyphernet"] }
 \ No newline at end of file
 radicle-crypto = { workspace = true, features = ["test", "cyphernet"] }

modified crates/radicle-protocol/src/service.rs

@@ -406,7 +406,7 @@ where

     G: crypto::signature::Signer<crypto::Signature>,
 {
     pub fn new(
         config: Config,
         mut config: Config,
         db: Stores<D>,
         storage: S,
         policies: policy::Config<Write>,

@@ -429,6 +429,10 @@ where

                 .with_max_capacity(fetcher::MaxQueueSize::default());
             FetcherService::new(config)
         };
         // For backwards compatibility, ensure that we are announcer of our own Node ID.
         config.announcers.insert(*signer.public_key());
         Self {
             config,
             storage,

@@ -1550,12 +1554,15 @@ where

                 // from a new repository being initialized.
                 self.seed_discovered(message.rid, *announcer, message.timestamp);
                 // Update sync status of announcer for this repo.
                 if let Some(refs) = refs.iter().find(|r| &r.remote == self.nid()) {
                 // Update sync status of announcers for this repo.
                 for refs in refs
                     .iter()
                     .filter(|r| self.config.announcers.contains(&r.remote))
                 {
                     debug!(
                         target: "service",
                         "Refs announcement of {announcer} for {} contains our own remote at {} (t={})",
                         message.rid, refs.at, message.timestamp
                         "Refs announcement of {announcer} for {} contains announcer {} (t={})",
                         message.rid, refs, message.timestamp
                     );
                     match self.db.seeds_mut().synced(
                         &message.rid,

modified crates/radicle-protocol/src/service/limiter.rs

@@ -156,7 +156,7 @@ mod test {

     }
     #[test]
     fn test_limitter_refill() {
     fn test_limiter_refill() {
         let mut r = RateLimiter::default();
         let t = (3, 0.2); // Three tokens burst. One token every 5 seconds.
         let a = HostName::Dns(String::from("seed.radicle.example.com"));

@@ -188,7 +188,7 @@ mod test {

     #[test]
     #[rustfmt::skip]
     fn test_limitter_multi() {
     fn test_limiter_multi() {
         let t = (1, 1.0); // One token per second. One token burst.
         let n = arbitrary::gen::<NodeId>(1);
         let n = Some(&n);

@@ -208,7 +208,7 @@ mod test {

     #[test]
     #[rustfmt::skip]
     fn test_limitter_different_rates() {
     fn test_limiter_different_rates() {
         let t1 = (1, 1.0); // One token per second. One token burst.
         let t2 = (2, 2.0); // Two tokens per second. Two token burst.
         let n = arbitrary::gen::<NodeId>(1);

modified crates/radicle-protocol/src/wire.rs

@@ -16,7 +16,7 @@ use std::string::FromUtf8Error;

 use bytes::{Buf, BufMut};
 #[cfg(feature = "i2p")]
 use cyphernet::addr::i2p;
 use cypheraddr::i2p;
 #[cfg(feature = "tor")]
 use cypheraddr::tor;

modified crates/radicle-protocol/src/wire/message.rs

@@ -3,7 +3,7 @@ use std::{mem, net};

 use bytes::Buf;
 use bytes::BufMut;
 #[cfg(feature = "i2p")]
 use cyphernet::addr::i2p;
 use cypheraddr::i2p;
 #[cfg(feature = "tor")]
 use cypheraddr::tor;
 use cypheraddr::{HostName, NetAddr};

modified crates/radicle-schemars/src/main.rs

@@ -81,7 +81,7 @@ fn print_schema() -> io::Result<()> {

             #[derive(JsonSchema)]
             #[schemars(untagged)]
             #[allow(dead_code)]
             #[allow(dead_code, clippy::large_enum_variant)]
             enum CommandResult {
                 Nid(radicle::node::NodeId),
                 Config(Box<radicle::node::Config>),

modified crates/radicle-ssh/src/agent/client.rs

@@ -52,20 +52,7 @@ pub enum Error {

 impl Error {
     pub fn is_not_running(&self) -> bool {
         match self {
             Self::EnvVar { .. } | Self::BadAuthSock { .. } => true,
             #[cfg(windows)]
             Self::Connect { source, .. }
                 if source.kind() == std::io::ErrorKind::ConnectionRefused =>
             {
                 // On Windows, a named pipe might be used, and if no
                 // agent is running, we might get a "connection refused"
                 // error, even though the `SSH_AUTH_SOCK` environment
                 // variable is set and the named pipe exists.
                 true
             }
             _ => false,
         }
         matches!(self, Self::EnvVar { .. } | Self::BadAuthSock { .. })
     }
 }

modified crates/radicle-ssh/src/encoding.rs

@@ -56,7 +56,7 @@ pub trait Encoding {

     /// May panic if the argument is greater than [`u32::MAX`].
     /// This is a convenience method, to spare callers casting or converting
     /// [`usize`] to [`u32`]. If callers end up in a situation where they
     /// need to push a 32-bit unisgned integer, but the value they would
     /// need to push a 32-bit unsigned integer, but the value they would
     /// like to push does not fit 32 bits, then the implementation will not
     /// comply with the SSH format anyway.
     fn extend_usize(&mut self, u: usize) {

modified crates/radicle/src/node/config.rs

@@ -473,6 +473,8 @@ pub struct Config {

     /// the environment variable `RAD_PASSPHRASE`.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub secret: Option<std::path::PathBuf>,
     #[serde(default, skip_serializing_if = "HashSet::is_empty")]
     pub announcers: HashSet<super::NodeId>,
 }
 impl Config {

@@ -503,6 +505,7 @@ impl Config {

             seeding_policy: DefaultSeedingPolicy::default(),
             extra: json::Map::default(),
             secret: None,
             announcers: HashSet::new(),
         }
     }

modified flake.nix

@@ -235,6 +235,13 @@

               hooks =
                 {
                   alejandra.enable = true;
                   typos = {
                     enable = true;
                     settings = {
                       verbose = true;
                       write = true;
                     };
                   };
                   codespell = {
                     enable = true;
                     entry = "${lib.getExe pkgs.codespell} -w";