feat(ecstore): run bucket operations on the store's own instance context (#4642)

backlog#1052 S7 — the final piece: full bucket-namespace isolation
between embedded servers in one process.

Server B's requests resolved B's own ECStore (per-server dispatch landed
earlier), but the store's bucket operations still went through ambient
process facades, so both servers effectively operated on the FIRST
server's disks and metadata:

- LocalPeerS3Client::local_disks_for_pools() called all_local_disk()
  (the ambient disk registry = the first published store's context), so
  list/make/delete/heal bucket scanned and wrote the wrong volumes.
- BucketMetadata::save() persisted through the ambient object handle, so
  a second server's bucket metadata landed in the first server's
  .rustfs.sys; set/remove/get/created_at all used the ambient metadata
  system.

Now the whole chain is bound to the owning store's InstanceContext:

- S3PeerSys/LocalPeerS3Client gain *_with_instance_ctx constructors and
  operate on that context's registered disks; ECStore::new (and the test
  store builder) pass the store's context. The legacy constructors keep
  the bootstrap default.
- BucketMetadata::save_with_store persists through an explicit store;
  BucketMetadataSys::persist_and_set uses the system's own api handle.
- metadata_sys gains instance-scoped variants (get_in / created_at_in /
  set_bucket_metadata_in / remove_bucket_metadata_in) that resolve the
  context's metadata system and fall back to the ambient default before
  the instance cell is initialized (early startup, unchanged behavior).
- The store's bucket handlers (make/get_info/list/delete + the
  table-bucket delete guard and emptiness check) use the per-context
  variants and this instance's disks.

Acceptance (e2e): two embedded servers with different credentials are now
isolated end to end — each authenticates only its own key, neither sees
the other's buckets or objects, and both data planes stay intact. The
embedded module doc drops the shared-IAM caveat.

579 ecstore bucket/metadata/peer regressions plus the embedded basic and
deferred-IAM e2e stay green.

crates/ecstore/src/cluster/rpc/peer_s3_client.rs

@@ -20,9 +20,9 @@ use crate::disk::error::DiskError;
 use crate::disk::error::{Error, Result};
 use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs};
 use crate::disk::{DiskAPI, DiskStore, disk_store::get_max_timeout_duration};
+use crate::runtime::instance::{InstanceContext, bootstrap_ctx};
 use crate::runtime::sources as runtime_sources;
 use crate::storage_api_contracts::bucket::{BucketInfo, BucketOptions, DeleteBucketOptions, MakeBucketOptions};
-use crate::store::all_local_disk;
 use crate::store::utils::is_reserved_or_invalid_bucket;
 use crate::{
     disk::{
@@ -93,19 +93,30 @@ pub struct S3PeerSys {
 
 impl S3PeerSys {
     pub fn new(eps: &EndpointServerPools) -> Self {
+        Self::new_with_instance_ctx(eps, bootstrap_ctx())
+    }
+
+    /// Build the peer system bound to an explicit instance context
+    /// (backlog#1052 S7): the local peer client operates on that instance's
+    /// disks. [`S3PeerSys::new`] keeps the ambient bootstrap default.
+    pub fn new_with_instance_ctx(eps: &EndpointServerPools, instance_ctx: Arc<InstanceContext>) -> Self {
         Self {
-            clients: Self::new_clients(eps),
+            clients: Self::new_clients(eps, instance_ctx),
             pools_count: eps.as_ref().len(),
         }
     }
 
-    fn new_clients(eps: &EndpointServerPools) -> Vec<Client> {
+    fn new_clients(eps: &EndpointServerPools, instance_ctx: Arc<InstanceContext>) -> Vec<Client> {
         let nodes = eps.get_nodes();
         let v: Vec<Client> = nodes
             .iter()
             .map(|e| {
                 if e.is_local {
-                    let cli: Box<dyn PeerS3Client> = Box::new(LocalPeerS3Client::new(Some(e.clone()), Some(e.pools.clone())));
+                    let cli: Box<dyn PeerS3Client> = Box::new(LocalPeerS3Client::new_with_instance_ctx(
+                        Some(e.clone()),
+                        Some(e.pools.clone()),
+                        instance_ctx.clone(),
+                    ));
                     Arc::new(cli)
                 } else {
                     let cli: Box<dyn PeerS3Client> = Box::new(RemotePeerS3Client::new(Some(e.clone()), Some(e.pools.clone())));
@@ -384,15 +395,24 @@ pub struct LocalPeerS3Client {
     local_disks: Option<Vec<DiskStore>>,
     // pub node: Node,
     pub pools: Option<Vec<usize>>,
+    /// The owning store's runtime context (backlog#1052 S7): local bucket
+    /// operations list/create/delete on THIS instance's registered disks, not
+    /// on whatever the ambient process default resolves to.
+    instance_ctx: Arc<InstanceContext>,
 }
 
 impl LocalPeerS3Client {
-    pub fn new(_node: Option<Node>, pools: Option<Vec<usize>>) -> Self {
+    pub fn new(node: Option<Node>, pools: Option<Vec<usize>>) -> Self {
+        Self::new_with_instance_ctx(node, pools, bootstrap_ctx())
+    }
+
+    pub fn new_with_instance_ctx(_node: Option<Node>, pools: Option<Vec<usize>>, instance_ctx: Arc<InstanceContext>) -> Self {
         Self {
             #[cfg(test)]
             local_disks: None,
             // node,
             pools,
+            instance_ctx,
         }
     }
 
@@ -401,6 +421,7 @@ impl LocalPeerS3Client {
         Self {
             local_disks: Some(local_disks),
             pools,
+            instance_ctx: bootstrap_ctx(),
         }
     }
 
@@ -409,10 +430,10 @@ impl LocalPeerS3Client {
         let local_disks = if let Some(local_disks) = self.local_disks.as_ref() {
             local_disks.clone()
         } else {
-            all_local_disk().await
+            runtime_sources::local_disks_in(&self.instance_ctx).await
         };
         #[cfg(not(test))]
-        let local_disks = all_local_disk().await;
+        let local_disks = runtime_sources::local_disks_in(&self.instance_ctx).await;
         let Some(pools) = self.pools.as_ref() else {
             return local_disks;
         };