Fix race condition for ui adjustments while loading recordings from redap client (#11365)

Author: Wumpf

crates/store/re_data_loader/src/load_file.rs

@@ -1,7 +1,7 @@
 use std::borrow::Cow;
 
 use ahash::{HashMap, HashMapExt as _};
-use re_log_types::{FileSource, LogMsg};
+use re_log_types::{DataSourceMessage, FileSource, LogMsg};
 use re_smart_channel::Sender;
 
 use crate::{DataLoader as _, DataLoaderError, LoadedData, RrdLoader};
@@ -21,7 +21,7 @@ pub fn load_from_path(
     file_source: FileSource,
     path: &std::path::Path,
     // NOTE: This channel must be unbounded since we serialize all operations when running on wasm.
-    tx: &Sender<LogMsg>,
+    tx: &Sender<DataSourceMessage>,
 ) -> Result<(), DataLoaderError> {
     use re_log_types::ApplicationId;
 
@@ -71,7 +71,7 @@ pub fn load_from_file_contents(
     filepath: &std::path::Path,
     contents: std::borrow::Cow<'_, [u8]>,
     // NOTE: This channel must be unbounded since we serialize all operations when running on wasm.
-    tx: &Sender<LogMsg>,
+    tx: &Sender<DataSourceMessage>,
 ) -> Result<(), DataLoaderError> {
     re_tracing::profile_function!(filepath.to_string_lossy());
 
@@ -271,7 +271,7 @@ pub(crate) fn send(
     settings: crate::DataLoaderSettings,
     file_source: FileSource,
     rx_loader: std::sync::mpsc::Receiver<LoadedData>,
-    tx: &Sender<LogMsg>,
+    tx: &Sender<DataSourceMessage>,
 ) {
     spawn({
         re_tracing::profile_function!();
@@ -319,7 +319,7 @@ pub(crate) fn send(
                         continue;
                     }
                 };
-                tx.send(msg).ok();
+                tx.send(msg.into()).ok();
             }
 
             for (store_id, tracked) in store_info_tracker {
@@ -335,7 +335,7 @@ pub(crate) fn send(
 
                 if should_send_new_store_info {
                     let store_info = prepare_store_info(&store_id, file_source.clone());
-                    tx.send(store_info).ok();
+                    tx.send(store_info.into()).ok();
                 }
             }
 

crates/store/re_data_source/src/data_source.rs

@@ -1,4 +1,4 @@
-use re_log_types::{LogMsg, RecordingId};
+use re_log_types::{DataSourceMessage, RecordingId};
 use re_redap_client::ConnectionRegistryHandle;
 use re_smart_channel::{Receiver, SmartChannelSource, SmartMessageSource};
 
@@ -146,9 +146,8 @@ impl LogDataSource {
     pub fn stream(
         self,
         connection_registry: &ConnectionRegistryHandle,
-        on_ui_cmd: Option<Box<dyn Fn(re_redap_client::UiCommand) + Send + Sync>>,
         on_msg: Option<Box<dyn Fn() + Send + Sync>>,
-    ) -> anyhow::Result<Receiver<LogMsg>> {
+    ) -> anyhow::Result<Receiver<DataSourceMessage>> {
         re_tracing::profile_function!();
 
         match self {
@@ -254,7 +253,7 @@ impl LogDataSource {
                 let stream_partition = async move {
                     let client = connection_registry.client(uri_clone.origin.clone()).await?;
                     re_redap_client::stream_blueprint_and_partition_from_server(
-                        client, tx, uri_clone, on_ui_cmd, on_msg,
+                        client, tx, uri_clone, on_msg,
                     )
                     .await
                 };

crates/store/re_data_source/src/load_stdin.rs

@@ -1,11 +1,11 @@
-use re_log_types::LogMsg;
+use re_log_types::DataSourceMessage;
 use re_smart_channel::Sender;
 
 /// Asynchronously loads RRD data streaming in from standard input.
 ///
 /// This fails synchronously iff the standard input stream could not be opened, otherwise errors
 /// are handled asynchronously (as in: they're logged).
-pub fn load_stdin(tx: Sender<LogMsg>) -> anyhow::Result<()> {
+pub fn load_stdin(tx: Sender<DataSourceMessage>) -> anyhow::Result<()> {
     let stdin = std::io::BufReader::new(std::io::stdin());
     let decoder = re_log_encoding::decoder::Decoder::new_concatenated(stdin)?;
 
@@ -20,7 +20,7 @@ pub fn load_stdin(tx: Sender<LogMsg>) -> anyhow::Result<()> {
                     continue;
                 }
             };
-            if tx.send(msg).is_err() {
+            if tx.send(msg.into()).is_err() {
                 break; // The other end has decided to hang up, not our problem.
             }
         }

crates/store/re_grpc_client/src/read.rs

@@ -1,7 +1,7 @@
+use re_log_types::DataSourceMessage;
 use tokio_stream::StreamExt as _;
 
 use re_log_encoding::protobuf_conversions::log_msg_from_proto;
-use re_log_types::LogMsg;
 use re_protos::sdk_comms::v1alpha1::ReadMessagesRequest;
 use re_protos::sdk_comms::v1alpha1::ReadMessagesResponse;
 use re_protos::sdk_comms::v1alpha1::message_proxy_service_client::MessageProxyServiceClient;
@@ -16,7 +16,7 @@ use crate::TonicStatusError;
 pub fn stream(
     uri: re_uri::ProxyUri,
     on_msg: Option<Box<dyn Fn() + Send + Sync>>,
-) -> re_smart_channel::Receiver<LogMsg> {
+) -> re_smart_channel::Receiver<DataSourceMessage> {
     re_log::debug!("Loading {uri} via gRPC…");
 
     let (tx, rx) = re_smart_channel::smart_channel(
@@ -35,7 +35,7 @@ pub fn stream(
 
 async fn stream_async(
     uri: re_uri::ProxyUri,
-    tx: &re_smart_channel::Sender<LogMsg>,
+    tx: &re_smart_channel::Sender<DataSourceMessage>,
     on_msg: Option<Box<dyn Fn() + Send + Sync>>,
 ) -> Result<(), StreamError> {
     let mut client = {
@@ -78,7 +78,7 @@ async fn stream_async(
                     re_sorbet::timestamp_metadata::now_timestamp(),
                 );
 
-                if tx.send(log_msg).is_err() {
+                if tx.send(log_msg.into()).is_err() {
                     re_log::debug!("gRPC stream smart channel closed");
                     break;
                 }

crates/store/re_grpc_server/src/lib.rs

@@ -14,7 +14,7 @@ use tower_http::cors::CorsLayer;
 
 use re_byte_size::SizeBytes;
 use re_log_encoding::codec::wire::decoder::Decode as _;
-use re_log_types::TableMsg;
+use re_log_types::{DataSourceMessage, TableMsg};
 use re_protos::sdk_comms::v1alpha1::{
     ReadTablesRequest, ReadTablesResponse, WriteMessagesRequest, WriteTableRequest,
     WriteTableResponse,
@@ -274,7 +274,7 @@ pub fn spawn_from_rx_set(
     addr: SocketAddr,
     options: ServerOptions,
     shutdown: shutdown::Shutdown,
-    rxs: re_smart_channel::ReceiveSet<re_log_types::LogMsg>,
+    rxs: re_smart_channel::ReceiveSet<re_log_types::DataSourceMessage>,
 ) {
     let message_proxy = MessageProxy::new(options);
     let event_tx = message_proxy.event_tx.clone();
@@ -317,6 +317,16 @@ pub fn spawn_from_rx_set(
                 continue;
             };
 
+            let msg = match msg {
+                DataSourceMessage::LogMsg(log_msg) => log_msg,
+                DataSourceMessage::UiCommand(ui_command) => {
+                    re_log::warn!(
+                        "Received a UI command, grpc server can't forward these yet: {ui_command:?}"
+                    );
+                    continue;
+                }
+            };
+
             let msg = match re_log_encoding::protobuf_conversions::log_msg_to_proto(
                 msg,
                 re_log_encoding::Compression::LZ4,
@@ -352,7 +362,7 @@ pub fn spawn_with_recv(
     options: ServerOptions,
     shutdown: shutdown::Shutdown,
 ) -> (
-    re_smart_channel::Receiver<re_log_types::LogMsg>,
+    re_smart_channel::Receiver<re_log_types::DataSourceMessage>,
     crossbeam::channel::Receiver<re_log_types::TableMsg>,
 ) {
     let uri = re_uri::ProxyUri::new(re_uri::Origin::from_scheme_and_socket_addr(
@@ -402,7 +412,7 @@ pub fn spawn_with_recv(
                         re_sorbet::timestamp_metadata::now_timestamp(),
                     );
 
-                    if channel_log_tx.send(log_msg).is_err() {
+                    if channel_log_tx.send(log_msg.into()).is_err() {
                         re_log::debug!(
                             "message proxy smart channel receiver closed, closing sender"
                         );

crates/store/re_log_encoding/src/stream_rrd_from_http.rs

@@ -3,7 +3,7 @@ use std::ops::ControlFlow;
 use std::sync::Arc;
 
 use re_log::ResultExt as _;
-use re_log_types::LogMsg;
+use re_log_types::{DataSourceMessage, LogMsg};
 
 /// Stream an rrd file from a HTTP server.
 ///
@@ -15,7 +15,7 @@ pub fn stream_rrd_from_http_to_channel(
     url: String,
     follow: bool,
     on_msg: Option<Box<dyn Fn() + Send + Sync>>,
-) -> re_smart_channel::Receiver<LogMsg> {
+) -> re_smart_channel::Receiver<DataSourceMessage> {
     let (tx, rx) = re_smart_channel::smart_channel(
         re_smart_channel::SmartMessageSource::RrdHttpStream { url: url.clone() },
         re_smart_channel::SmartChannelSource::RrdHttpStream {
@@ -31,7 +31,7 @@ pub fn stream_rrd_from_http_to_channel(
             }
             match msg {
                 HttpMessage::LogMsg(msg) => {
-                    if tx.send(msg).is_ok() {
+                    if tx.send(msg.into()).is_ok() {
                         ControlFlow::Continue(())
                     } else {
                         re_log::info_once!("Closing connection to {url}");

crates/store/re_log_types/src/data_source_message.rs

@@ -0,0 +1,50 @@
+// TODO(andreas): Conceptually these should go to `re_data_source`.
+// However, `re_data_source` depends on everything that _implements_ a datasource, therefore we would get a circular dependency!
+
+use crate::{AbsoluteTimeRange, LogMsg, StoreId, TimelineName, impl_into_enum};
+
+/// Message from a data source.
+///
+/// May contain limited UI commands for instrumenting the state of the receiving end.
+#[derive(Clone, Debug)]
+pub enum DataSourceMessage {
+    /// See [`LogMsg`].
+    LogMsg(LogMsg),
+
+    /// A UI command that has to be ordered relative to [`LogMsg`]s.
+    ///
+    /// Non-ui receivers can safely ignore these.
+    UiCommand(DataSourceUiCommand),
+}
+
+impl_into_enum!(LogMsg, DataSourceMessage, LogMsg);
+impl_into_enum!(DataSourceUiCommand, DataSourceMessage, UiCommand);
+
+/// UI commands issued when streaming in datasets.
+///
+/// If you're not in a ui context you can safely ignore these.
+#[derive(Clone, Debug)]
+pub enum DataSourceUiCommand {
+    /// Mark a time range as valid.
+    ///
+    /// Everything outside can still be navigated to, but will be considered potentially lacking some data and therefore "invalid".
+    /// Visually, it is outside of the normal time range and shown greyed out.
+    ///
+    /// If timeline is `None`, this signals that all timelines are considered to be valid entirely.
+    AddValidTimeRange {
+        store_id: StoreId,
+
+        /// If `None`, signals that all timelines are entirely valid.
+        timeline: Option<TimelineName>,
+        time_range: AbsoluteTimeRange,
+    },
+
+    /// Navigate to time/entities/anchors/etc. that are set in a `re_uri::Fragment`.
+    SetUrlFragment {
+        store_id: StoreId,
+
+        /// Uri fragment, see `re_uri::Fragment` on how to parse it.
+        // Not using `re_uri::Fragment` to avoid further dependency entanglement.
+        fragment: String, //re_uri::Fragment,
+    },
+}

crates/store/re_log_types/src/lib.rs

@@ -18,6 +18,7 @@
 //! `foo.transform * foo/bar.transform * foo/bar/baz.transform`.
 
 pub mod arrow_msg;
+mod data_source_message;
 mod entry_id;
 pub mod example_components;
 pub mod hash;
@@ -39,6 +40,7 @@ use re_byte_size::SizeBytes;
 
 pub use self::{
     arrow_msg::{ArrowMsg, ArrowRecordBatchReleaseCallback},
+    data_source_message::{DataSourceMessage, DataSourceUiCommand},
     entry_id::{EntryId, EntryIdOrName},
     index::{
         AbsoluteTimeRange, AbsoluteTimeRangeF, Duration, NonMinI64, TimeCell, TimeInt, TimePoint,

crates/store/re_redap_client/Cargo.toml

@@ -34,9 +34,9 @@ all-features = true
 re_arrow_util.workspace = true
 re_auth.workspace = true
 re_chunk.workspace = true
-re_log.workspace = true
 re_log_encoding = { workspace = true, features = ["encoder", "decoder"] }
 re_log_types.workspace = true
+re_log.workspace = true
 re_protos.workspace = true
 re_smart_channel.workspace = true
 re_uri.workspace = true