refactor(startup): thread explicit InstanceContext through the storage startup path (#4611)

crates/ecstore/src/api/mod.rs

@@ -326,6 +326,7 @@ pub mod global {
 }
 
 pub mod runtime {
+    pub use crate::runtime::instance::{InstanceContext, bootstrap_ctx};
     pub use crate::runtime::sources::{
         boot_time, bucket_monitor, deployment_id, endpoint_pools, expiry_state_handle, first_cluster_node_is_local,
         global_lock_client, global_lock_clients, global_tier_config_mgr, local_disk_map_read, object_store_handle, region,
@@ -389,8 +390,9 @@ pub mod store_list {
 pub mod storage {
     pub use crate::store::HealWalkVersion;
     pub use crate::store::{
-        ECStore, all_local_disk, all_local_disk_path, find_local_disk_by_ref, init_local_disks, init_lock_clients,
-        prewarm_local_disk_id_map,
+        ECStore, all_local_disk, all_local_disk_path, find_local_disk_by_ref, init_local_disks,
+        init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map,
+        prewarm_local_disk_id_map_with_instance_ctx,
     };
 }
 

crates/ecstore/src/core/sets.rs

@@ -102,6 +102,21 @@ impl Sets {
         fm: &FormatV3,
         pool_idx: usize,
         parity_count: usize,
+    ) -> Result<Arc<Self>> {
+        Self::new_with_instance_ctx(disks, endpoints, fm, pool_idx, parity_count, bootstrap_ctx()).await
+    }
+
+    /// Build the pool's sets bound to an explicit instance context (Phase 5
+    /// follow-up, backlog#1052). The legacy [`Sets::new`] entry adopts the
+    /// process bootstrap context; a store constructed around its own context
+    /// passes it here so the whole object graph shares one cell.
+    pub async fn new_with_instance_ctx(
+        disks: Vec<Option<DiskStore>>,
+        endpoints: &PoolEndpoints,
+        fm: &FormatV3,
+        pool_idx: usize,
+        parity_count: usize,
+        instance_ctx: Arc<InstanceContext>,
     ) -> Result<Arc<Self>> {
         let set_count = fm.erasure.sets.len();
         let set_drive_count = fm.erasure.sets[0].len();
@@ -127,8 +142,8 @@ impl Sets {
                     continue;
                 }
 
-                if disk.as_ref().unwrap().is_local() && runtime_sources::setup_is_dist_erasure().await {
-                    let local_disk = runtime_sources::local_disk_set_drive(pool_idx, i, j).await;
+                if disk.as_ref().unwrap().is_local() && instance_ctx.is_dist_erasure().await {
+                    let local_disk = runtime_sources::local_disk_set_drive(&instance_ctx, pool_idx, i, j).await;
 
                     if local_disk.is_none() {
                         warn!("sets new set_drive {}-{} local_disk is none", i, j);
@@ -163,7 +178,7 @@ impl Sets {
                 .as_ref()
                 .map(|registry| registry.clients_for_endpoints(&set_endpoints))
                 .unwrap_or_default();
-            let set_disks = SetDisks::new(
+            let set_disks = SetDisks::new_with_instance_ctx(
                 runtime_sources::local_node_name().await,
                 Arc::new(RwLock::new(set_drive)),
                 set_drive_count,
@@ -173,6 +188,7 @@ impl Sets {
                 set_endpoints,
                 fm.clone(),
                 lockers,
+                instance_ctx.clone(),
             )
             .await;
 
@@ -193,10 +209,7 @@ impl Sets {
             default_parity_count: parity_count,
             distribution_algo: fm.erasure.distribution_algo.clone(),
             exit_signal: Some(tx),
-            // Single-instance: same bootstrap context the owning ECStore adopts
-            // (constructed before the store, so sourced here directly). Slice 8
-            // threads a per-instance context in for true multi-instance.
-            ctx: bootstrap_ctx(),
+            ctx: instance_ctx,
         });
 
         let asets = sets.clone();
@@ -1462,4 +1475,65 @@ mod tests {
             "unformatted disk must be Missing on its real index, not on a placeholder"
         );
     }
+
+    fn instance_ctx_test_pool_endpoints() -> (FormatV3, PoolEndpoints) {
+        let format = FormatV3::new(1, 2);
+        let endpoints = vec![
+            Endpoint::try_from("http://127.0.0.1:9000/data0").expect("first endpoint should parse"),
+            Endpoint::try_from("http://127.0.0.1:9001/data1").expect("second endpoint should parse"),
+        ];
+        let pool_endpoints = PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 2,
+            endpoints: Endpoints::from(endpoints),
+            cmd_line: "instance-ctx-adoption-test".to_string(),
+            platform: "test".to_string(),
+        };
+        (format, pool_endpoints)
+    }
+
+    // Phase 5 follow-up (backlog#1052): a pool built through the ctx-explicit
+    // constructor carries the caller's context through Sets AND every SetDisks,
+    // so nothing in the object graph silently binds to the process bootstrap.
+    #[tokio::test]
+    async fn sets_new_with_instance_ctx_threads_context_through_graph() {
+        let (format, pool_endpoints) = instance_ctx_test_pool_endpoints();
+        let instance_ctx = Arc::new(InstanceContext::new());
+
+        let sets = Sets::new_with_instance_ctx(vec![None, None], &pool_endpoints, &format, 0, 1, instance_ctx.clone())
+            .await
+            .expect("sets should build with empty disks");
+
+        assert!(
+            Arc::ptr_eq(sets.instance_ctx(), &instance_ctx),
+            "Sets must adopt the explicitly passed instance context"
+        );
+        for set_disks in &sets.disk_set {
+            assert!(
+                Arc::ptr_eq(set_disks.instance_ctx(), &instance_ctx),
+                "every SetDisks must adopt the explicitly passed instance context"
+            );
+        }
+        assert!(
+            !Arc::ptr_eq(sets.instance_ctx(), &bootstrap_ctx()),
+            "a fresh context must not alias the process bootstrap context"
+        );
+    }
+
+    // The legacy constructor keeps single-instance behavior byte-for-byte: it
+    // still adopts the process bootstrap context.
+    #[tokio::test]
+    async fn sets_new_legacy_adopts_bootstrap_context() {
+        let (format, pool_endpoints) = instance_ctx_test_pool_endpoints();
+
+        let sets = Sets::new(vec![None, None], &pool_endpoints, &format, 0, 1)
+            .await
+            .expect("sets should build with empty disks");
+
+        assert!(
+            Arc::ptr_eq(sets.instance_ctx(), &bootstrap_ctx()),
+            "legacy Sets::new must keep adopting the process bootstrap context"
+        );
+    }
 }

crates/ecstore/src/runtime/sources.rs

@@ -20,6 +20,7 @@ use std::{
 
 use crate::bucket::bandwidth::monitor::Monitor;
 use crate::disk::endpoint::Endpoint;
+use crate::runtime::instance::InstanceContext;
 use crate::{
     bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState},
     bucket::metadata_sys::{BucketMetadataSys, get_global_bucket_metadata_sys},
@@ -33,8 +34,8 @@ use crate::{
         TypeLocalDiskSetDrives, get_background_services_cancel_token, get_global_bucket_monitor, get_global_deployment_id,
         get_global_endpoints, get_global_endpoints_opt, get_global_lock_client, get_global_lock_clients, get_global_region,
         get_global_tier_config_mgr, global_rustfs_port, init_global_bucket_monitor, is_dist_erasure, is_erasure, is_erasure_sd,
-        is_first_cluster_node_local, resolve_object_store_handle, set_global_deployment_id, set_global_lock_client,
-        set_global_lock_clients, set_object_layer, update_erasure_type,
+        is_first_cluster_node_local, resolve_object_store_handle, set_global_lock_client, set_global_lock_clients,
+        set_object_layer, update_erasure_type,
     },
     services::batch_processor::{GlobalBatchProcessors, get_global_processors},
     services::event_notification::EventNotifier,
@@ -278,12 +279,6 @@ pub(crate) fn replication_runtime_initialized() -> bool {
     crate::runtime::global::current_ctx().replication_initialized()
 }
 
-pub(crate) fn ensure_deployment_id(deployment_id: Uuid) {
-    if get_global_deployment_id().is_none() {
-        set_global_deployment_id(deployment_id);
-    }
-}
-
 pub fn global_lock_client() -> Option<Arc<dyn LockClient>> {
     get_global_lock_client()
 }
@@ -419,8 +414,8 @@ pub(crate) async fn clear_local_disk_id_map_for_test() {
     local_disk_id_map_handle().write().await.clear();
 }
 
-pub(crate) async fn record_local_disk_id(disk_id: Uuid, endpoint: String) {
-    local_disk_id_map_handle().write().await.insert(disk_id, endpoint);
+pub(crate) async fn record_local_disk_id(instance_ctx: &Arc<InstanceContext>, disk_id: Uuid, endpoint: String) {
+    instance_ctx.local_disk_id_map().write().await.insert(disk_id, endpoint);
 }
 
 pub(crate) async fn replace_local_disk_id(previous: Option<Uuid>, current: Option<Uuid>, endpoint: String) {
@@ -434,8 +429,8 @@ pub(crate) async fn replace_local_disk_id(previous: Option<Uuid>, current: Optio
     }
 }
 
-pub(crate) async fn record_local_disks(disks: Vec<DiskStore>) {
-    let map = local_disk_map_handle();
+pub(crate) async fn record_local_disks(instance_ctx: &Arc<InstanceContext>, disks: Vec<DiskStore>) {
+    let map = instance_ctx.local_disk_map();
     let mut global_local_disk_map = map.write().await;
     for disk in disks {
         let path = disk.endpoint().to_string();
@@ -443,8 +438,13 @@ pub(crate) async fn record_local_disks(disks: Vec<DiskStore>) {
     }
 }
 
-pub(crate) async fn local_disk_set_drive(pool_idx: usize, set_idx: usize, disk_idx: usize) -> Option<DiskStore> {
-    local_disk_set_drives_handle().read().await[pool_idx][set_idx][disk_idx].clone()
+pub(crate) async fn local_disk_set_drive(
+    instance_ctx: &Arc<InstanceContext>,
+    pool_idx: usize,
+    set_idx: usize,
+    disk_idx: usize,
+) -> Option<DiskStore> {
+    instance_ctx.local_disk_set_drives().read().await[pool_idx][set_idx][disk_idx].clone()
 }
 
 pub(crate) async fn local_disk_for_endpoint(endpoint: &Endpoint) -> Option<DiskStore> {
@@ -488,8 +488,12 @@ pub(crate) async fn local_disk_entries() -> Vec<Option<DiskStore>> {
     local_disk_map_handle().read().await.values().cloned().collect()
 }
 
-pub(crate) async fn initialize_local_disk_maps(endpoint_pools: EndpointServerPools, opt: &DiskOption) -> Result<()> {
-    let set_drives = local_disk_set_drives_handle();
+pub(crate) async fn initialize_local_disk_maps(
+    instance_ctx: &Arc<InstanceContext>,
+    endpoint_pools: EndpointServerPools,
+    opt: &DiskOption,
+) -> Result<()> {
+    let set_drives = instance_ctx.local_disk_set_drives();
     let mut global_set_drives = set_drives.write().await;
     for pool_eps in endpoint_pools.as_ref().iter() {
         let mut set_count_drives = Vec::with_capacity(pool_eps.set_count);
@@ -500,7 +504,7 @@ pub(crate) async fn initialize_local_disk_maps(endpoint_pools: EndpointServerPoo
         global_set_drives.push(set_count_drives);
     }
 
-    let map = local_disk_map_handle();
+    let map = instance_ctx.local_disk_map();
     let mut global_local_disk_map = map.write().await;
 
     for pool_eps in endpoint_pools.as_ref().iter() {

crates/ecstore/src/set_disk/mod.rs

@@ -1744,10 +1744,39 @@ impl SetDisks {
         format: FormatV3,
         lockers: Vec<Arc<dyn LockClient>>,
     ) -> Arc<Self> {
-        // Single-instance sources the process bootstrap context (the one the
-        // owning ECStore adopts). Slice 8 threads a per-instance context in for
-        // true multi-instance.
-        let ctx = bootstrap_ctx();
+        Self::new_with_instance_ctx(
+            locker_owner,
+            disks,
+            set_drive_count,
+            default_parity_count,
+            set_index,
+            pool_index,
+            set_endpoints,
+            format,
+            lockers,
+            bootstrap_ctx(),
+        )
+        .await
+    }
+
+    /// Build a set bound to an explicit instance context (Phase 5 follow-up,
+    /// backlog#1052). The legacy [`SetDisks::new`] entry adopts the process
+    /// bootstrap context; a store constructed around its own context threads it
+    /// down here so the whole object graph shares one cell.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn new_with_instance_ctx(
+        locker_owner: String,
+        disks: Arc<RwLock<Vec<Option<DiskStore>>>>,
+        set_drive_count: usize,
+        default_parity_count: usize,
+        set_index: usize,
+        pool_index: usize,
+        set_endpoints: Vec<Endpoint>,
+        format: FormatV3,
+        lockers: Vec<Arc<dyn LockClient>>,
+        instance_ctx: Arc<InstanceContext>,
+    ) -> Arc<Self> {
+        let ctx = instance_ctx;
         Arc::new(SetDisks {
             locker_owner,
             disks,

crates/ecstore/src/store/init.rs

@@ -15,6 +15,7 @@
 use super::*;
 use crate::core::pools::local_decommission_queue_prefix;
 use crate::error::is_err_decommission_running;
+use crate::runtime::instance::InstanceContext;
 use crate::runtime::sources as runtime_sources;
 use crate::storage_api_contracts::object::EcstoreObjectIO;
 use tracing::{debug, error, info, warn};
@@ -165,6 +166,24 @@ impl ECStore {
     #[allow(clippy::new_ret_no_self)]
     #[instrument(level = "debug", skip(endpoint_pools))]
     pub async fn new(address: SocketAddr, endpoint_pools: EndpointServerPools, ctx: CancellationToken) -> Result<Arc<Self>> {
+        Self::new_with_instance_ctx(address, endpoint_pools, ctx, crate::runtime::instance::bootstrap_ctx()).await
+    }
+
+    /// Build a store around an explicit instance context (Phase 5 follow-up,
+    /// backlog#1052). The legacy [`ECStore::new`] entry adopts the process
+    /// bootstrap context, keeping single-instance startup byte-for-byte
+    /// unchanged; a caller that owns its own context (a future second embedded
+    /// server) passes it here so every construction-time write — pool sets,
+    /// local-disk registry, deployment id — lands on that context instead of
+    /// the shared bootstrap one.
+    #[allow(clippy::new_ret_no_self)]
+    #[instrument(level = "debug", skip(endpoint_pools, instance_ctx))]
+    pub async fn new_with_instance_ctx(
+        address: SocketAddr,
+        endpoint_pools: EndpointServerPools,
+        ctx: CancellationToken,
+        instance_ctx: Arc<InstanceContext>,
+    ) -> Result<Arc<Self>> {
         // let layouts = DisksLayout::from_volumes(endpoints.as_slice())?;
 
         let mut deployment_id = None;
@@ -308,15 +327,16 @@ impl ECStore {
                 }
             }
 
-            let sets = Sets::new(disks.clone(), pool_eps, &fm, i, common_parity_drives).await?;
+            let sets =
+                Sets::new_with_instance_ctx(disks.clone(), pool_eps, &fm, i, common_parity_drives, instance_ctx.clone()).await?;
             pools.push(sets);
 
             disk_map.insert(i, disks);
         }
 
         // Replace the local disk
-        if !runtime_sources::setup_is_dist_erasure().await {
-            runtime_sources::record_local_disks(local_disks).await;
+        if !instance_ctx.is_dist_erasure().await {
+            runtime_sources::record_local_disks(&instance_ctx, local_disks).await;
         }
 
         let peer_sys = S3PeerSys::new(&endpoint_pools);
@@ -334,14 +354,17 @@ impl ECStore {
             decommission_cancelers,
             start_gate: Mutex::new(()),
             pool_meta_save_gate: Mutex::new(()),
-            // Adopt the process bootstrap context so startup writes (erasure
-            // type recorded before this point) and later reads share one cell.
-            ctx: crate::runtime::instance::bootstrap_ctx(),
+            // Adopt the caller's context (the process bootstrap one on the
+            // legacy path) so startup writes (erasure type recorded before
+            // this point) and later reads share one cell.
+            ctx: instance_ctx.clone(),
         });
 
-        // Only set it when the global deployment ID is not yet configured
-        if let Some(dep_id) = deployment_id {
-            runtime_sources::ensure_deployment_id(dep_id);
+        // Only set it when this instance's deployment ID is not yet configured
+        if let Some(dep_id) = deployment_id
+            && instance_ctx.deployment_id().is_none()
+        {
+            instance_ctx.set_deployment_id(dep_id);
         }
 
         let wait_sec = 5;
@@ -759,4 +782,84 @@ mod tests {
             "the expanded pool should be initialized by its own first local endpoint"
         );
     }
+
+    // Phase 5 follow-up (backlog#1052): building a real store through the
+    // ctx-explicit constructor lands every construction-time write — object
+    // graph adoption, local-disk registry, deployment id — on the passed
+    // context, not on the process bootstrap one. This is the storage-layer
+    // seam a future second embedded server needs to stay isolated.
+    #[tokio::test]
+    async fn new_with_instance_ctx_threads_context_through_store_graph() {
+        use crate::runtime::instance::InstanceContext;
+
+        let temp_dir = tempfile::tempdir().expect("create temp store dir");
+        let disk_paths: Vec<_> = (1..=4).map(|i| temp_dir.path().join(format!("disk{i}"))).collect();
+        for path in &disk_paths {
+            tokio::fs::create_dir_all(path).await.expect("create disk dir");
+        }
+
+        let mut endpoints = Vec::new();
+        for (i, path) in disk_paths.iter().enumerate() {
+            let mut endpoint = Endpoint::try_from(path.to_str().expect("disk path should be utf-8")).expect("local endpoint");
+            endpoint.set_pool_index(0);
+            endpoint.set_set_index(0);
+            endpoint.set_disk_index(i);
+            endpoints.push(endpoint);
+        }
+        let endpoint_pools = EndpointServerPools(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 4,
+            endpoints: Endpoints::from(endpoints),
+            cmd_line: "instance-ctx-store-graph-test".to_string(),
+            platform: "test".to_string(),
+        }]);
+
+        let instance_ctx = Arc::new(InstanceContext::new());
+        crate::store::init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
+            .await
+            .expect("register local disks into the fresh context");
+
+        let store = crate::store::ECStore::new_with_instance_ctx(
+            "127.0.0.1:0".parse().expect("test address"),
+            endpoint_pools,
+            CancellationToken::new(),
+            instance_ctx.clone(),
+        )
+        .await
+        .expect("store should build around the fresh context");
+
+        assert!(
+            Arc::ptr_eq(&store.ctx, &instance_ctx),
+            "the store must adopt the explicitly passed instance context"
+        );
+        for sets in &store.pools {
+            assert!(
+                Arc::ptr_eq(sets.instance_ctx(), &instance_ctx),
+                "every pool's Sets must carry the passed instance context"
+            );
+        }
+        assert_eq!(
+            instance_ctx.deployment_id(),
+            Some(store.id),
+            "the deployment id must land on the passed context and mirror the store id"
+        );
+
+        let registered: Vec<String> = instance_ctx.local_disk_map().read().await.keys().cloned().collect();
+        assert_eq!(registered.len(), 4, "the passed context must register all four local disks");
+        let bootstrap = crate::runtime::instance::bootstrap_ctx();
+        assert_ne!(
+            bootstrap.deployment_id(),
+            Some(store.id),
+            "the bootstrap context must not absorb the fresh store's deployment id"
+        );
+        let bootstrap_map = bootstrap.local_disk_map();
+        let bootstrap_map = bootstrap_map.read().await;
+        for key in &registered {
+            assert!(
+                !bootstrap_map.contains_key(key),
+                "the bootstrap context must not absorb the fresh store's disks"
+            );
+        }
+    }
 }

crates/ecstore/src/store/mod.rs

@@ -164,8 +164,9 @@ pub(crate) mod utils;
 
 use peer::init_local_peer;
 pub use peer::{
-    all_local_disk, all_local_disk_path, find_local_disk_by_ref, get_disk_infos, init_local_disks, init_lock_clients,
-    prewarm_local_disk_id_map,
+    all_local_disk, all_local_disk_path, find_local_disk_by_ref, get_disk_infos, init_local_disks,
+    init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map,
+    prewarm_local_disk_id_map_with_instance_ctx,
 };
 
 pub struct ECStore {

crates/ecstore/src/store/peer.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use super::*;
+use crate::runtime::instance::InstanceContext;
 use crate::runtime::sources as runtime_sources;
 use tracing::{debug, error};
 
@@ -22,8 +23,12 @@ const EVENT_LOCAL_DISK_ID_PREWARM_SKIPPED: &str = "local_disk_id_prewarm_skipped
 const EVENT_LOCK_CLIENT_INITIALIZATION_FAILED: &str = "lock_client_initialization_failed";
 
 async fn remember_local_disk_id(disk: &DiskStore) -> Option<Uuid> {
+    remember_local_disk_id_with_instance_ctx(&crate::runtime::global::current_ctx(), disk).await
+}
+
+async fn remember_local_disk_id_with_instance_ctx(instance_ctx: &Arc<InstanceContext>, disk: &DiskStore) -> Option<Uuid> {
     let disk_id = disk.get_disk_id().await.ok().flatten()?;
-    runtime_sources::record_local_disk_id(disk_id, disk.endpoint().to_string()).await;
+    runtime_sources::record_local_disk_id(instance_ctx, disk_id, disk.endpoint().to_string()).await;
     Some(disk_id)
 }
 
@@ -69,7 +74,21 @@ pub async fn all_local_disk() -> Vec<DiskStore> {
 }
 
 pub async fn prewarm_local_disk_id_map() {
-    for disk in all_local_disk().await {
+    prewarm_local_disk_id_map_with_instance_ctx(&crate::runtime::global::current_ctx()).await
+}
+
+/// Prewarm the disk-id map of an explicit instance context (Phase 5 follow-up,
+/// backlog#1052): startup passes the context whose disk map it just populated
+/// instead of resolving the process-level default.
+pub async fn prewarm_local_disk_id_map_with_instance_ctx(instance_ctx: &Arc<InstanceContext>) {
+    let disks: Vec<DiskStore> = instance_ctx
+        .local_disk_map()
+        .read()
+        .await
+        .values()
+        .filter_map(|v| v.as_ref().cloned())
+        .collect();
+    for disk in disks {
         if let Err(err) = disk.get_disk_id().await {
             debug!(
                 event = EVENT_LOCAL_DISK_ID_PREWARM_SKIPPED,
@@ -82,17 +101,28 @@ pub async fn prewarm_local_disk_id_map() {
             continue;
         }
 
-        let _ = remember_local_disk_id(&disk).await;
+        let _ = remember_local_disk_id_with_instance_ctx(instance_ctx, &disk).await;
     }
 }
 
 pub async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Result<()> {
+    init_local_disks_with_instance_ctx(&crate::runtime::global::current_ctx(), endpoint_pools).await
+}
+
+/// Register the pools' local disks into an explicit instance context (Phase 5
+/// follow-up, backlog#1052). The legacy [`init_local_disks`] entry resolves the
+/// process-level default context; startup paths that own a context pass it here
+/// so a future second instance's disks cannot leak into the first one's registry.
+pub async fn init_local_disks_with_instance_ctx(
+    instance_ctx: &Arc<InstanceContext>,
+    endpoint_pools: EndpointServerPools,
+) -> Result<()> {
     let opt = &DiskOption {
         cleanup: true,
         health_check: true,
     };
 
-    runtime_sources::initialize_local_disk_maps(endpoint_pools, opt).await
+    runtime_sources::initialize_local_disk_maps(instance_ctx, endpoint_pools, opt).await
 }
 
 pub fn init_lock_clients(endpoint_pools: EndpointServerPools) {
@@ -182,3 +212,63 @@ pub async fn get_disk_infos(disks: &[Option<DiskStore>]) -> Vec<Option<DiskInfo>
 
     res
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::layout::endpoints::{Endpoints, PoolEndpoints};
+
+    fn single_local_disk_pools(dir: &std::path::Path) -> EndpointServerPools {
+        let mut endpoint = Endpoint::try_from(dir.to_str().expect("temp dir path should be utf-8")).expect("local endpoint");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+
+        EndpointServerPools(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 1,
+            endpoints: Endpoints::from(vec![endpoint]),
+            cmd_line: "instance-ctx-disk-registry-test".to_string(),
+            platform: "test".to_string(),
+        }])
+    }
+
+    // Phase 5 follow-up (backlog#1052): registering local disks through the
+    // ctx-explicit entry writes the passed context's registry only — the
+    // process bootstrap context (and any other instance) stays clean, so a
+    // future second server's disks cannot leak into the first one's registry.
+    #[tokio::test]
+    async fn init_local_disks_with_instance_ctx_isolates_disk_registry() {
+        let temp_dir = tempfile::tempdir().expect("create temp disk dir");
+        let endpoint_pools = single_local_disk_pools(temp_dir.path());
+        let instance_ctx = Arc::new(InstanceContext::new());
+
+        init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools)
+            .await
+            .expect("local disks should register into the passed context");
+
+        let registered: Vec<String> = instance_ctx.local_disk_map().read().await.keys().cloned().collect();
+        assert_eq!(registered.len(), 1, "the passed context must hold exactly the one local disk");
+        assert_eq!(
+            instance_ctx.local_disk_set_drives().read().await.len(),
+            1,
+            "the passed context must hold the pool/set/drive layout"
+        );
+
+        let bootstrap = crate::runtime::instance::bootstrap_ctx();
+        let bootstrap_map = bootstrap.local_disk_map();
+        let bootstrap_map = bootstrap_map.read().await;
+        let sibling = InstanceContext::new();
+        for key in &registered {
+            assert!(
+                !bootstrap_map.contains_key(key),
+                "bootstrap context must not absorb a disk registered into an explicit context"
+            );
+            assert!(
+                !sibling.local_disk_map().read().await.contains_key(key),
+                "a sibling context must not observe another instance's disks"
+            );
+        }
+    }
+}