fix(ecstore): make local disk map initialization replace stale topology (#5734)

initialize_local_disk_maps appended pool entries to local_disk_set_drives and inserted into local_disk_map without ever clearing previous state. Every caller (both production startup entry points and all tests) passes the FULL topology, so re-initializing the same InstanceContext left the pool/set vectors sized for the stale topology and panicked with index-out-of-bounds for wider disk indices.

This surfaced as deterministic cross-test contamination under single-process cargo test: in crates/heal heal_b920_subquorum_union_test, a 4-disk test initialized pool 0 as [None; 4] on the process-level default context, and the two 8-disk tests then panicked at disk_idx 4. Each test passed alone, and CI never caught it because cargo nextest isolates every test in its own process.

Fix: clear both registries at the start of initialize_local_disk_maps so initialization is idempotent and last-topology-wins. Add a regression unit test in ecstore (process-isolation-proof, unlike the heal integration binary) that re-initializes the same context with a wider topology; it fails with the pre-fix code.

crates/ecstore/src/runtime/sources.rs

@@ -549,8 +549,13 @@ pub(crate) async fn initialize_local_disk_maps(
     endpoint_pools: EndpointServerPools,
     opt: &DiskOption,
 ) -> Result<()> {
+    // Every caller passes the FULL topology, so (re)initialization must replace
+    // any previous registration wholesale: appending would leave the pool/set
+    // vectors sized for a stale topology and panic on wider disk indices (seen
+    // as cross-test contamination under single-process `cargo test`).
     let set_drives = instance_ctx.local_disk_set_drives();
     let mut global_set_drives = set_drives.write().await;
+    global_set_drives.clear();
     for pool_eps in endpoint_pools.as_ref().iter() {
         let mut set_count_drives = Vec::with_capacity(pool_eps.set_count);
         for _ in 0..pool_eps.set_count {
@@ -562,6 +567,7 @@ pub(crate) async fn initialize_local_disk_maps(
 
     let map = instance_ctx.local_disk_map();
     let mut global_local_disk_map = map.write().await;
+    global_local_disk_map.clear();
 
     for pool_eps in endpoint_pools.as_ref().iter() {
         for ep in pool_eps.endpoints.as_ref().iter() {
@@ -727,4 +733,69 @@ mod tests {
         process_ctx.local_disk_id_map().write().await.remove(&process_sentinel);
         bootstrap_ctx.local_disk_id_map().write().await.remove(&bootstrap_sentinel);
     }
+
+    /// Re-initializing the same context with a WIDER topology must replace the
+    /// previous registration, not append to it: the stale pool-0 drive vector
+    /// (sized for the narrow topology) made `global_set_drives[0][0][disk_idx]`
+    /// panic for the wider set's higher disk indices. CI's nextest
+    /// process-per-test isolation never exercises re-init, so this pins it.
+    #[tokio::test]
+    async fn reinitializing_local_disk_maps_replaces_previous_topology() {
+        use crate::disk::DiskOption;
+        use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
+
+        let temp_dir = tempfile::tempdir().expect("reinit test directory should be created");
+        let build_pools = |label: &str, disk_count: usize| {
+            let mut endpoints = Vec::new();
+            for disk_idx in 0..disk_count {
+                let disk_path = temp_dir.path().join(format!("{label}-disk{disk_idx}"));
+                std::fs::create_dir_all(&disk_path).expect("reinit test disk should be created");
+                let mut endpoint =
+                    Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8")).expect("endpoint should parse");
+                endpoint.set_pool_index(0);
+                endpoint.set_set_index(0);
+                endpoint.set_disk_index(disk_idx);
+                endpoints.push(endpoint);
+            }
+            EndpointServerPools(vec![PoolEndpoints {
+                legacy: false,
+                set_count: 1,
+                drives_per_set: disk_count,
+                endpoints: Endpoints::from(endpoints),
+                cmd_line: format!("reinit-test-{label}"),
+                platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
+            }])
+        };
+        let opt = DiskOption {
+            cleanup: false,
+            health_check: false,
+        };
+
+        let instance_ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+        super::initialize_local_disk_maps(&instance_ctx, build_pools("narrow", 2), &opt)
+            .await
+            .expect("narrow topology should initialize");
+        super::initialize_local_disk_maps(&instance_ctx, build_pools("wide", 4), &opt)
+            .await
+            .expect("re-initializing with a wider topology must not panic or fail");
+
+        let set_drives = instance_ctx.local_disk_set_drives();
+        let set_drives = set_drives.read().await;
+        assert_eq!(set_drives.len(), 1, "stale pools must not accumulate across re-inits");
+        assert_eq!(set_drives[0][0].len(), 4, "pool 0 set 0 must be sized for the new topology");
+        assert!(
+            set_drives[0][0].iter().all(Option::is_some),
+            "every wide-topology drive slot must be registered"
+        );
+        drop(set_drives);
+
+        let disk_map = instance_ctx.local_disk_map();
+        let disk_map = disk_map.read().await;
+        assert_eq!(disk_map.len(), 4, "stale narrow-topology disk entries must be dropped");
+        assert!(
+            disk_map.keys().all(|path| path.contains("wide-disk")),
+            "only the new topology's disks may remain registered: {:?}",
+            disk_map.keys().collect::<Vec<_>>()
+        );
+    }
 }