diff --git a/crates/ecstore/src/bucket/quota/reservation.rs b/crates/ecstore/src/bucket/quota/reservation.rs index 33aa5c876..d4a35c23e 100644 --- a/crates/ecstore/src/bucket/quota/reservation.rs +++ b/crates/ecstore/src/bucket/quota/reservation.rs @@ -38,6 +38,7 @@ const MAX_ORPHANS_REAPED_PER_WRITE: usize = 64; const MAX_ORPHAN_PROBES_PER_WRITE: usize = 128; const ORPHAN_PROBE_CONCURRENCY: usize = 32; const EVENT_QUOTA_LEDGER_SETTLEMENT: &str = "quota_ledger_settlement"; +const EVENT_QUOTA_ADMISSION: &str = "quota_admission"; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_QUOTA: &str = "quota"; @@ -513,6 +514,7 @@ pub(crate) async fn begin( .as_ref() .is_some_and(|quota| quota.has_unsupported_reservation_protocol()) { + log_admission_rejected(bucket, object, "unsupported_reservation_protocol"); return Err(StorageError::PartMissingOrCorrupt); } let durable_quota = quota.as_ref().filter(|quota| quota.uses_durable_reservations()); @@ -529,7 +531,16 @@ pub(crate) async fn begin( Some(quota) => match (quota.quota, snapshot_admission) { (Some(limit), Some(admission)) if admission.quota_limit() == limit => Some(admission), (Some(_), None) if data_movement => None, - (Some(_), _) => return Err(StorageError::PartMissingOrCorrupt), + (Some(_), _) => { + // A snapshot-protocol quota requires the request handler's + // admission on every write. Missing or mismatched admission + // means a caller rebuilt `ObjectOptions` without carrying it + // over (rustfs/rustfs#7674 lost it on CopyObject); fail closed + // but leave a diagnosable trace, because the storage error is + // the generic `PartMissingOrCorrupt`. + log_admission_rejected(bucket, object, "snapshot_quota_admission_missing"); + return Err(StorageError::PartMissingOrCorrupt); + } (None, _) => None, }, None => None, @@ -904,6 +915,18 @@ pub fn fail_next_quota_ledger_save_for_test() { FAIL_NEXT_LEDGER_SAVE.store(true, std::sync::atomic::Ordering::SeqCst); } +fn log_admission_rejected(bucket: &str, object: &str, state: &'static str) { + warn!( + event = EVENT_QUOTA_ADMISSION, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_QUOTA, + state, + bucket = %bucket, + object = %object, + "quota admission rejected the write before commit" + ); +} + fn log_deferred_settlement(data: &LedgerReservationData, state: &'static str, err: &StorageError) { warn!( event = EVENT_QUOTA_LEDGER_SETTLEMENT, diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index 487d85a77..a83c4d699 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -790,6 +790,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for Sets { version_id: dst_opts.version_id.clone(), mod_time: dst_opts.mod_time, http_preconditions: dst_opts.http_preconditions.clone(), + quota_admission: dst_opts.quota_admission, ..Default::default() }; diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index baee8e787..115121e15 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -3149,6 +3149,120 @@ mod tests { shutdown.cancel(); } + /// rustfs/rustfs#7674: a bucket carrying a legacy snapshot-protocol quota + /// (written before durable reservations existed, so `quota.json` has no + /// `reservation_protocol`) must accept a cross-key CopyObject whose + /// destination options carry the handler's quota admission. The storage + /// layer fails closed with `PartMissingOrCorrupt` when a quota-enforced + /// bucket sees a write without admission, so dropping the admission while + /// rebuilding the destination options turns every copy into a + /// deterministic "part missing or corrupt" failure even though the source + /// object is perfectly readable. + #[cfg(feature = "test-util")] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[serial_test::serial(storage_class_env)] + async fn copy_object_forwards_quota_admission_on_legacy_snapshot_quota_bucket() { + let temp_dir = tempfile::tempdir().expect("create legacy quota copy store dir"); + let (ctx, store, shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "legacy-quota-copy", &[1])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await; + + let bucket = format!("legacy-quota-copy-{}", Uuid::new_v4()); + let source_object = "docker/registry/v2/repositories/example/_uploads/upload-id/data"; + let target_object = "docker/registry/v2/blobs/sha256/a5/digest/data"; + let payload = vec![0x5A; 8178]; + let quota_limit = 1u64 << 30; + + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create bucket for legacy quota copy"); + // Legacy quota shape: no `reservation_protocol`, so the storage layer + // requires the handler-supplied snapshot admission on every write. + let legacy_quota = format!(r#"{{"quota":{quota_limit},"quota_type":"Hard"}}"#); + crate::bucket::metadata_sys::update_in( + &ctx, + &bucket, + crate::bucket::metadata::BUCKET_QUOTA_CONFIG_FILE, + legacy_quota.into_bytes(), + ) + .await + .expect("persist legacy snapshot quota"); + let (quota, _, _) = crate::bucket::metadata_sys::get_quota_config_and_incarnation_from_disk_in(&ctx, &bucket) + .await + .expect("legacy quota should load"); + let quota = quota.expect("legacy quota must be persisted"); + assert_eq!(quota.quota, Some(quota_limit)); + assert!(!quota.uses_durable_reservations(), "fixture must stay on the snapshot protocol"); + + let mut write_opts = ObjectOptions::default(); + assert!(write_opts.set_quota_admission(0, quota_limit)); + + let upload = store + .new_multipart_upload(&bucket, source_object, &write_opts) + .await + .expect("create source multipart upload"); + let mut part_reader = PutObjReader::from_vec(payload.clone()); + let part = store + .put_object_part(&bucket, source_object, &upload.upload_id, 1, &mut part_reader, &write_opts) + .await + .expect("stage multipart source part"); + store + .clone() + .complete_multipart_upload( + &bucket, + source_object, + &upload.upload_id, + vec![crate::storage_api_contracts::multipart::CompletePart { + part_num: part.part_num, + etag: part.etag, + ..Default::default() + }], + &write_opts, + ) + .await + .expect("complete the multipart source under the legacy quota"); + + let source_reader = store + .get_object_reader(&bucket, source_object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("completed multipart source should be readable"); + let mut copy_info = source_reader.object_info.clone(); + let actual_size = copy_info.get_actual_size().expect("copy source logical size should resolve"); + assert_eq!(actual_size, payload.len() as i64); + let copy_reader = rustfs_rio::HashReader::from_stream(source_reader.stream, actual_size, actual_size, None, None, false) + .expect("copy source hash reader should build"); + copy_info.put_object_reader = Some(PutObjReader::new(copy_reader)); + + let mut dst_opts = ObjectOptions::default(); + assert!(dst_opts.set_quota_admission(payload.len() as u64, quota_limit)); + store + .copy_object( + &bucket, + source_object, + &bucket, + target_object, + &mut copy_info, + &ObjectOptions::default(), + &dst_opts, + ) + .await + .expect("CopyObject must forward the handler quota admission to the destination write"); + + let mut target_reader = store + .get_object_reader(&bucket, target_object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("copied target should be readable"); + let mut target_body = Vec::new(); + target_reader + .stream + .read_to_end(&mut target_body) + .await + .expect("target body should stream"); + assert_eq!(target_body, payload); + shutdown.cancel(); + } + #[cfg(feature = "test-util")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial_test::serial(storage_class_env)] diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index c5fbe2d26..a801ffea1 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -4564,6 +4564,7 @@ impl ECStore { namespace_lock_fence: dst_opts.namespace_lock_fence.clone(), bucket_lifecycle_lock_fence: dst_opts.bucket_lifecycle_lock_fence.clone(), object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(), + quota_admission: dst_opts.quota_admission, ..Default::default() }; if !self.single_pool() { @@ -4606,6 +4607,7 @@ impl ECStore { namespace_lock_fence: dst_opts.namespace_lock_fence.clone(), bucket_lifecycle_lock_fence: dst_opts.bucket_lifecycle_lock_fence.clone(), object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(), + quota_admission: dst_opts.quota_admission, ..Default::default() }; if !self.single_pool() { @@ -4658,6 +4660,7 @@ impl ECStore { namespace_lock_fence: dst_opts.namespace_lock_fence.clone(), bucket_lifecycle_lock_fence: dst_opts.bucket_lifecycle_lock_fence.clone(), object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(), + quota_admission: dst_opts.quota_admission, ..Default::default() }; if !self.single_pool() {