fix(iroh-net): temp fix for progress bar when downloading a hash seq (#1658)

## Description

Fixes the download progress bar when downloading a hash seq where the
size is not known.

## Notes & open questions

Is this good enough? The only way to get a **proper** progress for a
hash seq where you don't know the overall size is to do a verified size
request before and give the sum of the sizes to the progress logic. But
that would be another roundtrip. Fine for GUI stuff, but in many cases
we don't really care, so I would hate to do this all the time.

## Change checklist

- [x] Self-review.
- [ ] Documentation updates if relevant.
- [ ] Tests if relevant.

Author: rklaehn

iroh-bytes/src/provider.rs

@@ -156,26 +156,26 @@ pub enum AddProgress {
 
 /// Progress updates for the get operation.
 #[derive(Debug, Serialize, Deserialize)]
-pub enum GetProgress {
+pub enum DownloadProgress {
     /// A new connection was established.
     Connected,
     /// An item was found with hash `hash`, from now on referred to via `id`.
     Found {
         /// A new unique id for this entry.
         id: u64,
+        /// child offset
+        child: u64,
         /// The name of the entry.
         hash: Hash,
         /// The size of the entry in bytes.
         size: u64,
     },
     /// An item was found with hash `hash`, from now on referred to via `id`.
-    FoundCollection {
+    FoundHashSeq {
         /// The name of the entry.
         hash: Hash,
         /// Number of children in the collection, if known.
-        num_blobs: Option<u64>,
-        /// The size of the entry in bytes, if known.
-        total_blobs_size: Option<u64>,
+        children: u64,
     },
     /// We got progress ingesting item `id`.
     Progress {
@@ -279,7 +279,7 @@ pub async fn transfer_collection<D: Map, E: EventSender>(
     // if the request is just for the root, we don't need to deserialize the collection
     let just_root = matches!(request.ranges.as_single(), Some((0, _)));
     let mut c = if !just_root {
-        // use the collection parser to parse the collection
+        // parse the hash seq
         let (stream, num_blobs) = parse_hash_seq(&mut data).await?;
         writer
             .events

iroh-bytes/src/store/flat.rs

@@ -1197,7 +1197,7 @@ impl Store {
         meta_path: PathBuf,
         rt: crate::util::runtime::Handle,
     ) -> anyhow::Result<Self> {
-        tracing::debug!(
+        tracing::info!(
             "loading database from {} {}",
             complete_path.display(),
             partial_path.display()

iroh-bytes/src/store/traits.rs

@@ -332,13 +332,13 @@ async fn gc_mark_task<'a>(
                 warn!("gc: {} creating data reader failed", hash);
                 continue;
             };
-            let Ok((mut iter, count)) = parse_hash_seq(reader).await else {
+            let Ok((mut stream, count)) = parse_hash_seq(reader).await else {
                 warn!("gc: {} parse failed", hash);
                 continue;
             };
             info!("parsed collection {} {:?}", hash, count);
             loop {
-                let item = match iter.next().await {
+                let item = match stream.next().await {
                     Ok(Some(item)) => item,
                     Ok(None) => break,
                     Err(_err) => {

iroh/src/client.rs

@@ -37,7 +37,7 @@ use crate::rpc_protocol::{
     DocDelResponse, DocDropRequest, DocGetManyRequest, DocGetOneRequest, DocImportRequest,
     DocLeaveRequest, DocListRequest, DocOpenRequest, DocSetHashRequest, DocSetRequest,
     DocShareRequest, DocStartSyncRequest, DocStatusRequest, DocSubscribeRequest, DocTicket,
-    GetProgress, ListTagsRequest, ListTagsResponse, NodeConnectionInfoRequest,
+    DownloadProgress, ListTagsRequest, ListTagsResponse, NodeConnectionInfoRequest,
     NodeConnectionInfoResponse, NodeConnectionsRequest, NodeShutdownRequest, NodeStatsRequest,
     NodeStatusRequest, NodeStatusResponse, ProviderService, SetTagOption, ShareMode, WrapOption,
 };
@@ -333,7 +333,7 @@ where
     pub async fn download(
         &self,
         req: BlobDownloadRequest,
-    ) -> Result<impl Stream<Item = Result<GetProgress>>> {
+    ) -> Result<impl Stream<Item = Result<DownloadProgress>>> {
         let stream = self.rpc.server_streaming(req).await?;
         Ok(stream.map_err(anyhow::Error::from))
     }

iroh/src/commands.rs

@@ -1,4 +1,3 @@
-use std::collections::BTreeMap;
 use std::str::FromStr;
 use std::{net::SocketAddr, path::PathBuf, time::Duration};
 
@@ -11,7 +10,8 @@ use console::style;
 use futures::{Stream, StreamExt};
 use human_time::ToHumanTimeString;
 use indicatif::{
-    HumanBytes, HumanDuration, ProgressBar, ProgressDrawTarget, ProgressState, ProgressStyle,
+    HumanBytes, HumanDuration, MultiProgress, ProgressBar, ProgressDrawTarget, ProgressState,
+    ProgressStyle,
 };
 use iroh::client::quic::Iroh;
 use iroh::dial::Ticket;
@@ -727,15 +727,24 @@ fn fmt_latency(latency: Option<Duration>) -> String {
     }
 }
 
-const PROGRESS_STYLE: &str =
-    "{msg}\n{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {bytes}/{total_bytes} ({eta})";
+fn make_overall_progress() -> ProgressBar {
+    let pb = ProgressBar::hidden();
+    pb.enable_steady_tick(std::time::Duration::from_millis(100));
+    pb.set_style(
+        ProgressStyle::with_template(
+            "{msg}{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {pos}/{len}",
+        )
+        .unwrap()
+        .progress_chars("#>-"),
+    );
+    pb
+}
 
-fn make_download_pb() -> ProgressBar {
+fn make_individual_progress() -> ProgressBar {
     let pb = ProgressBar::hidden();
-    pb.set_draw_target(ProgressDrawTarget::stderr());
-    pb.enable_steady_tick(std::time::Duration::from_millis(50));
+    pb.enable_steady_tick(std::time::Duration::from_millis(100));
     pb.set_style(
-        ProgressStyle::with_template(PROGRESS_STYLE)
+        ProgressStyle::with_template("{msg}{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {bytes}/{total_bytes} ({eta})")
             .unwrap()
             .with_key(
                 "eta",
@@ -750,75 +759,59 @@ fn make_download_pb() -> ProgressBar {
 
 pub async fn show_download_progress(
     hash: Hash,
-    mut stream: impl Stream<Item = Result<GetProgress>> + Unpin,
+    mut stream: impl Stream<Item = Result<DownloadProgress>> + Unpin,
 ) -> Result<()> {
     eprintln!("Fetching: {}", hash);
-    let pb = make_download_pb();
-    pb.set_message(format!("{} Connecting ...", style("[1/3]").bold().dim()));
-    let mut sizes = BTreeMap::new();
-    let mut downloading = false;
+    let mp = MultiProgress::new();
+    mp.set_draw_target(ProgressDrawTarget::stderr());
+    let op = mp.add(make_overall_progress());
+    let ip = mp.add(make_individual_progress());
+    op.set_message(format!("{} Connecting ...\n", style("[1/3]").bold().dim()));
+    let mut seq = false;
     while let Some(x) = stream.next().await {
         match x? {
-            GetProgress::Connected => {
-                pb.set_message(format!("{} Requesting ...", style("[2/3]").bold().dim()));
+            DownloadProgress::Connected => {
+                op.set_message(format!("{} Requesting ...\n", style("[2/3]").bold().dim()));
             }
-            GetProgress::FoundCollection {
-                total_blobs_size,
-                num_blobs,
-                ..
-            } => {
-                let count = num_blobs.unwrap_or_default();
-                let missing_bytes = total_blobs_size.unwrap_or_default();
-                pb.set_message(format!(
-                    "{} Downloading {} file(s) with total transfer size {}",
+            DownloadProgress::FoundHashSeq { children, .. } => {
+                op.set_message(format!(
+                    "{} Downloading {} blob(s)\n",
                     style("[3/3]").bold().dim(),
-                    count,
-                    HumanBytes(missing_bytes),
+                    children + 1,
                 ));
-                pb.set_length(missing_bytes);
-                pb.reset();
-                downloading = true;
+                op.set_length(children + 1);
+                op.reset();
+                seq = true;
             }
-            GetProgress::Found { id, size, .. } => {
-                if !downloading {
-                    pb.set_message(format!(
-                        "{} Downloading blob with size {}",
-                        style("[3/3]").bold().dim(),
-                        size,
-                    ));
-                    pb.set_length(size);
-                    pb.reset();
+            DownloadProgress::Found { size, child, .. } => {
+                if seq {
+                    op.set_position(child);
+                } else {
+                    op.finish_and_clear();
                 }
-                sizes.insert(id, (size, 0));
+                ip.set_length(size);
+                ip.reset();
             }
-            GetProgress::Progress { id, offset } => {
-                if let Some((_, current)) = sizes.get_mut(&id) {
-                    *current = offset;
-                    let total = sizes.values().map(|(_, current)| current).sum::<u64>();
-                    pb.set_position(total);
-                }
+            DownloadProgress::Progress { offset, .. } => {
+                ip.set_position(offset);
             }
-            GetProgress::Done { id } => {
-                if let Some((size, current)) = sizes.get_mut(&id) {
-                    *current = *size;
-                    let total = sizes.values().map(|(_, current)| current).sum::<u64>();
-                    pb.set_position(total);
-                }
+            DownloadProgress::Done { .. } => {
+                ip.finish_and_clear();
             }
-            GetProgress::NetworkDone {
+            DownloadProgress::NetworkDone {
                 bytes_read,
                 elapsed,
                 ..
             } => {
-                pb.finish_and_clear();
+                op.finish_and_clear();
                 eprintln!(
                     "Transferred {} in {}, {}/s",
                     HumanBytes(bytes_read),
                     HumanDuration(elapsed),
                     HumanBytes((bytes_read as f64 / elapsed.as_secs_f64()) as u64)
                 );
             }
-            GetProgress::AllDone => {
+            DownloadProgress::AllDone => {
                 break;
             }
             _ => {}

iroh/src/commands/get.rs

@@ -17,7 +17,7 @@ use iroh_bytes::{
     Hash,
 };
 use iroh_bytes::{
-    provider::GetProgress,
+    provider::DownloadProgress,
     util::{
         progress::{FlumeProgressSender, IdGenerator, ProgressSender},
         BlobFormat,
@@ -130,7 +130,7 @@ impl GetInteractive {
         let response = fsm::start(connection, request);
         let connected = response.next().await?;
         // we are connected
-        sender.send(GetProgress::Connected).await?;
+        sender.send(DownloadProgress::Connected).await?;
         let ConnectedNext::StartRoot(curr) = connected.next().await? else {
             anyhow::bail!("expected root to be present");
         };
@@ -140,13 +140,13 @@ impl GetInteractive {
             get_to_stdout_multi(curr, sender.clone()).await?
         };
         sender
-            .send(GetProgress::NetworkDone {
+            .send(DownloadProgress::NetworkDone {
                 bytes_written: stats.bytes_written,
                 bytes_read: stats.bytes_read,
                 elapsed: stats.elapsed,
             })
             .await?;
-        sender.send(GetProgress::AllDone).await?;
+        sender.send(DownloadProgress::AllDone).await?;
         display_task.await??;
 
         Ok(())
@@ -155,22 +155,31 @@ impl GetInteractive {
 
 async fn get_to_stdout_single(
     curr: get::fsm::AtStartRoot,
-    sender: FlumeProgressSender<GetProgress>,
+    sender: FlumeProgressSender<DownloadProgress>,
 ) -> Result<get::Stats> {
     let curr = curr.next();
     let id = sender.new_id();
     let hash = curr.hash();
     let (curr, size) = curr.next().await?;
-    sender.send(GetProgress::Found { id, hash, size }).await?;
+    sender
+        .send(DownloadProgress::Found {
+            id,
+            hash,
+            size,
+            child: 0,
+        })
+        .await?;
     let sender2 = sender.clone();
     let mut writer = ProgressSliceWriter::new(
         ConcatenateSliceWriter::new(tokio::io::stdout()),
         move |offset| {
-            sender2.try_send(GetProgress::Progress { id, offset }).ok();
+            sender2
+                .try_send(DownloadProgress::Progress { id, offset })
+                .ok();
         },
     );
     let curr = curr.write_all(&mut writer).await?;
-    sender.send(GetProgress::Done { id }).await?;
+    sender.send(DownloadProgress::Done { id }).await?;
     let EndBlobNext::Closing(curr) = curr.next() else {
         anyhow::bail!("expected end of stream")
     };
@@ -179,15 +188,14 @@ async fn get_to_stdout_single(
 
 async fn get_to_stdout_multi(
     curr: get::fsm::AtStartRoot,
-    sender: FlumeProgressSender<GetProgress>,
+    sender: FlumeProgressSender<DownloadProgress>,
 ) -> Result<get::Stats> {
     let hash = curr.hash();
-    let (mut next, _links, collection) = Collection::read_fsm(curr).await?;
+    let (mut next, links, collection) = Collection::read_fsm(curr).await?;
     sender
-        .send(GetProgress::FoundCollection {
+        .send(DownloadProgress::FoundHashSeq {
             hash,
-            num_blobs: Some(collection.total_entries()),
-            total_blobs_size: Some(collection.total_blobs_size()),
+            children: links.len() as u64,
         })
         .await?;
     let collection = collection.into_inner();
@@ -207,21 +215,24 @@ async fn get_to_stdout_multi(
         let id = sender.new_id();
         let (curr, size) = header.next().await?;
         sender
-            .send(GetProgress::Found {
+            .send(DownloadProgress::Found {
                 id,
                 hash: blob.hash,
                 size,
+                child: curr.offset(),
             })
             .await?;
         let sender2 = sender.clone();
         let mut io_writer = ProgressSliceWriter::new(
             ConcatenateSliceWriter::new(tokio::io::stdout()),
             move |offset| {
-                sender2.try_send(GetProgress::Progress { id, offset }).ok();
+                sender2
+                    .try_send(DownloadProgress::Progress { id, offset })
+                    .ok();
             },
         );
         let curr = curr.write_all(&mut io_writer).await?;
-        sender.send(GetProgress::Done { id }).await?;
+        sender.send(DownloadProgress::Done { id }).await?;
         // wait for the progress task to finish, only after dropping the writer
         next = curr.next();
     };