Fix a whole lot of crashes, all at once (#1780)

crates/re_log_types/src/arrow_msg.rs

@@ -87,7 +87,14 @@ impl<'de> serde::Deserialize<'de> for ArrowMsg {
                     (table_id, timepoint_min, buf)
                 {
                     let mut cursor = std::io::Cursor::new(buf);
-                    let metadata = read_stream_metadata(&mut cursor).unwrap();
+                    let metadata = match read_stream_metadata(&mut cursor) {
+                        Ok(metadata) => metadata,
+                        Err(err) => {
+                            return Err(serde::de::Error::custom(format!(
+                                "Failed to read stream metadata: {err}"
+                            )))
+                        }
+                    };
                     let mut stream = StreamReader::new(cursor, metadata, None);
                     let chunk = stream
                         .find_map(|state| match state {

crates/re_viewer/src/ui/view_text/scene.rs

@@ -15,7 +15,7 @@ use super::ui::ViewTextFilters;
 #[derive(Debug, Clone)]
 pub struct TextEntry {
     // props
-    pub msg_id: MsgId,
+    pub msg_id: Option<MsgId>,
 
     pub entity_path: EntityPath,
 
@@ -86,7 +86,7 @@ impl SceneText {
 
                         if is_visible {
                             self.text_entries.push(TextEntry {
-                                msg_id: msg_id.unwrap(), // always present
+                                msg_id,
                                 entity_path: entity_path.clone(),
                                 time: time.map(|time| time.as_i64()),
                                 color: color.map(|c| c.to_array()),

crates/re_viewer/src/ui/view_text/ui.rs

@@ -188,7 +188,7 @@ fn get_time_point(ctx: &ViewerContext<'_>, entry: &TextEntry) -> Option<TimePoin
         .log_db
         .entity_db
         .data_store
-        .get_msg_metadata(&entry.msg_id)
+        .get_msg_metadata(&entry.msg_id?)
     {
         Some(time_point.clone())
     } else {

crates/re_ws_comms/src/server.rs

@@ -121,7 +121,7 @@ async fn handle_connection(
     tcp_stream: TcpStream,
     history: Arc<Mutex<Vec<Arc<[u8]>>>>,
 ) -> tungstenite::Result<()> {
-    let ws_stream = accept_async(tcp_stream).await.expect("Failed to accept");
+    let ws_stream = accept_async(tcp_stream).await?;
     let (mut ws_sender, mut ws_receiver) = ws_stream.split();
 
     // Re-sending packet history - this is not water tight, but better than nothing.

crates/rerun/src/run.rs

@@ -532,7 +532,7 @@ pub fn setup_ctrl_c_handler() -> (tokio::sync::broadcast::Receiver<()>, Arc<Atom
     let shutdown = shutdown_return.clone();
     ctrlc::set_handler(move || {
         re_log::debug!("Ctrl-C detected, shutting down.");
-        sender.send(()).unwrap();
+        sender.send(()).ok();
         shutdown.store(true, std::sync::atomic::Ordering::Relaxed);
     })
     .expect("Error setting Ctrl-C handler");