Use irpc::channel::SendError as default sink error.

Author: rklaehn

src/api.rs

@@ -30,7 +30,7 @@ pub mod downloader;
 pub mod proto;
 pub mod remote;
 pub mod tags;
-use crate::api::proto::WaitIdleRequest;
+use crate::{api::proto::WaitIdleRequest, provider::events::ProgressError};
 pub use crate::{store::util::Tag, util::temp_tag::TempTag};
 
 pub(crate) type ApiClient = irpc::Client<proto::Request>;
@@ -97,6 +97,8 @@ pub enum ExportBaoError {
     ExportBaoIo { source: io::Error },
     #[snafu(display("encode error: {source}"))]
     ExportBaoInner { source: bao_tree::io::EncodeError },
+    #[snafu(display("client error: {source}"))]
+    ClientError { source: ProgressError },
 }
 
 impl From<ExportBaoError> for Error {
@@ -107,6 +109,7 @@ impl From<ExportBaoError> for Error {
             ExportBaoError::Request { source, .. } => Self::Io(source.into()),
             ExportBaoError::ExportBaoIo { source, .. } => Self::Io(source),
             ExportBaoError::ExportBaoInner { source, .. } => Self::Io(source.into()),
+            ExportBaoError::ClientError { source, .. } => Self::Io(source.into()),
         }
     }
 }
@@ -152,6 +155,12 @@ impl From<bao_tree::io::EncodeError> for ExportBaoError {
     }
 }
 
+impl From<ProgressError> for ExportBaoError {
+    fn from(value: ProgressError) -> Self {
+        ClientSnafu.into_error(value)
+    }
+}
+
 pub type ExportBaoResult<T> = std::result::Result<T, ExportBaoError>;
 
 #[derive(Debug, derive_more::Display, derive_more::From, Serialize, Deserialize)]

src/api/blobs.rs

@@ -57,6 +57,7 @@ use super::{
 };
 use crate::{
     api::proto::{BatchRequest, ImportByteStreamUpdate},
+    provider::events::ClientResult,
     store::IROH_BLOCK_SIZE,
     util::temp_tag::TempTag,
     BlobFormat, Hash, HashAndFormat,
@@ -1112,7 +1113,9 @@ impl ExportBaoProgress {
                         .write_chunk(leaf.data)
                         .await
                         .map_err(io::Error::from)?;
-                    progress.notify_payload_write(index, leaf.offset, len).await;
+                    progress
+                        .notify_payload_write(index, leaf.offset, len)
+                        .await?;
                 }
                 EncodedItem::Done => break,
                 EncodedItem::Error(cause) => return Err(cause.into()),
@@ -1158,7 +1161,7 @@ impl ExportBaoProgress {
 
 pub(crate) trait WriteProgress {
     /// Notify the progress writer that a payload write has happened.
-    async fn notify_payload_write(&mut self, index: u64, offset: u64, len: usize);
+    async fn notify_payload_write(&mut self, index: u64, offset: u64, len: usize) -> ClientResult;
 
     /// Log a write of some other data.
     fn log_other_write(&mut self, len: usize);

src/api/downloader.rs

@@ -3,7 +3,6 @@ use std::{
     collections::{HashMap, HashSet},
     fmt::Debug,
     future::{Future, IntoFuture},
-    io,
     sync::Arc,
 };
 
@@ -113,7 +112,7 @@ async fn handle_download_impl(
         SplitStrategy::Split => handle_download_split_impl(store, pool, request, tx).await?,
         SplitStrategy::None => match request.request {
             FiniteRequest::Get(get) => {
-                let sink = IrpcSenderRefSink(tx).with_map_err(io::Error::other);
+                let sink = IrpcSenderRefSink(tx);
                 execute_get(&pool, Arc::new(get), &request.providers, &store, sink).await?;
             }
             FiniteRequest::GetMany(_) => {
@@ -144,7 +143,7 @@ async fn handle_download_split_impl(
                 let (tx, rx) = tokio::sync::mpsc::channel::<(usize, DownloadProgessItem)>(16);
                 progress_tx.send(rx).await.ok();
                 let sink = TokioMpscSenderSink(tx)
-                    .with_map_err(io::Error::other)
+                    .with_map_err(|_| irpc::channel::SendError::ReceiverClosed)
                     .with_map(move |x| (id, x));
                 let res = execute_get(&pool, Arc::new(request), &providers, &store, sink).await;
                 (hash, res)
@@ -375,7 +374,7 @@ async fn split_request<'a>(
     providers: &Arc<dyn ContentDiscovery>,
     pool: &ConnectionPool,
     store: &Store,
-    progress: impl Sink<DownloadProgessItem, Error = io::Error>,
+    progress: impl Sink<DownloadProgessItem, Error = irpc::channel::SendError>,
 ) -> anyhow::Result<Box<dyn Iterator<Item = GetRequest> + Send + 'a>> {
     Ok(match request {
         FiniteRequest::Get(req) => {
@@ -431,7 +430,7 @@ async fn execute_get(
     request: Arc<GetRequest>,
     providers: &Arc<dyn ContentDiscovery>,
     store: &Store,
-    mut progress: impl Sink<DownloadProgessItem, Error = io::Error>,
+    mut progress: impl Sink<DownloadProgessItem, Error = irpc::channel::SendError>,
 ) -> anyhow::Result<()> {
     let remote = store.remote();
     let mut providers = providers.find_providers(request.content());

src/api/remote.rs

@@ -18,6 +18,7 @@ use crate::{
         GetManyRequest, ObserveItem, ObserveRequest, PushRequest, Request, RequestType,
         MAX_MESSAGE_SIZE,
     },
+    provider::events::{ClientResult, ProgressError},
     util::sink::{Sink, TokioMpscSenderSink},
 };
 
@@ -478,9 +479,7 @@ impl Remote {
         let content = content.into();
         let (tx, rx) = tokio::sync::mpsc::channel(64);
         let tx2 = tx.clone();
-        let sink = TokioMpscSenderSink(tx)
-            .with_map(GetProgressItem::Progress)
-            .with_map_err(io::Error::other);
+        let sink = TokioMpscSenderSink(tx).with_map(GetProgressItem::Progress);
         let this = self.clone();
         let fut = async move {
             let res = this.fetch_sink(conn, content, sink).await.into();
@@ -503,7 +502,7 @@ impl Remote {
         &self,
         mut conn: impl GetConnection,
         content: impl Into<HashAndFormat>,
-        progress: impl Sink<u64, Error = io::Error>,
+        progress: impl Sink<u64, Error = irpc::channel::SendError>,
     ) -> GetResult<Stats> {
         let content = content.into();
         let local = self
@@ -556,9 +555,7 @@ impl Remote {
     pub fn execute_push(&self, conn: Connection, request: PushRequest) -> PushProgress {
         let (tx, rx) = tokio::sync::mpsc::channel(64);
         let tx2 = tx.clone();
-        let sink = TokioMpscSenderSink(tx)
-            .with_map(PushProgressItem::Progress)
-            .with_map_err(io::Error::other);
+        let sink = TokioMpscSenderSink(tx).with_map(PushProgressItem::Progress);
         let this = self.clone();
         let fut = async move {
             let res = this.execute_push_sink(conn, request, sink).await.into();
@@ -577,7 +574,7 @@ impl Remote {
         &self,
         conn: Connection,
         request: PushRequest,
-        progress: impl Sink<u64, Error = io::Error>,
+        progress: impl Sink<u64, Error = irpc::channel::SendError>,
     ) -> anyhow::Result<Stats> {
         let hash = request.hash;
         debug!(%hash, "pushing");
@@ -632,9 +629,7 @@ impl Remote {
     pub fn execute_get_with_opts(&self, conn: Connection, request: GetRequest) -> GetProgress {
         let (tx, rx) = tokio::sync::mpsc::channel(64);
         let tx2 = tx.clone();
-        let sink = TokioMpscSenderSink(tx)
-            .with_map(GetProgressItem::Progress)
-            .with_map_err(io::Error::other);
+        let sink = TokioMpscSenderSink(tx).with_map(GetProgressItem::Progress);
         let this = self.clone();
         let fut = async move {
             let res = this.execute_get_sink(&conn, request, sink).await.into();
@@ -658,7 +653,7 @@ impl Remote {
         &self,
         conn: &Connection,
         request: GetRequest,
-        mut progress: impl Sink<u64, Error = io::Error>,
+        mut progress: impl Sink<u64, Error = irpc::channel::SendError>,
     ) -> GetResult<Stats> {
         let store = self.store();
         let root = request.hash;
@@ -721,9 +716,7 @@ impl Remote {
     pub fn execute_get_many(&self, conn: Connection, request: GetManyRequest) -> GetProgress {
         let (tx, rx) = tokio::sync::mpsc::channel(64);
         let tx2 = tx.clone();
-        let sink = TokioMpscSenderSink(tx)
-            .with_map(GetProgressItem::Progress)
-            .with_map_err(io::Error::other);
+        let sink = TokioMpscSenderSink(tx).with_map(GetProgressItem::Progress);
         let this = self.clone();
         let fut = async move {
             let res = this.execute_get_many_sink(conn, request, sink).await.into();
@@ -747,7 +740,7 @@ impl Remote {
         &self,
         conn: Connection,
         request: GetManyRequest,
-        mut progress: impl Sink<u64, Error = io::Error>,
+        mut progress: impl Sink<u64, Error = irpc::channel::SendError>,
     ) -> GetResult<Stats> {
         let store = self.store();
         let hash_seq = request.hashes.iter().copied().collect::<HashSeq>();
@@ -884,7 +877,7 @@ async fn get_blob_ranges_impl(
     header: AtBlobHeader,
     hash: Hash,
     store: &Store,
-    mut progress: impl Sink<u64, Error = io::Error>,
+    mut progress: impl Sink<u64, Error = irpc::channel::SendError>,
 ) -> GetResult<AtEndBlob> {
     let (mut content, size) = header.next().await?;
     let Some(size) = NonZeroU64::new(size) else {
@@ -1048,11 +1041,20 @@ struct StreamContext<S> {
 
 impl<S> WriteProgress for StreamContext<S>
 where
-    S: Sink<u64, Error = io::Error>,
+    S: Sink<u64, Error = irpc::channel::SendError>,
 {
-    async fn notify_payload_write(&mut self, _index: u64, _offset: u64, len: usize) {
+    async fn notify_payload_write(
+        &mut self,
+        _index: u64,
+        _offset: u64,
+        len: usize,
+    ) -> ClientResult {
         self.payload_bytes_sent += len as u64;
-        self.sender.send(self.payload_bytes_sent).await.ok();
+        self.sender
+            .send(self.payload_bytes_sent)
+            .await
+            .map_err(|e| ProgressError::Internal { source: e.into() })?;
+        Ok(())
     }
 
     fn log_other_write(&mut self, _len: usize) {}

src/provider.rs

@@ -25,7 +25,7 @@ use crate::{
     },
     hashseq::HashSeq,
     protocol::{GetManyRequest, GetRequest, ObserveItem, ObserveRequest, PushRequest, Request},
-    provider::events::{ClientConnected, ConnectionClosed, RequestTracker},
+    provider::events::{ClientConnected, ClientResult, ConnectionClosed, RequestTracker},
     Hash,
 };
 pub mod events;
@@ -264,11 +264,11 @@ impl WriterContext {
 }
 
 impl WriteProgress for WriterContext {
-    async fn notify_payload_write(&mut self, _index: u64, offset: u64, len: usize) {
+    async fn notify_payload_write(&mut self, _index: u64, offset: u64, len: usize) -> ClientResult {
         let len = len as u64;
         let end_offset = offset + len;
         self.payload_bytes_written += len;
-        self.tracker.transfer_progress(len, end_offset).await.ok();
+        self.tracker.transfer_progress(len, end_offset).await
     }
 
     fn log_other_write(&mut self, len: usize) {

src/provider/events.rs

@@ -1,4 +1,4 @@
-use std::{fmt::Debug, ops::Deref};
+use std::{fmt::Debug, io, ops::Deref};
 
 use irpc::{
     channel::{mpsc, none::NoSender, oneshot},
@@ -76,60 +76,70 @@ pub enum AbortReason {
 }
 
 #[derive(Debug, Snafu)]
-pub enum ClientError {
-    RateLimited,
+pub enum ProgressError {
+    Limit,
     Permission,
     #[snafu(transparent)]
-    Irpc {
+    Internal {
         source: irpc::Error,
     },
 }
 
-impl ClientError {
+impl From<ProgressError> for io::Error {
+    fn from(value: ProgressError) -> Self {
+        match value {
+            ProgressError::Limit => io::ErrorKind::QuotaExceeded.into(),
+            ProgressError::Permission => io::ErrorKind::PermissionDenied.into(),
+            ProgressError::Internal { source } => source.into(),
+        }
+    }
+}
+
+impl ProgressError {
     pub fn code(&self) -> quinn::VarInt {
         match self {
-            ClientError::RateLimited => ERR_LIMIT,
-            ClientError::Permission => ERR_PERMISSION,
-            ClientError::Irpc { .. } => ERR_INTERNAL,
+            ProgressError::Limit => ERR_LIMIT,
+            ProgressError::Permission => ERR_PERMISSION,
+            ProgressError::Internal { .. } => ERR_INTERNAL,
         }
     }
 
     pub fn reason(&self) -> &'static [u8] {
         match self {
-            ClientError::RateLimited => b"limit",
-            ClientError::Permission => b"permission",
-            ClientError::Irpc { .. } => b"internal",
+            ProgressError::Limit => b"limit",
+            ProgressError::Permission => b"permission",
+            ProgressError::Internal { .. } => b"internal",
         }
     }
 }
 
-impl From<AbortReason> for ClientError {
+impl From<AbortReason> for ProgressError {
     fn from(value: AbortReason) -> Self {
         match value {
-            AbortReason::RateLimited => ClientError::RateLimited,
-            AbortReason::Permission => ClientError::Permission,
+            AbortReason::RateLimited => ProgressError::Limit,
+            AbortReason::Permission => ProgressError::Permission,
         }
     }
 }
 
-impl From<irpc::channel::RecvError> for ClientError {
+impl From<irpc::channel::RecvError> for ProgressError {
     fn from(value: irpc::channel::RecvError) -> Self {
-        ClientError::Irpc {
+        ProgressError::Internal {
             source: value.into(),
         }
     }
 }
 
-impl From<irpc::channel::SendError> for ClientError {
+impl From<irpc::channel::SendError> for ProgressError {
     fn from(value: irpc::channel::SendError) -> Self {
-        ClientError::Irpc {
+        ProgressError::Internal {
             source: value.into(),
         }
     }
 }
 
 pub type EventResult = Result<(), AbortReason>;
-pub type ClientResult = Result<(), ClientError>;
+pub type ClientResult = Result<(), ProgressError>;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct EventMask {
@@ -407,7 +417,7 @@ impl EventSender {
         f: impl FnOnce() -> Req,
         connection_id: u64,
         request_id: u64,
-    ) -> Result<RequestTracker, ClientError>
+    ) -> Result<RequestTracker, ProgressError>
     where
         ProviderProto: From<RequestReceived<Req>>,
         ProviderMessage: From<WithChannels<RequestReceived<Req>, ProviderProto>>,
@@ -466,7 +476,7 @@ impl EventSender {
                     RequestUpdates::Active(tx)
                 }
                 RequestMode::Disabled => {
-                    return Err(ClientError::Permission);
+                    return Err(ProgressError::Permission);
                 }
                 _ => RequestUpdates::None,
             },