Fix initializing two recordings with the same recording id causing SDK hangs

rerun_py/src/dataframe.rs

@@ -36,6 +36,7 @@ use re_sorbet::{
     ColumnSelector, ComponentColumnSelector, SorbetColumnDescriptors, TimeColumnSelector,
 };
 
+use super::utils::py_rerun_warn_cstr;
 use crate::catalog::to_py_err;
 use crate::{catalog::PyCatalogClientInternal, utils::get_tokio_runtime};
 
@@ -58,16 +59,6 @@ pub(crate) fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
     Ok(())
 }
 
-fn py_rerun_warn(msg: &std::ffi::CStr) -> PyResult<()> {
-    Python::with_gil(|py| {
-        let warning_type = PyModule::import(py, "rerun")?
-            .getattr("error_utils")?
-            .getattr("RerunWarning")?;
-        PyErr::warn(py, &warning_type, msg, 0)?;
-        Ok(())
-    })
-}
-
 /// The descriptor of an index column.
 ///
 /// Index columns contain the index values for when the data was updated. They
@@ -816,7 +807,7 @@ impl PyRecordingView {
                     && all_contents_are_static
                     && any_selected_data_is_static
                 {
-                    py_rerun_warn(c"RecordingView::select: tried to select static data, but no non-static contents generated an index value on this timeline. No results will be returned. Either include non-static data or consider using `select_static()` instead.")?;
+                    py_rerun_warn_cstr(c"RecordingView::select: tried to select static data, but no non-static contents generated an index value on this timeline. No results will be returned. Either include non-static data or consider using `select_static()` instead.")?;
                 }
 
                 let schema = query_handle.schema().clone();

rerun_py/src/python_bridge.rs

@@ -14,6 +14,7 @@ use pyo3::{
     types::{PyBytes, PyDict},
 };
 
+use super::utils;
 use re_log::ResultExt as _;
 use re_log_types::LogMsg;
 use re_log_types::{BlueprintActivationCommand, EntityPathPart, StoreKind};
@@ -24,7 +25,6 @@ use re_sdk::{
     time::TimePoint,
 };
 use re_sdk::{TimeCell, external::re_log_encoding::encoder::encode_ref_as_bytes_local};
-
 #[cfg(feature = "web_viewer")]
 use re_web_viewer_server::WebViewerServerPort;
 
@@ -238,6 +238,17 @@ fn new_recording(
         default_store_id(py, StoreKind::Recording, &application_id)
     };
 
+    // NOTE: The Rust-side of the bindings must be in control of the lifetimes of the recordings!
+    // Check if recording already exists first.
+    // NOTE: This is scoped in order to release the Mutex locked by `all_recordings` as soon as possible
+    if let Some(existing_recording) = all_recordings().get(&recording_id) {
+        utils::py_rerun_warn(
+            format!("Recording with id: {} already exists, will ignore creation and return existing recording.",
+            &recording_id).as_str()
+        )?;
+        return Ok(PyRecordingStream(existing_recording.clone()));
+    }
+
     let mut batcher_config = re_chunk::ChunkBatcherConfig::from_env().unwrap_or_default();
     let on_release = |chunk| {
         GARBAGE_QUEUE.0.send(chunk).ok();

rerun_py/src/utils.rs

@@ -1,6 +1,6 @@
+use pyo3::{Python, prelude::*};
+use std::ffi::CString;
 use std::{future::Future, sync::OnceLock};
-
-use pyo3::Python;
 use tokio::runtime::Runtime;
 
 /// Utility to get the Tokio Runtime from Python
@@ -23,3 +23,20 @@ where
     let runtime: &Runtime = get_tokio_runtime();
     py.allow_threads(|| runtime.block_on(f))
 }
+
+/// Issues a warning to python runtime
+pub fn py_rerun_warn_cstr(msg: &std::ffi::CStr) -> PyResult<()> {
+    Python::with_gil(|py| {
+        let warning_type = PyModule::import(py, "rerun")?
+            .getattr("error_utils")?
+            .getattr("RerunWarning")?;
+        PyErr::warn(py, &warning_type, msg, 0)?;
+        Ok(())
+    })
+}
+
+/// Logs a warning using rerun logging system and issues the warning to python runtime.
+pub fn py_rerun_warn(msg: &str) -> PyResult<()> {
+    let cmsg = CString::new(msg)?;
+    py_rerun_warn_cstr(&cmsg)
+}

rerun_py/tests/unit/test_dataframe.py

@@ -11,14 +11,13 @@
 from rerun_bindings.types import AnyColumn, ComponentLike, ViewContentsLike
 
 APP_ID = "rerun_example_test_recording"
-RECORDING_ID = uuid.uuid4()
 
 
 def test_load_recording() -> None:
     with tempfile.TemporaryDirectory() as tmpdir:
         rrd = tmpdir + "/tmp.rrd"
 
-        with rr.RecordingStream("rerun_example_test_recording") as rec:
+        with rr.RecordingStream("rerun_example_test_recording", recording_id=uuid.uuid4()) as rec:
             rec.save(rrd)
             rec.set_time("my_index", sequence=1)
             rec.log("log", rr.TextLog("Hello"))
@@ -51,7 +50,8 @@ def setup_method(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             rrd = tmpdir + "/tmp.rrd"
 
-            with rr.RecordingStream(APP_ID, recording_id=RECORDING_ID) as rec:
+            self.expected_recording_id = uuid.uuid4()
+            with rr.RecordingStream(APP_ID, recording_id=self.expected_recording_id) as rec:
                 rec.save(rrd)
                 rec.set_time("my_index", sequence=1)
                 rec.log("points", rr.Points3D([[1, 2, 3], [4, 5, 6], [7, 8, 9]], radii=[]))
@@ -94,7 +94,7 @@ def setup_method(self) -> None:
 
     def test_recording_info(self) -> None:
         assert self.recording.application_id() == APP_ID
-        assert self.recording.recording_id() == str(RECORDING_ID)
+        assert self.recording.recording_id() == str(self.expected_recording_id)
 
     def test_schema_recording(self) -> None:
         schema: Schema = self.recording.schema()
@@ -432,7 +432,7 @@ def test_roundtrip_send(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             rrd = tmpdir + "/tmp.rrd"
 
-            with rr.RecordingStream("rerun_example_test_recording") as rec:
+            with rr.RecordingStream("rerun_example_test_recording", recording_id=uuid.uuid4()) as rec:
                 rec.save(rrd)
                 rr.dataframe.send_dataframe(df, rec=rec)
 

rerun_py/tests/unit/test_expected_warnings.py

@@ -50,3 +50,27 @@ def test_expected_warnings() -> None:
         lambda: rr.log("world/image", rr.Pinhole(focal_length=3)),
         "Must provide one of principal_point, resolution, or width/height)",
     )
+
+
+def test_init_twice() -> None:
+    """Regression test for #9948: creating the same recording twice caused hangs in the past (should instead warn)."""
+    # Always set strict mode to false in case it leaked from another test
+    rr.set_strict_mode(False)
+
+    # Using default recording id
+    rr.init("rerun_example_test_app_id")
+    recording_id = rr.get_recording_id()
+    expect_warning(
+        lambda: rr.init("rerun_example_test_app_id"),
+        f"Recording with id: {recording_id} already exists, will ignore creation and return existing recording.",
+    )
+    assert recording_id == rr.get_recording_id()
+
+    # Using a custom recording id
+    recording_id = "test_recording_id"
+    rr.init("rerun_example_test_app_id", recording_id=recording_id)
+    expect_warning(
+        lambda: rr.init("rerun_example_test_app_id", recording_id=recording_id),
+        f"Recording with id: {recording_id} already exists, will ignore creation and return existing recording.",
+    )
+    assert recording_id == rr.get_recording_id()