diff --git a/crates/ecstore/src/bucket/metadata.rs b/crates/ecstore/src/bucket/metadata.rs index 31ac19520..b0ca13060 100644 --- a/crates/ecstore/src/bucket/metadata.rs +++ b/crates/ecstore/src/bucket/metadata.rs @@ -813,6 +813,14 @@ impl BucketMetadata { return Err(Error::other("errServerNotInitialized")); }; + self.save_with_store(store).await + } + + /// Persist this metadata through an explicit store (backlog#1052 S7): the + /// owning instance's metadata system passes its own store so a second + /// server's bucket metadata lands in that server's `.rustfs.sys`, not the + /// ambient (first) one. [`BucketMetadata::save`] keeps the ambient default. + pub async fn save_with_store(&mut self, store: std::sync::Arc) -> Result<()> { self.parse_all_configs()?; let mut buf: Vec = vec![0; 4]; diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 954c6e0b0..09358fb71 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -181,6 +181,44 @@ pub async fn get(bucket: &str) -> Result> { lock.get(bucket).await } +// ---- Instance-scoped variants (backlog#1052 S7) ---- +// +// A store's own bucket operations resolve the metadata system of *their* +// instance context so two servers in one process stay isolated; when the +// instance cell is not initialized yet (early startup) they fall back to the +// ambient default — the single-instance legacy behavior. + +fn bucket_metadata_sys_of(ctx: &crate::runtime::instance::InstanceContext) -> Result>> { + if let Some(sys) = ctx.bucket_metadata_sys() { + return Ok(sys); + } + get_bucket_metadata_sys() +} + +pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result> { + let sys = bucket_metadata_sys_of(ctx)?; + let lock = sys.read().await; + lock.get(bucket).await +} + +pub(crate) async fn created_at_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result { + let sys = bucket_metadata_sys_of(ctx)?; + let lock = sys.read().await; + lock.created_at(bucket).await +} + +pub(crate) async fn set_bucket_metadata_in(ctx: &crate::runtime::instance::InstanceContext, bm: BucketMetadata) -> Result<()> { + let sys = bucket_metadata_sys_of(ctx)?; + let lock = sys.read().await; + lock.persist_and_set(bm).await +} + +pub(crate) async fn remove_bucket_metadata_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result { + let sys = bucket_metadata_sys_of(ctx)?; + let lock = sys.read().await; + Ok(lock.remove(bucket).await) +} + pub async fn update(bucket: &str, config_file: &str, data: Vec) -> Result { let bucket_meta_sys_lock = get_bucket_metadata_sys()?; let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; @@ -531,9 +569,16 @@ impl BucketMetadataSys { return Err(Error::other("errInvalidArgument")); } + self.persist_and_set(bm).await + } + + /// Persist metadata through this system's own store and cache it here + /// (backlog#1052 S7). The store-scoped bucket path uses this so a second + /// server's metadata never leaks into the ambient (first) instance. + pub(crate) async fn persist_and_set(&self, bm: BucketMetadata) -> Result<()> { let mut bm = bm; - bm.save().await?; + bm.save_with_store(self.api.clone()).await?; self.set(bm.name.clone(), Arc::new(bm)).await; diff --git a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs index 8414a3dad..c16588362 100644 --- a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs +++ b/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) -> 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 { + fn new_clients(eps: &EndpointServerPools, instance_ctx: Arc) -> Vec { let nodes = eps.get_nodes(); let v: Vec = nodes .iter() .map(|e| { if e.is_local { - let cli: Box = Box::new(LocalPeerS3Client::new(Some(e.clone()), Some(e.pools.clone()))); + let cli: Box = Box::new(LocalPeerS3Client::new_with_instance_ctx( + Some(e.clone()), + Some(e.pools.clone()), + instance_ctx.clone(), + )); Arc::new(cli) } else { let cli: Box = Box::new(RemotePeerS3Client::new(Some(e.clone()), Some(e.pools.clone()))); @@ -384,15 +395,24 @@ pub struct LocalPeerS3Client { local_disks: Option>, // pub node: Node, pub pools: Option>, + /// 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, } impl LocalPeerS3Client { - pub fn new(_node: Option, pools: Option>) -> Self { + pub fn new(node: Option, pools: Option>) -> Self { + Self::new_with_instance_ctx(node, pools, bootstrap_ctx()) + } + + pub fn new_with_instance_ctx(_node: Option, pools: Option>, instance_ctx: Arc) -> 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; }; diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index a8ea5af2b..64e988767 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -475,6 +475,17 @@ pub(crate) async fn local_disk_paths() -> Vec { local_disk_map_handle().read().await.keys().cloned().collect() } +/// Local disks registered on an explicit instance context (backlog#1052 S7). +pub(crate) async fn local_disks_in(instance_ctx: &InstanceContext) -> Vec { + instance_ctx + .local_disk_map() + .read() + .await + .values() + .filter_map(|v| v.as_ref().cloned()) + .collect() +} + pub(crate) async fn local_disks() -> Vec { local_disk_map_handle() .read() diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index 73e062a3c..9a28d1905 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -40,8 +40,8 @@ fn validate_table_bucket_delete_allowed( Ok(()) } -async fn table_catalog_metadata_exists(bucket: &str) -> bool { - let local_disks = all_local_disk().await; +async fn table_catalog_metadata_exists(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> bool { + let local_disks = runtime_sources::local_disks_in(ctx).await; for disk in local_disks.iter() { let catalog_path = disk.path().join(bucket).join(BUCKET_TABLE_RESERVED_PREFIX); if has_xlmeta_files(&catalog_path).await { @@ -52,12 +52,12 @@ async fn table_catalog_metadata_exists(bucket: &str) -> bool { false } -async fn validate_table_bucket_delete_guard(bucket: &str) -> Result<()> { - let table_bucket_enabled = metadata_sys::get(bucket) +async fn validate_table_bucket_delete_guard(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result<()> { + let table_bucket_enabled = metadata_sys::get_in(ctx, bucket) .await .is_ok_and(|metadata| metadata.table_bucket_enabled()); if table_bucket_enabled { - validate_table_bucket_delete_allowed(bucket, true, table_catalog_metadata_exists(bucket).await)?; + validate_table_bucket_delete_allowed(bucket, true, table_catalog_metadata_exists(ctx, bucket).await)?; } Ok(()) @@ -106,7 +106,7 @@ impl ECStore { self.delete_all(RUSTFS_META_BUCKET, marker_prefix.as_str()).await?; } - metadata_sys::remove_bucket_metadata(bucket).await?; + metadata_sys::remove_bucket_metadata_in(&self.ctx, bucket).await?; runtime_sources::delete_bucket_monitor_entry(bucket); Ok(()) } @@ -186,9 +186,7 @@ impl ECStore { meta.versioning_config_xml = crate::bucket::utils::serialize::(&enableVersioningConfig)?; } - meta.save().await?; - - set_bucket_metadata(bucket.to_string(), meta).await?; + metadata_sys::set_bucket_metadata_in(&self.ctx, meta).await?; Ok(()) } @@ -197,7 +195,7 @@ impl ECStore { pub(super) async fn handle_get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result { let mut info = self.peer_sys.get_bucket_info(bucket, opts).await?; - if let Ok(sys) = metadata_sys::get(bucket).await { + if let Ok(sys) = metadata_sys::get_in(&self.ctx, bucket).await { if should_override_created_from_metadata(sys.created) { info.created = Some(sys.created); } @@ -216,7 +214,7 @@ impl ECStore { if !opts.no_metadata { for bucket in buckets.iter_mut() { - if let Ok(created) = metadata_sys::created_at(&bucket.name).await + if let Ok(created) = metadata_sys::created_at_in(&self.ctx, &bucket.name).await && should_override_created_from_metadata(created) { bucket.created = Some(created); @@ -270,7 +268,7 @@ impl ECStore { return Err(to_object_err(storage_err, vec![bucket])); } - validate_table_bucket_delete_guard(bucket).await?; + validate_table_bucket_delete_guard(&self.ctx, bucket).await?; let sr_mark_delete = opts.srdelete_op == SRBucketDeleteOp::MarkDelete; let sr_purge = opts.srdelete_op == SRBucketDeleteOp::Purge; @@ -280,7 +278,7 @@ impl ECStore { // is not set, return BucketNotEmpty error. // Note: Empty directories (left after object deletion) should NOT count as objects. if !opts.force && !sr_mark_delete { - let local_disks = all_local_disk().await; + let local_disks = runtime_sources::local_disks_in(&self.ctx).await; for disk in local_disks.iter() { // Check if bucket directory contains any xl.meta files (actual objects) // We recursively scan for xl.meta files to determine if bucket has objects diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index e5482581c..302913d94 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -339,7 +339,7 @@ impl ECStore { runtime_sources::record_local_disks(&instance_ctx, local_disks).await; } - let peer_sys = S3PeerSys::new(&endpoint_pools); + let peer_sys = S3PeerSys::new_with_instance_ctx(&endpoint_pools, instance_ctx.clone()); let mut pool_meta = PoolMeta::new(&pools, &PoolMeta::default()); pool_meta.dont_save = true; diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 3963e6514..eeb56476b 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -20,7 +20,7 @@ use crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc; use crate::bucket::lifecycle::bucket_lifecycle_ops::{ enqueue_immediate_expiry, enqueue_transition_immediate, init_background_expiry, }; -use crate::bucket::metadata_sys::{self, set_bucket_metadata}; +use crate::bucket::metadata_sys; use crate::bucket::utils::check_abort_multipart_args; use crate::bucket::utils::check_complete_multipart_args; use crate::bucket::utils::check_copy_obj_args; @@ -865,7 +865,7 @@ mod tests { id: uuid::Uuid::new_v4(), disk_map: std::collections::HashMap::new(), pools: Vec::new(), - peer_sys: crate::cluster::rpc::S3PeerSys::new(&endpoint_pools), + peer_sys: crate::cluster::rpc::S3PeerSys::new_with_instance_ctx(&endpoint_pools, ctx.clone()), pool_meta: RwLock::new(PoolMeta::default()), rebalance_meta: RwLock::new(None), decommission_cancelers: RwLock::new(Vec::new()), diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs index a482f480f..19b6860c9 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -40,15 +40,13 @@ //! } //! ``` //! -//! # Multi-instance status +//! # Multi-instance support //! -//! Multiple `RustFSServer`s may now coexist in one process on different ports -//! and volumes (backlog#1052). Each server's storage layer, request-path -//! dispatch, and app subsystem instances stay isolated. IAM and root-credential -//! validation still share a process-wide domain, so a second server whose -//! credentials differ from the first cannot authenticate S3 clients today; -//! wiring authentication per-server is tracked as a follow-up on the same -//! issue. +//! Multiple `RustFSServer`s may coexist in one process on different ports and +//! volumes (backlog#1052). Each server is fully isolated: its own storage +//! layer, request-path dispatch, root credentials, IAM domain, and bucket +//! namespace. Two servers with different credentials each authenticate their +//! own clients, and neither can see the other's buckets or objects. use crate::server::ShutdownHandle; use crate::startup_embedded::{EmbeddedStartedServer, EmbeddedStartupArgs, EmbeddedStartupError, run_embedded_startup}; diff --git a/rustfs/tests/embedded_multi_instance_test.rs b/rustfs/tests/embedded_multi_instance_test.rs index ababe757b..fb129c8bf 100644 --- a/rustfs/tests/embedded_multi_instance_test.rs +++ b/rustfs/tests/embedded_multi_instance_test.rs @@ -105,19 +105,13 @@ async fn two_embedded_servers_start_and_shutdown_independently() { server_a.shutdown().await; } -// backlog#1052 auth acceptance: two embedded servers with *different* -// credentials each authenticate against their own root identity. Server B -// accepts its own access key and rejects server A's. This exercises the -// per-server auth path (each request resolves its own AppContext for -// credential validation). -// -// NOTE: full bucket-namespace isolation is a separate, deeper follow-up: the -// ecstore data plane still resolves some lower-level reads (peer/disk/bucket -// metadata) through the process-global object handle, so the two servers do -// not yet present independent bucket listings even though each holds its own -// store object. That isolation is the remaining work on #1052. +// backlog#1052 full acceptance: two embedded servers with *different* +// credentials are isolated end to end — auth (each accepts its own key and +// rejects the other's) AND data plane (each server's buckets/objects are +// invisible to the other; each lists/creates/deletes only on its own disks +// and bucket-metadata system). #[tokio::test] -async fn two_embedded_servers_authenticate_with_their_own_credentials() { +async fn two_embedded_servers_isolate_auth_and_data_planes() { let port_a = match find_available_port() { Ok(port) => port, Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, @@ -176,6 +170,79 @@ async fn two_embedded_servers_authenticate_with_their_own_credentials() { .await .expect("server A must authenticate with its own credentials"); + // ---- Data-plane isolation (backlog#1052 S7) ---- + + // Server A owns a bucket + object. + client_a + .create_bucket() + .bucket("only-on-a") + .send() + .await + .expect("server A creates its bucket"); + client_a + .put_object() + .bucket("only-on-a") + .key("marker.txt") + .body(ByteStream::from_static(b"belongs to A")) + .send() + .await + .expect("server A writes its object"); + + // Server B's listing does not contain server A's bucket. + let b_buckets: Vec<_> = client_b + .list_buckets() + .send() + .await + .expect("server B lists buckets") + .buckets() + .iter() + .flat_map(|bucket| bucket.name.clone()) + .collect(); + assert!( + !b_buckets.contains(&"only-on-a".to_string()), + "server B must not see server A's bucket; saw {b_buckets:?}" + ); + + // Server B cannot resolve server A's object either. + let cross_head = client_b.head_object().bucket("only-on-a").key("marker.txt").send().await; + assert!(cross_head.is_err(), "server B must not resolve server A's object; got {cross_head:?}"); + + // Server B's own bucket is invisible to server A. + client_b + .create_bucket() + .bucket("only-on-b") + .send() + .await + .expect("server B creates its bucket"); + let a_buckets: Vec<_> = client_a + .list_buckets() + .send() + .await + .expect("server A lists buckets") + .buckets() + .iter() + .flat_map(|bucket| bucket.name.clone()) + .collect(); + assert!( + a_buckets.contains(&"only-on-a".to_string()), + "server A must keep seeing its own bucket; saw {a_buckets:?}" + ); + assert!( + !a_buckets.contains(&"only-on-b".to_string()), + "server A must not see server B's bucket; saw {a_buckets:?}" + ); + + // Server A's data plane is intact. + let a_get = client_a + .get_object() + .bucket("only-on-a") + .key("marker.txt") + .send() + .await + .expect("server A serves its own object"); + let a_data = a_get.body.collect().await.expect("read A body").into_bytes(); + assert_eq!(a_data.as_ref(), b"belongs to A"); + server_a.shutdown().await; server_b.shutdown().await; }