diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 279789719..081308754 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -137,6 +137,22 @@ pub const DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED: bool = false; const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE); const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED); +/// Request the object-transaction fencing contract used by storage-owned +/// cleanup receipts and lock-window optimizations. +/// +/// This is fail-closed: enabling the writer without a live fleet proof rejects +/// the commit rather than silently using a legacy-safe path. +pub const ENV_OBJECT_TRANSACTION_FENCING_WRITE: &str = "RUSTFS_OBJECT_TRANSACTION_FENCING_WRITE"; +pub const DEFAULT_OBJECT_TRANSACTION_FENCING_WRITE: bool = false; + +/// Operator-attested confirmation that every serving node understands the +/// object transaction fencing contract. +pub const ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED: &str = "RUSTFS_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED"; +pub const DEFAULT_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED: bool = false; + +const _: () = assert!(!DEFAULT_OBJECT_TRANSACTION_FENCING_WRITE); +const _: () = assert!(!DEFAULT_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED); + /// Request preserving legacy per-part checksum metadata during data movement. /// /// This remains ineffective until @@ -673,4 +689,13 @@ mod remote_version_state_tests { "RUSTFS_DATA_MOVEMENT_PART_CHECKSUMS_FLEET_CONFIRMED" ); } + + #[test] + fn object_transaction_fencing_gate_uses_stable_environment_names() { + assert_eq!(super::ENV_OBJECT_TRANSACTION_FENCING_WRITE, "RUSTFS_OBJECT_TRANSACTION_FENCING_WRITE"); + assert_eq!( + super::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, + "RUSTFS_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED" + ); + } } diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index b08211853..c63dbc401 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -206,6 +206,38 @@ pub(crate) fn remote_version_state_fleet_proof_matches(proof: &RemoteVersionStat }) } +#[cfg(test)] +pub(crate) struct RemoteVersionStateFleetProofGuard; + +#[cfg(test)] +impl Drop for RemoteVersionStateFleetProofGuard { + fn drop(&mut self) { + replace_remote_version_state_fleet_proof(None); + } +} + +#[cfg(test)] +pub(crate) fn install_remote_version_state_fleet_proof_for_test(topology_fingerprint: &str) -> RemoteVersionStateFleetProofGuard { + match REMOTE_VERSION_STATE_PROBE_TOPOLOGY.set(topology_fingerprint.to_string()) { + Ok(()) => {} + Err(_) + if REMOTE_VERSION_STATE_PROBE_TOPOLOGY + .get() + .is_some_and(|current| current == topology_fingerprint) => {} + Err(_) => panic!("remote version state test topology is already bound to another fingerprint"), + } + let peer_epochs = BTreeMap::new(); + if let Some(err) = publish_remote_version_state_probe_result( + remote_version_state_fleet_proof_slot(), + topology_fingerprint, + Ok(peer_epochs), + Instant::now(), + ) { + panic!("test proof installation must not fail: {err}"); + } + RemoteVersionStateFleetProofGuard +} + fn remote_version_state_fleet_proof_valid_at( proof: Option<&RemoteVersionStateFleetProof>, expected_topology: &str, diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index e5ca4fbec..089515ec6 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -1425,6 +1425,33 @@ impl SetDisks { /// post-heal tail — reclaim identically. Never fails the heal: delete errors /// are logged and swallowed. Callers must gate this on `!opts.dry_run`. async fn reclaim_orphan_data_dirs_best_effort(&self, bucket: &str, object: &str) { + match self.reconcile_old_data_cleanup_receipts(bucket, object).await { + Ok(removed) if removed > 0 => { + debug!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + removed, + state = "old_data_cleanup_receipt_reconciled", + "Set disk old-data cleanup receipts reconciled" + ); + } + Ok(_) => {} + Err(e) => { + warn!( + event = EVENT_SET_DISK_HEAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + error = %e, + state = "old_data_cleanup_receipt_reconcile_failed", + "Set disk old-data cleanup receipt reconcile failed" + ); + } + } match self.reclaim_orphan_data_dirs(bucket, object).await { Ok(removed) if removed > 0 => { debug!( diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 93770ca2e..7ecde7910 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -22,6 +22,11 @@ use super::super::*; use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards}; +use super::object::{ + assign_object_transaction_epoch, object_transaction_fencing_fleet_proof, object_transaction_fencing_fleet_proof_matches, + object_transaction_fencing_requested, old_data_cleanup_receipt_path, read_object_transaction_epoch_fence, + verify_object_transaction_epoch_fence, +}; use crate::crash_inject::{self, CrashPoint}; use crate::multipart_listing::paginate_multipart_listing; use futures::{StreamExt, stream}; @@ -63,6 +68,7 @@ pub(crate) enum MultipartCommitPause { PutPartBeforeLockLost, PutPartAfterRename, BeforeLockLost, + BeforeTransactionEpochVerify, AfterRename, } @@ -153,13 +159,24 @@ impl Drop for MultipartCommitBarrier { #[cfg(test)] async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartCommitPause) { - let barrier = MULTIPART_COMMIT_BARRIER - .get_or_init(|| std::sync::Mutex::new(None)) - .lock() - .expect("multipart commit barrier mutex should not poison") - .as_ref() - .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause) - .cloned(); + let barrier = { + let mut slot = MULTIPART_COMMIT_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("multipart commit barrier mutex should not poison"); + if slot + .as_ref() + .is_some_and(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause) + { + if pause == MultipartCommitPause::BeforeTransactionEpochVerify { + slot.take() + } else { + slot.clone() + } + } else { + None + } + }; if let Some(barrier) = barrier && let Ok(previous) = barrier.arrivals.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| { (current < barrier.expected_arrivals).then_some(current + 1) @@ -2296,6 +2313,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { } ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?; + let transaction_fencing_proof = object_transaction_fencing_fleet_proof(); + if object_transaction_fencing_requested() && transaction_fencing_proof.is_none() { + return Err(Error::other("object transaction fencing requires a live fleet capability proof")); + } + let transaction_epoch_fence = if transaction_fencing_proof.is_some() { + Some(read_object_transaction_epoch_fence(self.as_ref(), bucket, object).await?) + } else { + None + }; + let transaction_epoch = + transaction_epoch_fence.map(|_| assign_object_transaction_epoch(&shuffle_disks, &mut parts_metadatas)); + let commit_set = self.clone(); let commit_bucket = bucket.to_owned(); let commit_object = object.to_owned(); @@ -2323,6 +2352,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { // The trailing `_` drops the rename_data old-size backfill // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit // `get_object_info` lookup, so the backfill has no consumer here yet. + if let Some(proof) = transaction_fencing_proof.as_ref() + && !object_transaction_fencing_fleet_proof_matches(proof) + { + return Err(Error::other( + "object transaction fencing fleet capability changed during complete_multipart_upload", + )); + } + if let Some(expected) = transaction_epoch_fence { + #[cfg(test)] + pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::BeforeTransactionEpochVerify).await; + verify_object_transaction_epoch_fence(&commit_set, &commit_bucket, &commit_object, expected).await?; + } let (online_disks, convergence, op_old_dir, cleanup_disks, _) = SetDisks::rename_data( &shuffle_disks, RUSTFS_META_MULTIPART_BUCKET, @@ -2354,6 +2395,19 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { }); } + if let Some(old_dir) = op_old_dir { + commit_set + .persist_old_data_cleanup_receipts( + &cleanup_disks, + &commit_bucket, + &commit_object, + old_dir, + fi.data_dir, + transaction_epoch, + ) + .await; + } + // Crash-consistency injection: hard power loss after the authoritative // rename_data commit succeeded but before the stale part.N.meta cleanup. // The new version is durably committed and visible, so a crash here must @@ -2469,9 +2523,10 @@ fn resolve_complete_etag(opts: &ObjectOptions, uploaded_parts: &[CompletePart]) mod tests { use super::*; use crate::config::storageclass::lookup_config_for_pools_without_env; - use crate::disk::DiskAPI as _; + use crate::disk::{DiskAPI as _, ReadOptions}; use crate::disk::{endpoint::Endpoint, format::FormatV3}; use crate::layout::endpoints::SetupType; + use crate::services::notification_sys::install_remote_version_state_fleet_proof_for_test; // No-locker helpers resolve to the isolated-context variants (see // `hermetic_set_disks_isolated`); the guard-based tests build through // `hermetic_set_disks_with_lockers`, which stays on the bootstrap context @@ -2882,6 +2937,208 @@ mod tests { } } + async fn object_transaction_epochs(disks: &[DiskStore], bucket: &str, object: &str) -> Vec> { + let mut epochs = Vec::with_capacity(disks.len()); + for (disk_index, disk) in disks.iter().enumerate() { + let file_info = disk + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .unwrap_or_else(|err| panic!("disk {disk_index} should persist object metadata: {err}")); + epochs.push( + file_info + .object_transaction_epoch() + .unwrap_or_else(|err| panic!("disk {disk_index} transaction epoch should decode: {err}")), + ); + } + epochs + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn object_transaction_fencing_requires_live_fleet_proof_before_multipart_commit() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-transaction-fencing-no-proof"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let (upload_id, parts) = stage_upload_with_create_opts( + &set_disks, + bucket, + object, + b"must-not-complete-without-proof", + &ObjectOptions::default(), + ) + .await; + + let err = temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default()) + .await + }, + ) + .await + .expect_err("multipart completion must fail closed without a live fleet proof"); + + assert!( + err.to_string() + .contains("object transaction fencing requires a live fleet capability proof"), + "unexpected error: {err:?}" + ); + set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect_err("failed fenced completion must not publish object metadata"); + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn object_transaction_fencing_persists_epoch_on_multipart_commit() { + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-object-transaction-epoch"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, b"multipart fenced epoch", &ObjectOptions::default()).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + set_disks + .clone() + .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default()) + .await + .expect("fenced multipart completion should commit with a live proof"); + }, + ) + .await; + + let epochs = object_transaction_epochs(&disk_stores, bucket, object).await; + let first = epochs[0].expect("fenced multipart completion should persist an epoch"); + assert!(!first.is_nil()); + assert!(epochs.into_iter().all(|epoch| epoch == Some(first))); + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn object_transaction_fencing_rejects_stale_multipart_epoch() { + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-object-transaction-stale-epoch"; + let object = "object"; + make_bucket_on_all(&disk_stores, bucket).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut initial_reader = PutObjReader::from_vec(b"initial fenced object".to_vec()); + set_disks + .put_object( + bucket, + object, + &mut initial_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("initial fenced PUT should commit"); + let initial_epoch = object_transaction_epochs(&disk_stores, bucket, object) + .await + .into_iter() + .next() + .flatten() + .expect("initial fenced PUT should persist an epoch"); + + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, b"stale multipart body", &ObjectOptions::default()) + .await; + let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::BeforeTransactionEpochVerify); + let stale_set = Arc::clone(&set_disks); + let stale = tokio::spawn(async move { + stale_set + .clone() + .complete_multipart_upload( + bucket, + object, + &upload_id, + parts, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + }); + barrier.wait_until_paused().await; + + let mut winner_reader = PutObjReader::from_vec(b"winning put body".to_vec()); + set_disks + .put_object( + bucket, + object, + &mut winner_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("concurrent fenced PUT should advance the epoch"); + let winning_epoch = object_transaction_epochs(&disk_stores, bucket, object) + .await + .into_iter() + .next() + .flatten() + .expect("winning fenced PUT should persist an epoch"); + assert_ne!(winning_epoch, initial_epoch); + + barrier.release(); + let err = stale + .await + .expect("stale multipart task should not panic") + .expect_err("stale epoch multipart completion must be rejected"); + assert_eq!(err, StorageError::PreconditionFailed); + + let final_epochs = object_transaction_epochs(&disk_stores, bucket, object).await; + assert!(final_epochs.into_iter().all(|epoch| epoch == Some(winning_epoch))); + let mut reader = set_disks + .get_object_reader( + bucket, + object, + None, + HeaderMap::new(), + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("winning object should remain readable"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("winning body should stream"); + assert_eq!(restored, b"winning put body"); + }, + ) + .await; + } + #[tokio::test] async fn complete_multipart_quota_rejection_preserves_destination_and_upload() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; @@ -6193,6 +6450,24 @@ mod tests { (body, etag) } + async fn current_data_dir(disk: &DiskStore, bucket: &str, object: &str) -> Uuid { + disk.read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("current object metadata should read") + .data_dir + .expect("test object should be stored out-of-line") + } + + async fn data_dir_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool { + disk.read_all(bucket, &format!("{object}/{data_dir}/part.1")).await.is_ok() + } + + async fn cleanup_receipt_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool { + disk.read_all(bucket, &old_data_cleanup_receipt_path(object, data_dir)) + .await + .is_ok() + } + async fn upload_is_listed(set_disks: &Arc, bucket: &str, object: &str, upload_id: &str) -> bool { let page = set_disks .list_multipart_uploads_for_incarnation(bucket, object, None, None, None, 1000, None) @@ -6313,6 +6588,106 @@ mod tests { let (body_after, _) = read_object(&set_disks, bucket, object).await; assert_eq!(body_after, new, "reclaiming the leftover upload must not disturb the committed object"); } + + #[tokio::test] + #[serial(storage_class_env)] + async fn post_commit_crash_receipt_reclaims_old_data_after_restart() { + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-crash-old-data-receipt"; + let object = "crash-old-data-object"; + make_bucket_on_all(&disk_stores, bucket).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let old = payload(0x51); + let (u_old, parts_old) = stage_upload(&set_disks, bucket, object, &old).await; + complete(&set_disks, bucket, object, &u_old, parts_old) + .await + .expect("the old version should commit"); + let old_dir = current_data_dir(&disk_stores[0], bucket, object).await; + + let new = payload(0x52); + let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await; + crash_inject::arm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object); + let crashed = complete(&set_disks, bucket, object, &u_new, parts_new).await; + assert!( + matches!(crashed, Err(StorageError::Unexpected)), + "the post-commit crash point must surface as unexpected, got {crashed:?}" + ); + crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object); + + let (body, _) = read_object(&set_disks, bucket, object).await; + assert_eq!(body, new, "the committed replacement must remain readable after the crash"); + for disk in &disk_stores { + assert!( + cleanup_receipt_exists(disk, bucket, object, old_dir).await, + "post-commit crash must leave a durable old-data cleanup receipt" + ); + assert!( + data_dir_exists(disk, bucket, object, old_dir).await, + "post-commit crash must leave old data for restart reconciliation" + ); + } + + let restarted_endpoints = temp_dirs + .iter() + .enumerate() + .map(|(disk_idx, dir)| { + let mut endpoint = Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")) + .expect("endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(disk_idx); + endpoint + }) + .collect::>(); + let mut reloaded = Vec::with_capacity(restarted_endpoints.len()); + for endpoint in &restarted_endpoints { + reloaded.push( + new_disk( + endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("disk should restart"), + ); + } + let restarted_set = SetDisks::new_with_instance_ctx( + "restart-cleanup-receipt-test-owner".to_string(), + Arc::new(RwLock::new(reloaded.iter().cloned().map(Some).collect())), + 4, + 2, + 0, + 0, + restarted_endpoints, + set_disks.format.clone(), + Vec::new(), + Arc::new(crate::runtime::instance::InstanceContext::new()), + ) + .await; + let removed = restarted_set + .reconcile_old_data_cleanup_receipts(bucket, object) + .await + .expect("restart receipt reconciliation should succeed"); + assert_eq!(removed, 4, "restart reconciler should delete all receipt targets"); + for disk in &reloaded { + assert!( + !data_dir_exists(disk, bucket, object, old_dir).await, + "restart reconciler must reclaim the old data dir" + ); + } + }, + ) + .await; + } } #[test] diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 2363319b3..cd450dac3 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -44,10 +44,11 @@ use crate::bucket::replication::{ DeleteReplicationConfigSnapshot, VersionPurgeStatusType, replication_state_to_filemeta, version_purge_status_to_filemeta, }; use crate::diagnostics::get::GetObjectFailureReason; -use crate::disk::OldCurrentSize; +use crate::disk::{DataDirDeleteStatus, OldCurrentSize}; use crate::error::is_err_invalid_upload_id; use crate::object_api::NamespaceLockFence; use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed}; +use crate::services::notification_sys::RemoteVersionStateFleetProofToken; use crate::services::tier::tier::{TierConfigMgr, TierOperationLease}; use crate::store::ECStore; use crate::store::utils::clean_metadata; @@ -55,6 +56,9 @@ use futures::FutureExt as _; use http::HeaderValue; use rustfs_utils::path::decode_dir_object; use std::future::Future; +use std::sync::OnceLock; + +const OLD_DATA_CLEANUP_RECEIPT_FILE: &str = ".rustfs-old-data-cleanup-receipt.json"; #[inline] fn duration_millis_f64(duration: std::time::Duration) -> f64 { @@ -65,6 +69,133 @@ fn committed_response_metadata_slot(committed_disks: &[Option], fallback_s committed_disks.iter().position(Option::is_some).unwrap_or(fallback_slot) } +pub(in crate::set_disk::ops) fn assign_object_transaction_epoch( + shuffle_disks: &[Option], + parts_metadatas: &mut [FileInfo], +) -> Uuid { + let epoch = Uuid::new_v4(); + for (disk, file_info) in shuffle_disks.iter().zip(parts_metadatas.iter_mut()) { + if disk.is_some() { + file_info.set_object_transaction_epoch(epoch); + } + } + epoch +} + +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct OldDataCleanupReceiptRecord { + epoch: String, + old_data_dir: String, + committed_data_dir: Option, +} + +#[derive(Clone, Copy)] +struct OldDataCleanupReceipt { + epoch: Uuid, + old_data_dir: Uuid, + committed_data_dir: Option, +} + +impl OldDataCleanupReceipt { + fn new(epoch: Uuid, old_data_dir: Uuid, committed_data_dir: Option) -> Self { + Self { + epoch, + old_data_dir, + committed_data_dir, + } + } + + fn encode(self) -> disk::error::Result { + let record = OldDataCleanupReceiptRecord { + epoch: self.epoch.to_string(), + old_data_dir: self.old_data_dir.to_string(), + committed_data_dir: self.committed_data_dir.map(|dir| dir.to_string()), + }; + Ok(Bytes::from(serde_json::to_vec(&record)?)) + } + + fn decode(data: &[u8]) -> disk::error::Result { + let record: OldDataCleanupReceiptRecord = serde_json::from_slice(data)?; + let epoch = Uuid::parse_str(&record.epoch).map_err(DiskError::other)?; + let old_data_dir = Uuid::parse_str(&record.old_data_dir).map_err(DiskError::other)?; + let committed_data_dir = record + .committed_data_dir + .as_deref() + .map(Uuid::parse_str) + .transpose() + .map_err(DiskError::other)?; + if epoch.is_nil() || old_data_dir.is_nil() || committed_data_dir.is_some_and(|dir| dir.is_nil()) { + return Err(DiskError::FileCorrupt); + } + Ok(Self::new(epoch, old_data_dir, committed_data_dir)) + } +} + +pub(in crate::set_disk::ops) fn old_data_cleanup_receipt_path(object: &str, old_data_dir: Uuid) -> String { + path_join_buf(&[object, &old_data_dir.to_string(), OLD_DATA_CLEANUP_RECEIPT_FILE]) +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(in crate::set_disk::ops) enum ObjectTransactionEpochFence { + Absent, + Present(Uuid), +} + +impl ObjectTransactionEpochFence { + fn from_file_info(file_info: &FileInfo) -> Result { + match file_info.object_transaction_epoch() { + Ok(Some(epoch)) => Ok(Self::Present(epoch)), + Ok(None) => Ok(Self::Absent), + Err(_) => Err(StorageError::FileCorrupt), + } + } +} + +pub(in crate::set_disk::ops) async fn read_object_transaction_epoch_fence( + set: &SetDisks, + bucket: &str, + object: &str, +) -> Result { + let current = set + .get_object_fileinfo( + bucket, + object, + &ObjectOptions { + no_lock: true, + metadata_cache_safe: false, + versioned: true, + ..Default::default() + }, + false, + false, + ) + .await; + match current { + Ok(snapshot) => ObjectTransactionEpochFence::from_file_info(snapshot.fi()), + Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => Ok(ObjectTransactionEpochFence::Absent), + Err(err) => Err(err), + } +} + +pub(in crate::set_disk::ops) async fn verify_object_transaction_epoch_fence( + set: &SetDisks, + bucket: &str, + object: &str, + expected: ObjectTransactionEpochFence, +) -> Result<()> { + let current = read_object_transaction_epoch_fence(set, bucket, object).await?; + if current == expected { + Ok(()) + } else { + Err(StorageError::PreconditionFailed) + } +} + +fn old_data_cleanup_receipt_epoch_matches_current(receipt: OldDataCleanupReceipt, current: ObjectTransactionEpochFence) -> bool { + matches!(current, ObjectTransactionEpochFence::Present(epoch) if epoch == receipt.epoch) +} + #[cfg(test)] mod duration_metrics_tests { use super::duration_millis_f64; @@ -1080,6 +1211,114 @@ fn delete_file_info_with_replication_transport_metadata(fi: &FileInfo) -> FileIn } impl SetDisks { + pub(in crate::set_disk) async fn persist_old_data_cleanup_receipts( + &self, + disks: &[Option], + bucket: &str, + object: &str, + old_data_dir: Uuid, + committed_data_dir: Option, + epoch: Option, + ) { + let Some(epoch) = epoch else { return }; + if committed_data_dir == Some(old_data_dir) { + return; + } + let receipt = OldDataCleanupReceipt::new(epoch, old_data_dir, committed_data_dir); + let Ok(encoded) = receipt.encode() else { + return; + }; + let path = old_data_cleanup_receipt_path(object, old_data_dir); + let futures = disks.iter().filter_map(|disk| { + disk.as_ref().map(|disk| { + let disk = disk.clone(); + let encoded = encoded.clone(); + let bucket = bucket.to_owned(); + let path = path.clone(); + async move { disk.write_all(&bucket, &path, encoded).await } + }) + }); + for result in join_all(futures).await { + if let Err(err) = result { + debug!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + bucket, + object, + old_dir = %old_data_dir, + error = %err, + state = "cleanup_receipt_persist_failed", + "SetDisk old-data cleanup receipt persist failed" + ); + } + } + } + + pub(in crate::set_disk) async fn reconcile_old_data_cleanup_receipts( + &self, + bucket: &str, + object: &str, + ) -> disk::error::Result { + if object_transaction_fencing_fleet_proof().is_none() { + return Ok(0); + } + let current = read_object_transaction_epoch_fence(self, bucket, object) + .await + .map_err(DiskError::from)?; + let disks = self.get_disks_internal().await; + let mut removed = 0usize; + + for disk in disks.iter().flatten() { + let entries = match disk.list_dir("", bucket, object, 0).await { + Ok(entries) => entries, + Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => continue, + Err(err) => return Err(err), + }; + for entry in entries { + let Some(name) = entry.strip_suffix(SLASH_SEPARATOR) else { continue }; + let Ok(data_dir) = Uuid::parse_str(name) else { continue }; + if data_dir.is_nil() { + continue; + } + + let receipt_path = old_data_cleanup_receipt_path(object, data_dir); + let receipt_bytes = match disk.read_all(bucket, &receipt_path).await { + Ok(bytes) => bytes, + Err(DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound) => continue, + Err(err) => return Err(err), + }; + let receipt = OldDataCleanupReceipt::decode(&receipt_bytes)?; + if receipt.old_data_dir != data_dir + || receipt.committed_data_dir == Some(receipt.old_data_dir) + || !old_data_cleanup_receipt_epoch_matches_current(receipt, current) + { + continue; + } + + let old_path = format!("{object}/{}", receipt.old_data_dir); + match disk + .delete_data_dir( + bucket, + &old_path, + DeleteOptions { + recursive: true, + immediate: true, + ..Default::default() + }, + ) + .await + { + Ok(DataDirDeleteStatus::Deleted) => removed += 1, + Ok(DataDirDeleteStatus::Deferred) => {} + Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => {} + Err(err) => return Err(err), + } + } + } + + Ok(removed) + } + async fn validate_bucket_incarnation(&self, bucket: &str, expected: Uuid) -> Result<()> { let current = metadata_sys::get_bucket_incarnation_id_in(&self.ctx, bucket).await?; if current != expected { @@ -1681,6 +1920,18 @@ impl SetDisks { }); } + let transaction_fencing_proof = object_transaction_fencing_fleet_proof(); + if object_transaction_fencing_requested() && transaction_fencing_proof.is_none() { + return Err(Error::other("object transaction fencing requires a live fleet capability proof")); + } + let transaction_epoch_fence = if transaction_fencing_proof.is_some() { + Some(read_object_transaction_epoch_fence(self, bucket, object).await?) + } else { + None + }; + let transaction_epoch = + transaction_epoch_fence.map(|_| assign_object_transaction_epoch(&shuffle_disks, &mut parts_metadatas)); + let commit_set = self.clone(); let commit_bucket = bucket.to_owned(); let commit_object = object.to_owned(); @@ -1698,6 +1949,39 @@ impl SetDisks { let _object_lock_guard = commit_object_lock_guard; let _bucket_lifecycle_guard = commit_bucket_lifecycle_guard; let rename_stage_start = Instant::now(); + let pre_rename_result: Result<()> = async { + if let Some(proof) = transaction_fencing_proof.as_ref() + && !object_transaction_fencing_fleet_proof_matches(proof) + { + return Err(Error::other("object transaction fencing fleet capability changed during put_object")); + } + if let Some(expected) = transaction_epoch_fence { + #[cfg(any(test, feature = "test-util"))] + pause_put_object_commit( + &commit_bucket, + &commit_object, + PutObjectCommitPause::BeforeTransactionEpochVerify, + ) + .await; + verify_object_transaction_epoch_fence(&commit_set, &commit_bucket, &commit_object, expected).await?; + } + Ok(()) + } + .await; + if let Err(err) = pre_rename_result { + if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await { + warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data"); + } else if issue3031_diag_enabled() { + warn!( + target: "rustfs_ecstore::set_disk", + bucket = %commit_bucket, + object = %commit_object, + tmp_dir = %commit_tmp_dir, + "issue3031_put_object_tmp_cleanup_done" + ); + } + return Err(err); + } let rename_result = SetDisks::rename_data( &shuffle_disks, RUSTFS_META_TMP_BUCKET, @@ -1746,6 +2030,19 @@ impl SetDisks { let rename_stage_elapsed = rename_stage_start.elapsed(); let rename_stage_ms = rename_stage_elapsed.as_millis() as u64; + if let Some(old_dir) = op_old_dir { + commit_set + .persist_old_data_cleanup_receipts( + &cleanup_disks, + &commit_bucket, + &commit_object, + old_dir, + committed_data_dir, + transaction_epoch, + ) + .await; + } + commit_set .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object) .await; @@ -2842,14 +3139,12 @@ fn remote_version_state_writer_enabled() -> bool { remote_version_state_writer_fleet_proof().is_some() } -fn remote_version_state_writer_fleet_proof() -> Option { - remote_version_state_writer_requested() - .then(crate::services::notification_sys::acquire_remote_version_state_fleet_proof) - .flatten() +fn remote_version_state_writer_fleet_proof() -> Option { + transaction_fencing_fleet_proof(remote_version_state_writer_requested()) } fn remote_version_state_writer_requested() -> bool { - remote_version_state_writer_enabled_for( + transaction_fencing_gate_requested_for( rustfs_utils::get_env_bool( rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_WRITE, rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE, @@ -2862,20 +3157,66 @@ fn remote_version_state_writer_requested() -> bool { ) } -fn remote_version_state_writer_fleet_proof_matches( - proof: &crate::services::notification_sys::RemoteVersionStateFleetProofToken, -) -> bool { - remote_version_state_writer_fleet_proof_matches_for( +fn remote_version_state_writer_fleet_proof_matches(proof: &RemoteVersionStateFleetProofToken) -> bool { + transaction_fencing_fleet_proof_matches_for( remote_version_state_writer_requested(), crate::services::notification_sys::remote_version_state_fleet_proof_matches(proof), ) } -fn remote_version_state_writer_fleet_proof_matches_for(requested: bool, fleet_proof_matches: bool) -> bool { +pub(in crate::set_disk::ops) fn object_transaction_fencing_fleet_proof() -> Option { + transaction_fencing_fleet_proof(object_transaction_fencing_requested()) +} + +pub(in crate::set_disk::ops) fn object_transaction_fencing_requested() -> bool { + object_transaction_fencing_requested_cached() +} + +#[cfg(not(test))] +fn object_transaction_fencing_requested_cached() -> bool { + static REQUESTED: OnceLock = OnceLock::new(); + *REQUESTED.get_or_init(load_object_transaction_fencing_requested) +} + +#[cfg(test)] +fn object_transaction_fencing_requested_cached() -> bool { + load_object_transaction_fencing_requested() +} + +fn load_object_transaction_fencing_requested() -> bool { + transaction_fencing_gate_requested_for( + rustfs_utils::get_env_bool( + rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, + rustfs_config::DEFAULT_OBJECT_TRANSACTION_FENCING_WRITE, + ), + rustfs_utils::get_env_bool( + rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, + rustfs_config::DEFAULT_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, + ), + true, + ) +} + +pub(in crate::set_disk::ops) fn object_transaction_fencing_fleet_proof_matches( + proof: &RemoteVersionStateFleetProofToken, +) -> bool { + transaction_fencing_fleet_proof_matches_for( + object_transaction_fencing_requested(), + crate::services::notification_sys::remote_version_state_fleet_proof_matches(proof), + ) +} + +fn transaction_fencing_fleet_proof(requested: bool) -> Option { + requested + .then(crate::services::notification_sys::acquire_remote_version_state_fleet_proof) + .flatten() +} + +fn transaction_fencing_fleet_proof_matches_for(requested: bool, fleet_proof_matches: bool) -> bool { requested && fleet_proof_matches } -fn remote_version_state_writer_enabled_for(requested: bool, fleet_confirmed: bool, fleet_proof_valid: bool) -> bool { +fn transaction_fencing_gate_requested_for(requested: bool, fleet_confirmed: bool, fleet_proof_valid: bool) -> bool { requested && fleet_confirmed && fleet_proof_valid } @@ -2885,6 +3226,7 @@ pub enum PutObjectCommitPause { BeforeNamespace, AfterNamespace, BeforeMetadata, + BeforeTransactionEpochVerify, } #[cfg(any(test, feature = "test-util"))] @@ -2966,13 +3308,24 @@ impl Drop for PutObjectCommitBarrier { #[cfg(any(test, feature = "test-util"))] async fn pause_put_object_commit(bucket: &str, object: &str, pause: PutObjectCommitPause) { - let barrier = PUT_OBJECT_COMMIT_BARRIER - .get_or_init(|| std::sync::Mutex::new(Vec::new())) - .lock() - .expect("put object commit barrier mutex should not poison") - .iter() - .find(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause) - .cloned(); + let barrier = { + let mut slot = PUT_OBJECT_COMMIT_BARRIER + .get_or_init(|| std::sync::Mutex::new(Vec::new())) + .lock() + .expect("put object commit barrier mutex should not poison"); + if let Some(index) = slot + .iter() + .position(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause) + { + if pause == PutObjectCommitPause::BeforeTransactionEpochVerify { + Some(slot.remove(index)) + } else { + Some(Arc::clone(&slot[index])) + } + } else { + None + } + }; if let Some(barrier) = barrier { barrier.arrived.notify_one(); barrier.release.notified().await; @@ -3332,7 +3685,7 @@ mod transition_upload_completion_tests { mod transition_version_id_tests { use super::{ TransitionUploadCandidate, persisted_transition_version, persisted_transition_version_with_gate, - remote_version_state_writer_enabled_for, remote_version_state_writer_fleet_proof_matches_for, + transaction_fencing_fleet_proof_matches_for, transaction_fencing_gate_requested_for, }; use rustfs_filemeta::TransitionVersionState; use uuid::Uuid; @@ -3385,7 +3738,24 @@ mod transition_version_id_tests { ("fully upgraded fleet", true, true, true, true), ] { assert_eq!( - remote_version_state_writer_enabled_for(requested, fleet_confirmed, fleet_proof_valid), + transaction_fencing_gate_requested_for(requested, fleet_confirmed, fleet_proof_valid), + expected, + "{case}" + ); + } + } + + #[test] + fn object_transaction_fencing_gate_requires_request_confirmation_and_live_proof() { + for (case, requested, fleet_confirmed, fleet_proof_valid, expected) in [ + ("old defaults", false, false, false, false), + ("missing fleet confirmation", true, false, true, false), + ("missing local opt-in", false, true, true, false), + ("missing fleet proof", true, true, false, false), + ("fully upgraded fleet", true, true, true, true), + ] { + assert_eq!( + transaction_fencing_gate_requested_for(requested, fleet_confirmed, fleet_proof_valid), expected, "{case}" ); @@ -3400,7 +3770,7 @@ mod transition_version_id_tests { ("current authorization", true, true, true), ] { assert_eq!( - remote_version_state_writer_fleet_proof_matches_for(requested, fleet_proof_matches), + transaction_fencing_fleet_proof_matches_for(requested, fleet_proof_matches), expected, "{case}" ); @@ -5933,6 +6303,7 @@ mod inline_put_commit_path_tests { use crate::disk::{DiskAPI as _, ReadOptions}; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; use rustfs_config::server_config::KVS; + use serial_test::serial; use tokio::io::AsyncReadExt; async fn make_bucket(disks: &[DiskStore], bucket: &str) { @@ -7838,6 +8209,139 @@ mod transition_commit_failure_tests { ); } + #[tokio::test] + #[serial_test::serial] + async fn no_lock_restore_finalize_requires_live_namespace_fence() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "restore-finalize-fence-bucket"; + let object = "object.bin"; + let payload = b"restore finalize no_lock fence must fail closed".repeat(1024); + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let operation_id = Uuid::new_v4(); + let mut reader = PutObjReader::from_vec(payload); + set_disks + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("source object should be written"); + set_disks + .put_object_metadata( + bucket, + object, + &ObjectOptions { + eval_metadata: Some(restore_metadata(operation_id, true)), + ..Default::default() + }, + ) + .await + .expect("restore metadata should be installed"); + let restoring = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("restore metadata should be readable"); + + let err = set_disks + .finalize_restore_metadata( + bucket, + object, + &restoring, + &ObjectOptions { + no_lock: true, + namespace_lock_fence: Some(NamespaceLockFence::lost_for_test()), + user_defined: restore_operation_id_metadata(operation_id), + ..Default::default() + }, + ) + .await + .expect_err("lost outer namespace fence must reject no_lock restore finalization"); + assert!(matches!(err, Error::NamespaceLockQuorumUnavailable { .. })); + + let current = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("restore metadata should remain readable"); + let restore_status = parse_restore_obj_status( + current + .user_defined + .get(s3s::header::X_AMZ_RESTORE.as_str()) + .expect("restore header must remain pending"), + ) + .expect("restore header should remain parseable"); + assert!( + restore_status.on_going(), + "lost no_lock finalization must not publish restored completion metadata" + ); + } + + #[tokio::test] + #[serial_test::serial] + async fn no_lock_restore_cleanup_requires_live_namespace_fence() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "restore-cleanup-fence-bucket"; + let object = "object.bin"; + let payload = b"restore cleanup no_lock fence must fail closed".repeat(1024); + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let operation_id = Uuid::new_v4(); + let mut reader = PutObjReader::from_vec(payload); + set_disks + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("source object should be written"); + set_disks + .put_object_metadata( + bucket, + object, + &ObjectOptions { + eval_metadata: Some(restore_metadata(operation_id, true)), + ..Default::default() + }, + ) + .await + .expect("restore metadata should be installed"); + let restoring = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("restore metadata should be readable"); + + let err = set_disks + .update_restore_metadata( + bucket, + object, + &restoring, + &ObjectOptions { + no_lock: true, + namespace_lock_fence: Some(NamespaceLockFence::lost_for_test()), + user_defined: restore_operation_id_metadata(operation_id), + ..Default::default() + }, + ) + .await + .expect_err("lost outer namespace fence must reject no_lock restore cleanup"); + assert!(matches!(err, Error::NamespaceLockQuorumUnavailable { .. })); + + let current = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("restore metadata should remain readable"); + assert!( + current.user_defined.contains_key(s3s::header::X_AMZ_RESTORE.as_str()), + "lost no_lock cleanup must not remove the restore header" + ); + assert_eq!( + rustfs_utils::http::metadata_compat::get_consistent_str( + current.user_defined.as_ref(), + rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, + ), + Some(operation_id.to_string().as_str()), + "lost no_lock cleanup must not remove the restore operation id" + ); + } + #[tokio::test] #[serial_test::serial] async fn restore_worker_propagates_operation_id_to_final_put_commit() { @@ -8365,6 +8869,48 @@ mod transition_commit_failure_tests { .expect("the transitioned object body should drain"); assert_eq!(restored, payload); } + + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn object_transaction_fencing_requires_live_fleet_proof_before_put_commit() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "transaction-fencing-no-proof"; + let object = "object.bin"; + for disk in &disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + + let err = temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut reader = PutObjReader::from_vec(b"must-not-commit-without-proof".to_vec()); + set_disks + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + }, + ) + .await + .expect_err("object fencing must fail closed without a live fleet proof"); + + assert!( + err.to_string() + .contains("object transaction fencing requires a live fleet capability proof"), + "unexpected error: {err:?}" + ); + for disk in &disk_stores { + let missing = disk + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect_err("failed fenced PUT must not publish object metadata"); + assert!( + matches!(missing, DiskError::FileNotFound | DiskError::FileVersionNotFound), + "failed fenced PUT left unexpected disk state: {missing:?}" + ); + } + } } #[cfg(all(test, feature = "test-util"))] @@ -9831,13 +10377,62 @@ mod transition_source_identity_matrix_tests { #[cfg(test)] mod heterogeneous_pool_put_tests { - use super::hermetic_set_disks_support::hermetic_set_disks_for_pool_with_default_parity_isolated as hermetic_set_disks_for_pool_with_default_parity; + use super::hermetic_set_disks_support::{ + hermetic_set_disks_for_pool_with_default_parity_isolated as hermetic_set_disks_for_pool_with_default_parity, + hermetic_set_disks_isolated as hermetic_set_disks, + }; use super::*; use crate::config::storageclass::lookup_config_for_pools_without_env; use crate::disk::{DiskAPI as _, ReadOptions}; + use crate::services::notification_sys::install_remote_version_state_fleet_proof_for_test; use rustfs_config::server_config::KVS; + use serial_test::serial; use tokio::io::AsyncReadExt; + async fn make_bucket(disks: &[DiskStore], bucket: &str) { + for disk in disks { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + } + + async fn object_transaction_epochs(disks: &[DiskStore], bucket: &str, object: &str) -> Vec> { + let mut epochs = Vec::with_capacity(disks.len()); + for (disk_index, disk) in disks.iter().enumerate() { + let file_info = disk + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .unwrap_or_else(|err| panic!("disk {disk_index} should persist object metadata: {err}")); + epochs.push( + file_info + .object_transaction_epoch() + .unwrap_or_else(|err| panic!("disk {disk_index} transaction epoch should decode: {err}")), + ); + } + epochs + } + + async fn current_data_dir(disk: &DiskStore, bucket: &str, object: &str) -> Uuid { + disk.read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("current object metadata should read") + .data_dir + .expect("test object should be stored out-of-line") + } + + async fn data_dir_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool { + disk.read_all(bucket, &format!("{object}/{data_dir}/part.1")).await.is_ok() + } + + async fn cleanup_receipt_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool { + disk.read_all(bucket, &old_data_cleanup_receipt_path(object, data_dir)) + .await + .is_ok() + } + + fn large_payload(fill: u8) -> Vec { + vec![fill; 1024 * 1024] + } + #[tokio::test] async fn second_pool_regular_put_uses_its_own_layout_and_round_trips() { // Deliberately inject the first pool's invalid scalar fallback. The @@ -9882,6 +10477,317 @@ mod heterogeneous_pool_put_tests { .expect("second-pool regular PUT should stream"); assert_eq!(restored, payload); } + + #[tokio::test] + #[serial(storage_class_env)] + async fn object_transaction_fencing_persists_epoch_only_when_gate_is_enabled() { + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-object-transaction-epoch"; + make_bucket(&disk_stores, bucket).await; + + let mut default_reader = PutObjReader::from_vec(b"default gate stays epoch-free".to_vec()); + set_disks + .put_object(bucket, "default.bin", &mut default_reader, &ObjectOptions::default()) + .await + .expect("default PUT should commit"); + assert_eq!( + object_transaction_epochs(&disk_stores, bucket, "default.bin").await, + vec![None, None, None, None], + "live proof alone must not write epoch metadata while the opt-in gate is disabled" + ); + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut fenced_reader = PutObjReader::from_vec(b"fenced epoch commit".to_vec()); + set_disks + .put_object(bucket, "fenced.bin", &mut fenced_reader, &ObjectOptions::default()) + .await + .expect("fenced PUT should commit with a live proof"); + }, + ) + .await; + + let epochs = object_transaction_epochs(&disk_stores, bucket, "fenced.bin").await; + let first = epochs[0].expect("fenced PUT should persist an epoch"); + assert!(!first.is_nil()); + assert!(epochs.into_iter().all(|epoch| epoch == Some(first))); + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn old_data_cleanup_receipt_reconciles_failed_put_cleanup_idempotently() { + use crate::set_disk::core::io_primitives::cleanup_fault_injection; + + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-cleanup-receipt-reconcile"; + let object = "object.bin"; + make_bucket(&disk_stores, bucket).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut first_reader = PutObjReader::from_vec(large_payload(0x11)); + set_disks + .put_object(bucket, object, &mut first_reader, &ObjectOptions::default()) + .await + .expect("first fenced PUT should commit"); + let old_dir = current_data_dir(&disk_stores[0], bucket, object).await; + + let fault = cleanup_fault_injection::fail_cleanup_on(object, &[0, 1, 2, 3]); + let mut overwrite_reader = PutObjReader::from_vec(large_payload(0x22)); + set_disks + .put_object(bucket, object, &mut overwrite_reader, &ObjectOptions::default()) + .await + .expect("overwrite should commit even when old-data cleanup fails"); + for disk in &disk_stores { + assert!( + cleanup_receipt_exists(disk, bucket, object, old_dir).await, + "fenced failed cleanup should leave a durable receipt" + ); + assert!( + data_dir_exists(disk, bucket, object, old_dir).await, + "injected cleanup failure should leave the old data dir for reconciliation" + ); + } + drop(fault); + + let removed = set_disks + .reconcile_old_data_cleanup_receipts(bucket, object) + .await + .expect("receipt reconciliation should succeed"); + assert_eq!(removed, 4, "all receipt-targeted old dirs should be reclaimed"); + let repeated = set_disks + .reconcile_old_data_cleanup_receipts(bucket, object) + .await + .expect("receipt reconciliation should be idempotent"); + assert_eq!(repeated, 0, "a drained receipt must not be counted again"); + for disk in &disk_stores { + assert!( + !data_dir_exists(disk, bucket, object, old_dir).await, + "reconciled old data dir must be gone" + ); + } + }, + ) + .await; + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn old_data_cleanup_receipt_noops_after_epoch_mismatch() { + use crate::set_disk::core::io_primitives::cleanup_fault_injection; + + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-cleanup-receipt-epoch-mismatch"; + let object = "object.bin"; + make_bucket(&disk_stores, bucket).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut first_reader = PutObjReader::from_vec(large_payload(0x31)); + set_disks + .put_object(bucket, object, &mut first_reader, &ObjectOptions::default()) + .await + .expect("first fenced PUT should commit"); + let old_dir = current_data_dir(&disk_stores[0], bucket, object).await; + + let fault = cleanup_fault_injection::fail_cleanup_on(object, &[0, 1, 2, 3]); + let mut second_reader = PutObjReader::from_vec(large_payload(0x32)); + set_disks + .put_object(bucket, object, &mut second_reader, &ObjectOptions::default()) + .await + .expect("second fenced PUT should commit and leave a receipt"); + drop(fault); + + let mut third_reader = PutObjReader::from_vec(large_payload(0x33)); + set_disks + .put_object(bucket, object, &mut third_reader, &ObjectOptions::default()) + .await + .expect("third fenced PUT should advance the current epoch"); + + let removed = set_disks + .reconcile_old_data_cleanup_receipts(bucket, object) + .await + .expect("stale receipt reconciliation should succeed"); + assert_eq!(removed, 0, "stale receipt epoch must not reclaim after a newer commit"); + for disk in &disk_stores { + assert!( + cleanup_receipt_exists(disk, bucket, object, old_dir).await, + "stale receipt should remain as no-op evidence" + ); + assert!( + data_dir_exists(disk, bucket, object, old_dir).await, + "epoch mismatch must preserve the old receipt target" + ); + } + }, + ) + .await; + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn old_data_cleanup_receipt_is_not_persisted_without_epoch_gate() { + use crate::set_disk::core::io_primitives::cleanup_fault_injection; + + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-cleanup-receipt-gate-off"; + let object = "object.bin"; + make_bucket(&disk_stores, bucket).await; + + let mut first_reader = PutObjReader::from_vec(large_payload(0x41)); + set_disks + .put_object(bucket, object, &mut first_reader, &ObjectOptions::default()) + .await + .expect("default-gate first PUT should commit"); + let old_dir = current_data_dir(&disk_stores[0], bucket, object).await; + + let _fault = cleanup_fault_injection::fail_cleanup_on(object, &[0, 1, 2, 3]); + let mut second_reader = PutObjReader::from_vec(large_payload(0x42)); + set_disks + .put_object(bucket, object, &mut second_reader, &ObjectOptions::default()) + .await + .expect("default-gate overwrite should commit"); + + let removed = set_disks + .reconcile_old_data_cleanup_receipts(bucket, object) + .await + .expect("gate-off receipt reconciliation should be a no-op"); + assert_eq!(removed, 0, "mixed-version/default gate path must not consume epoch-dependent receipts"); + for disk in &disk_stores { + assert!( + !cleanup_receipt_exists(disk, bucket, object, old_dir).await, + "mixed-version/default gate path must not write epoch-dependent receipts" + ); + assert!( + data_dir_exists(disk, bucket, object, old_dir).await, + "cleanup fault should leave old dir without receipt" + ); + } + } + + #[tokio::test] + #[serial(storage_class_env)] + async fn object_transaction_fencing_rejects_stale_no_lock_put_epoch() { + let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test"); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "put-object-transaction-stale-epoch"; + let object = "object.bin"; + make_bucket(&disk_stores, bucket).await; + + temp_env::async_with_vars( + [ + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), + (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + ], + async { + let mut initial_reader = PutObjReader::from_vec(b"initial fenced body".to_vec()); + set_disks + .put_object( + bucket, + object, + &mut initial_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("initial fenced PUT should commit"); + let initial_epoch = object_transaction_epochs(&disk_stores, bucket, object) + .await + .into_iter() + .next() + .flatten() + .expect("initial fenced PUT should persist an epoch"); + + let barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeTransactionEpochVerify); + let stale_set = Arc::clone(&set_disks); + let stale = tokio::spawn(async move { + let mut stale_reader = PutObjReader::from_vec(b"stale writer body".to_vec()); + stale_set + .put_object( + bucket, + object, + &mut stale_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + }); + barrier.wait_until_paused().await; + + let mut winner_reader = PutObjReader::from_vec(b"winning writer body".to_vec()); + set_disks + .put_object( + bucket, + object, + &mut winner_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("concurrent fenced PUT should advance the epoch"); + let winning_epoch = object_transaction_epochs(&disk_stores, bucket, object) + .await + .into_iter() + .next() + .flatten() + .expect("winning fenced PUT should persist an epoch"); + assert_ne!(winning_epoch, initial_epoch); + + barrier.release(); + let err = stale + .await + .expect("stale PUT task should not panic") + .expect_err("stale epoch PUT must be rejected"); + assert_eq!(err, StorageError::PreconditionFailed); + + let final_epochs = object_transaction_epochs(&disk_stores, bucket, object).await; + assert!(final_epochs.into_iter().all(|epoch| epoch == Some(winning_epoch))); + let mut reader = set_disks + .get_object_reader( + bucket, + object, + None, + HeaderMap::new(), + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("winning object should remain readable"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("winning body should stream"); + assert_eq!(restored, b"winning writer body"); + }, + ) + .await; + } } #[cfg(test)] diff --git a/crates/ecstore/src/set_disk/replication.rs b/crates/ecstore/src/set_disk/replication.rs index d968cd487..86126c535 100644 --- a/crates/ecstore/src/set_disk/replication.rs +++ b/crates/ecstore/src/set_disk/replication.rs @@ -48,6 +48,23 @@ impl RestoreCleanupIdentity { } } +fn ensure_restore_metadata_lock_held(bucket: &str, object: &str, opts: &ObjectOptions, mode: &'static str) -> Result<()> { + if opts + .namespace_lock_fence + .as_ref() + .is_some_and(NamespaceLockFence::is_lock_lost) + { + return Err(StorageError::NamespaceLockQuorumUnavailable { + mode, + bucket: bucket.to_string(), + object: object.to_string(), + required: 1, + achieved: 0, + }); + } + Ok(()) +} + impl SetDisks { pub(super) async fn finalize_restore_metadata( &self, @@ -88,6 +105,7 @@ impl SetDisks { if !expected.matches_file_info(&fi, &expected_etag) { return Err(Error::other("restored object changed before restore metadata finalization")); } + ensure_restore_metadata_lock_held(bucket, object, opts, "restore_finalize_metadata")?; let restore_expiry = lifecycle::expected_expiry_time(OffsetDateTime::now_utc(), opts.transition.restore_request.days.unwrap_or(1)); fi.metadata.insert( @@ -159,6 +177,7 @@ impl SetDisks { if !expected.matches_file_info(&fi, &expected_etag) { return Ok(()); } + ensure_restore_metadata_lock_held(bucket, object, opts, "restore_cleanup_metadata")?; fi.metadata.remove(X_AMZ_RESTORE.as_str()); fi.metadata.remove(AMZ_RESTORE_EXPIRY_DAYS); fi.metadata.remove(AMZ_RESTORE_REQUEST_DATE); diff --git a/crates/filemeta/src/fileinfo.rs b/crates/filemeta/src/fileinfo.rs index 0f799a40e..393975c18 100644 --- a/crates/filemeta/src/fileinfo.rs +++ b/crates/filemeta/src/fileinfo.rs @@ -18,8 +18,9 @@ use rmp_serde::Serializer; use rustfs_utils::HashAlgorithm; use rustfs_utils::http::{ AMZ_OBJECT_TAGGING, SUFFIX_COMPRESSION, SUFFIX_DATA_MOVED, SUFFIX_DATA_MOVED_TAGS, SUFFIX_FREE_VERSION, SUFFIX_HEALING, - SUFFIX_INLINE_DATA, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TIER_SKIP_FV_ID, contains_key_str, get_str, - has_internal_suffix, insert_str, is_encryption_metadata_key, starts_with_ignore_ascii_case, + SUFFIX_INLINE_DATA, SUFFIX_OBJECT_TRANSACTION_EPOCH, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TIER_SKIP_FV_ID, + contains_key_str, get_consistent_str, get_str, has_internal_suffix, insert_str, is_encryption_metadata_key, + starts_with_ignore_ascii_case, }; use s3s::dto::{RestoreStatus, Timestamp}; use s3s::header::X_AMZ_RESTORE; @@ -1172,6 +1173,22 @@ impl FileInfo { insert_str(&mut self.metadata, SUFFIX_DATA_MOVED, String::new()); } + pub fn set_object_transaction_epoch(&mut self, epoch: Uuid) { + insert_str(&mut self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH, epoch.to_string()); + } + + pub fn object_transaction_epoch(&self) -> Result> { + if !contains_key_str(&self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH) { + return Ok(None); + } + let value = get_consistent_str(&self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH).ok_or(Error::FileCorrupt)?; + let epoch = Uuid::parse_str(value).map_err(|_| Error::FileCorrupt)?; + if epoch.is_nil() { + return Err(Error::FileCorrupt); + } + Ok(Some(epoch)) + } + pub fn inline_data(&self) -> bool { contains_key_str(&self.metadata, SUFFIX_INLINE_DATA) && !self.is_remote() } @@ -1484,6 +1501,46 @@ mod tests { assert_eq!(ei.get_checksum_info(99).algorithm, HashAlgorithm::HighwayHash256S); } + #[test] + fn object_transaction_epoch_uses_consistent_dual_internal_metadata() { + let mut fi = validation_test_fileinfo(); + assert_eq!(fi.object_transaction_epoch().expect("absent epoch should decode"), None); + + let epoch = Uuid::new_v4(); + let epoch_text = epoch.to_string(); + fi.set_object_transaction_epoch(epoch); + assert_eq!(fi.object_transaction_epoch().expect("written epoch should decode"), Some(epoch)); + assert_eq!(fi.metadata.get("x-rustfs-internal-object-transaction-epoch"), Some(&epoch_text)); + assert_eq!(fi.metadata.get("x-minio-internal-object-transaction-epoch"), Some(&epoch_text)); + + let mut rustfs_only = validation_test_fileinfo(); + rustfs_only + .metadata + .insert("x-rustfs-internal-object-transaction-epoch".to_string(), epoch_text); + assert_eq!( + rustfs_only + .object_transaction_epoch() + .expect("single compatibility key should decode"), + Some(epoch) + ); + + let mut conflicting = fi.clone(); + conflicting + .metadata + .insert("x-minio-internal-object-transaction-epoch".to_string(), Uuid::new_v4().to_string()); + assert_eq!(conflicting.object_transaction_epoch(), Err(Error::FileCorrupt)); + + let mut malformed = validation_test_fileinfo(); + malformed + .metadata + .insert("x-rustfs-internal-object-transaction-epoch".to_string(), "not-a-uuid".to_string()); + assert_eq!(malformed.object_transaction_epoch(), Err(Error::FileCorrupt)); + + let mut nil = validation_test_fileinfo(); + nil.set_object_transaction_epoch(Uuid::nil()); + assert_eq!(nil.object_transaction_epoch(), Err(Error::FileCorrupt)); + } + // backlog#949: distribution range/permutation validation. #[test] fn is_valid_distribution_accepts_permutation() { diff --git a/crates/utils/src/http/metadata_compat.rs b/crates/utils/src/http/metadata_compat.rs index bc71f81ec..a3cc07b6d 100644 --- a/crates/utils/src/http/metadata_compat.rs +++ b/crates/utils/src/http/metadata_compat.rs @@ -59,6 +59,7 @@ pub const SUFFIX_TRANSITION_TIER_DESTINATION_ID: &str = "transition-tier-destina pub const SUFFIX_TRANSITION_TRANSACTION_ID: &str = "transition-transaction-id"; pub const SUFFIX_RESTORE_OPERATION_ID: &str = "restore-operation-id"; pub const SUFFIX_BUCKET_INCARNATION_ID: &str = "bucket-incarnation-id"; +pub const SUFFIX_OBJECT_TRANSACTION_EPOCH: &str = "object-transaction-epoch"; pub const SUFFIX_FREE_VERSION: &str = "free-version"; pub const SUFFIX_PURGESTATUS: &str = "purgestatus"; pub const SUFFIX_REPLICA_STATUS: &str = "replica-status"; diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index cc4d0b798..8dd561da1 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -46,6 +46,7 @@ use super::storage_api::multipart_usecase::options::{ get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, parse_copy_source_range, put_opts_with_replication_authorization, validate_archive_content_encoding, }; +use super::storage_api::multipart_usecase::request_context::spawn_traced_join; use super::storage_api::multipart_usecase::s3_api::multipart::{ ListMultipartUploadsParams, build_list_multipart_uploads_output, build_list_parts_output, parse_list_multipart_uploads_params, parse_list_parts_params, parse_upload_part_number, @@ -588,56 +589,94 @@ impl DefaultMultipartUsecase { None => None, }; - let obj_info = store - .clone() - .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, &opts) - .await - .map_err(ApiError::from)?; - let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await; - record_capacity_write(Some(capacity_scope_token)).await; - - if let Some(metadata_sys) = quota_metadata_sys.as_ref() { - if opts.replication_request { - let quota_checker = QuotaChecker::new(metadata_sys.clone()); - match quota_checker - .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64) + let complete_commit = spawn_traced_join({ + let store = Arc::clone(&store); + let bucket = bucket.clone(); + let key = key.clone(); + let upload_id = upload_id.clone(); + let opts = opts.clone(); + let quota_metadata_sys = quota_metadata_sys.clone(); + async move { + let obj_info = store + .clone() + .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, &opts) .await - { - Ok(check_result) if !check_result.allowed => { - let _ = store.delete_object(&bucket, &key, ObjectOptions::default()).await; - let _ = invalidate_object_data_cache_after_delete_success(&cache_adapter, &bucket, &key).await; - return Err(S3Error::with_message( - S3ErrorCode::InvalidRequest, - format!( - "Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes", - check_result.current_usage.unwrap_or(0), - check_result.quota_limit.unwrap_or(0) - ), - )); + .map_err(ApiError::from)?; + let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await; + record_capacity_write(Some(capacity_scope_token)).await; + + if let Some(metadata_sys) = quota_metadata_sys.as_ref() { + if opts.replication_request { + let quota_checker = QuotaChecker::new(metadata_sys.clone()); + match quota_checker + .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64) + .await + { + Ok(check_result) if !check_result.allowed => { + let _ = store.delete_object(&bucket, &key, ObjectOptions::default()).await; + let _ = invalidate_object_data_cache_after_delete_success(&cache_adapter, &bucket, &key).await; + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!( + "Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes", + check_result.current_usage.unwrap_or(0), + check_result.quota_limit.unwrap_or(0) + ), + )); + } + Err(err) => { + warn!("Quota check failed for bucket {} after multipart completion: {}", bucket, err); + } + Ok(_) => {} + } } - Err(err) => { - warn!("Quota check failed for bucket {} after multipart completion: {}", bucket, err); + + let committed_size = if opts.replication_request { + obj_info.size.max(0) as u64 + } else { + quota_accounting_object_size(&obj_info, opts.quota_admission.is_some())? + }; + if versioned { + record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await; + } else { + record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await; } - Ok(_) => {} } + + enqueue_transition_immediate(&obj_info, LcEventSrc::S3CompleteMultipartUpload).await; + + let mt2 = obj_info.user_defined.clone(); + let dsc = must_replicate_object( + &bucket, + &key, + &mt2, + "".to_string(), + opts.delete_marker_replication_status(), + opts.clone(), + ) + .await; + + if dsc.replicate_any() { + warn!("need multipart replication"); + schedule_object_replication(obj_info.clone(), store, dsc).await; + } + + rustfs_scanner::record_dirty_usage_bucket(&bucket); + Ok::<_, S3Error>(obj_info) } + }); + let obj_info = complete_commit.await.map_err(|err| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("complete multipart upload commit owner task failed: {err}"), + ) + })??; - let committed_size = if opts.replication_request { - obj_info.size.max(0) as u64 - } else { - quota_accounting_object_size(&obj_info, opts.quota_admission.is_some())? - }; - if versioned { - record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await; - } else { - record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await; - } - } - - enqueue_transition_immediate(&obj_info, LcEventSrc::S3CompleteMultipartUpload).await; - - let raw_mpu_version = obj_info.version_id.map(|v| v.to_string()); - let mpu_version = if versioned { raw_mpu_version.clone() } else { None }; + let mpu_version = if versioned { + obj_info.version_id.map(|v| v.to_string()) + } else { + None + }; let mpu_version_for_event = mpu_version.clone(); // checksum: stored (decrypted) values take precedence over the request input; // additional algorithms (XXHash3/64/128, SHA-512, MD5), which have no typed @@ -660,28 +699,18 @@ impl DefaultMultipartUsecase { bucket: Some(bucket.clone()), key: Some(key.clone()), e_tag: obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)), - location: Some(location.clone()), + location: Some(location), server_side_encryption: server_side_encryption.clone(), ssekms_key_id: ssekms_key_id.clone(), - checksum_crc32: checksum_crc32.clone(), - checksum_crc32c: checksum_crc32c.clone(), - checksum_sha1: checksum_sha1.clone(), - checksum_sha256: checksum_sha256.clone(), - checksum_crc64nvme: checksum_crc64nvme.clone(), - checksum_type: checksum_type.clone(), + checksum_crc32, + checksum_crc32c, + checksum_sha1, + checksum_sha256, + checksum_crc64nvme, + checksum_type, version_id: mpu_version, ..Default::default() }; - let mt2 = obj_info.user_defined.clone(); - let dsc = - must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts.clone()) - .await; - - if dsc.replicate_any() { - warn!("need multipart replication"); - schedule_object_replication(obj_info.clone(), store, dsc).await; - } - // Set object info for event notification helper = helper.object(obj_info); if let Some(version_id) = &mpu_version_for_event { @@ -712,7 +741,6 @@ impl DefaultMultipartUsecase { } let result = Ok(response); let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); result } diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 0e3bbb031..b48430d28 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -2989,6 +2989,11 @@ struct PutObjectChecksums { crc64nvme: Option, } +struct PutObjectCommitResult { + obj_info: ObjectInfo, + put_versioned: bool, +} + fn normalize_delete_objects_version_id( version_id: Option, ) -> std::result::Result<(Option, Option), String> { @@ -5932,7 +5937,7 @@ impl DefaultObjectUsecase { reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?; rustfs_io_metrics::record_put_object_stage_duration_from("app_encryption_prepare", encryption_stage_start); - let mut reader = PutObjReader::new(reader); + let reader = PutObjReader::new(reader); let mt2 = metadata.clone(); opts.user_defined.extend(metadata); @@ -6005,97 +6010,145 @@ impl DefaultObjectUsecase { } else { None }; - let object_traffic_progress = object_traffic_health - .as_deref() - .and_then(ObjectTrafficHealth::track_write_storage); - let store_put_stage_start = put_stage_metrics_enabled.then(Instant::now); - let (obj_info, backfilled_old_current_size) = match store - .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts) - .await - .map_err(ApiError::from) - { - Ok(obj_info) => { - store_put_watchdog.cancel(); - debug!( - target: "rustfs::app::object_usecase", - event = EVENT_PUT_OBJECT_STORE_RETURNED, - component = LOG_COMPONENT_APP, - subsystem = LOG_SUBSYSTEM_OBJECT, - request_id = %request_id, - bucket = %bucket, - key = %key, - put_path = put_path, - object_size = actual_size, - duration_ms = start_time.elapsed().as_millis() as u64, - result = "success", - "PutObject store write returned" - ); - obj_info + let put_commit = spawn_traced_join({ + let store = Arc::clone(&store); + let bucket = bucket.clone(); + let key = key.clone(); + let opts = opts.clone(); + let cache_adapter = cache_adapter.clone(); + let request_id = request_id.clone(); + let put_path = put_path.to_string(); + async move { + let object_traffic_progress = object_traffic_health + .as_deref() + .and_then(ObjectTrafficHealth::track_write_storage); + let mut reader = reader; + let store_put_stage_start = put_stage_metrics_enabled.then(Instant::now); + let (obj_info, backfilled_old_current_size) = match store + .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts) + .await + .map_err(ApiError::from) + { + Ok(obj_info) => { + store_put_watchdog.cancel(); + debug!( + target: "rustfs::app::object_usecase", + event = EVENT_PUT_OBJECT_STORE_RETURNED, + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_OBJECT, + request_id = %request_id, + bucket = %bucket, + key = %key, + put_path = %put_path, + object_size = actual_size, + duration_ms = start_time.elapsed().as_millis() as u64, + result = "success", + "PutObject store write returned" + ); + obj_info + } + Err(err) => { + store_put_watchdog.cancel(); + rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start); + warn!( + target: "rustfs::app::object_usecase", + event = EVENT_PUT_OBJECT_STORE_RETURNED, + component = LOG_COMPONENT_APP, + subsystem = LOG_SUBSYSTEM_OBJECT, + request_id = %request_id, + bucket = %bucket, + key = %key, + put_path = %put_path, + object_size = actual_size, + duration_ms = start_time.elapsed().as_millis() as u64, + result = "error", + error = %err, + "PutObject store write returned" + ); + return Err(err.into()); + } + }; + rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start); + drop(object_traffic_progress); + #[cfg(test)] + wait_for_put_post_store_test_hook(&bucket).await; + + let post_store_stage_start = put_stage_metrics_enabled.then(Instant::now); + maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await; + let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &key).await; + + let put_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; + // Fast in-memory update for immediate quota and admin usage consistency. + // The previous current size comes from the prelookup when it ran, + // otherwise from the rename_data backfill (rustfs/backlog#1009); the + // backfill reproduces the lookup's observation bit for bit (latest + // version's ObjectInfo.size — 0 for a delete-marker latest — or + // not-found → None). + match prelookup_previous_current_size.or_else(|| previous_current_size_from_backfill(backfilled_old_current_size)) + { + Some(previous_current_size) => { + if put_versioned { + record_bucket_object_version_write_memory( + &bucket, + previous_current_size, + obj_info.size.max(0) as u64, + ) + .await; + } else { + record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; + } + } + None => { + // Neither source could determine the previous state (peers + // predating the backfill field during a rolling upgrade, or + // sub-quorum metadata divergence). Record the components that + // are correct regardless; the next authoritative scanner + // refresh replaces the in-memory numbers. + debug!( + target: "rustfs::app::object_usecase", + bucket = %bucket, + key = %key, + put_versioned, + "put_object old-size backfill unknown; recording degraded usage delta" + ); + record_bucket_object_write_unknown_previous_memory(&bucket, obj_info.size.max(0) as u64, put_versioned) + .await; + } + } + + if dsc.replicate_any() { + schedule_object_replication(obj_info.clone(), store, dsc).await; + } + + rustfs_scanner::record_dirty_usage_bucket(&bucket); + rustfs_io_metrics::record_put_object_stage_duration_from("app_post_store_bookkeeping", post_store_stage_start); + + let capacity_update_stage_start = put_stage_metrics_enabled.then(Instant::now); + let manager = get_capacity_manager(); + manager.record_write_operation().await; + rustfs_io_metrics::record_put_object_stage_duration_from("app_capacity_update", capacity_update_stage_start); + + Ok::<_, S3Error>(PutObjectCommitResult { obj_info, put_versioned }) + } + }); + let PutObjectCommitResult { obj_info, put_versioned } = match put_commit.await { + Ok(Ok(result)) => result, + Ok(Err(err)) => { + let result: S3Result> = Err(err); + put_request_guard.finish_err(); + let _ = helper.complete(&result); + return result; } Err(err) => { - store_put_watchdog.cancel(); - rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start); - warn!( - target: "rustfs::app::object_usecase", - event = EVENT_PUT_OBJECT_STORE_RETURNED, - component = LOG_COMPONENT_APP, - subsystem = LOG_SUBSYSTEM_OBJECT, - request_id = %request_id, - bucket = %bucket, - key = %key, - put_path = put_path, - object_size = actual_size, - duration_ms = start_time.elapsed().as_millis() as u64, - result = "error", - error = %err, - "PutObject store write returned" - ); - let result: S3Result> = Err(err.into()); + let result: S3Result> = Err(S3Error::with_message( + S3ErrorCode::InternalError, + format!("put object commit owner task failed: {err}"), + )); put_request_guard.finish_err(); let _ = helper.complete(&result); return result; } }; - rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start); - drop(object_traffic_progress); - #[cfg(test)] - wait_for_put_post_store_test_hook(&bucket).await; - - let post_store_stage_start = put_stage_metrics_enabled.then(Instant::now); - maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await; - let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &key).await; - - let put_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await; - // Fast in-memory update for immediate quota and admin usage consistency. - // The previous current size comes from the prelookup when it ran, - // otherwise from the rename_data backfill (rustfs/backlog#1009); the - // backfill reproduces the lookup's observation bit for bit (latest - // version's ObjectInfo.size — 0 for a delete-marker latest — or - // not-found → None). - match prelookup_previous_current_size.or_else(|| previous_current_size_from_backfill(backfilled_old_current_size)) { - Some(previous_current_size) => { - if put_versioned { - record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; - } else { - record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await; - } - } - None => { - // Neither source could determine the previous state (peers - // predating the backfill field during a rolling upgrade, or - // sub-quorum metadata divergence). Record the components that - // are correct regardless; the next authoritative scanner - // refresh replaces the in-memory numbers. - debug!( - target: "rustfs::app::object_usecase", - bucket = %bucket, - key = %key, - put_versioned, - "put_object old-size backfill unknown; recording degraded usage delta" - ); - record_bucket_object_write_unknown_previous_memory(&bucket, obj_info.size.max(0) as u64, put_versioned).await; - } - } let raw_version = obj_info.version_id.map(|v| v.to_string()); @@ -6110,17 +6163,6 @@ impl DefaultObjectUsecase { let expiration = resolve_put_object_expiration(&bucket, &obj_info).await; - // Reuse the single replication decision computed before commit (see `dsc` - // above) so the pending metadata persisted with the object and the - // post-commit schedule always derive from the same immutable decision. - // Recomputing here would repeat the versioning/config/target traversal and, - // worse, allow a replication-config hot update between the two phases to - // produce a pending-without-schedule or schedule-without-pending divergence - // (https://github.com/rustfs/backlog/issues/1320). - if dsc.replicate_any() { - schedule_object_replication(obj_info.clone(), store, dsc).await; - } - let mut checksums = PutObjectChecksums { crc32: input.checksum_crc32, crc32c: input.checksum_crc32c, @@ -6159,14 +6201,6 @@ impl DefaultObjectUsecase { inject_additional_checksum_headers(&mut response.headers, &put_extra_checksum_headers); let result = Ok(response); let _ = helper.complete(&result); - rustfs_scanner::record_dirty_usage_bucket(&bucket); - rustfs_io_metrics::record_put_object_stage_duration_from("app_post_store_bookkeeping", post_store_stage_start); - - // Record write operation for capacity management (inline to avoid per-request tokio::spawn overhead) - let capacity_update_stage_start = put_stage_metrics_enabled.then(Instant::now); - let manager = get_capacity_manager(); - manager.record_write_operation().await; - rustfs_io_metrics::record_put_object_stage_duration_from("app_capacity_update", capacity_update_stage_start); // Record PutObject metrics via zero-copy-metrics { @@ -11507,6 +11541,76 @@ mod tests { assert!(!recovered.write_stalled); } + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn cancelled_put_request_completes_post_commit_publication() { + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + + let (store, context) = real_cold_fill_test_context().await; + let bucket = format!("put-owner-tail-{}", Uuid::new_v4()); + let object = "object.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("PUT owner-tail bucket must be created"); + + let old_body = Bytes::from_static(b"old body that must be invalidated"); + let old_info = put_real_cold_fill_object(&store, &bucket, object, &old_body).await; + let adapter = context.object_data_cache(); + let old_plan = real_cold_fill_plan(&adapter, &bucket, object, &old_info); + + let post_store_entered = Arc::new(tokio::sync::Barrier::new(2)); + let post_store_resume = Arc::new(tokio::sync::Barrier::new(2)); + install_put_post_store_test_hook(bucket.clone(), Arc::clone(&post_store_entered), Arc::clone(&post_store_resume)); + + let payload = Bytes::from_static(b"published despite caller cancellation"); + let put_input = PutObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .body(Some(StreamingBlob::from(s3s::Body::from(payload.clone())))) + .content_length(Some(i64::try_from(payload.len()).expect("test payload length must fit i64"))) + .build() + .expect("PUT input must build"); + let put_usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context))); + let put = tokio::spawn(async move { + put_usecase + .execute_put_object(&FS::new(), build_request(put_input, Method::PUT)) + .await + }); + + tokio::time::timeout(Duration::from_secs(10), post_store_entered.wait()) + .await + .expect("PUT must reach the post-store owner-tail hook"); + assert_eq!( + adapter.fill_body(&old_plan, old_body.clone()).await, + rustfs_object_data_cache::ObjectDataCacheFillResult::Inserted, + "test must republish the old body while the owner tail is paused" + ); + put.abort(); + post_store_resume.wait().await; + let _ = put.await.expect_err("outer request task must be cancelled"); + + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if matches!( + adapter.lookup_body(&old_plan).await, + rustfs_object_data_cache::ObjectDataCacheLookup::Miss + ) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("post-commit owner tail must invalidate stale body cache after caller cancellation"); + + let recovered = store + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .expect("cancelled request's owned commit must still publish the object"); + assert_eq!(recovered.size, i64::try_from(payload.len()).expect("test payload length must fit i64")); + } + #[tokio::test] async fn object_progress_tracks_zero_byte_and_zero_copy_put_lock_waits() { use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index c18e8c3b3..39d7d7685 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -1150,7 +1150,9 @@ pub(crate) mod multipart_usecase { } } - pub(crate) use super::{access, bucket, data_usage, error, helper, io, object_utils, options, s3_api, set_disk, sse}; + pub(crate) use super::{ + access, bucket, data_usage, error, helper, io, object_utils, options, request_context, s3_api, set_disk, sse, + }; pub(crate) use crate::storage::storage_api::{ECStore, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader}; }