rad:z3gqcJUoA1n9HaHKufZs5FCSGazv5 heartwood72c715016f4b59a5677a4c97e4ce2c58bf9114ec
{
"request": "trigger",
"version": 1,
"event_type": "patch",
"repository": {
"id": "rad:z3gqcJUoA1n9HaHKufZs5FCSGazv5",
"name": "heartwood",
"description": "Radicle Heartwood Protocol & Stack",
"private": false,
"default_branch": "master",
"delegates": [
"did:key:z6MksFqXN3Yhqk8pTJdUGLwATkRfQvwZXPqR2qMEhbS9wzpT",
"did:key:z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM",
"did:key:z6MkkPvBfjP4bQmco5Dm7UGsX2ruDBieEHi8n9DVJWX5sTEz"
]
},
"action": "Created",
"patch": {
"id": "331186490f62ad8e54d357eba48b0578ecb6aee4",
"author": {
"id": "did:key:z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM",
"alias": null
},
"title": "Improve Uploading Experience",
"state": {
"status": "merged",
"conflicts": []
},
"before": "cadd996a763b4490ef177524d21e5059820ff833",
"after": "72c715016f4b59a5677a4c97e4ce2c58bf9114ec",
"commits": [
"72c715016f4b59a5677a4c97e4ce2c58bf9114ec",
"f038b7ee5caa5c776d022c9a76f4c688aabff898",
"344fe3a90121c214c3fd83ca44ddce56c15bce42",
"027bfbef67fd2af6ec1ae41a07b1b245c525d943",
"c8a24d558416661fc3b39e9dffd323b71a23f224",
"89627a8b6a58cad55267296c666d7c4b246e6c53",
"f0648c2a184732134e38f8ce1eeae5f6b2f5a04d",
"fd4c4cd0af8836213a0e846e674fb6554c123e96"
],
"target": "d285eccb23f3986bfa88fc1af033e6dcb0dc3a65",
"labels": [],
"assignees": [],
"revisions": [
{
"id": "331186490f62ad8e54d357eba48b0578ecb6aee4",
"author": {
"id": "did:key:z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM",
"alias": null
},
"description": "The main goal of the patch is to improve the upload experience of repositories\nthat are larger in size, e.g. 300MB and 3GB.\n\nThis is achieved by a series of changes, outlined in each commit, and summarised\nhere:\n\n1. Dynamic timeouts for fetching by passing the timeout through the use\n `Io::Fetch` and `Control::Open`. This removes the `DEFAULT_CHANNEL_TIMEOUT`\n and we now default to `FETCH_TIMEOUT` when it's not specified.\n2. The `Emitter` is moved to the `radicle` crate for re-use in `radicle-fetch`\n and `radicle-node`.\n3. Introduce events for the `upload-pack` process.\n4. Change the `call` method for the socket stream to use `crossbeam-channel` so\n that events can be subscribed to indefinitely but received with a timeout.\n5. The default value of `FETCH_TIMEOUT` is increased to 1 hour to allow the\n larger repositories to finish announcing and being cloned by seeds when\n initialised.\n6. Improve `rad init` to report the `upload-pack` events.",
"base": "c607619683e859fd715a23377c03fb08dc8090f4",
"oid": "29a14ec6076340916a57733de6402be365a5c9c5",
"timestamp": 1713969930
},
{
"id": "50cfa50d09770e78e97c459fa8ebe9d9f69d5f28",
"author": {
"id": "did:key:z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM",
"alias": null
},
"description": "Changes:\n- remove timeout from control frame",
"base": "abf89438ea3e3b61dda68677761ba43772eac27d",
"oid": "85aeb2e0052ee25d565265665f0d86eafb15cd7e",
"timestamp": 1714574622
},
{
"id": "a0a8da7471550785f4c9d9902f971934bcfc7099",
"author": {
"id": "did:key:z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM",
"alias": null
},
"description": "Changes:\n- Use `LineIter` for streaming semantics\n- Improve the `Progress` type to be an `enum`\n- Minor clean up from other review feedback",
"base": "6dd52c94fb1472e251110e439fe656aaed74019f",
"oid": "748bdf74bd1a97b23fa8c6b0146035172785cf12",
"timestamp": 1714752324
},
{
"id": "46914356db4dc8f5a3c6a7b73f25e51bc3b33f69",
"author": {
"id": "did:key:z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM",
"alias": null
},
"description": "Changes:\n- Simplify error reporting\n- Remove logging of RemoteProgress\n- Simplify UploadPack event enum",
"base": "6ab3bfcba0577fabdcb84498441c6605391290f4",
"oid": "62d3d045ec954306cb83ff4b41fd998819918e48",
"timestamp": 1715090357
},
{
"id": "556c2fe9eb8beae1ec0d928d7c29d794bdce0f84",
"author": {
"id": "did:key:z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM",
"alias": null
},
"description": "Changes:\n- remove unused upload pack event errors",
"base": "25c6660a59c621de1ec406e4f9b2ccfbd21e1b90",
"oid": "55e905613ddf81fb04c69ab6fa6f4fd99622d789",
"timestamp": 1715700841
},
{
"id": "bd112dd67b24571a668f60199b1ec6de360fd783",
"author": {
"id": "did:key:z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM",
"alias": null
},
"description": "Changes:\n- rebased\n- fixed documentation",
"base": "cadd996a763b4490ef177524d21e5059820ff833",
"oid": "72c715016f4b59a5677a4c97e4ce2c58bf9114ec",
"timestamp": 1715764091
}
]
}
}
{
"response": "triggered",
"run_id": {
"id": "2778bad7-777f-4219-9132-0d9b61d6468a"
},
"info_url": "https://cci.rad.levitte.org//2778bad7-777f-4219-9132-0d9b61d6468a.html"
}
Started at: 2026-09-07 20:19:32.778403+02:00
Commands:
$ rad clone rad:z3gqcJUoA1n9HaHKufZs5FCSGazv5 .
✓ Creating checkout in ./...
✓ Remote cloudhead@z6MksFqXN3Yhqk8pTJdUGLwATkRfQvwZXPqR2qMEhbS9wzpT added
✓ Remote-tracking branch cloudhead@z6MksFqXN3Yhqk8pTJdUGLwATkRfQvwZXPqR2qMEhbS9wzpT/master created for z6MksFqXN3Yhqk8pTJdUGLwATkRfQvwZXPqR2qMEhbS9wzpT
✓ Remote fintohaps@z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM added
✓ Remote-tracking branch fintohaps@z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM/master created for z6MkireRatUThvd3qzfKht1S44wpm4FEWSSa4PRMTSQZ3voM
✓ Remote lorenz@z6MkkPvBfjP4bQmco5Dm7UGsX2ruDBieEHi8n9DVJWX5sTEz added
✓ Remote-tracking branch lorenz@z6MkkPvBfjP4bQmco5Dm7UGsX2ruDBieEHi8n9DVJWX5sTEz/master created for z6MkkPvBfjP4bQmco5Dm7UGsX2ruDBieEHi8n9DVJWX5sTEz
✓ Repository successfully cloned under /opt/radcis/ci.rad.levitte.org/cci/state/2778bad7-777f-4219-9132-0d9b61d6468a/w/
╭────────────────────────────────────╮
│ heartwood │
│ Radicle Heartwood Protocol & Stack │
│ 189 issues · 38 patches │
╰────────────────────────────────────╯
Run `cd ./.` to go to the repository directory.
Exit code: 0
$ rad patch checkout 331186490f62ad8e54d357eba48b0578ecb6aee4
✓ Switched to branch patch/3311864 at revision bd112dd
✓ Branch patch/3311864 setup to track rad/patches/331186490f62ad8e54d357eba48b0578ecb6aee4
Exit code: 0
$ git config advice.detachedHead false
Exit code: 0
$ git checkout 72c715016f4b59a5677a4c97e4ce2c58bf9114ec
HEAD is now at 72c71501 fetch: add trace logging to fetch protocol
Exit code: 0
$ rad patch show 331186490f62ad8e54d357eba48b0578ecb6aee4 -p
╭──────────────────────────────────────────────────────────────────────────────────╮
│ Title Improve Uploading Experience │
│ Patch 331186490f62ad8e54d357eba48b0578ecb6aee4 │
│ Author fintohaps z6Mkire…SQZ3voM │
│ Head 72c715016f4b59a5677a4c97e4ce2c58bf9114ec │
│ Base cadd996a763b4490ef177524d21e5059820ff833 │
│ Target master │
│ Branches patch/3311864 │
│ Commits ahead 0, behind 1352 │
│ Status merged │
│ │
│ The main goal of the patch is to improve the upload experience of repositories │
│ that are larger in size, e.g. 300MB and 3GB. │
│ │
│ This is achieved by a series of changes, outlined in each commit, and summarised │
│ here: │
│ │
│ 1. Dynamic timeouts for fetching by passing the timeout through the use │
│ `Io::Fetch` and `Control::Open`. This removes the `DEFAULT_CHANNEL_TIMEOUT` │
│ and we now default to `FETCH_TIMEOUT` when it's not specified. │
│ 2. The `Emitter` is moved to the `radicle` crate for re-use in `radicle-fetch` │
│ and `radicle-node`. │
│ 3. Introduce events for the `upload-pack` process. │
│ 4. Change the `call` method for the socket stream to use `crossbeam-channel` so │
│ that events can be subscribed to indefinitely but received with a timeout. │
│ 5. The default value of `FETCH_TIMEOUT` is increased to 1 hour to allow the │
│ larger repositories to finish announcing and being cloned by seeds when │
│ initialised. │
│ 6. Improve `rad init` to report the `upload-pack` events. │
├──────────────────────────────────────────────────────────────────────────────────┤
│ 72c7150 fetch: add trace logging to fetch protocol │
│ f038b7e cli: watch for upload-pack events in `rad init` │
│ 344fe3a node: increase `FETCH_TIMEOUT` │
│ 027bfbe node: stream timeout semantics for `Node::call` │
│ c8a24d5 node: report upload-pack progress │
│ 89627a8 radicle: add upload-pack events │
│ f0648c2 radicle: move Emitter to events module │
│ fd4c4cd node: dynamic fetch timeout via channels │
├──────────────────────────────────────────────────────────────────────────────────┤
│ ● Revision 3311864 @ c607619..29a14ec by fintohaps z6Mkire…SQZ3voM 2 years ago │
│ ↑ Revision 50cfa50 @ abf8943..85aeb2e by fintohaps z6Mkire…SQZ3voM 2 years ago │
│ ↑ Revision a0a8da7 @ 6dd52c9..748bdf7 by fintohaps z6Mkire…SQZ3voM 2 years ago │
│ ↑ Revision 4691435 @ 6ab3bfc..62d3d04 by fintohaps z6Mkire…SQZ3voM 2 years ago │
│ ↑ Revision 556c2fe @ 25c6660..55e9056 by fintohaps z6Mkire…SQZ3voM 2 years ago │
│ ↑ Revision bd112dd @ cadd996..72c7150 by fintohaps z6Mkire…SQZ3voM 2 years ago │
│ └─ ✓ merged by cloudhead z6MksFq…bS9wzpT │
╰──────────────────────────────────────────────────────────────────────────────────╯
commit 72c715016f4b59a5677a4c97e4ce2c58bf9114ec
Author: Fintan Halpenny <fintan.halpenny@gmail.com>
Date: Wed Apr 24 15:26:54 2024 +0100
fetch: add trace logging to fetch protocol
Log the following at the trace level:
- Receiving the pack along with the amount of time it took
- The closing of the connection
Also organising the imports, while in the neighbourhood.
Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett
diff --git a/radicle-fetch/src/transport/fetch.rs b/radicle-fetch/src/transport/fetch.rs
index 5dfc1535b..f56d1dadb 100644
--- a/radicle-fetch/src/transport/fetch.rs
+++ b/radicle-fetch/src/transport/fetch.rs
@@ -1,23 +1,21 @@
-use std::{
- borrow::Cow,
- io::{self, BufRead},
- path::PathBuf,
- sync::{atomic::AtomicBool, Arc},
-};
+use std::borrow::Cow;
+use std::io;
+use std::io::BufRead;
+use std::path::PathBuf;
+use std::sync::{atomic::AtomicBool, Arc};
use gix_features::progress::NestedProgress;
use gix_pack as pack;
-use gix_protocol::{
- fetch::{self, Delegate, DelegateBlocking},
- handshake::{self, Ref},
- ls_refs, FetchConnection,
-};
-use gix_transport::{
- bstr::BString,
- client,
- client::{ExtendedBufRead, MessageKind},
- Protocol,
-};
+use gix_protocol::fetch;
+use gix_protocol::fetch::{Delegate, DelegateBlocking};
+use gix_protocol::handshake;
+use gix_protocol::handshake::Ref;
+use gix_protocol::ls_refs;
+use gix_protocol::FetchConnection;
+use gix_transport::bstr::BString;
+use gix_transport::client;
+use gix_transport::client::{ExtendedBufRead, MessageKind};
+use gix_transport::Protocol;
use super::{agent_name, indicate_end_of_interaction, Connection, WantsHaves};
@@ -253,6 +251,7 @@ where
if !sideband_all {
setup_remote_progress(progress, &mut reader);
}
+ let timer = std::time::Instant::now();
// TODO: remove delegate in favor of functional style to fix progress-hack,
// needed as it needs `'static`. As the top-level seems to pass `Discard`,
// there should be no repercussions right now.
@@ -262,6 +261,7 @@ where
&[],
&response,
)?;
+ log::trace!(target: "fetch", "Received pack ({}ms)", timer.elapsed().as_millis());
assert_eq!(
reader.stopped_at(),
None,
@@ -286,6 +286,7 @@ where
if matches!(protocol, Protocol::V2)
&& matches!(conn.mode, FetchConnection::TerminateOnSuccessfulCompletion)
{
+ log::trace!(target: "fetch", "Indicating end of interaction");
indicate_end_of_interaction(&mut conn)?;
}
commit f038b7ee5caa5c776d022c9a76f4c688aabff898
Author: Fintan Halpenny <fintan.halpenny@gmail.com>
Date: Wed Apr 24 15:20:27 2024 +0100
cli: watch for upload-pack events in `rad init`
To improve the user experience of `rad init`, the events listener waits for
upload-pack messages indefinitely. The spinner reports back the upload progress
over time.
This process can be cancelled via Ctrl+C without affecting the announcement,
since that is happening in the background.
To also improve the experience, the remote NIDs that have successfully synced
are reported to the terminal.
Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett
diff --git a/radicle-cli/examples/rad-init-sync-preferred.md b/radicle-cli/examples/rad-init-sync-preferred.md
index c0dbb309e..7698ccc54 100644
--- a/radicle-cli/examples/rad-init-sync-preferred.md
+++ b/radicle-cli/examples/rad-init-sync-preferred.md
@@ -10,6 +10,7 @@ Initializing public radicle 👾 repository in [..]
Your Repository ID (RID) is rad:z3Rry7rpdWuGpfjPYGzdJKQADsoNW.
You can show it any time by running `rad .` from this directory.
+✓ Repository successfully synced to z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi
✓ Repository successfully synced to 1 node(s).
Your repository has been synced to the network and is now discoverable by peers.
diff --git a/radicle-cli/src/commands/init.rs b/radicle-cli/src/commands/init.rs
index af955ead5..f6b8cc662 100644
--- a/radicle-cli/src/commands/init.rs
+++ b/radicle-cli/src/commands/init.rs
@@ -2,10 +2,10 @@
#![allow(clippy::collapsible_else_if)]
use std::collections::HashSet;
use std::convert::TryFrom;
+use std::env;
use std::ffi::OsString;
use std::path::PathBuf;
use std::str::FromStr;
-use std::{env, time};
use anyhow::{anyhow, bail, Context as _};
use serde_json as json;
@@ -14,8 +14,9 @@ use radicle::crypto::{ssh, Verified};
use radicle::explorer::ExplorerUrl;
use radicle::git::RefString;
use radicle::identity::{RepoId, Visibility};
+use radicle::node::events::UploadPack;
use radicle::node::policy::Scope;
-use radicle::node::{Event, Handle, NodeId};
+use radicle::node::{Event, Handle, NodeId, DEFAULT_SUBSCRIBE_TIMEOUT};
use radicle::prelude::Doc;
use radicle::{profile, Node};
@@ -364,7 +365,9 @@ fn sync(
return Ok(SyncResult::NodeStopped);
}
let mut spinner = term::spinner("Updating inventory..");
- let events = node.subscribe(time::Duration::from_secs(3))?;
+ // N.b. indefinitely subscribe to events and set a lower timeout on events
+ // below.
+ let events = node.subscribe(DEFAULT_SUBSCRIBE_TIMEOUT)?;
let sessions = node.sessions()?;
node.update_inventory(rid)?;
@@ -392,17 +395,40 @@ fn sync(
spinner.message("Syncing..");
let mut replicas = HashSet::new();
+
for e in events {
match e {
Ok(Event::RefsSynced {
remote, rid: rid_, ..
}) if rid == rid_ => {
+ term::success!("Repository successfully synced to {remote}");
replicas.insert(remote);
// If we manage to replicate to one of our preferred seeds, we can stop waiting.
if config.preferred_seeds.iter().any(|s| s.id == remote) {
break;
}
}
+ Ok(Event::UploadPack(UploadPack::Write {
+ rid: rid_,
+ remote,
+ progress,
+ })) if rid == rid_ => {
+ spinner.message(format!("Uploading {rid} to {remote} {progress}"));
+ }
+ Ok(Event::UploadPack(UploadPack::Done {
+ rid: rid_,
+ remote,
+ status,
+ })) if rid == rid_ => {
+ spinner.message(format!("Upload done for {rid} to {remote}: {status}"));
+ }
+ Ok(Event::UploadPack(UploadPack::Error {
+ rid: rid_,
+ remote,
+ err,
+ })) if rid == rid_ => {
+ spinner.message(format!("Upload error for {rid} to {remote}: {err}"));
+ }
Ok(_) => {
// Some other irrelevant event received.
}
commit 344fe3a90121c214c3fd83ca44ddce56c15bce42
Author: Fintan Halpenny <fintan.halpenny@gmail.com>
Date: Wed Apr 24 15:11:22 2024 +0100
node: increase `FETCH_TIMEOUT`
After empirically testing HardenedBSD[[0]], which has a total blob size of
42GiB, and its largest checkout is 1.26GiB (using `git-sizer`[[1]]), it was
found that letting the background process fetch the repository with a timeout of
1 hour is enough (a very generous estimate).
This deemed sufficient, since cases smaller than this will report back that they
are done before this time. For cases larger than this, more research will be
needed.
[0]: https://git.hardenedbsd.org/hardenedbsd/HardenedBSD
[1]: https://github.com/github/git-sizer
Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett
diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs
index 5ca26b25c..9f2109c23 100644
--- a/radicle-node/src/service.rs
+++ b/radicle-node/src/service.rs
@@ -97,8 +97,8 @@ pub const MIN_RECONNECTION_DELTA: LocalDuration = LocalDuration::from_secs(3);
pub const MAX_RECONNECTION_DELTA: LocalDuration = LocalDuration::from_mins(60);
/// Connection retry delta used for ephemeral peers that failed to connect previously.
pub const CONNECTION_RETRY_DELTA: LocalDuration = LocalDuration::from_mins(10);
-/// How long to wait for a fetch to stall before aborting.
-pub const FETCH_TIMEOUT: time::Duration = time::Duration::from_secs(9);
+/// How long to wait for a fetch to stall before aborting, default is 1 hour.
+pub const FETCH_TIMEOUT: time::Duration = time::Duration::from_secs(3600);
/// Maximum external address limit imposed by message size limits.
pub use message::ADDRESS_LIMIT;
commit 027bfbef67fd2af6ec1ae41a07b1b245c525d943
Author: Fintan Halpenny <fintan.halpenny@gmail.com>
Date: Tue Apr 23 18:11:12 2024 +0100
node: stream timeout semantics for `Node::call`
The timeout for a `UnixStream` will interrupt a `read` even when more data is
available. The preference would be that the timeout applies to each read, where
the timeout is reset again once a read is successful. This better supports
streaming semantics expected for the `Handle::subscribe` method.
To achieve this, the `LineIter` type holds the `UnixStream` and timeout
`Duration`. The `Iterator` implementation can then use `read_line` to progress
the stream one line at a time, and the timeout only applies to each read.
Co-authored-by: Alexis Sellier <alexis@radicle.xyz>
Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett
diff --git a/radicle-cli/src/commands/node/events.rs b/radicle-cli/src/commands/node/events.rs
index 7e822c644..a90ef920f 100644
--- a/radicle-cli/src/commands/node/events.rs
+++ b/radicle-cli/src/commands/node/events.rs
@@ -1,10 +1,13 @@
use std::time;
-use radicle::node::Handle;
+use radicle::node::{Event, Handle};
-pub fn run(node: impl Handle, count: usize, timeout: time::Duration) -> anyhow::Result<()> {
+pub fn run<H>(node: H, count: usize, timeout: time::Duration) -> anyhow::Result<()>
+where
+ H: Handle<Event = Result<Event, <H as Handle>::Error>>,
+{
let events = node.subscribe(timeout)?;
- for (i, event) in events.enumerate() {
+ for (i, event) in events.into_iter().enumerate() {
let event = event?;
let obj = serde_json::to_string(&event)?;
diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs
index e491e519b..23d087fc8 100644
--- a/radicle-node/src/control.rs
+++ b/radicle-node/src/control.rs
@@ -30,12 +30,12 @@ pub enum Error {
}
/// Listen for commands on the control socket, and process them.
-pub fn listen<H: Handle<Error = runtime::HandleError> + 'static>(
- listener: UnixListener,
- handle: H,
-) -> Result<(), Error>
+pub fn listen<E, H>(listener: UnixListener, handle: H) -> Result<(), Error>
where
+ H: Handle<Error = runtime::HandleError> + 'static,
H::Sessions: serde::Serialize,
+ CommandResult<E>: From<H::Event>,
+ E: serde::Serialize,
{
log::debug!(target: "control", "Control thread listening on socket..");
let nid = handle.nid()?;
@@ -74,12 +74,12 @@ enum CommandError {
Io(#[from] io::Error),
}
-fn command<H: Handle<Error = runtime::HandleError> + 'static>(
- stream: &UnixStream,
- mut handle: H,
-) -> Result<(), CommandError>
+fn command<E, H>(stream: &UnixStream, mut handle: H) -> Result<(), CommandError>
where
+ H: Handle<Error = runtime::HandleError> + 'static,
H::Sessions: serde::Serialize,
+ CommandResult<E>: From<H::Event>,
+ E: serde::Serialize,
{
let mut reader = BufReader::new(stream);
let mut writer = LineWriter::new(stream);
@@ -185,8 +185,7 @@ where
Command::Subscribe => match handle.subscribe(MAX_TIMEOUT) {
Ok(events) => {
for e in events {
- let event = e?;
- CommandResult::Okay(event).to_writer(&mut writer)?;
+ CommandResult::from(e).to_writer(&mut writer)?;
}
}
Err(e) => return Err(CommandError::Runtime(e)),
diff --git a/radicle-node/src/runtime/handle.rs b/radicle-node/src/runtime/handle.rs
index 160541292..ac329979b 100644
--- a/radicle-node/src/runtime/handle.rs
+++ b/radicle-node/src/runtime/handle.rs
@@ -124,6 +124,8 @@ impl Handle {
impl radicle::node::Handle for Handle {
type Sessions = Vec<radicle::node::Session>;
+ type Events = Events;
+ type Event = Event;
type Error = Error;
fn nid(&self) -> Result<NodeId, Self::Error> {
@@ -263,11 +265,8 @@ impl radicle::node::Handle for Handle {
receiver.recv().map_err(Error::from)
}
- fn subscribe(
- &self,
- _timeout: time::Duration,
- ) -> Result<Box<dyn Iterator<Item = Result<Event, Error>>>, Error> {
- Ok(Box::new(self.events().into_iter().map(Ok)))
+ fn subscribe(&self, _timeout: time::Duration) -> Result<Self::Events, Self::Error> {
+ Ok(self.events())
}
fn sessions(&self) -> Result<Self::Sessions, Error> {
diff --git a/radicle-node/src/test/handle.rs b/radicle-node/src/test/handle.rs
index 87de45ee7..52efaa596 100644
--- a/radicle-node/src/test/handle.rs
+++ b/radicle-node/src/test/handle.rs
@@ -22,6 +22,8 @@ pub struct Handle {
impl radicle::node::Handle for Handle {
type Error = HandleError;
type Sessions = Vec<radicle::node::Session>;
+ type Events = Vec<Self::Event>;
+ type Event = Result<Event, Self::Error>;
fn nid(&self) -> Result<NodeId, Self::Error> {
Ok(NodeId::from_str("z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK").unwrap())
@@ -81,11 +83,8 @@ impl radicle::node::Handle for Handle {
Ok(self.following.lock().unwrap().insert(id))
}
- fn subscribe(
- &self,
- _timeout: time::Duration,
- ) -> Result<Box<dyn Iterator<Item = Result<Event, Self::Error>>>, Self::Error> {
- Ok(Box::new(std::iter::empty()))
+ fn subscribe(&self, _timeout: time::Duration) -> Result<Self::Events, Self::Error> {
+ Ok(vec![])
}
fn unfollow(&mut self, id: NodeId) -> Result<bool, Self::Error> {
diff --git a/radicle/src/node.rs b/radicle/src/node.rs
index 9874b27b1..97364267c 100644
--- a/radicle/src/node.rs
+++ b/radicle/src/node.rs
@@ -15,6 +15,7 @@ pub mod timestamp;
use std::collections::{BTreeSet, HashMap, HashSet, VecDeque};
use std::io::{BufRead, BufReader};
+use std::marker::PhantomData;
use std::ops::{ControlFlow, Deref};
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
@@ -50,6 +51,9 @@ pub const DEFAULT_SOCKET_NAME: &str = "control.sock";
pub const DEFAULT_PORT: u16 = 8776;
/// Default timeout when waiting for the node to respond with data.
pub const DEFAULT_TIMEOUT: time::Duration = time::Duration::from_secs(30);
+/// Default timeout when waiting for an event to be received on the
+/// [`Handle::subscribe`] channel.
+pub const DEFAULT_SUBSCRIBE_TIMEOUT: time::Duration = time::Duration::from_secs(5);
/// Maximum length in bytes of a node alias.
pub const MAX_ALIAS_LENGTH: usize = 32;
/// Penalty threshold at which point we avoid connecting to this node.
@@ -338,6 +342,26 @@ pub enum CommandResult<T> {
},
}
+impl<T, E> From<Result<T, E>> for CommandResult<T>
+where
+ E: std::error::Error,
+{
+ fn from(result: Result<T, E>) -> Self {
+ match result {
+ Ok(t) => Self::Okay(t),
+ Err(e) => Self::Error {
+ reason: e.to_string(),
+ },
+ }
+ }
+}
+
+impl From<Event> for CommandResult<Event> {
+ fn from(event: Event) -> Self {
+ Self::Okay(event)
+ }
+}
+
/// A success response.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Success {
@@ -856,6 +880,8 @@ pub enum ConnectResult {
pub trait Handle: Clone + Sync + Send {
/// The peer sessions type.
type Sessions;
+ type Events: IntoIterator<Item = Self::Event>;
+ type Event;
/// The error returned by all methods.
type Error: std::error::Error + Send + Sync + 'static;
@@ -905,10 +931,53 @@ pub trait Handle: Clone + Sync + Send {
/// Query the peer session state.
fn sessions(&self) -> Result<Self::Sessions, Self::Error>;
/// Subscribe to node events.
- fn subscribe(
- &self,
- timeout: time::Duration,
- ) -> Result<Box<dyn Iterator<Item = Result<Event, Self::Error>>>, Self::Error>;
+ fn subscribe(&self, timeout: time::Duration) -> Result<Self::Events, Self::Error>;
+}
+
+/// Iterator of results `T` when passing a [`Command`] to [`Node::call`].
+///
+/// The iterator blocks for a `timeout` duration, returning [`Error::TimedOut`]
+/// if the duration is reached.
+pub struct LineIter<T> {
+ stream: BufReader<UnixStream>,
+ timeout: time::Duration,
+ witness: PhantomData<T>,
+}
+
+impl<T: DeserializeOwned> Iterator for LineIter<T> {
+ type Item = Result<T, Error>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ let mut l = String::new();
+
+ self.stream
+ .get_ref()
+ .set_read_timeout(Some(self.timeout))
+ .ok();
+
+ match self.stream.read_line(&mut l) {
+ Ok(0) => None,
+ Ok(_) => {
+ let result: CommandResult<T> = match json::from_str(&l) {
+ Err(e) => {
+ return Some(Err(Error::InvalidJson {
+ response: l.clone(),
+ error: e,
+ }))
+ }
+ Ok(result) => result,
+ };
+ match result {
+ CommandResult::Okay(result) => Some(Ok(result)),
+ CommandResult::Error { reason } => Some(Err(Error::Command { reason })),
+ }
+ }
+ Err(e) => match e.kind() {
+ io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut => Some(Err(Error::TimedOut)),
+ _ => Some(Err(Error::Io(e))),
+ },
+ }
+ }
}
/// Public node & device identifier.
@@ -929,33 +998,19 @@ impl Node {
}
/// Call a command on the node.
- pub fn call<T: DeserializeOwned>(
+ pub fn call<T: DeserializeOwned + Send + 'static>(
&self,
cmd: Command,
timeout: time::Duration,
- ) -> Result<impl Iterator<Item = Result<T, Error>>, Error> {
+ ) -> Result<LineIter<T>, Error> {
let stream = UnixStream::connect(&self.socket)
.map_err(|e| Error::Connect(self.socket.clone(), e.kind()))?;
cmd.to_writer(&stream)?;
-
- stream.set_read_timeout(Some(timeout))?;
-
- Ok(BufReader::new(stream).lines().map(move |l| {
- let l = l.map_err(|e| match e.kind() {
- io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut => Error::TimedOut,
- _ => Error::Io(e),
- })?;
-
- let result: CommandResult<T> = json::from_str(&l).map_err(|e| Error::InvalidJson {
- response: l.clone(),
- error: e,
- })?;
-
- match result {
- CommandResult::Okay(result) => Ok(result),
- CommandResult::Error { reason } => Err(Error::Command { reason }),
- }
- }))
+ Ok(LineIter {
+ stream: BufReader::new(stream),
+ timeout,
+ witness: PhantomData,
+ })
}
/// Announce refs of the given `rid` to the given seeds.
@@ -1029,6 +1084,8 @@ impl Node {
// attempt to return iterators instead of allocating vecs.
impl Handle for Node {
type Sessions = Vec<Session>;
+ type Events = LineIter<Event>;
+ type Event = Result<Event, Error>;
type Error = Error;
fn nid(&self) -> Result<NodeId, Error> {
@@ -1049,6 +1106,7 @@ impl Handle for Node {
let Ok(mut lines) = self.call::<Success>(Command::Status, DEFAULT_TIMEOUT) else {
return false;
};
+
let Some(Ok(_)) = lines.next() else {
return false;
};
@@ -1122,29 +1180,29 @@ impl Handle for Node {
}
fn follow(&mut self, nid: NodeId, alias: Option<Alias>) -> Result<bool, Error> {
- let mut line = self.call::<Success>(Command::Follow { nid, alias }, DEFAULT_TIMEOUT)?;
- let response = line.next().ok_or(Error::EmptyResponse)??;
+ let mut lines = self.call::<Success>(Command::Follow { nid, alias }, DEFAULT_TIMEOUT)?;
+ let response = lines.next().ok_or(Error::EmptyResponse)??;
Ok(response.updated)
}
fn seed(&mut self, rid: RepoId, scope: policy::Scope) -> Result<bool, Error> {
- let mut line = self.call::<Success>(Command::Seed { rid, scope }, DEFAULT_TIMEOUT)?;
- let response = line.next().ok_or(Error::EmptyResponse)??;
+ let mut lines = self.call::<Success>(Command::Seed { rid, scope }, DEFAULT_TIMEOUT)?;
+ let response = lines.next().ok_or(Error::EmptyResponse)??;
Ok(response.updated)
}
fn unfollow(&mut self, nid: NodeId) -> Result<bool, Error> {
- let mut line = self.call::<Success>(Command::Unfollow { nid }, DEFAULT_TIMEOUT)?;
- let response = line.next().ok_or(Error::EmptyResponse)??;
+ let mut lines = self.call::<Success>(Command::Unfollow { nid }, DEFAULT_TIMEOUT)?;
+ let response = lines.next().ok_or(Error::EmptyResponse)??;
Ok(response.updated)
}
fn unseed(&mut self, rid: RepoId) -> Result<bool, Error> {
- let mut line = self.call::<Success>(Command::Unseed { rid }, DEFAULT_TIMEOUT)?;
- let response = line.next().ok_or(Error::EmptyResponse {})??;
+ let mut lines = self.call::<Success>(Command::Unseed { rid }, DEFAULT_TIMEOUT)?;
+ let response = lines.next().ok_or(Error::EmptyResponse)??;
Ok(response.updated)
}
@@ -1166,26 +1224,21 @@ impl Handle for Node {
}
fn update_inventory(&mut self, rid: RepoId) -> Result<bool, Error> {
- let mut line = self.call::<Success>(Command::UpdateInventory { rid }, DEFAULT_TIMEOUT)?;
- let response = line.next().ok_or(Error::EmptyResponse {})??;
+ let mut lines = self.call::<Success>(Command::UpdateInventory { rid }, DEFAULT_TIMEOUT)?;
+ let response = lines.next().ok_or(Error::EmptyResponse)??;
Ok(response.updated)
}
- fn subscribe(
- &self,
- timeout: time::Duration,
- ) -> Result<Box<dyn Iterator<Item = Result<Event, Error>>>, Error> {
- let events = self.call(Command::Subscribe, timeout)?;
-
- Ok(Box::new(events))
+ fn subscribe(&self, timeout: time::Duration) -> Result<LineIter<Event>, Error> {
+ self.call(Command::Subscribe, timeout)
}
fn sessions(&self) -> Result<Self::Sessions, Error> {
let sessions = self
.call::<Vec<Session>>(Command::Sessions, DEFAULT_TIMEOUT)?
.next()
- .ok_or(Error::EmptyResponse {})??;
+ .ok_or(Error::EmptyResponse)??;
Ok(sessions)
}
commit c8a24d558416661fc3b39e9dffd323b71a23f224
Author: Fintan Halpenny <fintan.halpenny@gmail.com>
Date: Tue Apr 23 17:40:53 2024 +0100
node: report upload-pack progress
Emit the `UploadPack` events from the `upload_pack` method.
A wrapper type, `Reporter`, is introduced for implementing a `Write` instance
that will parse the given bytes as `gix_protocol::RemoteProgress`, if possible,
and emit it via the `Emitter`, and then write the bytes to the underlying
writer. This allows any subscribers to see the progress of an upload-pack, i.e.
`Compressing objects`, `Counting object`, `Enumerating objects`.
Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett
diff --git a/Cargo.lock b/Cargo.lock
index f1e9c3dd5..9343e5cb4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2537,6 +2537,7 @@ dependencies = [
"crossbeam-channel",
"cyphernet",
"fastrand",
+ "gix-protocol",
"io-reactor",
"lexopt",
"libc",
diff --git a/radicle-node/Cargo.toml b/radicle-node/Cargo.toml
index b873f4f87..cda98b6d1 100644
--- a/radicle-node/Cargo.toml
+++ b/radicle-node/Cargo.toml
@@ -21,6 +21,7 @@ colored = { version = "2.1.0" }
crossbeam-channel = { version = "0.5.6" }
cyphernet = { version = "0.5.0", features = ["tor", "dns", "ed25519", "p2p-ed25519"] }
fastrand = { version = "2.0.0" }
+gix-protocol = { version = "0.41.1", features = ["blocking-client"] }
io-reactor = { version = "0.5.1", features = ["popol"] }
lexopt = { version = "0.3.0" }
libc = { version = "0.2.137" }
diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs
index bb1f36e6a..5ca26b25c 100644
--- a/radicle-node/src/service.rs
+++ b/radicle-node/src/service.rs
@@ -425,6 +425,10 @@ where
pub fn local_time(&self) -> LocalTime {
self.clock
}
+
+ pub fn emitter(&self) -> Emitter<Event> {
+ self.emitter.clone()
+ }
}
impl<D, S, G> Service<D, S, G>
diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs
index 442e3e5bc..b9575fbba 100644
--- a/radicle-node/src/wire/protocol.rs
+++ b/radicle-node/src/wire/protocol.rs
@@ -721,7 +721,10 @@ where
};
let task = Task {
- fetch: FetchRequest::Responder { remote: *nid },
+ fetch: FetchRequest::Responder {
+ remote: *nid,
+ emitter: self.service.emitter(),
+ },
stream,
channels,
};
diff --git a/radicle-node/src/worker.rs b/radicle-node/src/worker.rs
index fce17e942..f2340fa88 100644
--- a/radicle-node/src/worker.rs
+++ b/radicle-node/src/worker.rs
@@ -11,14 +11,14 @@ use std::path::PathBuf;
use crossbeam_channel as chan;
use radicle::identity::RepoId;
-use radicle::node::notifications;
+use radicle::node::{notifications, Event};
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, Handle};
+use crate::runtime::{thread, Emitter, Handle};
use crate::service::policy;
use crate::service::policy::Policy;
use crate::wire::StreamId;
@@ -113,13 +113,15 @@ pub enum FetchRequest {
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,
+ Self::Initiator { remote, .. } | Self::Responder { remote, .. } => *remote,
}
}
}
@@ -233,7 +235,7 @@ impl Worker {
let result = self.fetch(rid, remote, refs_at, channels, notifs);
FetchResult::Initiator { rid, result }
}
- FetchRequest::Responder { remote } => {
+ FetchRequest::Responder { remote, emitter } => {
log::debug!(target: "worker", "Worker processing incoming fetch for {remote} on stream {stream}..");
let (mut stream_r, stream_w) = channels.split();
@@ -255,10 +257,17 @@ impl Worker {
};
}
- let result =
- upload_pack::upload_pack(&self.nid, &self.storage, &header, stream_r, stream_w)
- .map(|_| ())
- .map_err(|e| e.into());
+ let result = upload_pack::upload_pack(
+ &self.nid,
+ remote,
+ &self.storage,
+ &emitter,
+ &header,
+ stream_r,
+ stream_w,
+ )
+ .map(|_| ())
+ .map_err(|e| e.into());
log::debug!(target: "worker", "Upload process on stream {stream} exited with result {result:?}");
FetchResult::Responder {
diff --git a/radicle-node/src/worker/upload_pack.rs b/radicle-node/src/worker/upload_pack.rs
index c977a7fd5..731b278da 100644
--- a/radicle-node/src/worker/upload_pack.rs
+++ b/radicle-node/src/worker/upload_pack.rs
@@ -1,8 +1,13 @@
use std::io;
use std::io::Write;
use std::process::{Command, ExitStatus, Stdio};
+use std::time::Instant;
-use radicle::node::NodeId;
+use gix_protocol::transport::bstr::ByteSlice;
+use radicle::identity::RepoId;
+use radicle::node::events;
+use radicle::node::events::Emitter;
+use radicle::node::{Event, NodeId};
use radicle::storage::git::paths;
use radicle::Storage;
@@ -16,15 +21,18 @@ use crate::runtime::thread;
/// send the EOF file message.
pub fn upload_pack<R, W>(
nid: &NodeId,
+ remote: NodeId,
storage: &Storage,
+ emitter: &Emitter<Event>,
header: &pktline::GitRequest,
mut recv: R,
- mut send: W,
+ send: W,
) -> io::Result<ExitStatus>
where
R: io::Read + Send,
W: io::Write + Send,
{
+ let timer = Instant::now();
let protocol_version = header
.extra
.iter()
@@ -75,14 +83,21 @@ where
let mut stdin = child.stdin.take().unwrap();
let mut stdout = io::BufReader::new(child.stdout.take().unwrap());
+ let mut reporter = Reporter {
+ rid: header.repo,
+ remote,
+ emitter: emitter.clone(),
+ send,
+ };
thread::scope(|s| {
thread::spawn_scoped(nid, "upload-pack", s, || {
// N.b. we indefinitely copy stdout to the sender,
// i.e. there's no need for a loop.
- match io::copy(&mut stdout, &mut send) {
+ match io::copy(&mut stdout, &mut reporter) {
Ok(_) => {}
Err(e) => {
log::error!(target: "worker", "Worker channel disconnected for {}; aborting: {e}", header.repo);
+ emitter.emit(events::UploadPack::error(header.repo, remote, e).into());
}
}
});
@@ -104,6 +119,7 @@ where
}
Err(e) => {
log::error!(target: "worker", "Error on upload-pack channel read for {}: {e}", header.repo);
+ emitter.emit(events::UploadPack::error(header.repo, remote, e).into());
break;
}
}
@@ -124,9 +140,62 @@ where
})?;
let status = child.wait()?;
+ emitter.emit(events::UploadPack::done(header.repo, remote, status).into());
+ log::debug!(target: "worker", "Upload pack finished ({}ms)", timer.elapsed().as_millis());
Ok(status)
}
+/// A combination of the upload-pack sender with an [`Emitter`] for reporting
+/// the progress events to subscribers.
+struct Reporter<W> {
+ rid: RepoId,
+ remote: NodeId,
+ emitter: Emitter<Event>,
+ send: W,
+}
+
+impl<W> Reporter<W> {
+ fn emit(&self, buf: &[u8]) {
+ if let Some(progress) = Self::as_upload_pack_progress(buf) {
+ log::trace!(target: "worker", "upload-pack progress: {progress}");
+ self.emitter
+ .emit(events::UploadPack::write(self.rid, self.remote, progress).into());
+ }
+ }
+
+ fn as_upload_pack_progress(buf: &[u8]) -> Option<events::upload_pack::Progress> {
+ use events::upload_pack::Progress::*;
+ let gix_protocol::RemoteProgress {
+ action, step, max, ..
+ } = gix_protocol::RemoteProgress::from_bytes(buf)?;
+ if action.contains_str("Counting objects") {
+ step.and_then(|processed| max.map(|total| Counting { processed, total }))
+ } else if action.contains_str("Compressing objects") {
+ step.and_then(|processed| max.map(|total| Compressing { processed, total }))
+ } else if action.contains_str("Enumerating objects") {
+ max.map(|total| Enumerating { total })
+ } else {
+ None
+ }
+ }
+}
+
+impl<W> io::Write for Reporter<W>
+where
+ W: io::Write,
+{
+ fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+ let n = self.send.write(buf)?;
+ self.send.flush()?;
+ self.emit(buf);
+ Ok(n)
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ self.send.flush()
+ }
+}
+
pub(super) mod pktline {
use std::io;
use std::io::Read;
commit 89627a8b6a58cad55267296c666d7c4b246e6c53
Author: Fintan Halpenny <fintan.halpenny@gmail.com>
Date: Tue Apr 23 17:20:48 2024 +0100
radicle: add upload-pack events
Add a variant for upload-pack events to allow reporting from the upload-pack
process.
Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett
diff --git a/radicle/src/node/events.rs b/radicle/src/node/events.rs
index 34d9f43e7..945195e58 100644
--- a/radicle/src/node/events.rs
+++ b/radicle/src/node/events.rs
@@ -1,3 +1,7 @@
+//! Events for `upload-pack` processes.
+pub mod upload_pack;
+pub use upload_pack::UploadPack;
+
use std::ops::Deref;
use std::sync::Arc;
use std::sync::Mutex;
@@ -62,6 +66,13 @@ pub enum Event {
features: node::Features,
addresses: Vec<node::Address>,
},
+ UploadPack(upload_pack::UploadPack),
+}
+
+impl From<upload_pack::UploadPack> for Event {
+ fn from(value: upload_pack::UploadPack) -> Self {
+ Self::UploadPack(value)
+ }
}
/// Events feed.
diff --git a/radicle/src/node/events/upload_pack.rs b/radicle/src/node/events/upload_pack.rs
new file mode 100644
index 000000000..284947fe1
--- /dev/null
+++ b/radicle/src/node/events/upload_pack.rs
@@ -0,0 +1,96 @@
+use std::fmt;
+use std::io;
+use std::process::ExitStatus;
+
+use crate::node::NodeId;
+use crate::prelude::RepoId;
+
+/// Events that can occur when an upload-pack process is running.
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum UploadPack {
+ /// The upload-pack process finished with `status`.
+ Done {
+ /// The repository being fetched.
+ rid: RepoId,
+ /// The node being fetched from.
+ remote: NodeId,
+ /// The status code of the upload-pack process.
+ ///
+ /// N.b. `ExitStatus` can not be de/serialized, so the `Display` of the
+ /// status is used instead.
+ status: String,
+ },
+ /// The upload-pack process emitted some [`Progress`] data.
+ Write {
+ /// The repository being fetched.
+ rid: RepoId,
+ /// The node being fetched from.
+ remote: NodeId,
+ /// The progress metadata of the upload-pack.
+ progress: Progress,
+ },
+ Error {
+ /// The repository being fetched.
+ rid: RepoId,
+ /// The node being fetched from.
+ remote: NodeId,
+ /// The error that occurred during the upload-pack.
+ err: String,
+ },
+}
+
+impl UploadPack {
+ /// Construct a `UploadPack::Write` event.
+ pub fn write(rid: RepoId, remote: NodeId, progress: Progress) -> Self {
+ Self::Write {
+ rid,
+ remote,
+ progress,
+ }
+ }
+
+ /// Construct a `UploadPack::Done` event.
+ ///
+ /// If `error` is `None` the process finished successfully, otherwise it
+ /// finished with an error.
+ pub fn done(rid: RepoId, remote: NodeId, status: ExitStatus) -> Self {
+ Self::Done {
+ rid,
+ remote,
+ status: status.to_string(),
+ }
+ }
+
+ pub fn error(rid: RepoId, remote: NodeId, err: io::Error) -> Self {
+ Self::Error {
+ rid,
+ remote,
+ err: err.to_string(),
+ }
+ }
+}
+
+/// Progress updates emitted from the `git-upload-pack` process.
+#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum Progress {
+ Enumerating { total: usize },
+ Counting { processed: usize, total: usize },
+ Compressing { processed: usize, total: usize },
+}
+
+impl fmt::Display for Progress {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Progress::Enumerating { total } => write!(f, "Enumerating objects: {total}"),
+ Progress::Counting { processed, total } => {
+ let percent = (processed / total) * 100;
+ write!(f, "Counting objects: {percent}% ({processed}/{total})")
+ }
+ Progress::Compressing { processed, total } => {
+ let percent = (processed / total) * 100;
+ write!(f, "Compressing objects: {percent}% ({processed}/{total})")
+ }
+ }
+ }
+}
commit f0648c2a184732134e38f8ce1eeae5f6b2f5a04d
Author: Fintan Halpenny <fintan.halpenny@gmail.com>
Date: Tue Apr 23 17:17:20 2024 +0100
radicle: move Emitter to events module
Move the events Emitter to the `radicle::node::events` module to allow it to be
used outside of `radicle-node`.
Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett
diff --git a/radicle-node/src/runtime.rs b/radicle-node/src/runtime.rs
index 9406f1298..9f94a93a5 100644
--- a/radicle-node/src/runtime.rs
+++ b/radicle-node/src/runtime.rs
@@ -3,7 +3,6 @@ pub mod thread;
use std::os::unix::net::UnixListener;
use std::path::PathBuf;
-use std::sync::{Arc, Mutex};
use std::{fs, io, net};
use crossbeam_channel as chan;
@@ -35,6 +34,7 @@ use crate::{service, LocalTime};
pub use handle::Error as HandleError;
pub use handle::Handle;
+pub use node::events::Emitter;
/// A client error.
#[derive(Error, Debug)]
@@ -84,39 +84,6 @@ pub enum Error {
GitVersion(#[from] git::VersionError),
}
-/// Publishes events to subscribers.
-#[derive(Debug, Clone)]
-pub struct Emitter<T> {
- subscribers: Arc<Mutex<Vec<chan::Sender<T>>>>,
-}
-
-impl<T> Default for Emitter<T> {
- fn default() -> Emitter<T> {
- Emitter {
- subscribers: Default::default(),
- }
- }
-}
-
-impl<T: Clone> Emitter<T> {
- /// Emit event to subscribers and drop those who can't receive it.
- pub(crate) fn emit(&self, event: T) {
- self.subscribers
- .lock()
- .unwrap()
- .retain(|s| s.try_send(event.clone()).is_ok());
- }
-
- /// Subscribe to events stream.
- pub fn subscribe(&self) -> chan::Receiver<T> {
- let (sender, receiver) = chan::unbounded();
- let mut subs = self.subscribers.lock().unwrap();
- subs.push(sender);
-
- receiver
- }
-}
-
/// Holds join handles to the client threads, as well as a client handle.
pub struct Runtime {
pub id: NodeId,
diff --git a/radicle/src/node/events.rs b/radicle/src/node/events.rs
index d5c3eab30..34d9f43e7 100644
--- a/radicle/src/node/events.rs
+++ b/radicle/src/node/events.rs
@@ -1,4 +1,6 @@
use std::ops::Deref;
+use std::sync::Arc;
+use std::sync::Mutex;
use std::time;
use crossbeam_channel as chan;
@@ -119,3 +121,36 @@ impl Events {
}
}
}
+
+/// Publishes events to subscribers.
+#[derive(Debug, Clone)]
+pub struct Emitter<T> {
+ subscribers: Arc<Mutex<Vec<chan::Sender<T>>>>,
+}
+
+impl<T> Default for Emitter<T> {
+ fn default() -> Emitter<T> {
+ Emitter {
+ subscribers: Default::default(),
+ }
+ }
+}
+
+impl<T: Clone> Emitter<T> {
+ /// Emit event to subscribers and drop those who can't receive it.
+ pub fn emit(&self, event: T) {
+ self.subscribers
+ .lock()
+ .unwrap()
+ .retain(|s| s.try_send(event.clone()).is_ok());
+ }
+
+ /// Subscribe to events stream.
+ pub fn subscribe(&self) -> chan::Receiver<T> {
+ let (sender, receiver) = chan::unbounded();
+ let mut subs = self.subscribers.lock().unwrap();
+ subs.push(sender);
+
+ receiver
+ }
+}
commit fd4c4cd0af8836213a0e846e674fb6554c123e96
Author: Fintan Halpenny <fintan.halpenny@gmail.com>
Date: Tue Apr 23 14:43:49 2024 +0100
node: dynamic fetch timeout via channels
Instead of using `DEFAULT_CHANNEL_TIMEOUT`, the timeout is dynamically set
per-fetch by the caller.
This passed via the `Io::Fetch` variant -- when initiating a fetch, and the
`Control::Open` variant when responding to a fetch.
Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett
diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs
index 085bd5533..bb1f36e6a 100644
--- a/radicle-node/src/service.rs
+++ b/radicle-node/src/service.rs
@@ -280,6 +280,8 @@ struct QueuedFetch {
from: NodeId,
/// Refs being fetched.
refs_at: Vec<RefsAt>,
+ /// The timeout given for the fetch request.
+ timeout: time::Duration,
/// Result channel.
channel: Option<chan::Sender<FetchResult>>,
}
@@ -948,6 +950,7 @@ where
rid,
refs_at,
from,
+ timeout,
channel,
};
if self.queue.contains(&fetch) {
@@ -964,6 +967,7 @@ where
rid,
refs_at,
from,
+ timeout,
channel,
});
}
@@ -1128,6 +1132,7 @@ where
rid,
from,
refs_at,
+ timeout,
channel,
}) = self.queue.pop_front()
{
@@ -1140,12 +1145,12 @@ where
.expect("Service::dequeue_fetch: error accessing repo seeding configuration");
// Keep dequeueing if there was nothing to fetch, otherwise break.
- if self.fetch_refs_at(rid, from, refs, repo_entry.scope, FETCH_TIMEOUT, channel) {
+ if self.fetch_refs_at(rid, from, refs, repo_entry.scope, timeout, channel) {
break;
}
} else {
// If no refs are specified, always do a full fetch.
- self.fetch(rid, from, FETCH_TIMEOUT, channel);
+ self.fetch(rid, from, timeout, channel);
break;
}
}
diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs
index 2ac20c893..442e3e5bc 100644
--- a/radicle-node/src/wire/protocol.rs
+++ b/radicle-node/src/wire/protocol.rs
@@ -29,6 +29,7 @@ use crate::crypto::Signer;
use crate::prelude::Deserializer;
use crate::service;
use crate::service::io::Io;
+use crate::service::FETCH_TIMEOUT;
use crate::service::{session, DisconnectReason, Service};
use crate::wire::frame;
use crate::wire::frame::{Frame, FrameData, StreamId};
@@ -43,10 +44,6 @@ pub const NOISE_XK: HandshakePattern = HandshakePattern {
responder: cyphernet::encrypt::noise::OneWayPattern::Known,
};
-/// Default time to wait to receive something from a worker channel. Applies to
-/// workers waiting for data from remotes as well.
-pub const DEFAULT_CHANNEL_TIMEOUT: time::Duration = time::Duration::from_secs(30);
-
/// Default time to wait until a network connection is considered inactive.
pub const DEFAULT_CONNECTION_TIMEOUT: time::Duration = time::Duration::from_secs(6);
@@ -131,22 +128,22 @@ impl Streams {
}
/// Open a new stream.
- fn open(&mut self) -> (StreamId, worker::Channels) {
+ fn open(&mut self, timeout: time::Duration) -> (StreamId, worker::Channels) {
self.seq += 1;
let id = StreamId::git(self.link)
.nth(self.seq)
.expect("Streams::open: too many streams");
let channels = self
- .register(id)
+ .register(id, timeout)
.expect("Streams::open: stream was already open");
(id, channels)
}
/// Register an open stream.
- fn register(&mut self, stream: StreamId) -> Option<worker::Channels> {
- let (wire, worker) = worker::Channels::pair(DEFAULT_CHANNEL_TIMEOUT)
+ fn register(&mut self, stream: StreamId, timeout: time::Duration) -> Option<worker::Channels> {
+ let (wire, worker) = worker::Channels::pair(timeout)
.expect("Streams::register: fatal: unable to create channels");
match self.streams.entry(stream) {
@@ -718,7 +715,7 @@ where
})) => {
log::debug!(target: "wire", "Received `open` command for stream {stream} from {nid}");
- let Some(channels) = streams.register(stream) else {
+ let Some(channels) = streams.register(stream, FETCH_TIMEOUT) else {
log::warn!(target: "wire", "Peer attempted to open already-open stream stream {stream}");
continue;
};
@@ -1017,7 +1014,7 @@ where
log::error!(target: "wire", "Peer {remote} is not connected: dropping fetch");
continue;
};
- let (stream, channels) = streams.open();
+ let (stream, channels) = streams.open(timeout);
log::debug!(target: "wire", "Opened new stream with id {stream} for {rid} and remote {remote}");
@@ -1027,7 +1024,6 @@ where
rid,
remote,
refs_at,
- timeout,
},
stream,
channels,
diff --git a/radicle-node/src/worker.rs b/radicle-node/src/worker.rs
index 087fe5b8c..fce17e942 100644
--- a/radicle-node/src/worker.rs
+++ b/radicle-node/src/worker.rs
@@ -5,8 +5,8 @@ mod upload_pack;
pub mod fetch;
pub mod garbage;
+use std::io;
use std::path::PathBuf;
-use std::{io, time};
use crossbeam_channel as chan;
@@ -107,8 +107,6 @@ pub enum FetchRequest {
remote: NodeId,
/// If this fetch is for a particular set of `rad/sigrefs`.
refs_at: Option<Vec<RefsAt>>,
- /// Fetch timeout.
- timeout: time::Duration,
},
/// Server is responding to a fetch request by uploading the
/// specified `refspecs` sent by the client.
@@ -230,8 +228,6 @@ impl Worker {
rid,
remote,
refs_at,
- // TODO: nowhere to use this currently
- timeout: _timeout,
} => {
log::debug!(target: "worker", "Worker processing outgoing fetch for {rid}");
let result = self.fetch(rid, remote, refs_at, channels, notifs);
Exit code: 0
shell: 'cargo --version rustc --version cargo fmt --check cargo clippy --all-targets --workspace -- --deny clippy::all cargo build --all-targets --workspace cargo doc --workspace cargo test --workspace --no-fail-fast '
Commands:
$ podman run --name 2778bad7-777f-4219-9132-0d9b61d6468a -v /opt/radcis/ci.rad.levitte.org/cci/state/2778bad7-777f-4219-9132-0d9b61d6468a/s:/2778bad7-777f-4219-9132-0d9b61d6468a/s:ro -v /opt/radcis/ci.rad.levitte.org/cci/state/2778bad7-777f-4219-9132-0d9b61d6468a/w:/2778bad7-777f-4219-9132-0d9b61d6468a/w -w /2778bad7-777f-4219-9132-0d9b61d6468a/w -v /opt/radcis/ci.rad.levitte.org/.radicle:/${id}/.radicle:ro -e RAD_HOME=/${id}/.radicle rust:trixie bash /2778bad7-777f-4219-9132-0d9b61d6468a/s/script.sh
+ cargo --version
info: syncing channel updates for '1.77-x86_64-unknown-linux-gnu'
info: latest update on 2024-04-09, rust version 1.77.2 (25ef9e3d8 2024-04-09)
info: downloading component 'cargo'
info: downloading component 'rust-std'
info: downloading component 'rustc'
info: installing component 'cargo'
info: installing component 'rust-std'
info: installing component 'rustc'
cargo 1.77.2 (e52e36006 2024-03-26)
+ rustc --version
rustc 1.77.2 (25ef9e3d8 2024-04-09)
+ cargo fmt --check
error: 'cargo-fmt' is not installed for the toolchain '1.77-x86_64-unknown-linux-gnu'.
To install, run `rustup component add --toolchain 1.77-x86_64-unknown-linux-gnu rustfmt`
Exit code: 1
{
"response": "finished",
"result": "failure"
}