Address PR review comments

This commit addresses feedback from PR rerun-io#11798:

1. Use get_or_insert_with pattern for worker initialization
   - Replace manual if-check with idiomatic Option::get_or_insert_with
   - Better expresses intent and reduces nesting

2. Encapsulate ingestion worker processing in EntityDb
   - Add on_frame_start() method to EntityDb for worker processing
   - Add on_frame_start() to StoreBundle to process all workers
   - Add on_frame_start() to StoreHub as public interface
   - Remove process_ingestion_worker_output() from App
   - Move all worker-related logic out of App into EntityDb layer
   - App now calls store_hub.on_frame_start() instead of manually
     polling workers and processing output

3. Introduce IngestionWorkerResults type alias
   - Simplify complex return types
   - Export from re_entity_db crate

Benefits:
- Cleaner separation of concerns
- EntityDb fully encapsulates its ingestion worker
- Easier to understand and maintain
- App layer no longer needs to know about worker internals

Author: joelreymont

crates/store/re_entity_db/src/entity_db.rs

@@ -680,30 +680,32 @@ impl EntityDb {
         re_tracing::profile_function!();
 
         // Lazy-init worker on first use
-        if self.ingestion_worker.is_none() {
+        let worker = self.ingestion_worker.get_or_insert_with(|| {
             re_log::debug!("Creating ingestion worker for store {:?}", self.store_id);
-            self.ingestion_worker = Some(crate::ingestion_worker::IngestionWorker::new());
-        }
+            crate::ingestion_worker::IngestionWorker::new()
+        });
 
-        if let Some(worker) = &self.ingestion_worker {
-            worker.submit_arrow_msg_blocking(
-                self.store_id.clone(),
-                arrow_msg,
-                channel_source,
-                msg_will_add_new_store,
-            );
-        }
+        worker.submit_arrow_msg_blocking(
+            self.store_id.clone(),
+            arrow_msg,
+            channel_source,
+            msg_will_add_new_store,
+        );
     }
 
-    /// Poll the background ingestion worker for processed chunks (native only).
+    /// Process pending work from the background ingestion worker (native only).
+    ///
+    /// This should be called once per frame to process chunks that have been
+    /// converted by the background worker. The processed chunks are automatically
+    /// added to this `EntityDb`'s store.
     ///
-    /// Returns a vector of (`ProcessedChunk`, `was_empty_before`, `StoreEvents`) tuples.
-    /// Each chunk has been successfully added to the store.
+    /// Returns a vector of (`ProcessedChunk`, `was_empty_before`, `StoreEvents`) tuples
+    /// for each chunk that was successfully added.
     ///
-    /// On native: Returns chunks processed by the background worker
+    /// On native: Polls and processes chunks from the background worker
     /// On Wasm: This method doesn't exist (compile error if called)
     #[cfg(not(target_arch = "wasm32"))]
-    pub fn poll_worker_output(
+    pub fn on_frame_start(
         &mut self,
     ) -> Vec<(
         crate::ingestion_worker::ProcessedChunk,

crates/store/re_entity_db/src/lib.rs

@@ -26,6 +26,9 @@ pub use self::{
     versioned_instance_path::{VersionedInstancePath, VersionedInstancePathHash},
 };
 
+#[cfg(not(target_arch = "wasm32"))]
+pub use self::store_bundle::IngestionWorkerResults;
+
 #[doc(no_inline)]
 pub use re_log_types::{EntityPath, EntityPathPart, TimeInt, Timeline};
 

crates/store/re_entity_db/src/store_bundle.rs

@@ -4,6 +4,17 @@ use re_log_types::{StoreId, StoreKind};
 
 use crate::EntityDb;
 
+/// Result type for ingestion worker output processing.
+#[cfg(not(target_arch = "wasm32"))]
+pub type IngestionWorkerResults = indexmap::IndexMap<
+    StoreId,
+    Vec<(
+        crate::ingestion_worker::ProcessedChunk,
+        bool,
+        Result<Vec<re_chunk_store::ChunkStoreEvent>, crate::Error>,
+    )>,
+>;
+
 #[derive(thiserror::Error, Debug)]
 pub enum StoreLoadError {
     #[error(transparent)]
@@ -145,4 +156,29 @@ impl StoreBundle {
 
         entity_dbs.first().map(|db| db.store_id().clone())
     }
+
+    /// Process pending work from background ingestion workers for all stores (native only).
+    ///
+    /// This should be called once per frame to process chunks from all `EntityDb` workers.
+    /// Each `EntityDb` polls its worker and adds processed chunks to its store.
+    ///
+    /// Returns a map of [`StoreId`] to a vector of processing results for that store.
+    ///
+    /// On native: Processes worker output for all `EntityDb`s
+    /// On Wasm: This method doesn't exist (compile error if called)
+    #[cfg(not(target_arch = "wasm32"))]
+    pub fn on_frame_start(&mut self) -> IngestionWorkerResults {
+        re_tracing::profile_function!();
+
+        let mut all_results = indexmap::IndexMap::new();
+
+        for (store_id, entity_db) in &mut self.recording_store {
+            let results = entity_db.on_frame_start();
+            if !results.is_empty() {
+                all_results.insert(store_id.clone(), results);
+            }
+        }
+
+        all_results
+    }
 }

crates/viewer/re_viewer/src/app.rs

@@ -2497,7 +2497,42 @@ impl App {
         });
 
         #[cfg(not(target_arch = "wasm32"))]
-        self.process_ingestion_worker_output(store_hub, egui_ctx);
+        {
+            // Process background ingestion workers for all stores
+            let all_results = store_hub.on_frame_start();
+
+            for (_store_id, results) in all_results {
+                for (processed, was_empty_before, entity_db_add_result) in results {
+                    self.finalize_arrow_chunk_ingestion(
+                        store_hub,
+                        egui_ctx,
+                        processed.channel_source.as_ref(),
+                        &processed.store_id,
+                        processed.msg_will_add_new_store,
+                        was_empty_before,
+                        entity_db_add_result,
+                    );
+                }
+            }
+
+            // Check if all messages have been ingested after processing worker output
+            if self.rrd_loading_metrics.is_tracking {
+                let all_ingested = self.rrd_loading_metrics.total_messages > 0
+                    && self.rrd_loading_metrics.total_messages
+                        == self.rrd_loading_metrics.total_ingested;
+
+                if all_ingested
+                    && self.rrd_loading_metrics.all_messages_ingested.is_none()
+                    && self.rx_log.sources().is_empty()
+                {
+                    re_log::info!(
+                        "🎉 All {} messages have been ingested!",
+                        self.rrd_loading_metrics.total_messages
+                    );
+                    self.rrd_loading_metrics.mark_all_ingested();
+                }
+            }
+        }
 
         let start = web_time::Instant::now();
 
@@ -2765,99 +2800,6 @@ impl App {
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
-    fn process_ingestion_worker_output(
-        &mut self,
-        store_hub: &mut StoreHub,
-        egui_ctx: &egui::Context,
-    ) {
-        re_tracing::profile_function!();
-
-        // Debug: Log every time this function is called
-        static CALL_COUNT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
-        let count = CALL_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        if count % 100 == 0 || self.rrd_loading_metrics.total_messages > 100000 {
-            re_log::debug!(
-                "[process_ingestion_worker_output] Called #{}, is_tracking={}, all_ingested={:?}, total_msg={}, total_ingested={}",
-                count,
-                self.rrd_loading_metrics.is_tracking,
-                self.rrd_loading_metrics.all_messages_ingested.is_some(),
-                self.rrd_loading_metrics.total_messages,
-                self.rrd_loading_metrics.total_ingested
-            );
-        }
-
-        // Collect store IDs first to avoid borrowing issues
-        let store_ids: Vec<StoreId> = store_hub
-            .store_bundle()
-            .entity_dbs()
-            .map(|entity_db| entity_db.store_id().clone())
-            .collect();
-
-        // Poll each EntityDb's worker for processed chunks
-        for store_id in store_ids {
-            let results = {
-                let entity_db = store_hub.entity_db_mut(&store_id);
-                entity_db.poll_worker_output()
-            };
-
-            for (processed, was_empty_before, entity_db_add_result) in results {
-                self.finalize_arrow_chunk_ingestion(
-                    store_hub,
-                    egui_ctx,
-                    processed.channel_source.as_ref(),
-                    &processed.store_id,
-                    processed.msg_will_add_new_store,
-                    was_empty_before,
-                    entity_db_add_result,
-                );
-            }
-        }
-
-        // Check if all messages have been ingested after processing worker output
-        // Debug: Always evaluate and log guard conditions
-        let is_tracking = self.rrd_loading_metrics.is_tracking;
-        let all_ingested_is_none = self.rrd_loading_metrics.all_messages_ingested.is_none();
-        let all_channels_closed = self.rx_log.sources().is_empty();
-
-        if self.rrd_loading_metrics.total_messages > 100000 {
-            re_log::debug!(
-                "[process_ingestion_worker_output] Detection guards: is_tracking={}, all_ingested_is_none={}, channels_closed={}, total_msg={}, total_ingested={}",
-                is_tracking,
-                all_ingested_is_none,
-                all_channels_closed,
-                self.rrd_loading_metrics.total_messages,
-                self.rrd_loading_metrics.total_ingested
-            );
-        }
-
-        if is_tracking && all_ingested_is_none {
-            if all_channels_closed && self.rrd_loading_metrics.total_messages > 0 {
-                re_log::debug!(
-                    "Ingestion check: total_msg={}, total_ingested={}",
-                    self.rrd_loading_metrics.total_messages,
-                    self.rrd_loading_metrics.total_ingested
-                );
-            }
-
-            if all_channels_closed
-                && self.rrd_loading_metrics.total_messages > 0
-                && self.rrd_loading_metrics.total_messages
-                    == self.rrd_loading_metrics.total_ingested
-            {
-                re_log::info!(
-                    "✓ All {} messages ingested! Marking ingestion complete.",
-                    self.rrd_loading_metrics.total_ingested
-                );
-                self.rrd_loading_metrics.mark_all_ingested();
-            }
-        } else if self.rrd_loading_metrics.total_messages > 100000 {
-            re_log::debug!(
-                "[process_ingestion_worker_output] Guard conditions failed - skipping detection"
-            );
-        }
-    }
-
     #[cfg(not(target_arch = "wasm32"))]
     #[expect(clippy::too_many_arguments)]
     fn finalize_arrow_chunk_ingestion(

crates/viewer/re_viewer_context/src/store_hub.rs

@@ -295,6 +295,20 @@ impl StoreHub {
         self.store_bundle.get(store_id)
     }
 
+    /// Process pending work from background ingestion workers for all stores (native only).
+    ///
+    /// This should be called once per frame to process chunks from all `EntityDb` workers.
+    /// Each `EntityDb` polls its worker and adds processed chunks to its store.
+    ///
+    /// Returns a map of [`StoreId`] to a vector of processing results for that store.
+    ///
+    /// On native: Processes worker output for all `EntityDb`s
+    /// On Wasm: This method doesn't exist (compile error if called)
+    #[cfg(not(target_arch = "wasm32"))]
+    pub fn on_frame_start(&mut self) -> re_entity_db::IngestionWorkerResults {
+        self.store_bundle.on_frame_start()
+    }
+
     // ---------------------
     // Add and remove stores