Better stream errors (#11335)

IsseW · web-flow · commit bb7d135aeb8f · 2025-09-26T15:00:46.000+02:00
### Related Closes RR-2338 ### What Splits the `StreamError::Tonic` error into multiple different variants, making the errors more readable and useful. Also adds a details section to notification, which are automatically created if the notification contains a "<details>" part.
diff --git a/crates/store/re_data_source/src/data_source.rs b/crates/store/re_data_source/src/data_source.rs
@@ -261,10 +261,7 @@ impl LogDataSource {
 
                 spawn_future(async move {
                     if let Err(err) = stream_partition.await {
-                        re_log::warn!(
-                            "Error while streaming {uri}: {}",
-                            re_error::format_ref(&err)
-                        );
+                        re_log::warn!("Error while streaming: {}", re_error::format_ref(&err));
                     }
                 });
                 Ok(rx)
diff --git a/crates/store/re_redap_client/src/connection_client.rs b/crates/store/re_redap_client/src/connection_client.rs
@@ -33,7 +33,7 @@ use re_protos::{
     missing_field,
 };
 
-use crate::StreamError;
+use crate::{StreamEntryError, StreamError};
 
 /// Expose an ergonomic API over the gRPC redap client.
 ///
@@ -81,7 +81,8 @@ where
             .find_entries(FindEntriesRequest {
                 filter: Some(filter),
             })
-            .await?
+            .await
+            .map_err(|err| StreamEntryError::Find(err.into()))?
             .into_inner()
             .entries;
 
@@ -97,7 +98,8 @@ where
             .delete_entry(DeleteEntryRequest {
                 id: Some(entry_id.into()),
             })
-            .await?;
+            .await
+            .map_err(|err| StreamEntryError::Delete(err.into()))?;
 
         Ok(())
     }
@@ -117,7 +119,8 @@ where
                 }
                 .into(),
             ))
-            .await?
+            .await
+            .map_err(|err| StreamEntryError::Update(err.into()))?
             .into_inner()
             .try_into()?;
 
@@ -136,7 +139,8 @@ where
                 name: Some(name),
                 id: entry_id.map(Into::into),
             })
-            .await?
+            .await
+            .map_err(|err| StreamEntryError::Create(err.into()))?
             .into_inner()
             .try_into()?;
 
@@ -151,9 +155,12 @@ where
         let response: ReadDatasetEntryResponse = self
             .inner()
             .read_dataset_entry(
-                tonic::Request::new(ReadDatasetEntryRequest {}).with_entry_id(entry_id)?,
+                tonic::Request::new(ReadDatasetEntryRequest {})
+                    .with_entry_id(entry_id)
+                    .map_err(|err| StreamEntryError::InvalidId(err.into()))?,
             )
-            .await?
+            .await
+            .map_err(|err| StreamEntryError::Read(err.into()))?
             .into_inner()
             .try_into()?;
 
@@ -175,7 +182,8 @@ where
                 }
                 .into(),
             ))
-            .await?
+            .await
+            .map_err(|err| StreamEntryError::Update(err.into()))?
             .into_inner()
             .try_into()?;
 
@@ -189,7 +197,8 @@ where
             .read_table_entry(ReadTableEntryRequest {
                 id: Some(entry_id.into()),
             })
-            .await?
+            .await
+            .map_err(|err| StreamEntryError::Read(err.into()))?
             .into_inner()
             .try_into()?;
 
@@ -210,15 +219,20 @@ where
                 tonic::Request::new(ScanPartitionTableRequest {
                     columns: vec![COLUMN_NAME.to_owned()],
                 })
-                .with_entry_id(entry_id)?,
+                .with_entry_id(entry_id)
+                .map_err(|err| StreamEntryError::InvalidId(err.into()))?,
             )
-            .await?
+            .await
+            .map_err(|err| StreamEntryError::ReadPartitions(err.into()))?
             .into_inner();
 
         let mut partition_ids = Vec::new();
 
         while let Some(resp) = stream.next().await {
-            let record_batch = resp?.data()?.decode()?;
+            let record_batch = resp
+                .map_err(|err| StreamEntryError::ReadPartitions(err.into()))?
+                .data()?
+                .decode()?;
 
             let partition_id_col = record_batch
                 .column_by_name(COLUMN_NAME)
@@ -252,12 +266,14 @@ where
             data_sources,
             on_duplicate,
         })
-        .with_entry_id(dataset_id)?;
+        .with_entry_id(dataset_id)
+        .map_err(|err| StreamEntryError::InvalidId(err.into()))?;
 
         let response = self
             .inner()
             .register_with_dataset(req.map(Into::into))
-            .await?
+            .await
+            .map_err(|err| StreamEntryError::RegisterData(err.into()))?
             .into_inner()
             .data
             .ok_or_else(|| missing_field!(RegisterWithDatasetResponse, "data"))?
@@ -346,7 +362,8 @@ where
         let response: RegisterTableResponse = self
             .inner()
             .register_table(tonic::Request::new(request.into()))
-            .await?
+            .await
+            .map_err(|err| StreamEntryError::RegisterTable(err.into()))?
             .into_inner()
             .try_into()?;
 
@@ -375,7 +392,8 @@ where
                 }
                 .into(),
             ))
-            .await?;
+            .await
+            .map_err(|err| StreamEntryError::Maintenance(err.into()))?;
 
         Ok(())
     }
@@ -385,7 +403,8 @@ where
             .do_global_maintenance(tonic::Request::new(
                 re_protos::cloud::v1alpha1::DoGlobalMaintenanceRequest {},
             ))
-            .await?;
+            .await
+            .map_err(|err| StreamEntryError::Maintenance(err.into()))?;
 
         Ok(())
     }
diff --git a/crates/store/re_redap_client/src/connection_registry.rs b/crates/store/re_redap_client/src/connection_registry.rs
@@ -58,19 +58,19 @@ impl ConnectionRegistry {
 pub enum ClientConnectionError {
     /// Native connection error
     #[cfg(not(target_arch = "wasm32"))]
-    #[error("Connection error: {0}")]
+    #[error("Connection error\nDetails:{0}")]
     Tonic(#[from] tonic::transport::Error),
 
     #[error("server is expecting an unencrypted connection (try `rerun+http://` if you are sure)")]
     UnencryptedServer,
 
-    #[error("the server requires an authentication token, but none was provided: {0}")]
+    #[error("the server requires an authentication token, but none was provided\nDetails:{0}")]
     UnauthenticatedMissingToken(TonicStatusError),
 
-    #[error("the server rejected the provided authentication token: {0}")]
+    #[error("the server rejected the provided authentication token\nDetails:{0}")]
     UnauthenticatedBadToken(TonicStatusError),
 
-    #[error("failed to obtain server version: {0}")]
+    #[error("failed to obtain server version\nDetails:{0}")]
     AuthCheckError(TonicStatusError),
 }
 
diff --git a/crates/store/re_redap_client/src/grpc.rs b/crates/store/re_redap_client/src/grpc.rs
@@ -14,7 +14,7 @@ use tokio_stream::{Stream, StreamExt as _};
 
 use crate::{
     ConnectionClient, ConnectionRegistryHandle, MAX_DECODING_MESSAGE_SIZE, StreamError,
-    spawn_future,
+    StreamPartitionError, spawn_future,
 };
 
 /// UI commands issued when streaming in datasets.
@@ -248,13 +248,15 @@ pub(crate) async fn client(
 pub fn get_chunks_response_to_chunk_and_partition_id(
     response: tonic::Streaming<re_protos::cloud::v1alpha1::GetChunksResponse>,
 ) -> impl Stream<Item = Result<Vec<(Chunk, Option<String>)>, StreamError>> {
+    use crate::StreamPartitionError;
+
     response
         .then(|resp| {
             // We want to make sure to offload that compute-heavy work to the compute worker pool: it's
             // not going to make this one single pipeline any faster, but it will prevent starvation of
             // the Tokio runtime (which would slow down every other futures currently scheduled!).
             tokio::task::spawn_blocking(move || {
-                let r = resp.map_err(Into::<StreamError>::into)?;
+                let r = resp.map_err(|err| StreamPartitionError::StreamingChunks(err.into()))?;
                 let _span =
                     tracing::trace_span!("get_chunks::batch_decode", num_chunks = r.chunks.len())
                         .entered();
@@ -292,13 +294,15 @@ pub fn get_chunks_response_to_chunk_and_partition_id(
 pub fn fetch_chunks_response_to_chunk_and_partition_id(
     response: tonic::Streaming<re_protos::cloud::v1alpha1::FetchChunksResponse>,
 ) -> impl Stream<Item = Result<Vec<(Chunk, Option<String>)>, StreamError>> {
+    use crate::StreamPartitionError;
+
     response
         .then(|resp| {
             // We want to make sure to offload that compute-heavy work to the compute worker pool: it's
             // not going to make this one single pipeline any faster, but it will prevent starvation of
             // the Tokio runtime (which would slow down every other futures currently scheduled!).
             tokio::task::spawn_blocking(move || {
-                let r = resp.map_err(Into::<StreamError>::into)?;
+                let r = resp.map_err(|err| StreamPartitionError::StreamingChunks(err.into()))?;
                 let _span =
                     tracing::trace_span!("get_chunks::batch_decode", num_chunks = r.chunks.len())
                         .entered();
@@ -332,7 +336,7 @@ pub fn get_chunks_response_to_chunk_and_partition_id(
     response: tonic::Streaming<re_protos::cloud::v1alpha1::GetChunksResponse>,
 ) -> impl Stream<Item = Result<Vec<(Chunk, Option<String>)>, StreamError>> {
     response.map(|resp| {
-        let resp = resp?;
+        let resp = resp.map_err(|err| StreamPartitionError::StreamingChunks(err.into()))?;
 
         let _span =
             tracing::trace_span!("get_chunks::batch_decode", num_chunks = resp.chunks.len())
@@ -362,7 +366,7 @@ pub fn fetch_chunks_response_to_chunk_and_partition_id(
     response: tonic::Streaming<re_protos::cloud::v1alpha1::FetchChunksResponse>,
 ) -> impl Stream<Item = Result<Vec<(Chunk, Option<String>)>, StreamError>> {
     response.map(|resp| {
-        let resp = resp?;
+        let resp = resp.map_err(|err| StreamPartitionError::StreamingChunks(err.into()))?;
 
         let _span =
             tracing::trace_span!("get_chunks::batch_decode", num_chunks = resp.chunks.len())
@@ -514,7 +518,8 @@ async fn stream_partition_from_server(
                 exclude_temporal_data: true,
                 query: None,
             })
-            .await?
+            .await
+            .map_err(|err| StreamPartitionError::StreamingChunks(err.into()))?
             .into_inner()
     };
 
@@ -551,7 +556,8 @@ async fn stream_partition_from_server(
                     .into()
                 }),
             })
-            .await?
+            .await
+            .map_err(|err| StreamPartitionError::StreamingChunks(err.into()))?
             .into_inner()
     };
 
diff --git a/crates/store/re_redap_client/src/lib.rs b/crates/store/re_redap_client/src/lib.rs
@@ -78,13 +78,55 @@ impl std::error::Error for TonicStatusError {
     }
 }
 
+#[derive(thiserror::Error, Debug)]
+pub enum StreamEntryError {
+    #[error("Failed reading entry\nDetails:{0}")]
+    Read(TonicStatusError),
+
+    #[error("Failed finding entry\nDetails:{0}")]
+    Find(TonicStatusError),
+
+    #[error("Failed deleting entry\nDetails:{0}")]
+    Delete(TonicStatusError),
+
+    #[error("Failed updating entry\nDetails:{0}")]
+    Update(TonicStatusError),
+
+    #[error("Failed creating entry\nDetails:{0}")]
+    Create(TonicStatusError),
+
+    #[error("Failed reading entry's partitions\nDetails:{0}")]
+    ReadPartitions(TonicStatusError),
+
+    #[error("Failed registering data source with entry\nDetails:{0}")]
+    RegisterData(TonicStatusError),
+
+    #[error("Failed registering table\nDetails:{0}")]
+    RegisterTable(TonicStatusError),
+
+    #[error("Error while doing maintenance on entry\nDetails:{0}")]
+    Maintenance(TonicStatusError),
+
+    #[error("Invalid entry id\nDetails:{0}")]
+    InvalidId(TonicStatusError),
+}
+
+#[derive(thiserror::Error, Debug)]
+pub enum StreamPartitionError {
+    #[error("Failed streaming partition chunks\nDetails:{0}")]
+    StreamingChunks(TonicStatusError),
+}
+
 #[derive(thiserror::Error, Debug)]
 pub enum StreamError {
     #[error(transparent)]
     ClientConnectionError(#[from] ClientConnectionError),
 
     #[error(transparent)]
-    TonicStatus(#[from] TonicStatusError),
+    EntryError(#[from] StreamEntryError),
+
+    #[error(transparent)]
+    PartitionError(#[from] StreamPartitionError),
 
     #[error(transparent)]
     Tokio(#[from] tokio::task::JoinError),
@@ -116,12 +158,6 @@ const _: () = assert!(
     "Error type is too large. Try to reduce its size by boxing some of its variants.",
 );
 
-impl From<tonic::Status> for StreamError {
-    fn from(value: tonic::Status) -> Self {
-        Self::TonicStatus(value.into())
-    }
-}
-
 // TODO(ab, andreas): This should be replaced by the use of `AsyncRuntimeHandle`. However, this
 // requires:
 // - `AsyncRuntimeHandle` to be moved lower in the crate hierarchy to be available here (unsure
diff --git a/crates/utils/re_log/src/lib.rs b/crates/utils/re_log/src/lib.rs
@@ -11,6 +11,9 @@
 //!
 //! The `warn_once` etc macros are for when you want to suppress repeated
 //! logging of the exact same message.
+//!
+//! In the viewer these logs, if >= info, become notifications. See
+//! `re_ui::notifications` for more information.
 
 mod channel_logger;
 mod result_extensions;
diff --git a/crates/viewer/re_redap_browser/src/entries.rs b/crates/viewer/re_redap_browser/src/entries.rs
@@ -72,7 +72,8 @@ impl EntryError {
                 | StreamError::CodecError(_)
                 | StreamError::ChunkError(_)
                 | StreamError::DecodeError(_)
-                | StreamError::TonicStatus(_)
+                | StreamError::EntryError(_)
+                | StreamError::PartitionError(_)
                 | StreamError::TypeConversionError(_)
                 | StreamError::MissingDataframeColumn(_)
                 | StreamError::MissingData(_)
diff --git a/crates/viewer/re_ui/src/notifications.rs b/crates/viewer/re_ui/src/notifications.rs
