Merge branch 'main' into fix/ecstore-debug-log-boundary

.github/workflows/build.yml

@@ -1065,7 +1065,7 @@ jobs:
           while IFS= read -r preview_tag; do
             [[ -n "$preview_tag" ]] || continue
             echo "🧹 Deleting preview release $preview_tag (tag kept)"
-            gh release delete "$preview_tag" --yes
+            gh release delete "$preview_tag" --repo "${GITHUB_REPOSITORY}" --yes
             DELETED=$((DELETED + 1))
           done < <(
             jq -r --arg tag "$TAG" '

crates/ecstore/src/bucket/lifecycle/mod.rs

@@ -20,7 +20,7 @@ mod durable_namespace;
 pub mod evaluator;
 pub mod manual_transition_job;
 mod metadata_boundary;
-pub(crate) use metadata_boundary::{LifecycleExpiryConfigs, get_expiry_configs};
+pub(crate) use metadata_boundary::{LifecycleExpiryConfigs, get_expiry_configs, get_lifecycle_config};
 mod object_handlers_common;
 mod object_lock_boundary;
 pub use self::core as lifecycle;

crates/ecstore/src/data_usage/mod.rs

@@ -1161,11 +1161,45 @@ fn select_admin_data_usage_snapshot(
         authoritative.usage_snapshot_converged = Some(true);
     }
     match observed {
+        Some(observed)
+            if observed.usage_snapshot_partial
+                && authoritative.is_complete_bucket_usage_snapshot()
+                && observed_data_usage_is_newer(&observed, &authoritative) =>
+        {
+            (merge_partial_observation_for_admin(authoritative, observed), true)
+        }
         Some(observed) if observed_data_usage_is_newer(&observed, &authoritative) => (observed, true),
         _ => (authoritative, authoritative_format),
     }
 }
 
+fn merge_partial_observation_for_admin(mut authoritative: DataUsageInfo, observed: DataUsageInfo) -> DataUsageInfo {
+    for (bucket, usage) in observed.buckets_usage {
+        authoritative.buckets_usage.insert(bucket, usage);
+    }
+
+    authoritative.last_update = observed.last_update;
+    authoritative.scanner_cycle = observed.scanner_cycle;
+    authoritative.scanner_epoch = observed.scanner_epoch;
+    authoritative.usage_snapshot_complete = false;
+    authoritative.usage_snapshot_partial = true;
+    authoritative.usage_snapshot_converged = Some(false);
+    authoritative.usage_snapshot_authoritative_baseline = observed.usage_snapshot_authoritative_baseline;
+    authoritative.usage_snapshot_set_states = observed.usage_snapshot_set_states;
+    authoritative.usage_snapshot_bootstrap_pending = false;
+    authoritative.buckets_count = authoritative.buckets_usage.len() as u64;
+    authoritative.bucket_sizes = authoritative
+        .buckets_usage
+        .iter()
+        .map(|(bucket, usage)| (bucket.clone(), usage.size))
+        .collect();
+    authoritative.replication_info.clear();
+    authoritative.tier_stats = None;
+    authoritative.unknown_tier_stats = None;
+    authoritative.calculate_totals();
+    authoritative
+}
+
 async fn load_admin_data_usage_from_backend(store: Arc<ECStore>) -> Result<DataUsageInfo, Error> {
     let (authoritative, source) = load_data_usage_snapshot(store.clone()).await?;
     let observed = load_observed_data_usage_snapshot(store).await;
@@ -3217,6 +3251,108 @@ mod tests {
         assert_eq!(selected.buckets_usage.get("bucket").map(|usage| usage.size), Some(100));
     }
 
+    #[test]
+    fn partial_admin_observation_preserves_authoritative_cold_buckets() {
+        let baseline_time = SystemTime::UNIX_EPOCH + Duration::from_secs(10);
+        let mut authoritative = data_usage_info_for_test("cold", 152_318, 80 * 1024 * 1024 * 1024, baseline_time);
+        authoritative.scanner_epoch = Some(4);
+        authoritative.scanner_cycle = Some(10);
+        authoritative.buckets_usage.insert(
+            "hot".to_string(),
+            BucketUsageInfo {
+                objects_count: 3_000,
+                versions_count: 3_000,
+                size: 400 * 1024 * 1024,
+                ..Default::default()
+            },
+        );
+        authoritative.buckets_count = 2;
+        authoritative.bucket_sizes = authoritative
+            .buckets_usage
+            .iter()
+            .map(|(bucket, usage)| (bucket.clone(), usage.size))
+            .collect();
+        authoritative.calculate_totals();
+        authoritative.replication_info.insert(
+            "stale-target".to_string(),
+            BucketTargetUsageInfo {
+                replicated_size: 400 * 1024 * 1024,
+                replicated_count: 3_000,
+                ..Default::default()
+            },
+        );
+        authoritative.tier_stats = Some(rustfs_data_usage::AllTierStats {
+            tiers: HashMap::from([(
+                "WARM".to_string(),
+                rustfs_data_usage::TierStats {
+                    total_size: 80 * 1024 * 1024 * 1024,
+                    num_versions: 152_318,
+                    num_objects: 152_318,
+                },
+            )]),
+        });
+
+        let mut observed = DataUsageInfo {
+            last_update: Some(baseline_time + Duration::from_secs(1)),
+            scanner_epoch: Some(4),
+            scanner_cycle: Some(11),
+            usage_snapshot_complete: false,
+            usage_snapshot_partial: true,
+            usage_snapshot_converged: Some(false),
+            usage_snapshot_authoritative_baseline: Some(authoritative.snapshot_identity()),
+            usage_snapshot_set_states: vec![rustfs_data_usage::DataUsageSnapshotSetState {
+                pool_index: 0,
+                set_index: 0,
+                scanner_cycle: Some(11),
+                scanner_epoch: Some(4),
+                scan_plan_digest: Some([1; 32]),
+                complete: true,
+                tombstone: false,
+            }],
+            ..Default::default()
+        };
+        observed.buckets_usage.insert(
+            "hot".to_string(),
+            BucketUsageInfo {
+                objects_count: 34,
+                versions_count: 34,
+                size: 8 * 1024 * 1024,
+                ..Default::default()
+            },
+        );
+        observed.buckets_count = 1;
+        observed.bucket_sizes.insert("hot".to_string(), 8 * 1024 * 1024);
+        observed.calculate_totals();
+
+        let (selected, current_format) = select_admin_data_usage_snapshot(authoritative, true, Some(observed));
+
+        assert!(current_format);
+        assert!(!selected.usage_snapshot_complete);
+        assert!(selected.usage_snapshot_partial);
+        assert!(selected.is_valid_partial_snapshot());
+        assert_eq!(selected.usage_snapshot_converged, Some(false));
+        assert_eq!(selected.buckets_count, 2);
+        assert_eq!(
+            selected
+                .buckets_usage
+                .get("cold")
+                .map(|usage| (usage.objects_count, usage.size)),
+            Some((152_318, 80 * 1024 * 1024 * 1024))
+        );
+        assert_eq!(
+            selected
+                .buckets_usage
+                .get("hot")
+                .map(|usage| (usage.objects_count, usage.size)),
+            Some((34, 8 * 1024 * 1024))
+        );
+        assert_eq!(selected.objects_total_count, 152_352);
+        assert_eq!(selected.objects_total_size, 80 * 1024 * 1024 * 1024 + 8 * 1024 * 1024);
+        assert!(selected.replication_info.is_empty());
+        assert!(selected.tier_stats.is_none());
+        assert!(selected.unknown_tier_stats.is_none());
+    }
+
     #[tokio::test]
     async fn authoritative_save_cleanup_removes_observed_snapshot_best_effort() {
         let store = UsageCasStore::default();

crates/ecstore/src/services/tier/tier.rs

@@ -65,6 +65,7 @@ use crate::storage_api_contracts::{
 };
 use crate::{
     bucket::lifecycle::{
+        get_lifecycle_config,
         tier_delete_journal::{TIER_DELETE_JOURNAL_PREFIX, decode_tier_delete_journal_entry},
         transition_transaction::{TRANSITION_TRANSACTION_RECORD_PREFIX, decode_transition_transaction_record},
     },
@@ -81,7 +82,7 @@ use rustfs_filemeta::FileInfo;
 use rustfs_rio::HashReader;
 use rustfs_s3_client::{admin_handler_utils::AdminError, provider_versions::ProviderVersionCapabilities};
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join};
-use s3s::S3ErrorCode;
+use s3s::{S3ErrorCode, dto::BucketLifecycleConfiguration};
 
 use super::{
     tier_handlers::{ERR_TIER_BUCKET_NOT_FOUND, ERR_TIER_CONNECT_ERR, ERR_TIER_INVALID_CREDENTIALS, ERR_TIER_PERM_ERR},
@@ -694,6 +695,7 @@ fn tier_backend_identity_admin_error(err: io::Error) -> AdminError {
     admin_err
 }
 
+#[async_trait::async_trait]
 trait TierReferenceProofStore:
     EcstoreObjectIO
     + BucketOperations<Error = Error>
@@ -705,23 +707,21 @@ trait TierReferenceProofStore:
         WalkOptions = TierReferenceProofWalkOptions,
         WalkCancellation = tokio_util::sync::CancellationToken,
         WalkResultSender = tokio::sync::mpsc::Sender<StorageObjectInfoOrErr<ObjectInfo, Error>>,
-    >
+    > + Send
+    + Sync
 {
+    async fn lifecycle_config_for_reference_proof(&self, bucket: &str) -> Result<Option<BucketLifecycleConfiguration>>;
 }
 
-impl<T> TierReferenceProofStore for T where
-    T: EcstoreObjectIO
-        + BucketOperations<Error = Error>
-        + ListOperations<
-            Error = Error,
-            ListObjectsV2Info = StorageListObjectsV2Info<ObjectInfo>,
-            ListObjectVersionsInfo = StorageListObjectVersionsInfo<ObjectInfo>,
-            ObjectInfoOrErr = StorageObjectInfoOrErr<ObjectInfo, Error>,
-            WalkOptions = TierReferenceProofWalkOptions,
-            WalkCancellation = tokio_util::sync::CancellationToken,
-            WalkResultSender = tokio::sync::mpsc::Sender<StorageObjectInfoOrErr<ObjectInfo, Error>>,
-        >
-{
+#[async_trait::async_trait]
+impl TierReferenceProofStore for ECStore {
+    async fn lifecycle_config_for_reference_proof(&self, bucket: &str) -> Result<Option<BucketLifecycleConfiguration>> {
+        match get_lifecycle_config(bucket).await {
+            Ok((config, _updated_at)) => Ok(Some(config)),
+            Err(Error::ConfigNotFound) => Ok(None),
+            Err(err) => Err(err),
+        }
+    }
 }
 
 async fn ensure_no_authoritative_tier_object_references<S>(
@@ -754,6 +754,7 @@ where
         .await
         .map_err(tier_reference_proof_admin_error)?;
     for bucket in buckets {
+        ensure_no_authoritative_lifecycle_references(api.as_ref(), &bucket.name, targets).await?;
         let mut marker = None;
         let mut version_marker = None;
         loop {
@@ -789,6 +790,66 @@ where
     ensure_no_authoritative_persisted_references(api, targets).await
 }
 
+async fn ensure_no_authoritative_lifecycle_references(
+    api: &impl TierReferenceProofStore,
+    bucket: &str,
+    targets: &[TierMutationIntentTarget],
+) -> std::result::Result<(), AdminError> {
+    match api.lifecycle_config_for_reference_proof(bucket).await {
+        Ok(Some(config)) => {
+            if let Some(reference) = lifecycle_config_target_reference(&config, targets) {
+                return Err(tier_reference_proof_lifecycle_in_use_error(
+                    reference.tier_name,
+                    bucket,
+                    reference.rule_id,
+                ));
+            }
+        }
+        Ok(None) => {}
+        Err(err) => {
+            return Err(tier_reference_proof_admin_error(format!("bucket {bucket} lifecycle config: {err}")));
+        }
+    }
+    Ok(())
+}
+
+struct TierLifecycleReference<'a> {
+    tier_name: &'a str,
+    rule_id: Option<&'a str>,
+}
+
+fn lifecycle_config_target_reference<'a>(
+    config: &'a BucketLifecycleConfiguration,
+    targets: &[TierMutationIntentTarget],
+) -> Option<TierLifecycleReference<'a>> {
+    for rule in &config.rules {
+        let rule_id = rule.id.as_deref();
+        if let Some(transitions) = &rule.transitions {
+            for transition in transitions {
+                let Some(storage_class) = &transition.storage_class else {
+                    continue;
+                };
+                let tier_name = storage_class.as_str();
+                if !tier_name.is_empty() && targets.iter().any(|target| target.tier_name == tier_name) {
+                    return Some(TierLifecycleReference { tier_name, rule_id });
+                }
+            }
+        }
+        if let Some(noncurrent_version_transitions) = &rule.noncurrent_version_transitions {
+            for transition in noncurrent_version_transitions {
+                let Some(storage_class) = &transition.storage_class else {
+                    continue;
+                };
+                let tier_name = storage_class.as_str();
+                if !tier_name.is_empty() && targets.iter().any(|target| target.tier_name == tier_name) {
+                    return Some(TierLifecycleReference { tier_name, rule_id });
+                }
+            }
+        }
+    }
+    None
+}
+
 async fn ensure_no_authoritative_persisted_references<S>(
     api: Arc<S>,
     targets: &[TierMutationIntentTarget],
@@ -920,6 +981,15 @@ fn tier_reference_proof_persisted_in_use_error(tier_name: &str, object: &str) ->
     err
 }
 
+fn tier_reference_proof_lifecycle_in_use_error(tier_name: &str, bucket: &str, rule_id: Option<&str>) -> AdminError {
+    let mut err = ERR_TIER_BACKEND_IN_USE.clone();
+    err.message = match rule_id {
+        Some(rule_id) => format!("Remote tier {tier_name} is still referenced by lifecycle rule {rule_id} in bucket {bucket}"),
+        None => format!("Remote tier {tier_name} is still referenced by a lifecycle rule in bucket {bucket}"),
+    };
+    err
+}
+
 fn tier_reference_proof_admin_error(err: impl std::fmt::Display) -> AdminError {
     let mut admin_err = ERR_TIER_INVALID_CONFIG.clone();
     admin_err.message = format!("Remote tier reference proof failed: {err}");
@@ -5308,6 +5378,10 @@ mod tests {
         endpoints::{Endpoints, PoolEndpoints, SetupType},
     };
     use crate::services::tier::tier_mutation_intent::TIER_MUTATION_INTENT_RECORD_PREFIX;
+    use s3s::dto::{
+        BucketLifecycleConfiguration, ExpirationStatus, LifecycleRule, NoncurrentVersionTransition, Transition,
+        TransitionStorageClass,
+    };
 
     struct SetupTypeGuard {
         previous: SetupType,
@@ -6093,6 +6167,13 @@ mod tests {
         }
     }
 
+    #[async_trait::async_trait]
+    impl TierReferenceProofStore for LockingTierConfigStore {
+        async fn lifecycle_config_for_reference_proof(&self, _bucket: &str) -> Result<Option<BucketLifecycleConfiguration>> {
+            Ok(None)
+        }
+    }
+
     #[tokio::test]
     async fn tier_config_update_path_acquires_meta_namespace_sidecar_lock_before_save() {
         let manager = TierConfigMgr::new();
@@ -11296,6 +11377,7 @@ mod tests {
         lock_manager: Arc<rustfs_lock::GlobalLockManager>,
         lock_requests: Mutex<Vec<(String, String)>>,
         listed_versions: Mutex<Vec<ObjectInfo>>,
+        lifecycle_configs: Mutex<HashMap<String, BucketLifecycleConfiguration>>,
     }
 
     impl Default for CasConfigStore {
@@ -11316,6 +11398,7 @@ mod tests {
                 lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()),
                 lock_requests: Mutex::new(Vec::new()),
                 listed_versions: Mutex::new(Vec::new()),
+                lifecycle_configs: Mutex::new(HashMap::new()),
             }
         }
     }
@@ -11342,6 +11425,13 @@ mod tests {
                 .push(object);
         }
 
+        fn add_lifecycle_config(&self, bucket: &str, config: BucketLifecycleConfiguration) {
+            self.lifecycle_configs
+                .lock()
+                .expect("tier reference fixture should not poison")
+                .insert(bucket.to_string(), config);
+        }
+
         async fn insert_config_object(&self, object: String, data: Vec<u8>) {
             self.objects
                 .lock()
@@ -11917,6 +12007,18 @@ mod tests {
         }
     }
 
+    #[async_trait::async_trait]
+    impl TierReferenceProofStore for CasConfigStore {
+        async fn lifecycle_config_for_reference_proof(&self, bucket: &str) -> Result<Option<BucketLifecycleConfiguration>> {
+            Ok(self
+                .lifecycle_configs
+                .lock()
+                .expect("tier reference fixture should not poison")
+                .get(bucket)
+                .cloned())
+        }
+    }
+
     #[tokio::test]
     async fn remove_and_clear_full_update_paths_preserve_force() {
         let remove_store = Arc::new(CasConfigStore::default());
@@ -12090,6 +12192,96 @@ mod tests {
         assert!(err.message.contains("photos/2026/old-destination.jpg"), "{}", err.message);
     }
 
+    #[tokio::test]
+    async fn zero_reference_proof_blocks_lifecycle_transition_references() {
+        let current = build_rustfs_tier("COLD-A");
+        let current_identity = tier_backend_identity(&current).expect("current identity should encode");
+        let target = TierMutationIntentTarget {
+            tier_name: "COLD-A".to_string(),
+            old_backend_identity: Some(current_identity),
+            new_backend_identity: None,
+        };
+        let config = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("move-current".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: None,
+                prefix: None,
+                transitions: Some(vec![Transition {
+                    days: Some(1),
+                    date: None,
+                    storage_class: Some(TransitionStorageClass::from_static("COLD-A")),
+                }]),
+            }],
+        };
+        let store = Arc::new(CasConfigStore::default());
+        store.add_listed_version(ObjectInfo {
+            bucket: "photos".to_string(),
+            name: "safe.txt".to_string(),
+            ..Default::default()
+        });
+        store.add_lifecycle_config("photos", config);
+
+        let err = ensure_no_authoritative_tier_object_references(store, &[target])
+            .await
+            .expect_err("lifecycle rule should block deletion");
+
+        assert_eq!(err.code, ERR_TIER_BACKEND_IN_USE.code);
+        assert!(err.message.contains("move-current"), "{}", err.message);
+        assert!(err.message.contains("photos"), "{}", err.message);
+    }
+
+    #[tokio::test]
+    async fn zero_reference_proof_blocks_lifecycle_noncurrent_transition_references() {
+        let current = build_rustfs_tier("COLD-A");
+        let current_identity = tier_backend_identity(&current).expect("current identity should encode");
+        let target = TierMutationIntentTarget {
+            tier_name: "COLD-A".to_string(),
+            old_backend_identity: Some(current_identity),
+            new_backend_identity: None,
+        };
+        let config = BucketLifecycleConfiguration {
+            expiry_updated_at: None,
+            rules: vec![LifecycleRule {
+                status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
+                expiration: None,
+                abort_incomplete_multipart_upload: None,
+                del_marker_expiration: None,
+                filter: None,
+                id: Some("move-noncurrent".to_string()),
+                noncurrent_version_expiration: None,
+                noncurrent_version_transitions: Some(vec![NoncurrentVersionTransition {
+                    noncurrent_days: Some(1),
+                    newer_noncurrent_versions: None,
+                    storage_class: Some(TransitionStorageClass::from_static("COLD-A")),
+                }]),
+                prefix: None,
+                transitions: None,
+            }],
+        };
+        let store = Arc::new(CasConfigStore::default());
+        store.add_listed_version(ObjectInfo {
+            bucket: "photos".to_string(),
+            name: "safe.txt".to_string(),
+            ..Default::default()
+        });
+        store.add_lifecycle_config("photos", config);
+
+        let err = ensure_no_authoritative_tier_object_references(store, &[target])
+            .await
+            .expect_err("lifecycle rule should block deletion");
+
+        assert_eq!(err.code, ERR_TIER_BACKEND_IN_USE.code);
+        assert!(err.message.contains("move-noncurrent"), "{}", err.message);
+        assert!(err.message.contains("photos"), "{}", err.message);
+    }
+
     #[tokio::test]
     async fn zero_reference_proof_blocks_persisted_journal_transaction_and_free_version_references() {
         let current = build_rustfs_tier("COLD-A");

crates/ecstore/src/set_disk/ops/object.rs

@@ -2592,29 +2592,19 @@ impl SetDisks {
 
         let mut object_lock_guard = None;
         let mut bucket_lifecycle_guard = None;
-        let deferred_data_movement_precondition = opts.data_movement && opts.http_preconditions.is_some();
 
-        if opts.http_preconditions.is_some() && !deferred_data_movement_precondition {
-            if !opts.no_lock {
-                if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id
-                    && opts.bucket_lifecycle_lock_fence.is_none()
-                {
-                    bucket_lifecycle_guard = Some(
-                        metadata_sys::object_store_in(&self.ctx)
-                            .await?
-                            .acquire_bucket_incarnation_fence(bucket, expected_incarnation_id)
-                            .await?,
-                    );
-                }
-                object_lock_guard = Some(
-                    self.acquire_write_lock_diag("put_object_precondition", bucket, object)
-                        .await?,
-                );
-            }
-
-            if let Some(err) = self.check_write_precondition(bucket, object, opts).await {
-                return Err(err);
-            }
+        // This pre-body check is advisory fast-fail only: the authoritative
+        // precondition evaluation happens under the commit namespace lock
+        // below, so the namespace write lock must NOT be taken here — holding
+        // it across client-paced body ingestion starves concurrent reads of
+        // the same object into lock-timeout 503s (rustfs/backlog#2074).
+        // Data movement skips the advisory read: its staleness predicate is
+        // only meaningful at commit time.
+        if opts.http_preconditions.is_some()
+            && !opts.data_movement
+            && let Some(err) = self.check_write_precondition(bucket, object, opts).await
+        {
+            return Err(err);
         }
 
         let expected_restore_operation_id = restore_commit_operation_id_from_metadata(&opts.user_defined)?;
@@ -3084,7 +3074,9 @@ impl SetDisks {
             #[cfg(any(test, feature = "test-util"))]
             pause_put_object_commit(bucket, object, PutObjectCommitPause::AfterNamespace).await;
 
-            if deferred_data_movement_precondition && let Some(err) = self.check_write_precondition(bucket, object, opts).await {
+            if opts.http_preconditions.is_some()
+                && let Some(err) = self.check_write_precondition(bucket, object, opts).await
+            {
                 return Err(err);
             }
 
@@ -15843,6 +15835,161 @@ mod put_object_tmp_cleanup_tests {
         assert_eq!(body, b"new client body");
     }
 
+    #[tokio::test]
+    async fn conditional_put_does_not_block_reads_during_body_ingestion() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "conditional-put-nonblocking-read";
+        let object = "object";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let mut initial_reader = PutObjReader::from_vec(b"old body".to_vec());
+        let initial = set_disks
+            .put_object(bucket, object, &mut initial_reader, &ObjectOptions::default())
+            .await
+            .expect("initial object should be written");
+        let initial_etag = initial.etag.clone().expect("initial object should have an etag");
+
+        let body = vec![b'c'; 64 * 1024];
+        let split = body.len() / 2;
+        let (mut source, stream) = tokio::io::duplex(64);
+        let hash_reader = HashReader::from_stream(
+            stream,
+            i64::try_from(body.len()).expect("body length should fit i64"),
+            i64::try_from(body.len()).expect("body length should fit i64"),
+            None,
+            None,
+            false,
+        )
+        .expect("conditional hash reader should be created");
+        let writer_store = Arc::clone(&set_disks);
+        let etag_for_put = initial_etag.clone();
+        let put = tokio::spawn(async move {
+            let mut reader = PutObjReader::new(hash_reader);
+            writer_store
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        http_preconditions: Some(HTTPPreconditions {
+                            if_match: Some(etag_for_put),
+                            ..Default::default()
+                        }),
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+
+        source
+            .write_all(&body[..split])
+            .await
+            .expect("conditional PUT should consume the first half of the body");
+        let info = tokio::time::timeout(
+            Duration::from_secs(5),
+            set_disks.get_object_info(bucket, object, &ObjectOptions::default()),
+        )
+        .await
+        .expect("reads must not wait for the conditional PUT body")
+        .expect("the old version must stay readable during body ingestion");
+        assert_eq!(info.etag.as_deref(), Some(initial_etag.as_str()));
+
+        source
+            .write_all(&body[split..])
+            .await
+            .expect("conditional PUT should consume the remaining body");
+        drop(source);
+        put.await
+            .expect("conditional PUT task should join")
+            .expect("conditional PUT should commit after the body completes");
+    }
+
+    #[tokio::test]
+    async fn conditional_put_precondition_is_rechecked_at_commit() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "conditional-put-commit-recheck";
+        let object = "object";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let mut initial_reader = PutObjReader::from_vec(b"old body".to_vec());
+        let initial = set_disks
+            .put_object(bucket, object, &mut initial_reader, &ObjectOptions::default())
+            .await
+            .expect("initial object should be written");
+        let initial_etag = initial.etag.clone().expect("initial object should have an etag");
+
+        let body = vec![b'c'; 64 * 1024];
+        let split = body.len() / 2;
+        let (mut source, stream) = tokio::io::duplex(64);
+        let hash_reader = HashReader::from_stream(
+            stream,
+            i64::try_from(body.len()).expect("body length should fit i64"),
+            i64::try_from(body.len()).expect("body length should fit i64"),
+            None,
+            None,
+            false,
+        )
+        .expect("conditional hash reader should be created");
+        let writer_store = Arc::clone(&set_disks);
+        let put = tokio::spawn(async move {
+            let mut reader = PutObjReader::new(hash_reader);
+            writer_store
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        http_preconditions: Some(HTTPPreconditions {
+                            if_match: Some(initial_etag),
+                            ..Default::default()
+                        }),
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+
+        source
+            .write_all(&body[..split])
+            .await
+            .expect("conditional PUT should consume the first half of the body");
+        let mut interloper_reader = PutObjReader::from_vec(b"interloper body".to_vec());
+        tokio::time::timeout(
+            Duration::from_secs(5),
+            set_disks.put_object(bucket, object, &mut interloper_reader, &ObjectOptions::default()),
+        )
+        .await
+        .expect("the interloper write must not wait for the conditional PUT body")
+        .expect("the interloper write should commit while the conditional PUT streams");
+
+        source
+            .write_all(&body[split..])
+            .await
+            .expect("conditional PUT should consume the remaining body");
+        drop(source);
+        let err = put
+            .await
+            .expect("conditional PUT task should join")
+            .expect_err("the conditional PUT must recheck its precondition under the commit lock");
+        assert_eq!(err, StorageError::PreconditionFailed);
+
+        let mut reader = set_disks
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("the interloper object should remain readable");
+        let mut body = Vec::new();
+        reader
+            .stream
+            .read_to_end(&mut body)
+            .await
+            .expect("the interloper object should drain");
+        assert_eq!(body, b"interloper body");
+    }
+
     #[tokio::test]
     async fn metadata_copy_no_lock_aborts_after_outer_namespace_lock_loss() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;

crates/s3-client/src/api_s3_datatypes.rs

@@ -396,3 +396,30 @@ impl DeleteMultiObjects {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::ListBucketV2Result;
+
+    #[test]
+    fn list_bucket_v2_common_prefix_accepts_s3_pascal_case_xml() {
+        let xml = r#"
+            <ListBucketResult>
+                <Name>tier-bucket</Name>
+                <Prefix></Prefix>
+                <KeyCount>1</KeyCount>
+                <MaxKeys>1000</MaxKeys>
+                <Delimiter>/</Delimiter>
+                <IsTruncated>false</IsTruncated>
+                <CommonPrefixes>
+                    <Prefix>tenant-a/</Prefix>
+                </CommonPrefixes>
+            </ListBucketResult>
+        "#;
+
+        let result = quick_xml::de::from_str::<ListBucketV2Result>(xml).expect("S3 list response should decode");
+
+        assert_eq!(result.common_prefixes.len(), 1);
+        assert_eq!(result.common_prefixes[0].prefix, "tenant-a/");
+    }
+}

rustfs/src/app/admin_usecase.rs

@@ -237,7 +237,7 @@ impl DefaultAdminUsecase {
     /// namespace still contains, carries no usable information and is dropped
     /// exactly as before.
     fn narrow_data_usage_snapshot_to_measured_buckets(info: &mut DataUsageInfo, buckets: impl IntoIterator<Item = String>) {
-        if !info.is_complete_bucket_usage_snapshot() {
+        if !info.is_complete_bucket_usage_snapshot() && !info.is_valid_partial_snapshot() {
             *info = DataUsageInfo::default();
             return;
         }
@@ -759,6 +759,28 @@ mod tests {
         DefaultAdminUsecase::narrow_data_usage_snapshot_to_measured_buckets(&mut info, ["bucket-a".to_string()]);
         assert_eq!(info, DataUsageInfo::default());
 
+        // A structurally valid partial admin view can still carry useful
+        // conservative totals.
+        let mut info = measured("bucket-a");
+        info.usage_snapshot_complete = false;
+        info.usage_snapshot_partial = true;
+        info.usage_snapshot_converged = Some(false);
+        info.scanner_cycle = Some(11);
+        info.scanner_epoch = Some(4);
+        info.usage_snapshot_set_states = vec![rustfs_data_usage::DataUsageSnapshotSetState {
+            pool_index: 0,
+            set_index: 0,
+            scanner_cycle: Some(11),
+            scanner_epoch: Some(4),
+            scan_plan_digest: Some([1; 32]),
+            complete: true,
+            tombstone: false,
+        }];
+        DefaultAdminUsecase::narrow_data_usage_snapshot_to_measured_buckets(&mut info, ["bucket-a".to_string()]);
+        assert_eq!(info.usage_snapshot_converged, Some(false));
+        assert_eq!(info.buckets_count, 1);
+        assert_eq!(info.objects_total_count, 7);
+
         // An empty namespace with an empty snapshot stays a confirmed zero.
         let mut info = DataUsageInfo {
             last_update: Some(std::time::SystemTime::UNIX_EPOCH),

rustfs/src/app/object/get.rs

@@ -3820,7 +3820,15 @@ impl DefaultObjectUsecase {
         Box::pin(self.execute_get_object_inner(req))
     }
 
+    fn complete_get_object_error<T>(helper: OperationHelper, err: S3Error) -> S3Result<S3Response<T>> {
+        let result = Err(err);
+        let _ = helper.complete(&result);
+        result
+    }
+
     async fn execute_get_object_inner(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
+        let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).suppress_event();
+
         if let Some(context) = &self.context {
             let _ = context.object_store();
         }
@@ -3840,7 +3848,10 @@ impl DefaultObjectUsecase {
                 context.start_time.elapsed().as_secs_f64(),
             );
         }
-        let bootstrap = self.init_get_object_bootstrap(&req.input.bucket, &req.input.key, &request_id)?;
+        let bootstrap = match self.init_get_object_bootstrap(&req.input.bucket, &req.input.key, &request_id) {
+            Ok(bootstrap) => bootstrap,
+            Err(err) => return Self::complete_get_object_error(helper, err),
+        };
         record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_REQUEST_SHAPE, request_shape_start);
         let timeout_config = bootstrap.timeout_config;
         let wrapper = bootstrap.wrapper;
@@ -3848,7 +3859,6 @@ impl DefaultObjectUsecase {
         let concurrent_requests = bootstrap.concurrent_requests;
         let mut lifecycle = GetObjectBodyLifecycle::tracked(bootstrap.request_guard);
 
-        let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).suppress_event();
         // mc get 3
 
         // Cheap request-shape validations run first so invalid requests keep
@@ -3858,7 +3868,7 @@ impl DefaultObjectUsecase {
             Ok(validated) => validated,
             Err(err) => {
                 lifecycle.finish_err();
-                return Err(err);
+                return Self::complete_get_object_error(helper, err);
             }
         };
         record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_REQUEST_VALIDATION, request_validation_start);
@@ -3875,7 +3885,10 @@ impl DefaultObjectUsecase {
         let store_lookup_start = stage_metrics_enabled.then(std::time::Instant::now);
         let Some(store) = self.object_store() else {
             lifecycle.finish_err();
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+            return Self::complete_get_object_error(
+                helper,
+                S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()),
+            );
         };
         if let Some(store_lookup_start) = store_lookup_start {
             rustfs_io_metrics::record_get_object_stage_duration(
@@ -3887,7 +3900,7 @@ impl DefaultObjectUsecase {
         let bucket_validation_start = stage_metrics_enabled.then(std::time::Instant::now);
         if let Err(err) = validate_bucket_exists(&store, &req.input.bucket).await {
             lifecycle.finish_err();
-            return Err(err);
+            return Self::complete_get_object_error(helper, err);
         }
         record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BUCKET_VALIDATION, bucket_validation_start);
 
@@ -3896,7 +3909,7 @@ impl DefaultObjectUsecase {
             Ok(request_context) => request_context,
             Err(err) => {
                 lifecycle.finish_err();
-                return Err(err);
+                return Self::complete_get_object_error(helper, err);
             }
         };
         if let Some(request_context_start) = request_context_start {
@@ -3951,7 +3964,7 @@ impl DefaultObjectUsecase {
                     return result;
                 }
                 lifecycle.finish_err();
-                return Err(err);
+                return Self::complete_get_object_error(helper.version_id(version_id_for_event), err);
             }
         };
         let GetObjectPreparedRead { io_planning, read_setup } = prepared_read;
@@ -4040,7 +4053,7 @@ impl DefaultObjectUsecase {
             .await;
         let output_context = match output_context {
             Ok(output_context) => output_context,
-            Err(err) => return Err(err),
+            Err(err) => return Self::complete_get_object_error(helper.version_id(version_id_for_event), err),
         };
         if let Some(output_build_start) = output_build_start {
             rustfs_io_metrics::record_get_object_stage_duration(

rustfs/src/storage/helper.rs

@@ -23,7 +23,7 @@ use http::StatusCode;
 use metrics::counter;
 use rustfs_audit::{
     ObjectVersion,
-    entity::{ApiDetails, ApiDetailsBuilder, AuditEntryBuilder},
+    entity::{ApiDetailsBuilder, AuditEntryBuilder},
     global::AuditLogger,
 };
 use rustfs_io_metrics::record_s3_op;
@@ -136,7 +136,7 @@ impl OperationHelper {
 
         record_s3_op(op);
 
-        // Fast path: when both chains are disabled, avoid all request parsing/builder work.
+        // Fast path: when both chains are disabled, avoid audit/notify builder work.
         if !audit_enabled && !notify_enabled {
             return Self::Disabled;
         }
@@ -180,7 +180,7 @@ impl OperationHelper {
 
         let audit_builder = if audit_enabled {
             Some(
-                AuditEntryBuilder::new("1.0", event, trigger, ApiDetails::default())
+                AuditEntryBuilder::new("1.0", event, trigger, api_builder.clone().build())
                     .remote_host(remote_host)
                     .user_agent(get_request_user_agent(&req.headers))
                     .req_host(get_request_host(&req.headers))
@@ -453,8 +453,8 @@ mod tests {
     use rustfs_s3_ops::S3Operation;
     use rustfs_s3_types::EventName;
     use rustfs_utils::http::headers::{AMZ_REQUEST_ID, REQUEST_ID_HEADER};
-    use s3s::dto::{DeleteObjectTaggingInput, DeleteObjectTaggingOutput};
-    use s3s::{S3Request, S3Response};
+    use s3s::dto::{DeleteObjectTaggingInput, DeleteObjectTaggingOutput, GetObjectInput, GetObjectOutput};
+    use s3s::{S3Error, S3ErrorCode, S3Request, S3Response};
     use std::sync::{Arc, Mutex};
     use temp_env::{async_with_vars, with_vars};
 
@@ -615,6 +615,71 @@ mod tests {
         );
     }
 
+    #[test]
+    fn operation_helper_initializes_audit_api_details_before_completion() {
+        with_vars(
+            [
+                (rustfs_config::ENV_NOTIFY_ENABLE, Some("false")),
+                (rustfs_config::ENV_AUDIT_ENABLE, Some("true")),
+            ],
+            || {
+                refresh_notify_module_enabled();
+                refresh_audit_module_enabled();
+
+                let input = GetObjectInput::builder()
+                    .bucket("audit-bucket".to_string())
+                    .key("missing/object.txt".to_string())
+                    .build()
+                    .expect("get object input should build");
+                let req = build_request(input, Method::GET, Uri::from_static("/audit-bucket/missing/object.txt"));
+                let mut helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject);
+
+                let OperationHelper::Enabled(state) = &mut helper else {
+                    panic!("helper should be enabled when the audit switch is on");
+                };
+                let audit_entry = state.audit_builder.take().expect("audit builder should exist").build();
+
+                assert_eq!(audit_entry.api.name.as_deref(), Some("s3:GetObject"));
+                assert_eq!(audit_entry.api.bucket.as_deref(), Some("audit-bucket"));
+                assert_eq!(audit_entry.api.object.as_deref(), Some("missing/object.txt"));
+                assert!(audit_entry.api.status_code.is_none());
+            },
+        );
+    }
+
+    #[test]
+    fn operation_helper_complete_records_failed_status_code() {
+        with_vars(
+            [
+                (rustfs_config::ENV_NOTIFY_ENABLE, Some("false")),
+                (rustfs_config::ENV_AUDIT_ENABLE, Some("true")),
+            ],
+            || {
+                refresh_notify_module_enabled();
+                refresh_audit_module_enabled();
+
+                let input = GetObjectInput::builder()
+                    .bucket("audit-bucket".to_string())
+                    .key("missing/object.txt".to_string())
+                    .build()
+                    .expect("get object input should build");
+                let req = build_request(input, Method::GET, Uri::from_static("/audit-bucket/missing/object.txt"));
+                let result: Result<S3Response<GetObjectOutput>, S3Error> = Err(S3Error::new(S3ErrorCode::NoSuchKey));
+                let mut helper =
+                    OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).complete(&result);
+
+                let OperationHelper::Enabled(state) = &mut helper else {
+                    panic!("helper should be enabled when the audit switch is on");
+                };
+                let audit_entry = state.audit_builder.take().expect("audit builder should exist").build();
+
+                assert_eq!(audit_entry.api.name.as_deref(), Some("s3:GetObject"));
+                assert_eq!(audit_entry.api.status.as_deref(), Some("failure"));
+                assert_eq!(audit_entry.api.status_code, Some(404));
+            },
+        );
+    }
+
     #[test]
     fn operation_helper_prioritizes_request_context_for_request_id() {
         with_vars(