fix(python-node): seed the pool-id counter randomly so restarts don't collide

apis/python/node/src/lib.rs

@@ -134,8 +134,69 @@ def basicConfig(*pargs, **kwargs):
 static CUDA_HELPERS: LazyLock<std::sync::Mutex<Option<Py<PyModule>>>> =
     LazyLock::new(|| std::sync::Mutex::new(None));
 
+/// A random `u64` seed derived from the standard library's `RandomState`,
+/// which is seeded from the OS once per process. Used to make process-local
+/// counters unique across process restarts without pulling in a new
+/// dependency (dora-rs/dora#3015).
+fn random_u64_seed() -> u64 {
+    use std::hash::{BuildHasher, Hasher};
+    // `RandomState::new()` picks fresh random keys from the OS; hashing no
+    // input and finishing yields a value derived from those keys, so the
+    // result differs from process to process.
+    std::hash::RandomState::new().build_hasher().finish()
+}
+
+/// Build a memory pool's shared-memory name from the
+/// `(dataflow_id, node_id, counter)` triple.
+///
+/// This is the single source of truth for the on-wire pool-id format
+/// (`dora_pool_{dataflow_id}_{node_id}_{counter}`); [`parse_pool_counter`]
+/// recovers the counter from it. Keeping the two together means a change to
+/// the format can only be made in one place (dora-rs/dora#3015).
+fn pool_shmem_name(
+    dataflow_id: impl std::fmt::Display,
+    node_id: impl std::fmt::Display,
+    counter: u64,
+) -> String {
+    format!("dora_pool_{dataflow_id}_{node_id}_{counter}")
+}
+
+/// Recover the pool counter from a pool id of the form
+/// `dora_pool_{dataflow_id}_{node_id}_{counter}` or `pool_{node_id}_{counter}`.
+///
+/// The counter is always the final `_`-separated component — node ids may
+/// themselves contain `_` (legal in dora node ids) — so the last segment is
+/// taken and parsed as a `u64`. Returns `None` when the last segment is not a
+/// `u64`. Inverse of [`pool_shmem_name`]'s counter component.
+fn parse_pool_counter(pool_id: &str) -> Option<u64> {
+    pool_id
+        .rsplit_once('_')
+        .and_then(|(_, counter)| counter.parse::<u64>().ok())
+}
+
 /// Counter to make pinned memory buffer IDs unique across registrations.
-static PINNED_COUNTER: LazyLock<std::sync::Mutex<u64>> = LazyLock::new(|| std::sync::Mutex::new(0));
+///
+/// Seeded with a random value per process rather than `0` (dora-rs/dora#3015).
+/// The counter is combined with the `(dataflow_id, node_id)` pair — both of
+/// which are stable across a crash-restart — into the pool's shared-memory
+/// name, so a deterministic `0` seed makes a restarted node re-derive the
+/// exact name its previous incarnation used. If the old segment is still live
+/// (its receiver is still reading it, so #2881's reclaim deliberately kept it),
+/// `ShmemConf::create()` collides and the restarted node can never register its
+/// pool — a crash-restart loop that never recovers. A random per-incarnation
+/// seed keeps the id unique across restarts while the component stays a plain
+/// `u64`, so both existing name parsers keep working unchanged.
+///
+/// Trade-off (dora-rs/dora#3015 review): with the collision gone, a repeatedly
+/// crashing node whose receiver never calls `free_memory_pool` now leaks the
+/// previous incarnation's registry entry and `/dev/shm` segment on each
+/// restart, instead of failing fast on the first. `cleanup_orphans` only runs
+/// at dataflow spawn, and the daemon registry is capped, so a pathological
+/// crash loop can eventually exhaust it. The complete fix is an owner-death
+/// pool reclaim (tracked as follow-up); this change trades fail-fast for
+/// recover-on-transient-crash, which is the common case.
+static PINNED_COUNTER: LazyLock<std::sync::Mutex<u64>> =
+    LazyLock::new(|| std::sync::Mutex::new(random_u64_seed()));
 
 /// Tracks freed pool buffer IDs so the DORADMA fast path can detect
 /// read-after-free. Entries are inserted on free_memory_pool and never
@@ -301,6 +362,62 @@ mod pin_tests {
     }
 }
 
+#[cfg(test)]
+mod pool_id_tests {
+    use super::*;
+
+    /// dora-rs/dora#3015: the per-process pool counter is now seeded from
+    /// `random_u64_seed()` so a restarted node does not re-derive its previous
+    /// incarnation's shared-memory name and collide on a still-live segment.
+    /// The seed must vary from call to call (a proxy for varying from process
+    /// to process — `RandomState` reseeds on each `new()`).
+    #[test]
+    fn random_seed_is_not_constant() {
+        let seeds: std::collections::HashSet<u64> = (0..8).map(|_| random_u64_seed()).collect();
+        assert!(
+            seeds.len() > 1,
+            "random_u64_seed() must not return a constant value"
+        );
+    }
+
+    /// The on-wire pool-id format is a contract shared with the daemon
+    /// (`MemoryPoolManager` derives the same name). Pin the exact literal so a
+    /// change to `pool_shmem_name` that both in-process halves would still
+    /// agree on, yet breaks that cross-process contract, is caught here.
+    #[test]
+    fn pool_shmem_name_has_the_expected_wire_format() {
+        assert_eq!(
+            pool_shmem_name("dataflow-uuid", "cam", 7),
+            "dora_pool_dataflow-uuid_cam_7"
+        );
+    }
+
+    /// A pool name built from a large, random-seeded counter must round-trip
+    /// back through the *production* parser — including for a `node_id` that
+    /// itself contains underscores, where only taking the last segment as the
+    /// counter is correct. Exercising the real `pool_shmem_name` /
+    /// `parse_pool_counter` pair means a regression in either can turn this
+    /// RED (dora-rs/dora#3015).
+    #[test]
+    fn large_counter_round_trips_through_the_production_parser() {
+        for (node_id, counter) in [("my_node", u64::MAX - 3), ("cam_left", 1), ("plain", 0)] {
+            let shmem_name = pool_shmem_name("dataflow-uuid", node_id, counter);
+            assert_eq!(
+                parse_pool_counter(&shmem_name),
+                Some(counter),
+                "counter must round-trip for node id {node_id:?}"
+            );
+        }
+    }
+
+    /// A pool id whose final segment is not a `u64` yields `None` rather than a
+    /// wrong counter — the fall-back-to-daemon path relies on this.
+    #[test]
+    fn parse_pool_counter_rejects_a_non_numeric_tail() {
+        assert_eq!(parse_pool_counter("dora_pool_df_node_notanumber"), None);
+    }
+}
+
 // ---------------------------------------------------------------------------
 // GPU transport-path classification — pure decision logic extractable
 // from CUDA-runtime-embedded code so the full matrix can be exercised in
@@ -1969,16 +2086,15 @@ impl Node {
             );
         }
 
-        // Generate unique pool counter for this registration
+        // Generate unique pool counter for this registration. `wrapping_add`
+        // guards the (astronomically unlikely) overflow now that the counter
+        // starts from a random seed rather than 0 (dora-rs/dora#3015).
         let pool_counter = {
             let mut c = PINNED_COUNTER.lock().unwrap_or_else(|e| e.into_inner());
-            *c += 1;
+            *c = c.wrapping_add(1);
             *c
         };
-        let shmem_name = format!(
-            "dora_pool_{}_{}_{}",
-            self.dataflow_id, self.node_id, pool_counter
-        );
+        let shmem_name = pool_shmem_name(&self.dataflow_id, &self.node_id, pool_counter);
 
         let header_meta = PyDict::new(py);
         header_meta.set_item("size", size)?;
@@ -2475,9 +2591,7 @@ impl Node {
         if buffer_id.starts_with("pool_") {
             // Extract counter from the last underscore segment — node_id
             // may legitimately contain underscores.
-            if let Some((_, counter_str)) = buffer_id.rsplit_once('_')
-                && let Ok(counter) = counter_str.parse::<u64>()
-            {
+            if let Some(counter) = parse_pool_counter(&buffer_id) {
                 // Try PINNED_POOL cache first to avoid per-iteration mmap/munmap.
                 // register_memory_pool already stored the Shmem here; taking it
                 // prevents munmap, and storing it back keeps the mapping alive.
@@ -2503,10 +2617,7 @@ impl Node {
                     } else {
                         // Cache miss: open via ShmemConf, wrap immediately
                         // so the mapping stays alive until post-write re-insert.
-                        let shmem_name = format!(
-                            "dora_pool_{}_{}_{}",
-                            self.dataflow_id, self.node_id, counter
-                        );
+                        let shmem_name = pool_shmem_name(&self.dataflow_id, &self.node_id, counter);
                         match ShmemConf::new().os_id(&shmem_name).open() {
                             Ok(shmem) => {
                                 let cap = shmem.len();
@@ -2830,9 +2941,7 @@ impl Node {
 
                         if ipc_present == 1 && !is_cuda {
                             // Extract counter for the DMA slot from buffer_id.
-                            let slow_counter = buffer_id
-                                .rsplit_once('_')
-                                .and_then(|(_, c)| c.parse::<u64>().ok());
+                            let slow_counter = parse_pool_counter(&buffer_id);
                             let gen_ptr = unsafe { shmem_ptr.add(96) as *mut u64 };
                             let pre_write_gen = unsafe { seqlock_begin_if_even(gen_ptr) };
                             let mut copy_ok = true;
@@ -2877,19 +2986,17 @@ impl Node {
                                 let bound = helpers.bind(py);
                                 // Slow path transit look-up: PINNED_POOL
                                 // (contrast fast path which uses store_back).
-                                let (transit_ptr, pool_device) = if let Some((_, counter_str)) =
-                                    buffer_id.rsplit_once('_')
-                                    && let Ok(c) = counter_str.parse::<u64>()
-                                {
-                                    PINNED_POOL
-                                        .lock()
-                                        .unwrap_or_else(|e| e.into_inner())
-                                        .get(&c)
-                                        .map(|s| (s.transit_ptr, s.pool_device))
-                                        .unwrap_or((0, 0))
-                                } else {
-                                    (0, 0)
-                                };
+                                let (transit_ptr, pool_device) =
+                                    if let Some(c) = parse_pool_counter(&buffer_id) {
+                                        PINNED_POOL
+                                            .lock()
+                                            .unwrap_or_else(|e| e.into_inner())
+                                            .get(&c)
+                                            .map(|s| (s.transit_ptr, s.pool_device))
+                                            .unwrap_or((0, 0))
+                                    } else {
+                                        (0, 0)
+                                    };
                                 let write_path = classify_write_path(
                                     ipc_present,
                                     /*is_cuda=*/ true,
@@ -2906,10 +3013,7 @@ impl Node {
                                             .call_method1(
                                                 "_transit_copy_gpu_buf",
                                                 (
-                                                    buffer_id
-                                                        .rsplit_once('_')
-                                                        .and_then(|(_, cs)| cs.parse::<u64>().ok())
-                                                        .unwrap_or(0),
+                                                    parse_pool_counter(&buffer_id).unwrap_or(0),
                                                     ptr_val,
                                                     sender_dev,
                                                     transit_ptr,
@@ -2923,10 +3027,7 @@ impl Node {
                                         .call_method1(
                                             "_cuda_memcpy_gpu_buf",
                                             (
-                                                buffer_id
-                                                    .rsplit_once('_')
-                                                    .and_then(|(_, cs)| cs.parse::<u64>().ok())
-                                                    .unwrap_or(0),
+                                                parse_pool_counter(&buffer_id).unwrap_or(0),
                                                 ptr_val,
                                                 size,
                                             ),
@@ -3562,12 +3663,8 @@ impl Node {
         // Format: "pool_{node_id}_{counter}".
         // Use rsplit to extract the counter from the end — the node_id
         // portion may itself contain underscores (legal in dora node ids).
-        let counter: u64 = match buffer_id.rsplit_once('_') {
-            Some((_, c)) => match c.parse() {
-                Ok(c) => c,
-                Err(_) => return Ok(None),
-            },
-            None => return Ok(None),
+        let Some(counter) = parse_pool_counter(buffer_id) else {
+            return Ok(None);
         };
         let pool_node_id = buffer_id
             .strip_prefix("pool_")
@@ -3587,10 +3684,7 @@ impl Node {
             }
         }
 
-        let shmem_name = format!(
-            "dora_pool_{}_{}_{}",
-            self.dataflow_id, pool_node_id, counter
-        );
+        let shmem_name = pool_shmem_name(&self.dataflow_id, pool_node_id, counter);
 
         // Open shared memory
         let shmem = match ShmemConf::new().os_id(&shmem_name).open() {