diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index b72ce3e93..5d4bba0b9 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -366,6 +366,8 @@ pub mod data_usage { pub mod disk { pub use crate::disk::disk_store::get_object_disk_read_timeout; pub use crate::disk::local::ScanGuard; + #[cfg(all(feature = "test-util", not(windows)))] + pub use crate::disk::os::{LocalPublicationPause, LocalPublicationStage}; pub use crate::disk::{ BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, ConditionalFileUpdate, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, @@ -542,8 +544,8 @@ pub mod storage { pub use crate::core::pools::HealLifecycleExpiryContext; pub use crate::store::HealWalkVersion; pub use crate::store::{ - ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk, all_local_disk_path, - find_local_disk_by_ref, init_local_disks, init_local_disks_with_instance_ctx, init_lock_clients, + BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, 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, }; } diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index caf878742..87ec741b4 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -243,7 +243,7 @@ pub(crate) mod fsync_dir_recorder { } /// Pause a real namespace mutation inside its physical executor. -#[cfg(all(test, not(windows)))] +#[cfg(all(any(test, feature = "test-util"), not(windows)))] pub(crate) mod prepared_publication_test_hooks { use super::*; @@ -252,6 +252,7 @@ pub(crate) mod prepared_publication_test_hooks { PreparedRename, Rename, Remove, + #[cfg(test)] Rollback, DirFsync, } @@ -268,6 +269,7 @@ pub(crate) mod prepared_publication_test_hooks { } } + #[cfg(test)] pub(crate) fn install(path: &Path, hook: impl FnOnce() + Send + 'static) -> Guard { install_at(Stage::PreparedRename, path, hook) } @@ -286,6 +288,51 @@ pub(crate) mod prepared_publication_test_hooks { } } +/// Controlled application-test pause at an existing physical executor boundary. +#[cfg(all(feature = "test-util", not(windows)))] +pub struct LocalPublicationPause { + _hook: prepared_publication_test_hooks::Guard, + entered: oneshot::Receiver<()>, + _release: std::sync::mpsc::Sender<()>, +} + +#[cfg(all(feature = "test-util", not(windows)))] +#[derive(Clone, Copy)] +pub enum LocalPublicationStage { + PreparedRename, + Rename, + Remove, +} + +#[cfg(all(feature = "test-util", not(windows)))] +impl LocalPublicationPause { + pub fn install(disk: &crate::disk::Disk, volume: &str, path: &str, stage: LocalPublicationStage) -> Result { + let path = disk + .get_object_path_for_io_if_local(volume, path) + .ok_or(DiskError::DiskNotFound)??; + let stage = match stage { + LocalPublicationStage::PreparedRename => prepared_publication_test_hooks::Stage::PreparedRename, + LocalPublicationStage::Rename => prepared_publication_test_hooks::Stage::Rename, + LocalPublicationStage::Remove => prepared_publication_test_hooks::Stage::Remove, + }; + let (entered_tx, entered) = oneshot::channel(); + let (release, release_rx) = std::sync::mpsc::channel::<()>(); + let hook = prepared_publication_test_hooks::install_at(stage, &path, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + Ok(Self { + _hook: hook, + entered, + _release: release, + }) + } + + pub async fn entered(&mut self) -> std::result::Result<(), oneshot::error::RecvError> { + (&mut self.entered).await + } +} + #[cfg(all(test, windows))] pub(crate) mod windows_rename_test_hooks { use super::*; @@ -776,7 +823,7 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup, namespace_owners: Ve fsync_spawn_blocking(move || { // The batch worker may be cancelled while this syscall is still running. let _namespace_owners = namespace_owners; - #[cfg(all(test, not(windows)))] + #[cfg(all(any(test, feature = "test-util"), not(windows)))] prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::DirFsync, &dir); #[cfg(test)] { @@ -1912,7 +1959,7 @@ pub(crate) async fn remove_file_with_owner( let path = path.as_ref().to_path_buf(); let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await; run_blocking_namespace_operation(lease, move || { - #[cfg(all(test, not(windows)))] + #[cfg(all(any(test, feature = "test-util"), not(windows)))] prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Remove, &path); std::fs::remove_file(path) }) @@ -2136,7 +2183,7 @@ pub(crate) async fn rename_all_with_prepared_source( move || { validate_prepared_rename_source(&prepared_source, &src_file_path)?; let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?; - #[cfg(test)] + #[cfg(any(test, feature = "test-util"))] prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::PreparedRename, &dst_file_path); rename_prepared(&src_file_path, &dst_file_path, &preparation) } @@ -2267,7 +2314,7 @@ async fn reliable_rename_inner_with_lease( let base_dir = base_dir.clone(); move || { let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?; - #[cfg(all(test, not(windows)))] + #[cfg(all(any(test, feature = "test-util"), not(windows)))] { prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &src_file_path); prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &dst_file_path); diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index b54a69438..530362f39 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -442,7 +442,7 @@ 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, + BootstrapLocalTarget, 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, }; @@ -1787,7 +1787,7 @@ mod tests { // Build a minimal ECStore carrying an explicit instance context. Empty // pools/disks are sufficient: the Phase 5 accessors read only `self.ctx`. - fn build_store_with_ctx(ctx: Arc) -> Arc { + pub(super) fn build_store_with_ctx(ctx: Arc) -> Arc { let endpoint_pools = EndpointServerPools::default(); Arc::new(ECStore { id: uuid::Uuid::new_v4(), diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index d8ba54d1f..858b14be7 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -13,7 +13,10 @@ // limitations under the License. use super::*; -use crate::runtime::instance::InstanceContext; +use crate::bucket::utils::has_bad_path_component; +use crate::disk::error::{DiskError, Result as DiskResult}; +use crate::disk::{DeleteOptions, Disk, RenameDataGuards, RenameDataResp}; +use crate::runtime::instance::{InstanceContext, NamespaceCommitGuard}; use crate::runtime::sources as runtime_sources; use tracing::{debug, error}; @@ -22,6 +25,203 @@ const LOG_SUBSYSTEM_DISK_STARTUP: &str = "disk_startup"; const EVENT_LOCAL_DISK_ID_PREWARM_SKIPPED: &str = "local_disk_id_prewarm_skipped"; const EVENT_LOCK_CLIENT_INITIALIZATION_FAILED: &str = "lock_client_initialization_failed"; +/// An instance-bound capability for internal writes before ECStore/IAM startup. +/// Its private context and volume checks cannot be replaced by a caller guard. +#[derive(Clone)] +pub struct BootstrapLocalTarget { + ctx: Arc, +} + +impl BootstrapLocalTarget { + pub fn new(ctx: Arc) -> Self { + Self { ctx } + } + + pub fn is_for_store(&self, store: &ECStore) -> bool { + Arc::ptr_eq(&self.ctx, &store.ctx) + } + + pub async fn rename_local_data( + &self, + disk_ref: &str, + source: (&str, &str), + fi: &FileInfo, + destination: (&str, &str), + scanner_token: Option, + ) -> DiskResult { + if scanner_token.is_some() { + return Err(DiskError::other("bootstrap rename cannot use a scanner publication lease")); + } + validate_bootstrap_volume(source.0)?; + validate_bootstrap_volume(destination.0)?; + rename_local_data_with_ctx(&self.ctx, disk_ref, source, fi, destination, RenameDataGuards::default()).await + } + + pub async fn undo_local_write( + &self, + disk_ref: &str, + volume: &str, + path: &str, + fi: FileInfo, + opts: DeleteOptions, + ) -> DiskResult<()> { + validate_bootstrap_volume(volume)?; + undo_local_write_with_ctx(&self.ctx, disk_ref, volume, path, fi, opts).await + } +} + +fn validate_bootstrap_volume(volume: &str) -> DiskResult<()> { + // Prefix membership alone permits aliases such as .rustfs.sys/../bucket. + // Validate both raw rename volumes before any disk lookup or admission. + if has_bad_path_component(volume) || !is_meta_bucketname(volume) { + return Err(DiskError::FileAccessDenied); + } + Ok(()) +} + +impl ECStore { + /// Execute on this instance's active local disk through the physical owner. + pub async fn rename_local_data( + &self, + disk_ref: &str, + source: (&str, &str), + fi: &FileInfo, + destination: (&str, &str), + scanner_token: Option, + ) -> DiskResult { + let external_guard: Option> = if let Some(token) = scanner_token { + Some(Arc::new( + self.acquire_scanner_publication_lease_guard(token) + .await + .map_err(|err| DiskError::other(err.to_string()))?, + )) + } else { + None + }; + rename_local_data_with_ctx( + &self.ctx, + disk_ref, + source, + fi, + destination, + RenameDataGuards { + scanner_publication_lease_token: scanner_token, + external_guard, + namespace_owner: None, + }, + ) + .await + } + + pub async fn undo_local_write( + &self, + disk_ref: &str, + volume: &str, + path: &str, + fi: FileInfo, + opts: DeleteOptions, + ) -> DiskResult<()> { + undo_local_write_with_ctx(&self.ctx, disk_ref, volume, path, fi, opts).await + } +} + +// The optional ID is a cold lookup to cache only after final admission. +async fn local_disk_candidate(ctx: &Arc, disk_ref: &str) -> DiskResult<(DiskStore, Option)> { + let map = ctx.local_disk_map(); + if let Some(disk) = map.read().await.get(disk_ref).and_then(Option::as_ref).cloned() { + return Ok((disk, None)); + } + let disk_id = Uuid::parse_str(disk_ref).map_err(|_| DiskError::DiskNotFound)?; + let cached_path = ctx.local_disk_id_map().read().await.get(&disk_id).cloned(); + if let Some(path) = cached_path { + let cached_disk = map.read().await.get(&path).and_then(Option::as_ref).cloned(); + if let Some(disk) = cached_disk + && matches!(disk.as_ref(), Disk::Local(_)) + && disk.get_disk_id().await? == Some(disk_id) + { + return Ok((disk, None)); + } + } + let disks: Vec<_> = map.read().await.values().filter_map(Clone::clone).collect(); + // Disk identity may perform format I/O. No registry guard spans this await. + for disk in disks { + if matches!(disk.as_ref(), Disk::Local(_)) && disk.get_disk_id().await.ok().flatten() == Some(disk_id) { + return Ok((disk, Some(disk_id))); + } + } + Err(DiskError::DiskNotFound) +} + +async fn admit_local_disk( + ctx: &Arc, + disk: &DiskStore, + disk_id: Option, + volume: &str, +) -> DiskResult>> { + if !matches!(disk.as_ref(), Disk::Local(_)) { + return Err(DiskError::DiskNotFound); + } + let map = ctx.local_disk_map(); + let active = map.read().await; + if !active + .get(&disk.endpoint().to_string()) + .and_then(Option::as_ref) + .is_some_and(|current| Arc::ptr_eq(current, disk)) + { + return Err(DiskError::DiskNotFound); + } + // Preserve registry -> ID-cache lock order; no filesystem I/O under either. + if let Some(disk_id) = disk_id { + ctx.local_disk_id_map() + .write() + .await + .insert(disk_id, disk.endpoint().to_string()); + } + // Admission linearizes under the registry read: replacement/quarantine + // before this point rejects; later changes do not revoke physical I/O. + Ok((!is_meta_bucketname(volume)).then(|| ctx.begin_namespace_commit())) +} + +async fn rename_local_data_with_ctx( + ctx: &Arc, + disk_ref: &str, + source: (&str, &str), + fi: &FileInfo, + destination: (&str, &str), + mut guards: RenameDataGuards, +) -> DiskResult { + let (disk, disk_id) = local_disk_candidate(ctx, disk_ref).await?; + let owner = admit_local_disk(ctx, &disk, disk_id, destination.0).await?; + guards.namespace_owner = owner.as_ref().map(|owner| owner.clone() as Arc); + let result = disk + .rename_data_borrowed_with_fence_observed(source.0, source.1, fi, destination.0, destination.1, guards) + .await + .result; + drop(owner); + result +} + +async fn undo_local_write_with_ctx( + ctx: &Arc, + disk_ref: &str, + volume: &str, + path: &str, + fi: FileInfo, + opts: DeleteOptions, +) -> DiskResult<()> { + if !opts.undo_write { + return Err(DiskError::other("target undo requires undo_write")); + } + let (disk, disk_id) = local_disk_candidate(ctx, disk_ref).await?; + let owner = admit_local_disk(ctx, &disk, disk_id, volume).await?; + let physical_owner = owner.as_ref().map(|owner| owner.clone() as Arc); + let result = disk + .undo_write_with_namespace_owner(volume, path, fi, opts, physical_owner) + .await; + drop(owner); + result +} + async fn remember_local_disk_id(disk: &DiskStore) -> Option { remember_local_disk_id_with_instance_ctx(&crate::runtime::global::current_ctx(), disk).await } @@ -265,6 +465,522 @@ mod tests { }]) } + async fn target_disk(ctx: &Arc, root: &std::path::Path, id: Uuid) -> DiskStore { + let mut format = crate::layout::format::FormatV3::new(1, 1); + format.erasure.this = id; + format.erasure.sets[0][0] = id; + let meta = root.join(crate::disk::RUSTFS_META_BUCKET); + tokio::fs::create_dir_all(&meta).await.expect("create format volume"); + tokio::fs::write( + meta.join(crate::disk::FORMAT_CONFIG_FILE), + serde_json::to_vec(&format).expect("encode format"), + ) + .await + .expect("write real disk identity"); + let mut endpoint = Endpoint::try_from(root.to_str().expect("UTF-8 root")).expect("endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(0); + let disk = new_disk( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("open real local disk"); + assert_eq!(disk.get_disk_id().await.expect("read disk format identity"), Some(id)); + ctx.local_disk_map() + .write() + .await + .insert(disk.endpoint().to_string(), Some(disk.clone())); + disk + } + + fn target_file_info(object: &str, version: Uuid, body: &'static [u8]) -> FileInfo { + let mut fi = FileInfo::new(object, 1, 0); + fi.erasure.index = 1; + fi.version_id = Some(version); + fi.mod_time = Some(OffsetDateTime::now_utc()); + fi.size = i64::try_from(body.len()).expect("fixture length"); + fi.parts = vec![rustfs_filemeta::ObjectPartInfo { + number: 1, + size: body.len(), + actual_size: fi.size, + ..Default::default() + }]; + fi.data = Some(bytes::Bytes::from_static(body)); + fi.set_inline_data(); + fi + } + + async fn seed_target(disk: &DiskStore, volume: &str, object: &str, fi: FileInfo) -> Vec { + let dir = disk.path().join(volume); + tokio::fs::create_dir_all(&dir).await.expect("real fixture volume"); + disk.write_metadata(volume, volume, object, fi.clone()) + .await + .expect("seed real metadata"); + let read = disk + .read_version( + volume, + volume, + object, + &fi.version_id.expect("fixture version").to_string(), + &crate::disk::ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("read fixture before mutation"); + assert_eq!(read.data, fi.data, "fixture must contain readable inline bytes"); + tokio::fs::read(dir.join(object).join(crate::disk::STORAGE_FORMAT_FILE)) + .await + .expect("seeded metadata bytes") + } + + #[tokio::test] + async fn target_uuid_lookup_binds_real_disk_and_owner_to_one_instance() { + for warm in [false, true] { + let ctx_a = Arc::new(InstanceContext::new()); + let ctx_b = Arc::new(InstanceContext::new()); + let a = tempfile::tempdir().expect("A root"); + let b = tempfile::tempdir().expect("B root"); + let id = Uuid::new_v4(); + let disk_a = target_disk(&ctx_a, a.path(), id).await; + let disk_b = target_disk(&ctx_b, b.path(), id).await; + if warm { + assert!(record_local_disk_id_if_active(&ctx_a, &disk_a, id).await); + assert!(record_local_disk_id_if_active(&ctx_b, &disk_b, id).await); + } + let version = Uuid::new_v4(); + let fi = target_file_info("destination", version, b"new-A"); + for disk in [&disk_a, &disk_b] { + seed_target(disk, "target-bucket", "staged", fi.clone()).await; + } + let b_before = seed_target( + &disk_b, + "target-bucket", + "destination", + target_file_info("destination", version, b"old-B"), + ) + .await; + let store = super::super::tests::build_store_with_ctx(ctx_a.clone()); + store + .rename_local_data(&id.to_string(), ("target-bucket", "staged"), &fi, ("target-bucket", "destination"), None) + .await + .expect("rename on A"); + let read = disk_a + .read_version( + "target-bucket", + "target-bucket", + "destination", + &version.to_string(), + &crate::disk::ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("read committed A"); + assert_eq!(read.data, fi.data, "warm={warm}"); + assert_eq!( + tokio::fs::read(b.path().join("target-bucket/destination/xl.meta")) + .await + .expect("B metadata"), + b_before + ); + assert!(b.path().join("target-bucket/staged/xl.meta").exists()); + assert!(ctx_a.namespace_commit_generation() > 0); + assert_eq!(ctx_b.namespace_commit_generation(), 0); + assert!(!ctx_a.namespace_commits_pending()); + assert!(!ctx_b.namespace_commits_pending()); + assert_eq!(ctx_a.local_disk_id_map().read().await.get(&id), Some(&disk_a.endpoint().to_string())); + } + } + + #[tokio::test] + async fn target_admission_rejects_removed_quarantined_and_replaced_arcs() { + let ctx = Arc::new(InstanceContext::new()); + let root = tempfile::tempdir().expect("root"); + let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await; + let endpoint = disk.endpoint().to_string(); + for state in ["removed", "quarantined", "replaced"] { + let replacement = new_disk( + &disk.endpoint(), + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("separate active Arc"); + let map = ctx.local_disk_map(); + let mut entries = map.write().await; + match state { + "removed" => { + entries.remove(&endpoint); + } + "quarantined" => { + entries.insert(endpoint.clone(), None); + } + _ => { + entries.insert(endpoint.clone(), Some(replacement)); + } + } + drop(entries); + assert!( + matches!(admit_local_disk(&ctx, &disk, None, "target-bucket").await, Err(DiskError::DiskNotFound)), + "{state}" + ); + assert!(!ctx.namespace_commits_pending()); + assert_eq!(ctx.namespace_commit_generation(), 0); + } + } + + #[tokio::test] + async fn target_uuid_cache_cannot_admit_a_different_format_at_the_same_path() { + let ctx = Arc::new(InstanceContext::new()); + let root = tempfile::tempdir().expect("root"); + let old_id = Uuid::new_v4(); + let old = target_disk(&ctx, root.path(), old_id).await; + assert!(record_local_disk_id_if_active(&ctx, &old, old_id).await); + let replacement_id = Uuid::new_v4(); + let replacement = target_disk(&ctx, root.path(), replacement_id).await; + assert!(!Arc::ptr_eq(&old, &replacement)); + assert!(matches!( + local_disk_candidate(&ctx, &old_id.to_string()).await, + Err(DiskError::DiskNotFound) + )); + let (candidate, verified) = local_disk_candidate(&ctx, &replacement_id.to_string()) + .await + .expect("replacement UUID"); + assert!(Arc::ptr_eq(&candidate, &replacement)); + assert_eq!(verified, Some(replacement_id)); + assert!(!ctx.namespace_commits_pending()); + } + + #[tokio::test] + async fn bootstrap_rejects_user_volumes_aliases_and_scanner_tokens_without_mutation() { + let ctx = Arc::new(InstanceContext::new()); + let root = tempfile::tempdir().expect("root"); + let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await; + let target = BootstrapLocalTarget::new(ctx.clone()); + let fi = target_file_info("destination", Uuid::new_v4(), b"body"); + let user_before = seed_target(&disk, "victim", "staged", fi.clone()).await; + let meta_before = seed_target(&disk, ".rustfs.sys/tmp", "staged", fi.clone()).await; + for invalid in [ + "victim", + ".rustfs.sys/../victim", + ".rustfs.sys/./tmp", + ".rustfs.sys/ .. /victim", + ".rustfs.sys\\..\\victim", + ".minio.sys/../victim", + ] { + for (src, dst) in [(invalid, ".rustfs.sys/tmp"), (".rustfs.sys/tmp", invalid)] { + assert!( + target + .rename_local_data(&disk.endpoint().to_string(), (src, "staged"), &fi, (dst, "destination"), None) + .await + .is_err(), + "src={src}, dst={dst}" + ); + } + assert!( + target + .undo_local_write( + &disk.endpoint().to_string(), + invalid, + "staged", + fi.clone(), + DeleteOptions { + undo_write: true, + ..Default::default() + } + ) + .await + .is_err(), + "{invalid}" + ); + } + assert!( + target + .rename_local_data( + &disk.endpoint().to_string(), + (".rustfs.sys/tmp", "staged"), + &fi, + (".rustfs.sys/tmp", "destination"), + Some(Uuid::new_v4()) + ) + .await + .is_err() + ); + assert_eq!( + tokio::fs::read(root.path().join("victim/staged/xl.meta")) + .await + .expect("user source"), + user_before + ); + assert_eq!( + tokio::fs::read(root.path().join(".rustfs.sys/tmp/staged/xl.meta")) + .await + .expect("metadata source"), + meta_before + ); + assert!(!root.path().join("victim/destination").exists()); + assert!(!root.path().join(".rustfs.sys/tmp/destination").exists()); + assert_eq!(ctx.namespace_commit_generation(), 0); + assert!(!ctx.namespace_commits_pending()); + } + + #[tokio::test] + async fn bootstrap_allows_internal_multisegment_rename_without_namespace_owner() { + for volume in [".rustfs.sys/tmp", ".rustfs.sys/multipart", ".minio.sys/config"] { + let ctx = Arc::new(InstanceContext::new()); + let root = tempfile::tempdir().expect("root"); + let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await; + let fi = target_file_info("destination", Uuid::new_v4(), b"internal-CAS-body"); + seed_target(&disk, volume, "staged", fi.clone()).await; + BootstrapLocalTarget::new(ctx.clone()) + .rename_local_data(&disk.endpoint().to_string(), (volume, "staged"), &fi, (volume, "destination"), None) + .await + .expect("legitimate bootstrap metadata write"); + let read = disk + .read_version( + volume, + volume, + "destination", + &fi.version_id.expect("version").to_string(), + &crate::disk::ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("read bootstrap result"); + assert_eq!(read.data, fi.data); + assert_eq!(ctx.namespace_commit_generation(), 0); + assert!(!ctx.namespace_commits_pending()); + } + } + + #[cfg(not(windows))] + #[tokio::test] + async fn target_rename_cancellation_retains_real_namespace_and_scanner_owners() { + use crate::disk::os::prepared_publication_test_hooks as hooks; + let ctx = Arc::new(InstanceContext::new()); + let sibling = Arc::new(InstanceContext::new()); + let root = tempfile::tempdir().expect("root"); + let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await; + let store = super::super::tests::build_store_with_ctx(ctx.clone()); + let fi = target_file_info("destination", Uuid::new_v4(), b"physically-owned"); + seed_target(&disk, "target-bucket", "staged", fi.clone()).await; + let (token, _) = store + .acquire_scanner_publication_lease(0, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL) + .await + .expect("real scanner token in A"); + let destination = disk + .get_object_path_for_io_if_local("target-bucket", "destination/xl.meta") + .expect("local disk") + .expect("destination IO path"); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let _hook = hooks::install(&destination, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + let disk_ref = disk.endpoint().to_string(); + let mut rename = Box::pin(store.rename_local_data( + &disk_ref, + ("target-bucket", "staged"), + &fi, + ("target-bucket", "destination"), + Some(token), + )); + tokio::time::timeout(std::time::Duration::from_secs(10), async { + tokio::select! { + result = &mut rename => panic!("rename completed before physical pause: {result:?}"), + entered = entered_rx => entered.expect("physical rename entered"), + } + }) + .await + .expect("bounded physical entry"); + drop(rename); + assert!(store.scanner_data_usage_publication_blocked().await); + assert!(ctx.namespace_commits_pending()); + assert!(!sibling.namespace_commits_pending()); + assert!( + store + .rename_local_data(&disk_ref, ("target-bucket", "staged"), &fi, ("target-bucket", "another"), Some(token)) + .await + .is_err(), + "real pending rename blocks another scanner publication" + ); + assert!(store.release_scanner_publication_lease(token).await, "remove registered token"); + let gate = ctx.data_movement_operation_gate(); + assert!( + gate.clone().try_write_owned().is_err(), + "physical operation still owns the scanner read guard" + ); + drop(release_tx); + let _drained = tokio::time::timeout(std::time::Duration::from_secs(10), gate.write_owned()) + .await + .expect("physical tail must release scanner guard"); + tokio::time::timeout(std::time::Duration::from_secs(10), async { + while ctx.namespace_commits_pending() { + tokio::task::yield_now().await; + } + }) + .await + .expect("namespace owner drains"); + let read = disk + .read_version( + "target-bucket", + "target-bucket", + "destination", + &fi.version_id.expect("version").to_string(), + &crate::disk::ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("read actual late commit"); + assert_eq!(read.data, fi.data); + assert!(ctx.namespace_commit_generation() >= 2); + assert_eq!(sibling.namespace_commit_generation(), 0); + } + + #[tokio::test] + async fn target_ready_rejects_unknown_foreign_released_and_expired_scanner_tokens() { + let ctx = Arc::new(InstanceContext::new()); + let other = Arc::new(InstanceContext::new()); + let store = super::super::tests::build_store_with_ctx(ctx.clone()); + let other_store = super::super::tests::build_store_with_ctx(other); + let root = tempfile::tempdir().expect("root"); + let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await; + let fi = target_file_info("destination", Uuid::new_v4(), b"unchanged"); + let before = seed_target(&disk, "target-bucket", "staged", fi.clone()).await; + let ttl = crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL; + let (foreign, _) = other_store.acquire_scanner_publication_lease(0, ttl).await.expect("B token"); + let (released, _) = store.acquire_scanner_publication_lease(0, ttl).await.expect("A token"); + assert!(store.release_scanner_publication_lease(released).await); + let (valid, _) = store.acquire_scanner_publication_lease(0, ttl).await.expect("new A token"); + for token in [Uuid::new_v4(), foreign, released] { + assert!( + store + .rename_local_data( + &disk.endpoint().to_string(), + ("target-bucket", "staged"), + &fi, + ("target-bucket", "destination"), + Some(token) + ) + .await + .is_err() + ); + } + tokio::time::pause(); + tokio::time::advance(ttl + std::time::Duration::from_secs(1)).await; + tokio::time::resume(); + assert!( + store + .rename_local_data( + &disk.endpoint().to_string(), + ("target-bucket", "staged"), + &fi, + ("target-bucket", "destination"), + Some(valid) + ) + .await + .is_err(), + "expired real token" + ); + let _ = other_store.release_scanner_publication_lease(foreign).await; + assert_eq!( + tokio::fs::read(root.path().join("target-bucket/staged/xl.meta")) + .await + .expect("source bytes"), + before + ); + assert!(!root.path().join("target-bucket/destination").exists()); + assert!(!ctx.namespace_commits_pending()); + } + + #[cfg(not(windows))] + #[tokio::test] + #[serial_test::serial] + async fn target_ordinary_timeout_keeps_its_physical_namespace_owner() { + use crate::disk::os::prepared_publication_test_hooks as hooks; + temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("1"))], async { + let ctx = Arc::new(InstanceContext::new()); + let store = super::super::tests::build_store_with_ctx(ctx.clone()); + let root = tempfile::tempdir().expect("root"); + let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await; + let fi = target_file_info("destination", Uuid::new_v4(), b"timed-out-physical-commit"); + seed_target(&disk, "target-bucket", "staged", fi.clone()).await; + let path = disk + .get_object_path_for_io_if_local("target-bucket", "destination/xl.meta") + .expect("local") + .expect("destination IO path"); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let _hook = hooks::install(&path, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + let disk_ref = disk.endpoint().to_string(); + let mut rename = Box::pin(store.rename_local_data( + &disk_ref, + ("target-bucket", "staged"), + &fi, + ("target-bucket", "destination"), + None, + )); + tokio::time::timeout(std::time::Duration::from_secs(10), async { + tokio::select! { + result = &mut rename => panic!("completed before physical pause: {result:?}"), + entered = entered_rx => entered.expect("physical entry"), + } + }) + .await + .expect("bounded entry"); + tokio::time::pause(); + tokio::time::advance(std::time::Duration::from_secs(2)).await; + tokio::time::resume(); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), &mut rename) + .await + .expect("ordinary deadline remains enabled"); + assert!(matches!(result, Err(DiskError::Timeout)), "{result:?}"); + drop(rename); + assert!(ctx.namespace_commits_pending(), "timeout is not a physical drain"); + drop(release_tx); + tokio::time::timeout(std::time::Duration::from_secs(10), async { + while ctx.namespace_commits_pending() { + tokio::task::yield_now().await; + } + }) + .await + .expect("late physical owner drains"); + let read = disk + .read_version( + "target-bucket", + "target-bucket", + "destination", + &fi.version_id.expect("version").to_string(), + &crate::disk::ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("read actual timeout tail"); + assert_eq!(read.data, fi.data); + }) + .await; + } + #[test] fn endpoint_rpc_authority_preserves_port_and_ipv6_brackets() { let endpoint = Endpoint::try_from("https://127.0.0.1:9001/d1").expect("URL endpoint"); diff --git a/rustfs/src/app/context/server_slot.rs b/rustfs/src/app/context/server_slot.rs index 56ccedf57..99c260144 100644 --- a/rustfs/src/app/context/server_slot.rs +++ b/rustfs/src/app/context/server_slot.rs @@ -26,13 +26,14 @@ //! server is not ready rather than that another server's global context applies. use super::global::{AppContext, get_global_app_context}; -use crate::app::storage_api::context::ECStore; +use crate::app::storage_api::context::{BootstrapLocalTarget, ECStore, InstanceContext}; use std::sync::{Arc, OnceLock}; /// Late-bound, per-server handle to the application context. #[derive(Default)] pub struct ServerContextSlot { app_context: OnceLock>, + bootstrap_target: Option, heal_topology_fingerprint: Arc>, } @@ -50,15 +51,47 @@ impl ServerContextSlot { pub fn new() -> Arc { Arc::new(Self { app_context: OnceLock::new(), + bootstrap_target: None, heal_topology_fingerprint: Arc::new(tokio::sync::OnceCell::new()), }) } + /// Bind the listener to its foundation before it can accept requests. + pub fn with_instance_context(ctx: Arc) -> Arc { + Arc::new(Self { + bootstrap_target: Some(BootstrapLocalTarget::new(ctx)), + ..Self::default() + }) + } + /// Install this server's application context (once). Returns `false` if /// the slot was already installed; the first installation wins, matching /// the process-global singleton's `get_or_init` semantics. pub fn install(&self, context: Arc) -> bool { - self.app_context.set(context).is_ok() + self.try_install(context).is_ok() + } + + /// Claim the slot before any process-global application publication. + /// Repeated installation, even of the same Arc, is an explicit conflict. + pub fn try_install(&self, context: Arc) -> std::io::Result<()> { + if self + .bootstrap_target + .as_ref() + .is_some_and(|target| !target.is_for_store(&context.object_store())) + { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "application context does not belong to this server foundation", + )); + } + self.app_context.set(context).map_err(|_| { + std::io::Error::new(std::io::ErrorKind::AlreadyExists, "server application context is already installed") + }) + } + + /// Immutable, restricted startup capability; never resolves an ambient store. + pub fn bootstrap_target(&self) -> Option { + self.bootstrap_target.clone() } /// This server's installed application context, if startup has completed. diff --git a/rustfs/src/app/context/startup.rs b/rustfs/src/app/context/startup.rs index b19b1aca0..77966fe6c 100644 --- a/rustfs/src/app/context/startup.rs +++ b/rustfs/src/app/context/startup.rs @@ -37,8 +37,8 @@ impl AppContext { // also publishes to the process default (first server wins) so legacy // free-function readers keep resolving the first server's context. let context = Arc::new(AppContext::with_default_interfaces(store, iam, kms_interface)); - publish_global_app_context(context.clone()); - let _ = server_ctx.install(context); + server_ctx.try_install(context.clone())?; + publish_global_app_context(context); Ok(()) } } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 2719ccb72..2be6263bf 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -1259,7 +1259,7 @@ pub(crate) mod context { pub(crate) use super::EndpointServerPools; pub(crate) use super::bucket; pub(crate) use super::runtime; - pub(crate) use crate::storage::storage_api::{ECStore, EndpointServerPools}; + pub(crate) use crate::storage::storage_api::{BootstrapLocalTarget, ECStore, EndpointServerPools, InstanceContext}; #[cfg(test)] pub(crate) use crate::storage::storage_api::{Endpoint, Endpoints, PoolEndpoints}; } diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index d82d67f20..41a9e3519 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -36,7 +36,9 @@ use crate::server::{ }; use crate::storage_api::server::http as storage; use crate::storage_api::server::http::rpc::InternodeRpcService; +#[cfg(test)] use crate::storage_api::server::http::tonic_service::make_server; +use crate::storage_api::server::http::tonic_service::make_server_for_slot; use crate::storage_api::server::http::{ ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, verify_tonic_rpc_signature_with_bootstrap, @@ -1834,7 +1836,7 @@ fn process_connection( // each service in the auth interceptor. let rpc_max_message_size = rustfs_protos::internode_rpc_max_message_size(); let node_service = InterceptedService::new( - NodeServiceServer::new(make_server()) + NodeServiceServer::new(make_server_for_slot(Arc::clone(&server_ctx))) .max_decoding_message_size(rpc_max_message_size) .max_encoding_message_size(rpc_max_message_size), check_auth, diff --git a/rustfs/src/startup_embedded.rs b/rustfs/src/startup_embedded.rs index e23bf9563..7738f85ba 100644 --- a/rustfs/src/startup_embedded.rs +++ b/rustfs/src/startup_embedded.rs @@ -124,9 +124,6 @@ pub(crate) async fn run_embedded_startup(args: EmbeddedStartupArgs) -> Result Result Result<()> { // the storage path explicitly (Phase 5 follow-up, backlog#1052); a future // multi-instance server constructs its own context here instead. let instance_ctx = bootstrap_instance_ctx(); - // This server's request-path context slot (backlog#1052 S2): handed to the - // HTTP service now, installed once IAM bootstrap completes. - let server_ctx = ServerContextSlot::new(); - let StartupListenContext { readiness, server_addr, @@ -152,6 +148,7 @@ async fn run(config: Config) -> Result<()> { } = init_startup_listen_context(&config, &instance_ctx).await?; let endpoint_pools = init_startup_storage_foundation(&server_address, &config.volumes, &instance_ctx).await?; + let server_ctx = ServerContextSlot::with_instance_context(instance_ctx.clone()); let StartupHttpServers { state_manager, s3_shutdown_tx, diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 6b3596390..bf93a77e7 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -28,9 +28,9 @@ use crate::storage::storage_api::rpc_consumer::node_service::{ SCANNER_PUBLICATION_LEASE_TTL_MS, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt as _, StorageResult, all_local_disk_path, find_local_disk_by_ref, reload_transition_tier_config, }; -use crate::storage::storage_api::runtime_sources_consumer::{EndpointServerPools, runtime_sources}; +use crate::storage::storage_api::runtime_sources_consumer::{EndpointServerPools, ServerContextSlot, runtime_sources}; use crate::storage::storage_api::{ - sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, + BootstrapLocalTarget, sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_mutation_body_digest_reject_unsigned, }; use bytes::Bytes; @@ -482,6 +482,13 @@ mod metrics; pub struct NodeService { local_peer: LocalPeerS3Client, context: Option>, + server_ctx: Option>, +} + +enum LocalMutationTarget { + Ready(Arc), + Bootstrap(BootstrapLocalTarget), + Unbound, } impl std::fmt::Debug for NodeService { @@ -507,7 +514,19 @@ pub fn make_server() -> NodeService { pub fn make_server_for_context(context: Option>) -> NodeService { let local_peer = LocalPeerS3Client::new(None, None); - NodeService { local_peer, context } + NodeService { + local_peer, + context, + server_ctx: None, + } +} + +pub(crate) fn make_server_for_slot(server_ctx: Arc) -> NodeService { + // Unrelated RPCs retain their existing context policy. Target mutations + // resolve exclusively through this listener slot on each request. + let mut service = make_server(); + service.server_ctx = Some(server_ctx); + service } #[derive(Clone, Debug, Default)] @@ -1074,6 +1093,24 @@ impl heal_control_service_server::HealControlService for HealControlRpcService { } impl NodeService { + fn local_mutation_target(&self) -> LocalMutationTarget { + if let Some(slot) = &self.server_ctx { + // Capture exactly once per request, not at connection acceptance. + // A captured Bootstrap request cannot upgrade across a later await. + if let Some(store) = slot.installed_object_store() { + LocalMutationTarget::Ready(store) + } else if let Some(target) = slot.bootstrap_target() { + LocalMutationTarget::Bootstrap(target) + } else { + LocalMutationTarget::Unbound + } + } else if let Some(context) = &self.context { + LocalMutationTarget::Ready(context.object_store()) + } else { + LocalMutationTarget::Unbound + } + } + fn resolve_object_store(&self) -> Option> { let context = self.context.clone().or_else(runtime_sources::current_app_context); runtime_sources::current_object_store_handle_for_context(context.as_deref()) @@ -2680,6 +2717,7 @@ mod tests { validate_admin_heal_control_start, }; use crate::storage::rpc::node_service::heal::heal_topology_fingerprint; + use crate::storage::storage_api::ecstore_disk::DiskAPI as _; use crate::storage::storage_api::rpc_consumer::node_service::{DiskError, HealBucketInfo}; use crate::storage::storage_api::set_tonic_canonical_body_digest; use crate::storage::storage_api::{ @@ -4530,6 +4568,411 @@ mod tests { assert!(rename_response.error.is_some()); } + struct TargetRpcFixture { + _root: tempfile::TempDir, + env: rustfs_test_utils::TestECStoreEnv, + instance: Arc, + context: Arc, + iam: Arc>, + } + + async fn target_rpc_fixture() -> TargetRpcFixture { + super::timeout(Duration::from_secs(90), async { + let root = tempfile::tempdir().expect("target RPC root"); + let env = rustfs_test_utils::TestECStoreEnv::builder() + .base_dir(root.path()) + .init_bucket_metadata(false) + .build() + .await; + ObjectStore::new(env.ecstore.clone()) + .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX)) + .await + .expect("seed real IAM format"); + let iam = rustfs_iam::build_iam_sys(env.ecstore.clone()) + .await + .expect("build fixture IAM"); + let context = Arc::new(crate::runtime_sources::AppContext::with_default_interfaces( + env.ecstore.clone(), + iam.clone(), + Arc::new(KmsServiceManager::new()), + )); + let instance = crate::storage::storage_api::bootstrap_instance_ctx(); + assert!( + super::BootstrapLocalTarget::new(instance.clone()).is_for_store(&env.ecstore), + "the standard builder must use this exact instance context" + ); + super::timeout(Duration::from_secs(10), async { + while env.ecstore.scanner_data_usage_publication_blocked().await { + tokio::task::yield_now().await; + } + }) + .await + .expect("startup namespace commits drain before test"); + TargetRpcFixture { + _root: root, + env, + instance, + context, + iam, + } + }) + .await + .expect("bounded real fixture initialization") + } + + async fn stage_target_rpc(fixture: &TargetRpcFixture) -> (super::DiskStore, rustfs_filemeta::FileInfo, Vec) { + use crate::storage::storage_api::ecstore_disk::{DiskAPI, ReadOptions}; + let disk = fixture + .instance + .local_disk_map() + .read() + .await + .values() + .find_map(Clone::clone) + .expect("local target"); + let mut fi = rustfs_filemeta::FileInfo::new("destination", 1, 0); + fi.erasure.index = 1; + fi.version_id = Some(Uuid::new_v4()); + fi.mod_time = Some(OffsetDateTime::now_utc()); + fi.size = 17; + fi.parts = vec![rustfs_filemeta::ObjectPartInfo { + number: 1, + size: 17, + actual_size: 17, + ..Default::default() + }]; + fi.data = Some(Bytes::from_static(b"target-rpc-inline")); + fi.set_inline_data(); + disk.make_volume("target-rpc").await.expect("target volume"); + disk.write_metadata("target-rpc", "target-rpc", "staged", fi.clone()) + .await + .expect("stage real inline body"); + let read = disk + .read_version( + "target-rpc", + "target-rpc", + "staged", + &fi.version_id.expect("version").to_string(), + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("read staged body before mutation"); + assert_eq!(read.data, fi.data); + let before = tokio::fs::read(disk.path().join("target-rpc/staged/xl.meta")) + .await + .expect("staged bytes"); + (disk, fi, before) + } + + fn target_rename_request(disk: &super::DiskStore, fi: &rustfs_filemeta::FileInfo) -> Request { + let mut request = Request::new(RenameDataRequest { + disk: disk.endpoint().to_string(), + src_volume: "target-rpc".to_string(), + src_path: "staged".to_string(), + dst_volume: "target-rpc".to_string(), + dst_path: "destination".to_string(), + file_info: serde_json::to_string(fi).expect("real FileInfo JSON"), + ..Default::default() + }); + let body = rustfs_protos::canonical_rename_data_request_body(request.get_ref()).expect("canonical target body"); + set_tonic_canonical_body_digest(&mut request, &body).expect("body digest"); + // Direct-handler precondition only; this does not stand in for wire authentication. + mark_v2_authenticated(&mut request); + request + } + + #[tokio::test] + async fn target_slot_rejects_mismatched_and_repeated_install_before_global_publication() { + let fixture = target_rpc_fixture().await; + assert!( + crate::runtime_sources::current_app_context().is_none(), + "requires a separate nextest process" + ); + let wrong = super::ServerContextSlot::with_instance_context(crate::storage::storage_api::new_instance_ctx()); + let error = crate::runtime_sources::AppContext::ensure_startup_after_iam( + fixture.env.ecstore.clone(), + Arc::new(KmsServiceManager::new()), + &wrong, + fixture.iam.clone(), + ) + .expect_err("mismatched startup must fail"); + assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput); + assert!(wrong.installed_app_context().is_none()); + assert!( + crate::runtime_sources::current_app_context().is_none(), + "failed install must not publish globally" + ); + assert!(!wrong.install(fixture.context.clone()), "bool adapter cannot bypass identity checks"); + let slot = super::ServerContextSlot::with_instance_context(fixture.instance.clone()); + crate::runtime_sources::AppContext::ensure_startup_after_iam( + fixture.env.ecstore.clone(), + Arc::new(KmsServiceManager::new()), + &slot, + fixture.iam.clone(), + ) + .expect("matching startup installation"); + let installed = slot.installed_app_context().expect("installed A"); + assert!(Arc::ptr_eq( + &crate::runtime_sources::current_app_context().expect("published A"), + &installed + )); + assert_eq!( + slot.try_install(installed.clone()) + .expect_err("same Arc is still a duplicate") + .kind(), + std::io::ErrorKind::AlreadyExists + ); + assert!(!slot.install(installed.clone())); + assert!(Arc::ptr_eq(&slot.installed_app_context().expect("first winner retained"), &installed)); + } + + #[tokio::test] + async fn target_slot_captures_bootstrap_once_and_next_request_observes_ready() { + let fixture = target_rpc_fixture().await; + let (disk, fi, before) = stage_target_rpc(&fixture).await; + let slot = super::ServerContextSlot::with_instance_context(fixture.instance.clone()); + let service = super::make_server_for_slot(slot.clone()); + let captured = service.local_mutation_target(); + slot.try_install(fixture.context.clone()) + .expect("install after the request captures bootstrap"); + let super::LocalMutationTarget::Bootstrap(target) = captured else { + panic!("pre-install request must capture bootstrap"); + }; + assert!( + target + .rename_local_data( + &disk.endpoint().to_string(), + ("target-rpc", "staged"), + &fi, + ("target-rpc", "destination"), + None + ) + .await + .is_err(), + "captured request cannot acquire Ready privileges" + ); + assert_eq!( + tokio::fs::read(disk.path().join("target-rpc/staged/xl.meta")) + .await + .expect("original source"), + before + ); + assert!(!disk.path().join("target-rpc/destination").exists()); + assert!( + matches!(service.local_mutation_target(), super::LocalMutationTarget::Ready(_)), + "the same service must read the installed slot for its next request" + ); + let result = service + .rename_data(target_rename_request(&disk, &fi)) + .await + .expect("ready handler") + .into_inner(); + assert!(result.success, "{:?}", result.error); + } + + #[tokio::test] + async fn target_unbound_slot_never_mutates_a_published_global_store() { + let fixture = target_rpc_fixture().await; + let (disk, fi, before) = stage_target_rpc(&fixture).await; + let published = crate::runtime_sources::publish_test_app_context(fixture.context.clone()); + assert!(Arc::ptr_eq(&published, &fixture.context)); + let service = super::make_server_for_slot(super::ServerContextSlot::new()); + let result = service + .rename_data(target_rename_request(&disk, &fi)) + .await + .expect("handler reply") + .into_inner(); + assert!(!result.success); + assert!(result.error.is_some()); + assert_eq!( + tokio::fs::read(disk.path().join("target-rpc/staged/xl.meta")) + .await + .expect("source remains"), + before + ); + assert!(!disk.path().join("target-rpc/destination").exists()); + assert!(!fixture.env.ecstore.scanner_data_usage_publication_blocked().await); + } + + #[tokio::test] + async fn target_undo_rejects_force_delete_marker_before_mutation() { + let fixture = target_rpc_fixture().await; + let (disk, fi, before) = stage_target_rpc(&fixture).await; + let service = make_server_for_context(Some(fixture.context.clone())); + let opts = crate::storage::storage_api::ecstore_disk::DeleteOptions { + undo_write: true, + ..Default::default() + }; + let mut request = Request::new(DeleteVersionRequest { + disk: disk.endpoint().to_string(), + volume: "target-rpc".to_string(), + path: "staged".to_string(), + file_info: serde_json::to_string(&fi).expect("FileInfo"), + opts: serde_json::to_string(&opts).expect("opts"), + force_del_marker: true, + ..Default::default() + }); + let body = rustfs_protos::canonical_delete_version_request_body(request.get_ref()).expect("canonical undo body"); + set_tonic_canonical_body_digest(&mut request, &body).expect("body digest"); + mark_v2_authenticated(&mut request); + let result = service.delete_version(request).await.expect("handler reply").into_inner(); + assert!(!result.success); + assert!(result.error.is_some()); + assert_eq!( + tokio::fs::read(disk.path().join("target-rpc/staged/xl.meta")) + .await + .expect("source remains"), + before + ); + assert!(!fixture.env.ecstore.scanner_data_usage_publication_blocked().await); + } + + #[cfg(not(windows))] + #[tokio::test] + async fn target_handler_cancellation_retains_namespace_through_physical_rename() { + use crate::storage::storage_api::{ + LocalPublicationPause, LocalPublicationStage, + ecstore_disk::{DiskAPI, ReadOptions}, + }; + let fixture = target_rpc_fixture().await; + let (disk, fi, _) = stage_target_rpc(&fixture).await; + let slot = super::ServerContextSlot::with_instance_context(fixture.instance.clone()); + slot.try_install(fixture.context.clone()).expect("ready target"); + let service = super::make_server_for_slot(slot); + let mut pause = + LocalPublicationPause::install(&disk, "target-rpc", "destination/xl.meta", LocalPublicationStage::PreparedRename) + .expect("install scoped physical pause"); + let mut handler = Box::pin(service.rename_data(target_rename_request(&disk, &fi))); + super::timeout(Duration::from_secs(10), async { + tokio::select! { + result = &mut handler => panic!("handler completed before physical entry: {result:?}"), + entered = pause.entered() => entered.expect("physical executor entered"), + } + }) + .await + .expect("bounded physical entry"); + drop(handler); + assert!( + fixture.env.ecstore.scanner_data_usage_publication_blocked().await, + "dropping the actual target handler must not release its physical owner" + ); + drop(pause); + super::timeout(Duration::from_secs(10), async { + while fixture.env.ecstore.scanner_data_usage_publication_blocked().await { + tokio::task::yield_now().await; + } + }) + .await + .expect("physical owner must drain"); + let read = disk + .read_version( + "target-rpc", + "target-rpc", + "destination", + &fi.version_id.expect("version").to_string(), + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("read real late commit"); + assert_eq!(read.data, fi.data); + } + + #[cfg(not(windows))] + #[tokio::test] + async fn target_undo_handler_cancellation_retains_owner_until_backup_restoration() { + use crate::storage::storage_api::{ + LocalPublicationPause, LocalPublicationStage, + ecstore_disk::{DeleteOptions, DiskAPI, ReadOptions}, + }; + let fixture = target_rpc_fixture().await; + let (disk, fi, _) = stage_target_rpc(&fixture).await; + let mut old = fi.clone(); + old.data = Some(Bytes::from_static(b"previous-rpc-body")); + assert_eq!(old.data.as_ref().expect("old body").len(), 17); + disk.write_metadata("target-rpc", "target-rpc", "destination", old.clone()) + .await + .expect("old actual version"); + let old_bytes = tokio::fs::read(disk.path().join("target-rpc/destination/xl.meta")) + .await + .expect("old metadata bytes"); + let committed = fixture + .env + .ecstore + .rename_local_data( + &disk.endpoint().to_string(), + ("target-rpc", "staged"), + &fi, + ("target-rpc", "destination"), + None, + ) + .await + .expect("real overwrite creates rollback backup"); + let opts = DeleteOptions { + undo_write: true, + old_data_dir: Some(committed.rollback_data_dir.expect("real rollback backup")), + ..Default::default() + }; + let service = make_server_for_context(Some(fixture.context.clone())); + let mut request = Request::new(DeleteVersionRequest { + disk: disk.endpoint().to_string(), + volume: "target-rpc".to_string(), + path: "destination".to_string(), + file_info: serde_json::to_string(&fi).expect("FileInfo"), + opts: serde_json::to_string(&opts).expect("undo options"), + ..Default::default() + }); + let body = rustfs_protos::canonical_delete_version_request_body(request.get_ref()).expect("canonical undo body"); + set_tonic_canonical_body_digest(&mut request, &body).expect("body digest"); + mark_v2_authenticated(&mut request); + let mut pause = LocalPublicationPause::install(&disk, "target-rpc", "destination/xl.meta", LocalPublicationStage::Rename) + .expect("pause actual backup restoration"); + let mut handler = Box::pin(service.delete_version(request)); + super::timeout(Duration::from_secs(10), async { + tokio::select! { + result = &mut handler => panic!("undo completed before physical entry: {result:?}"), + entered = pause.entered() => entered.expect("physical restore entered"), + } + }) + .await + .expect("bounded physical restore entry"); + drop(handler); + assert!(fixture.env.ecstore.scanner_data_usage_publication_blocked().await); + drop(pause); + super::timeout(Duration::from_secs(10), async { + while fixture.env.ecstore.scanner_data_usage_publication_blocked().await { + tokio::task::yield_now().await; + } + }) + .await + .expect("restore owner drains"); + assert_eq!( + tokio::fs::read(disk.path().join("target-rpc/destination/xl.meta")) + .await + .expect("restored bytes"), + old_bytes + ); + let read = disk + .read_version( + "target-rpc", + "target-rpc", + "destination", + &fi.version_id.expect("version").to_string(), + &ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("restored readable version"); + assert_eq!(read.data, old.data); + } + #[tokio::test] async fn rename_data_same_uuid_uses_captured_instance_instead_of_global_disk() { use crate::storage::storage_api::{ diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index 19fb86c80..8dd011f2e 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::NodeService; +use super::{LocalMutationTarget, NodeService}; use crate::storage::storage_api::rpc_consumer::node_service::{ BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, FileInfoVersions, ReadMultipleReq, ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, validate_batch_read_version_item_count, @@ -39,6 +39,46 @@ use tonic::{Request, Response, Status}; use tracing::debug; use uuid::Uuid; +impl LocalMutationTarget { + async fn rename_local_data( + &self, + disk_ref: &str, + source: (&str, &str), + fi: &FileInfo, + destination: (&str, &str), + scanner_token: Option, + ) -> Result { + match self { + Self::Ready(store) => { + store + .rename_local_data(disk_ref, source, fi, destination, scanner_token) + .await + } + Self::Bootstrap(target) => { + target + .rename_local_data(disk_ref, source, fi, destination, scanner_token) + .await + } + Self::Unbound => Err(DiskError::other("target disk instance is unavailable")), + } + } + + async fn undo_local_write( + &self, + disk_ref: &str, + volume: &str, + path: &str, + fi: FileInfo, + opts: DeleteOptions, + ) -> Result<(), DiskError> { + match self { + Self::Ready(store) => store.undo_local_write(disk_ref, volume, path, fi, opts).await, + Self::Bootstrap(target) => target.undo_local_write(disk_ref, volume, path, fi, opts).await, + Self::Unbound => Err(DiskError::other("target disk instance is unavailable")), + } + } +} + /// Initial capacity hint (bytes) for typical small msgpack requests and responses. const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512; const FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT: usize = 1024; @@ -670,55 +710,59 @@ impl NodeService { "delete_version", )?; let request = request.into_inner(); - if let Some(disk) = self.find_disk(&request.disk).await { - let file_info = match decode_msgpack_or_json::(&request.file_info_bin, &request.file_info, "FileInfo") { - Ok(file_info) => file_info, - Err(err) => { - return Ok(Response::new(DeleteVersionResponse { - success: false, - raw_file_info: "".to_string(), - error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()), - })); - } - }; - let opts = match decode_msgpack_or_json::(&request.opts_bin, &request.opts, "DeleteOptions") { - Ok(opts) => opts, - Err(err) => { - return Ok(Response::new(DeleteVersionResponse { - success: false, - raw_file_info: "".to_string(), - error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()), - })); - } - }; - match disk - .delete_version(&request.volume, &request.path, file_info, request.force_del_marker, opts) + let file_info = match decode_msgpack_or_json::(&request.file_info_bin, &request.file_info, "FileInfo") { + Ok(file_info) => file_info, + Err(err) => { + return Ok(Response::new(DeleteVersionResponse { + success: false, + raw_file_info: "".to_string(), + error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()), + })); + } + }; + let opts = match decode_msgpack_or_json::(&request.opts_bin, &request.opts, "DeleteOptions") { + Ok(opts) => opts, + Err(err) => { + return Ok(Response::new(DeleteVersionResponse { + success: false, + raw_file_info: "".to_string(), + error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()), + })); + } + }; + let result = if opts.undo_write { + if request.force_del_marker { + Err(DiskError::other("undo_write cannot force a delete marker")) + } else { + let target = self.local_mutation_target(); + target + .undo_local_write(&request.disk, &request.volume, &request.path, file_info, opts) + .await + } + } else if let Some(disk) = self.find_disk(&request.disk).await { + disk.delete_version(&request.volume, &request.path, file_info, request.force_del_marker, opts) .await - { - Ok(raw_file_info) => match serde_json::to_string(&raw_file_info) { - Ok(raw_file_info) => Ok(Response::new(DeleteVersionResponse { - success: true, - raw_file_info, - error: None, - })), - Err(err) => Ok(Response::new(DeleteVersionResponse { - success: false, - raw_file_info: "".to_string(), - error: Some(DiskError::other(format!("encode data failed: {err}")).into()), - })), - }, + } else { + Err(DiskError::other("cannot find disk")) + }; + match result { + Ok(raw_file_info) => match serde_json::to_string(&raw_file_info) { + Ok(raw_file_info) => Ok(Response::new(DeleteVersionResponse { + success: true, + raw_file_info, + error: None, + })), Err(err) => Ok(Response::new(DeleteVersionResponse { success: false, raw_file_info: "".to_string(), - error: Some(err.into()), + error: Some(DiskError::other(format!("encode data failed: {err}")).into()), })), - } - } else { - Ok(Response::new(DeleteVersionResponse { + }, + Err(err) => Ok(Response::new(DeleteVersionResponse { success: false, raw_file_info: "".to_string(), - error: Some(DiskError::other("cannot find disk".to_string()).into()), - })) + error: Some(err.into()), + })), } } @@ -1206,98 +1250,59 @@ impl NodeService { "rename_data", )?; let request = request.into_inner(); - if let Some(disk) = self.find_disk(&request.disk).await { - let decoded_file_info = match decode_rename_data_request_file_info(&request.file_info_bin, &request.file_info) { - Ok(file_info) => file_info, - Err(err) => { - return Ok(Response::new(RenameDataResponse { - success: false, - rename_data_resp: String::new(), - rename_data_resp_bin: Vec::new().into(), - error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()), - })); - } - }; - let scanner_publication_lease_token = if request.scanner_publication_lease_token.is_empty() { - None - } else { - let token = Uuid::from_slice(&request.scanner_publication_lease_token) - .map_err(|_| Status::invalid_argument("scanner publication lease token must be a UUID"))?; - if token.is_nil() { - return Err(Status::invalid_argument("scanner publication lease token must not be nil")); - } - Some(token) - }; - // The target owns this read guard. It must span the complete - // disk rename, not merely the preflight, so a movement transition - // cannot restart after validation and before rename linearization. - let scanner_publication_lease_guard: Option> = - if let Some(token) = scanner_publication_lease_token { - let Some(store) = self.resolve_object_store() else { - return Ok(Response::new(RenameDataResponse { - success: false, - rename_data_resp: String::new(), - rename_data_resp_bin: Vec::new().into(), - error: Some(DiskError::other("scanner publication lease owner is unavailable").into()), - })); - }; - match store.acquire_scanner_publication_lease_guard(token).await { - Ok(guard) => Some(Arc::new(guard)), - Err(err) => { - return Ok(Response::new(RenameDataResponse { - success: false, - rename_data_resp: String::new(), - rename_data_resp_bin: Vec::new().into(), - error: Some(DiskError::other(err.to_string()).into()), - })); - } - } - } else { - None - }; - let request_decoded_from_msgpack = decoded_file_info.from_msgpack; - match disk - .rename_data_borrowed_with_fence_and_guard( - &request.src_volume, - &request.src_path, - &decoded_file_info.value, - &request.dst_volume, - &request.dst_path, - scanner_publication_lease_token, - scanner_publication_lease_guard, - ) - .await - { - Ok(rename_data_resp) => { - match encode_rename_data_response_payloads(&rename_data_resp, request_decoded_from_msgpack) { - Ok((rename_data_resp, rename_data_resp_bin)) => Ok(Response::new(RenameDataResponse { - success: true, - rename_data_resp, - rename_data_resp_bin: rename_data_resp_bin.into(), - error: None, - })), - Err(err) => Ok(Response::new(RenameDataResponse { - success: false, - rename_data_resp: String::new(), - rename_data_resp_bin: Vec::new().into(), - error: Some(err.into()), - })), - } - } + let target = self.local_mutation_target(); + let decoded_file_info = match decode_rename_data_request_file_info(&request.file_info_bin, &request.file_info) { + Ok(file_info) => file_info, + Err(err) => { + return Ok(Response::new(RenameDataResponse { + success: false, + rename_data_resp: String::new(), + rename_data_resp_bin: Vec::new().into(), + error: Some(DiskError::other(format!("decode FileInfo failed: {err}")).into()), + })); + } + }; + let scanner_publication_lease_token = if request.scanner_publication_lease_token.is_empty() { + None + } else { + let token = Uuid::from_slice(&request.scanner_publication_lease_token) + .map_err(|_| Status::invalid_argument("scanner publication lease token must be a UUID"))?; + if token.is_nil() { + return Err(Status::invalid_argument("scanner publication lease token must not be nil")); + } + Some(token) + }; + let request_decoded_from_msgpack = decoded_file_info.from_msgpack; + match target + .rename_local_data( + &request.disk, + (&request.src_volume, &request.src_path), + &decoded_file_info.value, + (&request.dst_volume, &request.dst_path), + scanner_publication_lease_token, + ) + .await + { + Ok(rename_data_resp) => match encode_rename_data_response_payloads(&rename_data_resp, request_decoded_from_msgpack) { + Ok((rename_data_resp, rename_data_resp_bin)) => Ok(Response::new(RenameDataResponse { + success: true, + rename_data_resp, + rename_data_resp_bin: rename_data_resp_bin.into(), + error: None, + })), Err(err) => Ok(Response::new(RenameDataResponse { success: false, rename_data_resp: String::new(), rename_data_resp_bin: Vec::new().into(), error: Some(err.into()), })), - } - } else { - Ok(Response::new(RenameDataResponse { + }, + Err(err) => Ok(Response::new(RenameDataResponse { success: false, rename_data_resp: String::new(), rename_data_resp_bin: Vec::new().into(), - error: Some(DiskError::other("cannot find disk".to_string()).into()), - })) + error: Some(err.into()), + })), } } diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index e35cf812d..39cdd312e 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -377,10 +377,12 @@ pub(crate) mod timeout_wrapper_consumer { } pub(crate) mod tonic_service_consumer { + #[cfg(test)] + pub(crate) use super::super::tonic_service::make_server; #[cfg(test)] pub(crate) use super::super::tonic_service::{heal_topology_fingerprint, make_heal_control_server_for_source}; pub(crate) use super::super::tonic_service::{ - make_heal_control_server_with_cache, make_scanner_control_server, make_server, make_tier_mutation_control_server, + make_heal_control_server_with_cache, make_scanner_control_server, make_server_for_slot, make_tier_mutation_control_server, }; } @@ -600,8 +602,8 @@ pub(crate) mod ecstore_storage { #[cfg(test)] pub(crate) use rustfs_ecstore::api::storage::init_local_disks; pub(crate) use rustfs_ecstore::api::storage::{ - ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk, all_local_disk_path, - find_local_disk_by_ref, init_local_disks_with_instance_ctx, init_lock_clients, + BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk, + all_local_disk_path, find_local_disk_by_ref, init_local_disks_with_instance_ctx, init_lock_clients, prewarm_local_disk_id_map_with_instance_ctx, }; } @@ -677,6 +679,9 @@ type EcstoreReplicationStats = ecstore_bucket::replication::ReplicationStats; pub(crate) type DynReplicationPool = StorageReplicationPoolHandle; pub(crate) type DynReader = ecstore_rio::DynReader; pub(crate) type ECStore = ecstore_storage::ECStore; +pub(crate) type BootstrapLocalTarget = ecstore_storage::BootstrapLocalTarget; +#[cfg(all(test, not(windows)))] +pub(crate) use rustfs_ecstore::api::disk::{LocalPublicationPause, LocalPublicationStage}; pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint; #[cfg(test)] pub(crate) type Endpoints = ecstore_layout::Endpoints; diff --git a/rustfs/src/storage/tonic_service.rs b/rustfs/src/storage/tonic_service.rs index c7d2ee28a..a86724486 100644 --- a/rustfs/src/storage/tonic_service.rs +++ b/rustfs/src/storage/tonic_service.rs @@ -13,8 +13,8 @@ // limitations under the License. pub(crate) use crate::storage::rpc::node_service::make_heal_control_server_with_cache; -pub(crate) use crate::storage::rpc::node_service::make_scanner_control_server; #[cfg(test)] pub(crate) use crate::storage::rpc::node_service::{heal::heal_topology_fingerprint, make_heal_control_server_for_source}; +pub(crate) use crate::storage::rpc::node_service::{make_scanner_control_server, make_server_for_slot}; pub use crate::storage::rpc::{make_heal_control_server, make_server, make_tier_mutation_control_server}; pub type NodeService = crate::storage::rpc::NodeService; diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 9abdbd5fd..2b1f38d3b 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -171,12 +171,15 @@ pub(crate) mod server { } pub(crate) mod tonic_service { + #[cfg(test)] + pub(crate) use crate::storage::storage_api::tonic_service_consumer::make_server; #[cfg(test)] pub(crate) use crate::storage::storage_api::tonic_service_consumer::{ heal_topology_fingerprint, make_heal_control_server_for_source, }; pub(crate) use crate::storage::storage_api::tonic_service_consumer::{ - make_heal_control_server_with_cache, make_scanner_control_server, make_server, make_tier_mutation_control_server, + make_heal_control_server_with_cache, make_scanner_control_server, make_server_for_slot, + make_tier_mutation_control_server, }; } }