refactor(object): extract internal put entry and ODM provenance keys (#7071)

* feat(utils): add on-demand migration provenance metadata suffixes

* refactor(object): extract internal put entry from the S3 PutObject path

crates/e2e_test/src/on_demand_migration/harness_self_test.rs

@@ -452,9 +452,9 @@ async fn fake_source_fault_actions_truncate_stall_and_status() -> TestResult {
     let stalled = client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
     assert!(started.elapsed() >= Duration::from_millis(350), "stall must delay the first byte");
     assert_eq!(stalled.content_length(), Some(4096));
-    let unstalled_started = Instant::now();
+    let post_stall_started = Instant::now();
     client.head_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;
-    assert!(unstalled_started.elapsed() < Duration::from_millis(350), "stall is consumed once");
+    assert!(post_stall_started.elapsed() < Duration::from_millis(350), "stall is consumed once");
 
     // The object is intact once the script is drained.
     let intact = client.get_object().bucket(SOURCE_BUCKET).key("faulty").send().await?;

crates/ecstore/src/api/mod.rs

@@ -128,7 +128,6 @@ pub mod bucket {
     }
 
     pub mod metadata {
-        pub use crate::bucket::metadata::BUCKET_DURABILITY_CONFIG;
         pub use crate::bucket::metadata::{
             BUCKET_ACCELERATE_CONFIG, BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_LOGGING_CONFIG,
             BUCKET_NOTIFICATION_CONFIG, BUCKET_POLICY_CONFIG, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, BUCKET_QUOTA_CONFIG_FILE,
@@ -137,6 +136,7 @@ pub mod bucket {
             BUCKET_TARGETS_FILE, BUCKET_VERSIONING_CONFIG, BUCKET_WEBSITE_CONFIG, BucketMetadata, OBJECT_LOCK_CONFIG,
             load_bucket_metadata, table_catalog_path_hash,
         };
+        pub use crate::bucket::metadata::{BUCKET_DURABILITY_CONFIG, BUCKET_ON_DEMAND_MIGRATION_CONFIG};
     }
 
     pub mod durability {
@@ -145,6 +145,21 @@ pub mod bucket {
         };
     }
 
+    pub mod on_demand_migration {
+        pub use crate::bucket::on_demand_migration::{
+            ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
+            OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy,
+            SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
+        };
+        pub mod source_client {
+            pub use crate::bucket::on_demand_migration::source_client::{
+                SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceObject, SourcePage, SourceProbe,
+                SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value,
+                resolve_path_style,
+            };
+        }
+    }
+
     pub mod metadata_sys {
         #[cfg(feature = "test-util")]
         pub use crate::bucket::metadata_sys::ConfigWriteLockProbe;
@@ -154,11 +169,11 @@ pub mod bucket {
             delete_if_incarnation, delete_under_transaction_lock, get, get_accelerate_config, get_bucket_policy,
             get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config,
             get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config,
-            get_object_lock_config, get_object_lock_config_state, get_public_access_block_config, get_quota_config,
+            get_object_lock_config, get_object_lock_config_state, get_on_demand_migration_config, get_public_access_block_config,
-            get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config,
+            get_quota_config, get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config,
-            get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata,
+            get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata,
-            set_bucket_metadata, update, update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation,
+            remove_bucket_metadata, set_bucket_metadata, update, update_bucket_targets_under_transaction_lock,
-            update_quota_if_incarnation, update_under_transaction_lock,
+            update_config_with, update_if_incarnation, update_quota_if_incarnation, update_under_transaction_lock,
         };
     }
 
@@ -166,16 +181,6 @@ pub mod bucket {
         pub use crate::bucket::migration::{LegacyBlobDecryptFn, try_migrate_bucket_metadata, try_migrate_iam_config};
     }
 
-    pub mod on_demand_migration {
-        pub mod source_client {
-            pub use crate::bucket::on_demand_migration::source_client::{
-                SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceObject, SourcePage, SourceProbe,
-                SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value,
-                resolve_path_style,
-            };
-        }
-    }
-
     pub mod object_lock {
         pub use crate::bucket::object_lock::{ObjectLockApi, ObjectLockStatusExt};
 

crates/ecstore/src/bucket/metadata.rs

@@ -270,6 +270,7 @@ pub const BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG: &str = "public-access-block.xml";
 pub const BUCKET_ACL_CONFIG: &str = "bucket-acl.json";
 pub const BUCKET_TABLE_CONFIG: &str = "table-bucket.json";
 pub const BUCKET_DURABILITY_CONFIG: &str = "durability.json";
+pub const BUCKET_ON_DEMAND_MIGRATION_CONFIG: &str = "on-demand-migration.json";
 pub const BUCKET_TABLE_RESERVED_PREFIX: &str = ".rustfs-table";
 pub const BUCKET_TABLE_CATALOG_META_PREFIX: &str = "s3tables/catalog";
 pub const BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX: &str = "table-buckets";
@@ -321,6 +322,7 @@ pub struct BucketMetadata {
     pub bucket_acl_config_json: Vec<u8>,
     pub table_bucket_config_json: Vec<u8>,
     pub durability_config_json: Vec<u8>,
+    pub on_demand_migration_config_json: Vec<u8>,
 
     pub policy_config_updated_at: OffsetDateTime,
     pub object_lock_config_updated_at: OffsetDateTime,
@@ -342,6 +344,7 @@ pub struct BucketMetadata {
     pub bucket_acl_config_updated_at: OffsetDateTime,
     pub table_bucket_config_updated_at: OffsetDateTime,
     pub durability_config_updated_at: OffsetDateTime,
+    pub on_demand_migration_config_updated_at: OffsetDateTime,
 
     pub new_field_updated_at: OffsetDateTime,
 
@@ -393,6 +396,7 @@ impl Default for BucketMetadata {
             bucket_acl_config_json: Default::default(),
             table_bucket_config_json: Default::default(),
             durability_config_json: Default::default(),
+            on_demand_migration_config_json: Default::default(),
             policy_config_updated_at: OffsetDateTime::UNIX_EPOCH,
             object_lock_config_updated_at: OffsetDateTime::UNIX_EPOCH,
             encryption_config_updated_at: OffsetDateTime::UNIX_EPOCH,
@@ -413,6 +417,7 @@ impl Default for BucketMetadata {
             bucket_acl_config_updated_at: OffsetDateTime::UNIX_EPOCH,
             table_bucket_config_updated_at: OffsetDateTime::UNIX_EPOCH,
             durability_config_updated_at: OffsetDateTime::UNIX_EPOCH,
+            on_demand_migration_config_updated_at: OffsetDateTime::UNIX_EPOCH,
             new_field_updated_at: OffsetDateTime::UNIX_EPOCH,
             policy_config: Default::default(),
             notification_config: Default::default(),
@@ -477,6 +482,23 @@ impl BucketMetadata {
     /// Absent/empty/unparsable payloads all mean "no override" (the bucket
     /// follows the global durability mode); a parse failure is logged so a
     /// corrupted entry cannot silently change fsync behavior.
+    /// Parsed on-demand migration config, if one is stored.
+    ///
+    /// `Ok(None)` means no config (absent or cleared). A stored payload that
+    /// does not parse is an error, never a default: the runtime must not
+    /// pull from a source it cannot describe.
+    pub fn on_demand_migration_config(
+        &self,
+    ) -> std::result::Result<
+        Option<super::on_demand_migration::OnDemandMigrationConfig>,
+        super::on_demand_migration::OnDemandMigrationConfigError,
+    > {
+        if self.on_demand_migration_config_json.is_empty() {
+            return Ok(None);
+        }
+        super::on_demand_migration::OnDemandMigrationConfig::from_json(&self.on_demand_migration_config_json).map(Some)
+    }
+
     pub fn durability_config(&self) -> Option<super::durability::BucketDurabilityConfig> {
         if self.durability_config_json.is_empty() {
             return None;
@@ -555,6 +577,9 @@ impl BucketMetadata {
                 "BucketAclConfigJSON" | "BucketAclConfigJson" => self.bucket_acl_config_json = read_msgp_bin(rd)?,
                 "TableBucketConfigJSON" | "TableBucketConfigJson" => self.table_bucket_config_json = read_msgp_bin(rd)?,
                 "DurabilityConfigJSON" | "DurabilityConfigJson" => self.durability_config_json = read_msgp_bin(rd)?,
+                "OnDemandMigrationConfigJSON" | "OnDemandMigrationConfigJson" => {
+                    self.on_demand_migration_config_json = read_msgp_bin(rd)?
+                }
                 "CorsConfigUpdatedAt" => self.cors_config_updated_at = read_msgp_time_value(rd)?,
                 "LoggingConfigUpdatedAt" => self.logging_config_updated_at = read_msgp_time_value(rd)?,
                 "WebsiteConfigUpdatedAt" => self.website_config_updated_at = read_msgp_time_value(rd)?,
@@ -564,6 +589,7 @@ impl BucketMetadata {
                 "BucketAclConfigUpdatedAt" => self.bucket_acl_config_updated_at = read_msgp_time_value(rd)?,
                 "TableBucketConfigUpdatedAt" => self.table_bucket_config_updated_at = read_msgp_time_value(rd)?,
                 "DurabilityConfigUpdatedAt" => self.durability_config_updated_at = read_msgp_time_value(rd)?,
+                "OnDemandMigrationConfigUpdatedAt" => self.on_demand_migration_config_updated_at = read_msgp_time_value(rd)?,
                 other => {
                     tracing::debug!(field = %other, "BucketMetadata decode_from: skipping unknown field");
                     skip_msgp_value(rd)?;
@@ -576,8 +602,8 @@ impl BucketMetadata {
 
     /// Encode to msgp bytes. Field order follows MinIO BucketMetadata for compatibility.
     pub fn encode_to<W: Write>(&self, wr: &mut W) -> Result<()> {
-        // Map size: MinIO fields (25) + RustFS extensions (19)
+        // Map size: MinIO fields (25) + RustFS extensions (21)
-        let map_len: u32 = 44;
+        let map_len: u32 = 46;
         rmp::encode::write_map_len(wr, map_len)?;
 
         // MinIO field order (same as Go struct)
@@ -637,6 +663,7 @@ impl BucketMetadata {
         write_bin_field(wr, "BucketAclConfigJSON", &self.bucket_acl_config_json)?;
         write_bin_field(wr, "TableBucketConfigJSON", &self.table_bucket_config_json)?;
         write_bin_field(wr, "DurabilityConfigJSON", &self.durability_config_json)?;
+        write_bin_field(wr, "OnDemandMigrationConfigJSON", &self.on_demand_migration_config_json)?;
         rmp::encode::write_str(wr, "CorsConfigUpdatedAt")?;
         write_msgp_time(wr, self.cors_config_updated_at)?;
         rmp::encode::write_str(wr, "LoggingConfigUpdatedAt")?;
@@ -655,6 +682,8 @@ impl BucketMetadata {
         write_msgp_time(wr, self.table_bucket_config_updated_at)?;
         rmp::encode::write_str(wr, "DurabilityConfigUpdatedAt")?;
         write_msgp_time(wr, self.durability_config_updated_at)?;
+        rmp::encode::write_str(wr, "OnDemandMigrationConfigUpdatedAt")?;
+        write_msgp_time(wr, self.on_demand_migration_config_updated_at)?;
 
         Ok(())
     }
@@ -756,6 +785,9 @@ impl BucketMetadata {
         if self.durability_config_updated_at == OffsetDateTime::UNIX_EPOCH {
             self.durability_config_updated_at = self.created
         }
+        if self.on_demand_migration_config_updated_at == OffsetDateTime::UNIX_EPOCH {
+            self.on_demand_migration_config_updated_at = self.created
+        }
     }
 
     pub fn update_config(&mut self, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
@@ -871,6 +903,17 @@ impl BucketMetadata {
                 self.durability_config_json = data;
                 self.durability_config_updated_at = updated;
             }
+            BUCKET_ON_DEMAND_MIGRATION_CONFIG => {
+                // Structural check only (shape, unknown fields); the
+                // deployment-relative rules run in the admin handler with a
+                // `ValidationContext`. A blob this build cannot read must not
+                // be persisted for every later reader to trip over.
+                if !data.is_empty() {
+                    super::on_demand_migration::OnDemandMigrationConfig::from_json(&data).map_err(Error::other)?;
+                }
+                self.on_demand_migration_config_json = data;
+                self.on_demand_migration_config_updated_at = updated;
+            }
             _ => return Err(Error::other(format!("config file not found : {config_file}"))),
         }
 
@@ -1779,6 +1822,117 @@ mod test {
         assert!(!bm.table_bucket_enabled());
     }
 
+    const ODM_JSON: &[u8] = br#"{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://legacy.example.com:9000","region":"auto","bucket":"legacy-bucket","credentials":{"access_key":"AK","secret_key":"SK"}}}"#;
+
+    /// rustfs/backlog#2148: the on-demand migration config is a RustFS
+    /// extension entry that round-trips through `update_config` and the
+    /// msgpack codec, clears on delete, and never parses corruption into a
+    /// default.
+    #[test]
+    fn on_demand_migration_config_round_trips_and_tracks_updates() {
+        use crate::bucket::on_demand_migration::{OnDemandMigrationConfig, OnDemandMigrationConfigError};
+
+        let mut bm = BucketMetadata::new("odm-bucket");
+        assert_eq!(bm.on_demand_migration_config(), Ok(None), "fresh metadata carries no config");
+
+        let expected = OnDemandMigrationConfig::from_json(ODM_JSON).unwrap();
+        bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
+            .expect("valid config is accepted");
+        assert_ne!(bm.on_demand_migration_config_updated_at, OffsetDateTime::UNIX_EPOCH);
+        assert_eq!(bm.on_demand_migration_config(), Ok(Some(expected.clone())));
+
+        let back = BucketMetadata::unmarshal(&bm.marshal_msg().unwrap()).unwrap();
+        assert_eq!(back.on_demand_migration_config_json, bm.on_demand_migration_config_json);
+        assert_eq!(
+            back.on_demand_migration_config_updated_at.unix_timestamp(),
+            bm.on_demand_migration_config_updated_at.unix_timestamp()
+        );
+        assert_eq!(back.on_demand_migration_config(), Ok(Some(expected)));
+
+        // A blob this build cannot read is rejected at the write boundary
+        // rather than persisted for every reader to trip over.
+        let before = bm.on_demand_migration_config_json.clone();
+        assert!(
+            bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, br#"{"source":{"provider":"s3"},"bogus":1}"#.to_vec())
+                .is_err()
+        );
+        assert_eq!(bm.on_demand_migration_config_json, before, "a rejected update leaves the blob untouched");
+
+        // Delete clears the entry.
+        let stamped = bm.on_demand_migration_config_updated_at;
+        bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, Vec::new()).unwrap();
+        assert!(bm.on_demand_migration_config_json.is_empty());
+        assert_eq!(bm.on_demand_migration_config(), Ok(None));
+        assert!(bm.on_demand_migration_config_updated_at >= stamped);
+
+        // Corruption that bypassed `update_config` (disk, another writer)
+        // is a typed error, never a default.
+        bm.on_demand_migration_config_json = b"not-json".to_vec();
+        assert!(matches!(bm.on_demand_migration_config(), Err(OnDemandMigrationConfigError::Malformed(_))));
+    }
+
+    /// rustfs/backlog#2148: a `.metadata.bin` written before the on-demand
+    /// migration keys existed decodes with an empty blob and an epoch
+    /// timestamp that `default_timestamps` back-fills from `created`.
+    #[test]
+    fn on_demand_migration_config_absent_in_legacy_blob_defaults_to_created() {
+        let blob = decode_hex(include_str!("../../tests/fixtures/minio/bucket_metadata.blob.hex"));
+        let mut bm = BucketMetadata::unmarshal(&blob[4..]).expect("unmarshal MinIO bucket metadata");
+        assert!(bm.on_demand_migration_config_json.is_empty());
+        assert_eq!(bm.on_demand_migration_config_updated_at, OffsetDateTime::UNIX_EPOCH);
+        assert_eq!(bm.on_demand_migration_config(), Ok(None));
+
+        bm.default_timestamps();
+        assert_ne!(bm.created, OffsetDateTime::UNIX_EPOCH, "fixture must carry a real creation time");
+        assert_eq!(bm.on_demand_migration_config_updated_at, bm.created);
+
+        // A metadata blob from this build with no config set stays
+        // indistinguishable from the legacy one for these fields.
+        let fresh = BucketMetadata::unmarshal(&BucketMetadata::new("fresh").marshal_msg().unwrap()).unwrap();
+        assert!(fresh.on_demand_migration_config_json.is_empty());
+        assert_eq!(fresh.on_demand_migration_config_updated_at, OffsetDateTime::UNIX_EPOCH);
+    }
+
+    /// rustfs/backlog#2148: a reader that predates the two on-demand
+    /// migration keys takes `decode_from`'s unknown-field branch, which is
+    /// `skip_msgp_value`. Walk the new-format blob with exactly that
+    /// primitive and prove both keys are skipped without desynchronising the
+    /// stream, so the fields that follow them still decode.
+    #[test]
+    fn old_decoder_skips_on_demand_migration_fields_without_desync() {
+        let mut bm = BucketMetadata::new("odm-skip");
+        bm.update_config(BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
+            .unwrap();
+        bm.update_config(BUCKET_DURABILITY_CONFIG, br#"{"mode":"relaxed"}"#.to_vec())
+            .unwrap();
+        let buf = bm.marshal_msg().unwrap();
+
+        let mut rd = std::io::Cursor::new(buf.as_slice());
+        let fields = rmp::decode::read_map_len(&mut rd).unwrap();
+        let mut skipped = Vec::new();
+        let mut durability_json = Vec::new();
+        for _ in 0..fields {
+            let key_len = rmp::decode::read_str_len(&mut rd).unwrap();
+            let mut key = vec![0u8; key_len as usize];
+            rd.read_exact(&mut key).unwrap();
+            let key = String::from_utf8(key).unwrap();
+            match key.as_str() {
+                // The field an old reader knows that is encoded *after* the
+                // unknown JSON key and *before* the unknown timestamp key.
+                "DurabilityConfigJSON" => durability_json = read_msgp_bin(&mut rd).unwrap(),
+                other => {
+                    if other.starts_with("OnDemandMigration") {
+                        skipped.push(other.to_string());
+                    }
+                    skip_msgp_value(&mut rd).unwrap();
+                }
+            }
+        }
+        assert_eq!(skipped, ["OnDemandMigrationConfigJSON", "OnDemandMigrationConfigUpdatedAt"]);
+        assert_eq!(durability_json, br#"{"mode":"relaxed"}"#);
+        assert_eq!(rd.position() as usize, buf.len(), "old-style walk must consume the blob exactly");
+    }
+
     /// HP-5b (rustfs/backlog#938): the durability override is a RustFS
     /// extension entry and must survive an encode/decode round trip.
     #[test]

crates/ecstore/src/bucket/metadata_sys.rs

@@ -19,6 +19,7 @@ use super::quota::BucketQuota;
 use super::target::BucketTargets;
 use crate::bucket::bucket_target_sys::BucketTargetSys;
 use crate::bucket::metadata::{load_bucket_metadata_parse, load_bucket_metadata_parse_with_presence};
+use crate::bucket::on_demand_migration::{ON_DEMAND_MIGRATION_CONFIG_HOOK, OnDemandMigrationConfig};
 use crate::bucket::utils::is_meta_bucketname;
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::error::{Error, Result, is_err_bucket_not_found, is_err_strict_volume_not_found};
@@ -384,6 +385,42 @@ fn clear_bucket_durability(bucket: &str) {
     crate::disk::local::bucket_durability::set(bucket, None);
 }
 
+/// Publish the bucket's on-demand migration config (or its absence) to the
+/// runtime registered in `ON_DEMAND_MIGRATION_CONFIG_HOOK`.
+///
+/// Called from the same five cache-install paths as
+/// [`sync_bucket_durability`]. A stored payload this build cannot parse is
+/// published as `None`: the runtime must stop pulling for that bucket rather
+/// than keep an older config or guess.
+fn sync_on_demand_migration(bucket: &str, bm: &BucketMetadata) {
+    let Some(hook) = ON_DEMAND_MIGRATION_CONFIG_HOOK.get() else {
+        return;
+    };
+    match bm.on_demand_migration_config() {
+        Ok(config) => hook(bucket, config.as_ref()),
+        Err(err) => {
+            warn!(
+                event = "bucket_metadata_parse_failed",
+                component = "ecstore",
+                subsystem = "bucket_metadata",
+                bucket = %bucket,
+                config = "on_demand_migration",
+                error = %err,
+                "Failed to parse bucket metadata config"
+            );
+            hook(bucket, None);
+        }
+    }
+}
+
+/// Withdraw a bucket's on-demand migration config when its metadata leaves
+/// the cache.
+fn clear_on_demand_migration(bucket: &str) {
+    if let Some(hook) = ON_DEMAND_MIGRATION_CONFIG_HOOK.get() {
+        hook(bucket, None);
+    }
+}
+
 pub async fn get(bucket: &str) -> Result<Arc<BucketMetadata>> {
     let sys = get_bucket_metadata_sys()?;
     let lock = sys.read().await;
@@ -970,6 +1007,16 @@ pub async fn get_durability_config(
     Ok((bm.durability_config(), bm.durability_config_updated_at))
 }
 
+/// The bucket's on-demand migration config with its update time, or
+/// `Ok(None)` when the bucket has none. A stored payload that does not parse
+/// is a typed error (`OnDemandMigrationConfigError` inside `Error::Io`).
+pub async fn get_on_demand_migration_config(bucket: &str) -> Result<Option<(OnDemandMigrationConfig, OffsetDateTime)>> {
+    let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
+    let bucket_meta_sys = bucket_meta_sys_lock.read().await;
+
+    bucket_meta_sys.get_on_demand_migration_config(bucket).await
+}
+
 pub async fn get_quota_config(bucket: &str) -> Result<(BucketQuota, OffsetDateTime)> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1492,6 +1539,7 @@ impl BucketMetadataSys {
                 if removed {
                     BucketTargetSys::get().delete(bucket).await;
                     clear_bucket_durability(bucket);
+                    clear_on_demand_migration(bucket);
                 }
             }
             return Ok(());
@@ -1529,6 +1577,7 @@ impl BucketMetadataSys {
                 self.missing_buckets.invalidate(bucket).await;
                 sync_bucket_target_sys(bucket, &bm).await;
                 sync_bucket_durability(bucket, &bm);
+                sync_on_demand_migration(bucket, &bm);
             }
             MetadataLoadMode::Initial => {
                 let _publish_guard = self
@@ -1575,6 +1624,7 @@ impl BucketMetadataSys {
             if removed {
                 BucketTargetSys::get().delete(bucket).await;
                 clear_bucket_durability(bucket);
+                clear_on_demand_migration(bucket);
             }
             return Ok(());
         }
@@ -1597,6 +1647,7 @@ impl BucketMetadataSys {
         self.missing_buckets.invalidate(bucket).await;
         sync_bucket_target_sys(bucket, &metadata).await;
         sync_bucket_durability(bucket, &metadata);
+        sync_on_demand_migration(bucket, &metadata);
         Ok(())
     }
 
@@ -1624,6 +1675,7 @@ impl BucketMetadataSys {
             self.missing_buckets.invalidate(&bucket).await;
             sync_bucket_target_sys(&bucket, &bm).await;
             sync_bucket_durability(&bucket, &bm);
+            sync_on_demand_migration(&bucket, &bm);
         }
     }
 
@@ -1644,6 +1696,7 @@ impl BucketMetadataSys {
         if removed {
             BucketTargetSys::get().delete(bucket).await;
             clear_bucket_durability(bucket);
+            clear_on_demand_migration(bucket);
         }
         removed || removed_fabricated
     }
@@ -1933,6 +1986,7 @@ impl BucketMetadataSys {
                 self.missing_buckets.invalidate(bucket).await;
                 sync_bucket_target_sys(bucket, &bm).await;
                 sync_bucket_durability(bucket, &bm);
+                sync_on_demand_migration(bucket, &bm);
             } else {
                 let exists = self
                     .bucket_exists(bucket, &guard, "lazy bucket metadata existence check")
@@ -2271,6 +2325,7 @@ impl BucketMetadataSys {
         self.missing_buckets.invalidate(bucket).await;
         sync_bucket_target_sys(bucket, &metadata).await;
         sync_bucket_durability(bucket, &metadata);
+        sync_on_demand_migration(bucket, &metadata);
         Ok(BucketMetadataAuthority::Authoritative(metadata))
     }
 
@@ -2463,6 +2518,17 @@ impl BucketMetadataSys {
             Err(Error::ConfigNotFound)
         }
     }
+
+    /// See [`get_on_demand_migration_config`].
+    pub async fn get_on_demand_migration_config(
+        &self,
+        bucket: &str,
+    ) -> Result<Option<(OnDemandMigrationConfig, OffsetDateTime)>> {
+        let (bm, _) = self.get_config(bucket).await?;
+
+        let config = bm.on_demand_migration_config().map_err(Error::other)?;
+        Ok(config.map(|config| (config, bm.on_demand_migration_config_updated_at)))
+    }
 }
 
 /// Test-only fixture shared with sibling modules (e.g. the quota checker
@@ -4043,6 +4109,151 @@ mod tests {
         assert_eq!(bucket_durability::lookup(bucket), None);
     }
 
+    const ODM_JSON: &[u8] = br#"{"source":{"provider":"minio","endpoint":"https://legacy.example.com:9000","region":"auto","bucket":"legacy-bucket","credentials":{"access_key":"AK","secret_key":"SK"}}}"#;
+
+    /// Every `(bucket, config)` the recording hook has seen. Tests filter by
+    /// their own bucket name; the hook is process-wide and set once.
+    static ODM_HOOK_CALLS: std::sync::Mutex<Vec<(String, Option<OnDemandMigrationConfig>)>> = std::sync::Mutex::new(Vec::new());
+
+    fn install_recording_odm_hook() {
+        ON_DEMAND_MIGRATION_CONFIG_HOOK.get_or_init(|| {
+            Box::new(|bucket, config| {
+                ODM_HOOK_CALLS.lock().unwrap().push((bucket.to_string(), config.cloned()));
+            })
+        });
+    }
+
+    fn odm_hook_calls(bucket: &str) -> Vec<Option<OnDemandMigrationConfig>> {
+        ODM_HOOK_CALLS
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|(name, _)| name == bucket)
+            .map(|(_, config)| config.clone())
+            .collect()
+    }
+
+    /// rustfs/backlog#2148: the accessor reports absence as `Ok(None)` and a
+    /// stored payload it cannot parse as a typed error, never as a default
+    /// and never as `ConfigNotFound`.
+    #[tokio::test]
+    async fn get_on_demand_migration_config_distinguishes_absent_from_corrupt() {
+        use crate::bucket::on_demand_migration::OnDemandMigrationConfigError;
+
+        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let sys = BucketMetadataSys::new(ecstore);
+        let bucket = "odm-accessor";
+
+        sys.set(bucket.to_string(), Arc::new(BucketMetadata::new(bucket))).await;
+        assert_eq!(sys.get_on_demand_migration_config(bucket).await.unwrap(), None);
+
+        let mut corrupt = BucketMetadata::new(bucket);
+        corrupt.on_demand_migration_config_json = br#"{"source":{"provider":"s3"},"bogus":1}"#.to_vec();
+        sys.set(bucket.to_string(), Arc::new(corrupt)).await;
+        let err = sys
+            .get_on_demand_migration_config(bucket)
+            .await
+            .expect_err("corrupt config must not read as a default");
+        assert_ne!(err, Error::ConfigNotFound, "corruption must not be reported as absence");
+        let typed = match &err {
+            Error::Io(io) => io
+                .get_ref()
+                .and_then(|source| source.downcast_ref::<OnDemandMigrationConfigError>()),
+            _ => None,
+        };
+        assert!(
+            matches!(typed, Some(OnDemandMigrationConfigError::Malformed(_))),
+            "typed parse error must survive the Result boundary, got: {err:?}"
+        );
+
+        let mut valid = BucketMetadata::new(bucket);
+        valid
+            .update_config(crate::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
+            .unwrap();
+        let stamped = valid.on_demand_migration_config_updated_at;
+        sys.set(bucket.to_string(), Arc::new(valid)).await;
+        let (config, updated_at) = sys
+            .get_on_demand_migration_config(bucket)
+            .await
+            .unwrap()
+            .expect("stored config is returned");
+        assert_eq!(config, OnDemandMigrationConfig::from_json(ODM_JSON).unwrap());
+        assert_eq!(updated_at, stamped);
+    }
+
+    /// rustfs/backlog#2148: the publish hook fires on every path that
+    /// installs bucket metadata into the cache (set, initial load, peer
+    /// reload, refresh loop, lazy load) and withdraws on removal, mirroring
+    /// `sync_bucket_durability`.
+    #[tokio::test]
+    async fn on_demand_migration_hook_fires_on_every_cache_install_path() {
+        install_recording_odm_hook();
+
+        let (dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let bucket = "odm-hook-paths";
+        for dir in &dirs {
+            std::fs::create_dir_all(dir.path().join(bucket)).expect("physical bucket should exist");
+        }
+        let expected = OnDemandMigrationConfig::from_json(ODM_JSON).unwrap();
+        let expect_publish = |before: usize, label: &str| {
+            let calls = odm_hook_calls(bucket);
+            assert_eq!(calls.len(), before + 1, "{label} must publish exactly once");
+            assert_eq!(calls.last().unwrap().as_ref(), Some(&expected), "{label} must publish the stored config");
+        };
+
+        // set (via persist_new_and_set, which installs through `set`).
+        let mut bm = BucketMetadata::new(bucket);
+        bm.update_config(crate::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
+            .unwrap();
+        let writer = BucketMetadataSys::new(ecstore.clone());
+        let before = odm_hook_calls(bucket).len();
+        writer.persist_new_and_set(bm).await.expect("metadata should persist");
+        expect_publish(before, "set");
+
+        // init (initial load on a cold system).
+        let mut cold = BucketMetadataSys::new(ecstore.clone());
+        let before = odm_hook_calls(bucket).len();
+        cold.init(vec![bucket.to_string()]).await;
+        assert!(cold.get(bucket).await.is_ok(), "initial load must cache the bucket");
+        expect_publish(before, "init");
+
+        // peer reload.
+        let before = odm_hook_calls(bucket).len();
+        cold.reload_from_store(bucket).await.expect("peer reload should publish");
+        expect_publish(before, "peer reload");
+
+        // refresh loop.
+        let before = odm_hook_calls(bucket).len();
+        let mut failed = HashSet::new();
+        cold.concurrent_load(&[bucket.to_string()], &mut failed, MetadataLoadMode::Refresh)
+            .await;
+        assert!(failed.is_empty(), "refresh must succeed");
+        expect_publish(before, "refresh loop");
+
+        // lazy load on another cold system.
+        let lazy = BucketMetadataSys::new(ecstore);
+        let before = odm_hook_calls(bucket).len();
+        let (_, loaded) = lazy.get_config(bucket).await.expect("lazy load should publish");
+        assert!(loaded, "the lazy path must have gone to disk");
+        expect_publish(before, "lazy load");
+
+        // Removal withdraws the config.
+        let before = odm_hook_calls(bucket).len();
+        assert!(lazy.remove(bucket).await);
+        let calls = odm_hook_calls(bucket);
+        assert_eq!(calls.len(), before + 1, "remove must withdraw exactly once");
+        assert_eq!(calls.last().unwrap(), &None);
+
+        // A corrupt payload is withdrawn, never published as a config.
+        let mut corrupt = BucketMetadata::new(bucket);
+        corrupt.on_demand_migration_config_json = b"not-json".to_vec();
+        let before = odm_hook_calls(bucket).len();
+        lazy.set(bucket.to_string(), Arc::new(corrupt)).await;
+        let calls = odm_hook_calls(bucket);
+        assert_eq!(calls.len(), before + 1);
+        assert_eq!(calls.last().unwrap(), &None, "unreadable config must publish absence");
+    }
+
     #[tokio::test]
     async fn refresh_wait_exits_when_cancelled() {
         let cancel_token = CancellationToken::new();

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

@@ -12,7 +12,17 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! On-demand migration (ODM): serve and back-fill objects from an external
+//! On-Demand Migration (ODM): a bucket can name an external S3-compatible
-//! S3-compatible source bucket.
+//! source bucket; GET misses are served from that source and backfilled
+//! locally. This module owns the bucket-level configuration model
+//! (`on-demand-migration.json` in the bucket metadata file); the runtime is
+//! layered on top of it by later tasks (rustfs/backlog#2147).
 
+pub mod config;
 pub mod source_client;
+
+pub use config::{
+    ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
+    OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, SourceConfig,
+    SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
+};

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

@@ -66,7 +66,6 @@ use crate::disk::new_disk;
 use crate::multipart_listing::paginate_multipart_listing;
 #[cfg(test)]
 use crate::object_api::ObjectLockConfigSnapshot;
-use crate::set_disk::core::io_primitives::finish_rename_tail_heal;
 use crate::set_disk::mem;
 use crate::set_disk::metadata_sys;
 use crate::set_disk::runtime_sources;
@@ -3124,8 +3123,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         let commit_object_lock_guard = object_lock_guard.take();
         let commit_decommission_object_lock_guard = decommission_object_lock_guard.take();
         let commit_decommission_capacity_guard = decommission_capacity_guard.take();
-        let commit_allows_early_ack = !(opts.data_movement && opts.has_decommission_capacity_reservation())
+        // CompleteMultipartUpload is an S3 publication boundary: after a
-            && (commit_object_lock_guard.is_some() || commit_decommission_object_lock_guard.is_some());
+        // successful response, the object must be immediately readable and
+        // usable as a CopyObject source. Do not return on rename quorum while
+        // a tail owner may still hold the object guard and finish shard moves.
+        let commit_allows_early_ack = false;
         let detach_commit_owner = commit_allows_early_ack || upload_guard.is_some() || quota_mutation_fence;
         let commit = async move {
             let mut _object_lock_guard = commit_object_lock_guard;
@@ -3256,105 +3258,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 commit_allows_early_ack,
             )
             .await;
-            let mut rename_guard_release = None;
-            let mut needs_immediate_heal = false;
-            let mut tail_owns_staging_cleanup = false;
             if let Ok(rename_commit) = rename_result.as_mut() {
                 commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &rename_commit.capacity_disks);
-                // Install the tail watcher before any post-commit await. The
+                debug_assert!(
-                // latch keeps namespace guards through their prior handoff point.
+                    rename_commit.tail_drain.is_none(),
-                needs_immediate_heal = rename_commit.needs_immediate_heal();
+                    "multipart completion disables early ACK and must not detach a rename tail"
-                if let Some(rename_tail_drain) = rename_commit.tail_drain.take() {
+                );
-                    tail_owns_staging_cleanup = true;
-                    let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options(
-                        commit_bucket.clone(),
-                        Some(commit_object.clone()),
-                        false,
-                        Some(HealChannelPriority::Normal),
-                        Some(commit_set.pool_index),
-                        Some(commit_set.set_index),
-                    );
-                    request.object_version_id = fi
-                        .version_id
-                        .or_else(|| commit_version_suspended.then(Uuid::nil))
-                        .map(|version_id| version_id.to_string());
-                    let object_lock_guard = _object_lock_guard.take();
-                    let upload_guard = _upload_guard.take();
-                    let decommission_object_lock_guard = _decommission_object_lock_guard.take();
-                    let decommission_capacity_guard = _decommission_capacity_guard.take();
-                    let cleanup_bucket = commit_bucket.clone();
-                    let cleanup_object = commit_object.clone();
-                    let heal_set = commit_set.clone();
-                    let cleanup_set = commit_set.clone();
-                    let committed_data_dir = fi.data_dir;
-                    let cleanup_parts = parts.clone();
-                    let cleanup_upload_path = commit_upload_id_path.clone();
-                    let cleanup_upload_id = commit_upload_id.clone();
-                    let fence_disks = commit_disks.clone();
-                    let fence_tokens = quota_fence_tokens.clone();
-                    let fence_bucket = commit_bucket.clone();
-                    let fence_object = commit_object.clone();
-                    let (guard_release_tx, guard_release_rx) = tokio::sync::oneshot::channel();
-                    rename_guard_release = Some(guard_release_tx);
-                    tokio::spawn(finish_rename_tail_heal(
-                        rename_tail_drain,
-                        guard_release_rx,
-                        (
-                            object_lock_guard,
-                            upload_guard,
-                            decommission_object_lock_guard,
-                            decommission_capacity_guard,
-                        ),
-                        request,
-                        move || async move {
-                            if quota_mutation_fence {
-                                let _ = SetDisks::release_quota_mutation_fences(
-                                    &fence_disks,
-                                    &fence_tokens,
-                                    &fence_bucket,
-                                    &fence_object,
-                                    write_quorum,
-                                )
-                                .await;
-                            }
-                        },
-                        move |(object_lock_guard, upload_guard, decommission_object_lock_guard, decommission_capacity_guard),
-                              targets| async move {
-                            drop(object_lock_guard);
-                            cleanup_set.cleanup_multipart_path(&cleanup_parts).await;
-                            cleanup_set
-                                .cleanup_rename_tail(
-                                    targets,
-                                    &cleanup_bucket,
-                                    &cleanup_object,
-                                    committed_data_dir,
-                                    transaction_epoch,
-                                )
-                                .await;
-                            if let Err(err) = cleanup_set
-                                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &cleanup_upload_path, write_quorum)
-                                .await
-                            {
-                                warn!(
-                                    bucket = %cleanup_bucket,
-                                    object = %cleanup_object,
-                                    upload_id = %cleanup_upload_id,
-                                    error = ?err,
-                                    "completed multipart upload staging cleanup did not reach write quorum"
-                                );
-                            }
-                            drop(upload_guard);
-                            drop(decommission_object_lock_guard);
-                            drop(decommission_capacity_guard);
-                        },
-                        |request| async move { heal_set.submit_rename_tail_heal(request).await },
-                    ));
-                }
             }
-            if !tail_owns_staging_cleanup {
+            drop(_decommission_capacity_guard.take());
-                drop(_decommission_capacity_guard.take());
+            if quota_mutation_fence {
-            }
-            if quota_mutation_fence && !tail_owns_staging_cleanup {
                 let _ = SetDisks::release_quota_mutation_fences(
                     &commit_disks,
                     &quota_fence_tokens,
@@ -3371,6 +3283,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 Ok(result) => result,
                 Err(err) => return Err(err.into()),
             };
+            let needs_immediate_heal = rename_commit.needs_immediate_heal();
             let op_old_dir = rename_commit.data_dir;
             let cleanup_disks = rename_commit.cleanup_disks;
             let committed_file_info = rename_commit.committed_file_info;
@@ -3413,9 +3326,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             // parts are swept by a retried completion or upload GC (rustfs/backlog#946).
             // Compiles to a no-op outside `#[cfg(test)]`.
             if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, &commit_object) {
-                if let Some(release) = rename_guard_release.take() {
-                    let _ = release.send(false);
-                }
                 return Err(StorageError::Unexpected);
             }
 
@@ -3431,10 +3341,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
                 .await;
 
-            if let Some(release) = rename_guard_release.take() {
+            drop(_object_lock_guard.take()); // release the object lock before multipart cleanup IO.
-                let _ = release.send(true);
-            }
-            drop(_object_lock_guard.take()); // release the object lock before multipart cleanup tail IO.
 
             #[cfg(test)]
             pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterObjectPublication).await;
@@ -3447,9 +3354,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             // parts; deleting them before the commit would strand the upload
             // permanently. This mirrors the "clean up only after commit" pattern
             // already used for the old data-dir GC and the upload-dir delete_all below.
-            if !tail_owns_staging_cleanup {
+            commit_set.cleanup_multipart_path(&parts).await;
-                commit_set.cleanup_multipart_path(&parts).await;
-            }
 
             if let Some(old_dir) = op_old_dir {
                 // backlog#898: best-effort reclaim of the dereferenced old data dir.
@@ -3480,10 +3385,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             #[cfg(test)]
             pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterRename).await;
 
-            if !tail_owns_staging_cleanup
+            if let Err(err) = commit_set
-                && let Err(err) = commit_set
+                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path, write_quorum)
-                    .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path, write_quorum)
+                .await
-                    .await
             {
                 warn!(
                     bucket = %commit_bucket,
@@ -4010,7 +3914,7 @@ mod tests {
 
     #[tokio::test]
     #[serial(capacity_dirty_scope)]
-    async fn early_ack_multipart_holds_quota_fences_and_re_marks_capacity_after_tail_drain() {
+    async fn complete_multipart_waits_for_tail_before_releasing_guards_and_marking_capacity() {
         use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes;
 
         temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
@@ -4056,10 +3960,9 @@ mod tests {
                 .collect::<HashSet<_>>();
             let _ = drain_global_dirty_scopes();
 
-            let rename_tasks = rename_fanout_barrier::observe_tasks(object);
             let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
             let complete_store = Arc::clone(&set_disks);
-            let complete = tokio::spawn(async move {
+            let mut complete = tokio::spawn(async move {
                 let mut opts = ObjectOptions::default();
                 assert!(opts.set_quota_admission(0, u64::MAX));
                 complete_store
@@ -4069,20 +3972,15 @@ mod tests {
             tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
                 .await
                 .expect("multipart completion should pause one tail disk during rename");
-            let cleanup_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_CLEANUP);
-            complete
-                .await
-                .expect("early-ACK multipart task should join before tail release")
-                .expect("multipart completion should return after write quorum");
             assert!(
-                rename_tasks.running() >= 1,
+                tokio::time::timeout(Duration::from_millis(100), &mut complete).await.is_err(),
-                "the paused multipart tail disk must remain in flight after quorum ACK"
+                "multipart completion must not publish success while a tail rename is still paused"
             );
 
             let initial = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
             assert!(
-                expected.is_subset(&initial),
+                initial.is_empty(),
-                "the multipart quorum ACK must mark every candidate disk dirty"
+                "capacity must not be marked as committed before the full multipart rename finishes"
             );
 
             let abort_store = Arc::clone(&set_disks);
@@ -4092,7 +3990,7 @@ mod tests {
                     .await
             });
             signaling.wait_for_attempts(2).await;
-            assert!(!abort.is_finished(), "the detached tail owner must retain the multipart upload guard");
+            assert!(!abort.is_finished(), "the in-flight completion must retain the multipart upload guard");
 
             let retained_staging = futures::future::join_all(
                 disk_stores
@@ -4117,25 +4015,20 @@ mod tests {
                     .await
             });
             signaling.wait_for_attempts(object_attempt).await;
-            assert!(!object_probe.is_finished(), "the detached tail owner must retain the object guard");
+            assert!(!object_probe.is_finished(), "the in-flight completion must retain the object guard");
 
             rename_barrier.release();
+            complete
+                .await
+                .expect("multipart task should join after tail release")
+                .expect("multipart completion should return after every tail rename finishes");
             object_probe
                 .await
-                .expect("object guard probe should join after the tail releases")
+                .expect("object guard probe should join after completion releases")
-                .expect("object guard probe should acquire after the tail releases");
+                .expect("object guard probe should acquire after completion releases");
-            tokio::time::timeout(Duration::from_secs(30), cleanup_barrier.wait_until_paused())
-                .await
-                .expect("the multipart tail should pause before reclaiming its old body");
-            let after_tail = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
-            assert!(
-                expected.is_subset(&after_tail),
-                "the multipart rename tail must re-mark capacity after the first scope was drained"
-            );
-            cleanup_barrier.release();
             let abort_err = abort
                 .await
-                .expect("abort task should join after the tail releases")
+                .expect("abort task should join after completion releases")
                 .expect_err("the committed upload should no longer exist");
             assert!(matches!(abort_err, StorageError::InvalidUploadID(..)));
 
@@ -4148,7 +4041,7 @@ mod tests {
             let after_cleanup = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
             assert!(
                 expected.is_subset(&after_cleanup),
-                "the multipart tail cleanup must re-mark capacity after its preceding scope was drained"
+                "the completed multipart commit must mark every candidate disk dirty"
             );
         })
         .await;
@@ -4283,23 +4176,26 @@ mod tests {
             ],
             async {
                 let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
-                set_disks
+                let complete_store = Arc::clone(&set_disks);
-                    .clone()
+                let mut complete = tokio::spawn(async move {
-                    .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default())
+                    complete_store
-                    .await
+                        .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default())
-                    .expect("fenced multipart completion should commit with a live proof");
+                        .await
+                });
 
                 tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
                     .await
-                    .expect("multipart completion should leave one rename tail in flight after quorum ACK");
+                    .expect("multipart completion should pause one tail disk during rename");
-                let disks = disk_stores.clone();
-                let mut epochs = tokio::spawn(async move { object_transaction_epochs(&disks, bucket, object).await });
                 assert!(
-                    tokio::time::timeout(Duration::from_millis(100), &mut epochs).await.is_err(),
+                    tokio::time::timeout(Duration::from_millis(100), &mut complete).await.is_err(),
-                    "epoch read-back should wait for the lagging rename tail"
+                    "fenced multipart completion must wait for every rename tail before returning"
                 );
                 rename_barrier.release();
-                epochs.await.expect("epoch read-back should finish after the rename tail")
+                complete
+                    .await
+                    .expect("fenced multipart task should join after tail release")
+                    .expect("fenced multipart completion should commit with a live proof");
+                object_transaction_epochs(&disk_stores, bucket, object).await
             },
         )
         .await;
@@ -8392,29 +8288,18 @@ mod tests {
             let new = payload(0xC3);
             let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await;
             let parts_retry = parts_new.clone();
-            let rename_tasks = rename_fanout_barrier::observe_tasks(object);
-            let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
             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 armed post-commit crash point must be the failure that surfaced, got {crashed:?}"
             );
-            assert!(rename_tasks.running() >= 1, "the crash must interrupt an actual early-ACK tail handoff");
             crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object);
-            rename_barrier.release();
-            tokio::time::timeout(Duration::from_secs(30), async {
-                while rename_tasks.running() != 0 {
-                    tokio::task::yield_now().await;
-                }
-            })
-            .await
-            .expect("the crash-interrupted rename tail should drain after release");
             drop(
                 set_disks
                     .acquire_write_lock_diag("post_commit_crash_tail_probe", bucket, object)
                     .await
-                    .expect("the crash-interrupted tail should release its object guard"),
+                    .expect("the failed post-commit completion should release its object guard"),
             );
 
             // The commit landed: the new version reads back whole and correct.
@@ -8487,29 +8372,18 @@ mod tests {
 
                     let new = payload(0x52);
                     let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await;
-                    let rename_tasks = rename_fanout_barrier::observe_tasks(object);
-                    let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
                     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:?}"
                     );
-                    assert!(rename_tasks.running() >= 1, "the crash must interrupt an actual early-ACK tail handoff");
                     crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object);
-                    rename_barrier.release();
-                    tokio::time::timeout(Duration::from_secs(30), async {
-                        while rename_tasks.running() != 0 {
-                            tokio::task::yield_now().await;
-                        }
-                    })
-                    .await
-                    .expect("the crash-interrupted rename tail should drain after release");
                     drop(
                         set_disks
                             .acquire_write_lock_diag("post_commit_receipt_tail_probe", bucket, object)
                             .await
-                            .expect("the crash-interrupted tail should release its object guard"),
+                            .expect("the failed post-commit completion should release its object guard"),
                     );
 
                     let (body, _) = read_object(&set_disks, bucket, object).await;
@@ -8523,8 +8397,8 @@ mod tests {
                         );
                     }
                     assert_eq!(
-                        receipts, 3,
+                        receipts, 4,
-                        "the committed quorum must persist receipts while the crash-interrupted tail preserves staging"
+                        "the completed rename must persist old-data cleanup receipts on every disk before surfacing the post-commit crash"
                     );
 
                     let restarted_endpoints = temp_dirs
@@ -8570,12 +8444,12 @@ mod tests {
                         .reconcile_old_data_cleanup_receipts(bucket, object)
                         .await
                         .expect("restart receipt reconciliation should succeed");
-                    assert_eq!(removed, 3, "restart receipt reconciliation should delete the committed quorum's targets");
+                    assert_eq!(removed, 4, "restart receipt reconciliation should delete every committed target");
                     let reclaimed = restarted_set
                         .reclaim_orphan_data_dirs(bucket, object)
                         .await
                         .expect("restart orphan reconciliation should succeed");
-                    assert_eq!(reclaimed, 1, "the late commit without a receipt must remain reclaimable as an orphan");
+                    assert_eq!(reclaimed, 0, "the post-commit crash should leave no receipt-less late commit orphan");
                     for disk in &reloaded {
                         assert!(
                             !data_dir_exists(disk, bucket, object, old_dir).await,

crates/ecstore/src/store/init.rs

@@ -2305,6 +2305,200 @@ mod tests {
         shutdown.cancel();
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[serial_test::serial(storage_class_env)]
+    async fn copy_object_immediately_reads_small_completed_multipart_source() {
+        let temp_dir = tempfile::tempdir().expect("create small multipart copy store dir");
+        let (_ctx, store, shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "small-multipart-copy", &[1])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+
+        let bucket = format!("small-multipart-copy-{}", Uuid::new_v4());
+        let source_object = "docker/registry/v2/repositories/example/_uploads/upload-id/data";
+        let target_object = "docker/registry/v2/blobs/sha256/c0/digest/data";
+        let payload = vec![0xAB; 273];
+
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket for small multipart copy");
+        let upload = store
+            .new_multipart_upload(&bucket, source_object, &ObjectOptions::default())
+            .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, &ObjectOptions::default())
+            .await
+            .expect("stage small multipart source part");
+        let completed = 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()
+                }],
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("complete the small multipart source");
+        assert_eq!(completed.get_actual_size().expect("completed object logical size"), payload.len() as i64);
+
+        let source_reader = store
+            .get_object_reader(&bucket, source_object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("completed multipart source should be immediately 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));
+
+        store
+            .copy_object(
+                &bucket,
+                source_object,
+                &bucket,
+                target_object,
+                &mut copy_info,
+                &ObjectOptions::default(),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("CopyObject should accept a freshly completed multipart source");
+
+        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)]
+    async fn complete_multipart_waits_for_tail_rename_before_copy_source_visibility() {
+        let temp_dir = tempfile::tempdir().expect("create early-ack multipart copy store dir");
+        let (_ctx, store, shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "early-ack-multipart-copy", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
+
+        let bucket = format!("early-ack-multipart-copy-{}", Uuid::new_v4());
+        let source_object = "docker/registry/v2/repositories/example/_uploads/upload-id/data";
+        let target_object = "docker/registry/v2/blobs/sha256/c0/digest/data";
+        let payload = vec![0xCD; 273];
+
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket for early-ack multipart copy");
+        let upload = store
+            .new_multipart_upload(&bucket, source_object, &ObjectOptions::default())
+            .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, &ObjectOptions::default())
+            .await
+            .expect("stage small multipart source part");
+        let completed_parts = vec![crate::storage_api_contracts::multipart::CompletePart {
+            part_num: part.part_num,
+            etag: part.etag,
+            ..Default::default()
+        }];
+
+        temp_env::async_with_vars([(crate::set_disk::ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            let rename_tasks = crate::set_disk::rename_fanout_barrier::observe_tasks(source_object);
+            let rename_barrier = crate::set_disk::rename_fanout_barrier::arm(
+                source_object,
+                0,
+                crate::set_disk::rename_fanout_barrier::PHASE_RENAME,
+            );
+            let complete_store = Arc::clone(&store);
+            let complete_bucket = bucket.clone();
+            let complete_upload_id = upload.upload_id.clone();
+            let mut complete = tokio::spawn(async move {
+                complete_store
+                    .complete_multipart_upload(
+                        &complete_bucket,
+                        source_object,
+                        &complete_upload_id,
+                        completed_parts,
+                        &ObjectOptions::default(),
+                    )
+                    .await
+            });
+            tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
+                .await
+                .expect("multipart completion should pause one tail disk during rename");
+            assert!(
+                rename_tasks.running() >= 1,
+                "the paused multipart tail disk must remain in flight before completion returns"
+            );
+            assert!(
+                tokio::time::timeout(Duration::from_millis(100), &mut complete).await.is_err(),
+                "CompleteMultipartUpload must not return while a copy source rename tail is still pending"
+            );
+            rename_barrier.release();
+            complete
+                .await
+                .expect("multipart completion task should join")
+                .expect("multipart completion should return after every rename tail finishes");
+
+            let source_reader = store
+                .get_object_reader(&bucket, source_object, None, HeaderMap::new(), &ObjectOptions::default())
+                .await
+                .expect("completed multipart source should be immediately readable after success");
+            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));
+
+            store
+                .copy_object(
+                    &bucket,
+                    source_object,
+                    &bucket,
+                    target_object,
+                    &mut copy_info,
+                    &ObjectOptions::default(),
+                    &ObjectOptions::default(),
+                )
+                .await
+                .expect("CopyObject should accept a freshly completed multipart source");
+        })
+        .await;
+
+        let mut target_reader = store
+            .get_object_reader(&bucket, target_object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("copied target should be readable after tail release");
+        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)]

crates/protocols/src/common/client/s3.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use async_trait::async_trait;
+use rustfs_credentials::Credentials;
 use s3s::dto::*;
 
 #[cfg(feature = "webdav")]
@@ -27,49 +28,33 @@ pub trait StorageBackend: Send + Sync {
         &self,
         bucket: &str,
         key: &str,
-        access_key: &str,
+        credentials: &Credentials,
-        secret_key: &str,
         start_pos: Option<u64>,
     ) -> Result<GetObjectOutput, Self::Error>;
     async fn get_object_range(
         &self,
         bucket: &str,
         key: &str,
-        access_key: &str,
+        credentials: &Credentials,
-        secret_key: &str,
         start_pos: u64,
         length: u64,
     ) -> Result<GetObjectOutput, Self::Error>;
     /// Put object content with metadata
-    async fn put_object(&self, input: PutObjectInput, access_key: &str, secret_key: &str)
+    async fn put_object(&self, input: PutObjectInput, credentials: &Credentials) -> Result<PutObjectOutput, Self::Error>;
-    -> Result<PutObjectOutput, Self::Error>;
     /// Delete an object
-    async fn delete_object(
+    async fn delete_object(&self, bucket: &str, key: &str, credentials: &Credentials) -> Result<DeleteObjectOutput, Self::Error>;
-        &self,
-        bucket: &str,
-        key: &str,
-        access_key: &str,
-        secret_key: &str,
-    ) -> Result<DeleteObjectOutput, Self::Error>;
     /// Get object metadata without content
-    async fn head_object(
+    async fn head_object(&self, bucket: &str, key: &str, credentials: &Credentials) -> Result<HeadObjectOutput, Self::Error>;
-        &self,
-        bucket: &str,
-        key: &str,
-        access_key: &str,
-        secret_key: &str,
-    ) -> Result<HeadObjectOutput, Self::Error>;
     /// Check if bucket exists and get metadata
-    async fn head_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<HeadBucketOutput, Self::Error>;
+    async fn head_bucket(&self, bucket: &str, credentials: &Credentials) -> Result<HeadBucketOutput, Self::Error>;
     /// List objects in a bucket with pagination
     async fn list_objects_v2(
         &self,
         input: ListObjectsV2Input,
-        access_key: &str,
+        credentials: &Credentials,
-        secret_key: &str,
     ) -> Result<ListObjectsV2Output, Self::Error>;
     /// List all buckets (requires authentication).
-    async fn list_buckets(&self, access_key: &str, secret_key: &str) -> Result<ListBucketsOutput, Self::Error>;
+    async fn list_buckets(&self, credentials: &Credentials) -> Result<ListBucketsOutput, Self::Error>;
     /// List buckets visible to the authenticated session.
     ///
     /// Backends that implement this must apply per-bucket authorization. The default denies the
@@ -87,20 +72,15 @@ pub trait StorageBackend: Send + Sync {
         ))
     }
     /// Create a new bucket
-    async fn create_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<CreateBucketOutput, Self::Error>;
+    async fn create_bucket(&self, bucket: &str, credentials: &Credentials) -> Result<CreateBucketOutput, Self::Error>;
     /// Delete a bucket (must be empty)
-    async fn delete_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<DeleteBucketOutput, Self::Error>;
+    async fn delete_bucket(&self, bucket: &str, credentials: &Credentials) -> Result<DeleteBucketOutput, Self::Error>;
     /// Server-side copy of an object from one bucket+key to another.
     /// The input carries the full S3 surface (content type, metadata map,
     /// metadata directive, storage class, SSE config, conditional-copy
     /// headers) so protocol drivers can map client-supplied metadata
     /// onto the destination object.
-    async fn copy_object(
+    async fn copy_object(&self, input: CopyObjectInput, credentials: &Credentials) -> Result<CopyObjectOutput, Self::Error>;
-        &self,
-        input: CopyObjectInput,
-        access_key: &str,
-        secret_key: &str,
-    ) -> Result<CopyObjectOutput, Self::Error>;
     /// Initiate a multipart upload. Returns an upload_id that identifies
     /// the in-progress upload for subsequent UploadPart, CompleteMultipartUpload,
     /// and AbortMultipartUpload calls. The input carries the full S3 surface
@@ -110,25 +90,18 @@ pub trait StorageBackend: Send + Sync {
     async fn create_multipart_upload(
         &self,
         input: CreateMultipartUploadInput,
-        access_key: &str,
+        credentials: &Credentials,
-        secret_key: &str,
     ) -> Result<CreateMultipartUploadOutput, Self::Error>;
     /// Upload one part of a multipart upload. The part_number must be in
     /// the range 1 to the 10 000-part S3 limit. The returned ETag
     /// identifies the part in the subsequent CompleteMultipartUpload call.
-    async fn upload_part(
+    async fn upload_part(&self, input: UploadPartInput, credentials: &Credentials) -> Result<UploadPartOutput, Self::Error>;
-        &self,
-        input: UploadPartInput,
-        access_key: &str,
-        secret_key: &str,
-    ) -> Result<UploadPartOutput, Self::Error>;
     /// Assemble the parts listed in the input into the final object.
     /// The parts list must be sorted by part_number with no duplicates.
     async fn complete_multipart_upload(
         &self,
         input: CompleteMultipartUploadInput,
-        access_key: &str,
+        credentials: &Credentials,
-        secret_key: &str,
     ) -> Result<CompleteMultipartUploadOutput, Self::Error>;
     /// Abort an in-progress multipart upload. Releases any storage
     /// associated with the upload_id. Idempotent: calling abort on an
@@ -138,8 +111,7 @@ pub trait StorageBackend: Send + Sync {
     async fn abort_multipart_upload(
         &self,
         input: AbortMultipartUploadInput,
-        access_key: &str,
+        credentials: &Credentials,
-        secret_key: &str,
     ) -> Result<AbortMultipartUploadOutput, Self::Error>;
     /// Copy a byte range from an existing object into a part of an
     /// in-progress multipart upload. Used by rename for objects larger
@@ -147,7 +119,6 @@ pub trait StorageBackend: Send + Sync {
     async fn upload_part_copy(
         &self,
         input: UploadPartCopyInput,
-        access_key: &str,
+        credentials: &Credentials,
-        secret_key: &str,
     ) -> Result<UploadPartCopyOutput, Self::Error>;
 }

crates/protocols/src/common/dummy_storage.rs

@@ -35,6 +35,7 @@ use crate::common::session::SessionContext;
 use async_trait::async_trait;
 use bytes::Bytes;
 use futures_util::stream::{self, StreamExt};
+use rustfs_credentials::Credentials;
 use s3s::dto::{
     AbortMultipartUploadInput, AbortMultipartUploadOutput, CompleteMultipartUploadInput, CompleteMultipartUploadOutput,
     CopyObjectInput, CopyObjectOutput, CopyPartResult, CreateBucketOutput, CreateMultipartUploadInput,
@@ -605,8 +606,7 @@ impl StorageBackend for DummyBackend {
         &self,
         bucket: &str,
         key: &str,
-        _ak: &str,
+        _credentials: &Credentials,
-        _sk: &str,
         _start_pos: Option<u64>,
     ) -> Result<GetObjectOutput, Self::Error> {
         match self.inner.lock().expect("lock").get_object.pop_front() {
@@ -619,8 +619,7 @@ impl StorageBackend for DummyBackend {
         &self,
         bucket: &str,
         key: &str,
-        _ak: &str,
+        _credentials: &Credentials,
-        _sk: &str,
         _start_pos: u64,
         _length: u64,
     ) -> Result<GetObjectOutput, Self::Error> {
@@ -630,7 +629,7 @@ impl StorageBackend for DummyBackend {
         }
     }
 
-    async fn put_object(&self, input: PutObjectInput, _ak: &str, _sk: &str) -> Result<PutObjectOutput, Self::Error> {
+    async fn put_object(&self, input: PutObjectInput, _credentials: &Credentials) -> Result<PutObjectOutput, Self::Error> {
         // Decide control flow while holding the lock. Release before
         // awaiting so the stall path does not hold the Mutex across
         // an await point.
@@ -659,7 +658,12 @@ impl StorageBackend for DummyBackend {
         }
     }
 
-    async fn delete_object(&self, bucket: &str, key: &str, _ak: &str, _sk: &str) -> Result<DeleteObjectOutput, Self::Error> {
+    async fn delete_object(
+        &self,
+        bucket: &str,
+        key: &str,
+        _credentials: &Credentials,
+    ) -> Result<DeleteObjectOutput, Self::Error> {
         let mut inner = self.inner.lock().expect("lock");
         inner.delete_object_calls.push(DeleteObjectCall {
             bucket: bucket.to_string(),
@@ -671,7 +675,7 @@ impl StorageBackend for DummyBackend {
         }
     }
 
-    async fn head_object(&self, bucket: &str, key: &str, _ak: &str, _sk: &str) -> Result<HeadObjectOutput, Self::Error> {
+    async fn head_object(&self, bucket: &str, key: &str, _credentials: &Credentials) -> Result<HeadObjectOutput, Self::Error> {
         {
             let mut inner = self.inner.lock().expect("lock");
             inner.head_object_calls.push(HeadObjectCall {
@@ -685,7 +689,7 @@ impl StorageBackend for DummyBackend {
         }
     }
 
-    async fn head_bucket(&self, bucket: &str, _ak: &str, _sk: &str) -> Result<HeadBucketOutput, Self::Error> {
+    async fn head_bucket(&self, bucket: &str, _credentials: &Credentials) -> Result<HeadBucketOutput, Self::Error> {
         match self.inner.lock().expect("lock").head_bucket.pop_front() {
             Some(r) => r,
             None => Err(DummyError::NoSuchBucket(bucket.to_string())),
@@ -695,8 +699,7 @@ impl StorageBackend for DummyBackend {
     async fn list_objects_v2(
         &self,
         _input: ListObjectsV2Input,
-        _ak: &str,
+        _credentials: &Credentials,
-        _sk: &str,
     ) -> Result<ListObjectsV2Output, Self::Error> {
         // Decide control flow while holding the lock. Release before
         // awaiting so the stall path does not hold the Mutex across
@@ -721,7 +724,7 @@ impl StorageBackend for DummyBackend {
         }
     }
 
-    async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result<ListBucketsOutput, Self::Error> {
+    async fn list_buckets(&self, _credentials: &Credentials) -> Result<ListBucketsOutput, Self::Error> {
         match self.inner.lock().expect("lock").list_buckets.pop_front() {
             Some(r) => r,
             None => Ok(ListBucketsOutput::default()),
@@ -744,14 +747,14 @@ impl StorageBackend for DummyBackend {
             .unwrap_or_else(|| Ok(ListBucketsOutput::default()))
     }
 
-    async fn create_bucket(&self, _bucket: &str, _ak: &str, _sk: &str) -> Result<CreateBucketOutput, Self::Error> {
+    async fn create_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result<CreateBucketOutput, Self::Error> {
         match self.inner.lock().expect("lock").create_bucket.pop_front() {
             Some(r) => r,
             None => Err(DummyError::Unconfigured("create_bucket")),
         }
     }
 
-    async fn delete_bucket(&self, bucket: &str, _ak: &str, _sk: &str) -> Result<DeleteBucketOutput, Self::Error> {
+    async fn delete_bucket(&self, bucket: &str, _credentials: &Credentials) -> Result<DeleteBucketOutput, Self::Error> {
         let mut inner = self.inner.lock().expect("lock");
         inner.delete_bucket_calls.push(bucket.to_string());
         match inner.delete_bucket.pop_front() {
@@ -760,7 +763,7 @@ impl StorageBackend for DummyBackend {
         }
     }
 
-    async fn copy_object(&self, _input: CopyObjectInput, _ak: &str, _sk: &str) -> Result<CopyObjectOutput, Self::Error> {
+    async fn copy_object(&self, _input: CopyObjectInput, _credentials: &Credentials) -> Result<CopyObjectOutput, Self::Error> {
         match self.inner.lock().expect("lock").copy_object.pop_front() {
             Some(r) => r,
             None => Err(DummyError::Unconfigured("copy_object")),
@@ -770,8 +773,7 @@ impl StorageBackend for DummyBackend {
     async fn create_multipart_upload(
         &self,
         input: CreateMultipartUploadInput,
-        _ak: &str,
+        _credentials: &Credentials,
-        _sk: &str,
     ) -> Result<CreateMultipartUploadOutput, Self::Error> {
         {
             let mut inner = self.inner.lock().expect("lock");
@@ -787,7 +789,7 @@ impl StorageBackend for DummyBackend {
         }
     }
 
-    async fn upload_part(&self, input: UploadPartInput, _ak: &str, _sk: &str) -> Result<UploadPartOutput, Self::Error> {
+    async fn upload_part(&self, input: UploadPartInput, _credentials: &Credentials) -> Result<UploadPartOutput, Self::Error> {
         // Record the call and decide the control flow while holding the
         // lock. Release the lock before awaiting so the stall path does
         // not hold the Mutex across an await point.
@@ -821,8 +823,7 @@ impl StorageBackend for DummyBackend {
     async fn complete_multipart_upload(
         &self,
         input: CompleteMultipartUploadInput,
-        _ak: &str,
+        _credentials: &Credentials,
-        _sk: &str,
     ) -> Result<CompleteMultipartUploadOutput, Self::Error> {
         let part_count = input
             .multipart_upload
@@ -847,8 +848,7 @@ impl StorageBackend for DummyBackend {
     async fn abort_multipart_upload(
         &self,
         input: AbortMultipartUploadInput,
-        _ak: &str,
+        _credentials: &Credentials,
-        _sk: &str,
     ) -> Result<AbortMultipartUploadOutput, Self::Error> {
         {
             let mut inner = self.inner.lock().expect("lock");
@@ -867,8 +867,7 @@ impl StorageBackend for DummyBackend {
     async fn upload_part_copy(
         &self,
         _input: UploadPartCopyInput,
-        _ak: &str,
+        _credentials: &Credentials,
-        _sk: &str,
     ) -> Result<UploadPartCopyOutput, Self::Error> {
         match self.inner.lock().expect("lock").upload_part_copy.pop_front() {
             Some(r) => r,
@@ -884,7 +883,8 @@ mod tests {
     #[tokio::test]
     async fn dummy_backend_reports_not_found_by_default() {
         let backend = DummyBackend::new();
-        let result = backend.head_object("b", "k", "ak", "sk").await;
+        let credentials = Credentials::default();
+        let result = backend.head_object("b", "k", &credentials).await;
         let Err(err) = result else {
             panic!("default head_object must return an error");
         };
@@ -897,21 +897,23 @@ mod tests {
     #[tokio::test]
     async fn dummy_backend_returns_queued_head_object_response() {
         let backend = DummyBackend::new();
+        let credentials = Credentials::default();
         backend.queue_head_object_ok(42, None);
-        let out = backend.head_object("b", "k", "ak", "sk").await.expect("queued Ok");
+        let out = backend.head_object("b", "k", &credentials).await.expect("queued Ok");
         assert_eq!(out.content_length, Some(42));
     }
 
     #[tokio::test]
     async fn dummy_backend_logs_abort_multipart_calls() {
         let backend = Arc::new(DummyBackend::new());
+        let credentials = Credentials::default();
         let input = AbortMultipartUploadInput::builder()
             .bucket("b".to_string())
             .key("k".to_string())
             .upload_id("UP-1".to_string())
             .build()
             .expect("build");
-        backend.abort_multipart_upload(input, "ak", "sk").await.expect("Ok");
+        backend.abort_multipart_upload(input, &credentials).await.expect("Ok");
         let calls = backend.abort_multipart_calls();
         assert_eq!(calls.len(), 1);
         assert_eq!(calls[0].upload_id, "UP-1");
@@ -920,6 +922,7 @@ mod tests {
     #[tokio::test]
     async fn dummy_backend_unconfigured_errors_loudly() {
         let backend = DummyBackend::new();
+        let credentials = Credentials::default();
         let err = backend
             .create_multipart_upload(
                 CreateMultipartUploadInput::builder()
@@ -927,8 +930,7 @@ mod tests {
                     .key("k".to_string())
                     .build()
                     .expect("build"),
-                "ak",
+                &credentials,
-                "sk",
             )
             .await
             .expect_err("default create_multipart_upload must error");

crates/protocols/src/common/gateway.rs

@@ -288,12 +288,7 @@ pub async fn is_authorized(
         }
     };
 
-    // Create policy arguments
+    let claims = policy_claims_for_session(session_context);
-    let mut claims = HashMap::new();
-    claims.insert(
-        "principal".to_string(),
-        serde_json::Value::String(session_context.principal.access_key().to_string()),
-    );
 
     let policy_action: rustfs_policy::policy::action::Action = action.clone().into();
 
@@ -315,6 +310,21 @@ pub async fn is_authorized(
     Ok(iam_sys.is_allowed(&args).await)
 }
 
+fn policy_claims_for_session(session_context: &SessionContext) -> HashMap<String, serde_json::Value> {
+    let mut claims = session_context
+        .principal
+        .user_identity
+        .credentials
+        .claims
+        .clone()
+        .unwrap_or_default();
+    claims.insert(
+        "principal".to_string(),
+        serde_json::Value::String(session_context.principal.access_key().to_string()),
+    );
+    claims
+}
+
 /// Authorize an operation and return an error if not authorized.
 /// AccessDenied covers both the protocol-not-supported case and the
 /// policy-denies case. IamUnavailable propagates from is_authorized
@@ -457,7 +467,9 @@ pub use test_auth_override::{with_test_auth_override, with_test_iam_unavailable}
 mod tests {
     use super::*;
     use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext};
+    use rustfs_credentials::{IAM_POLICY_CLAIM_NAME_SA, INHERITED_POLICY_TYPE};
     use rustfs_policy::auth::UserIdentity;
+    use serde_json::Value;
     use std::net::{IpAddr, Ipv4Addr};
     use std::sync::Arc;
 
@@ -466,6 +478,44 @@ mod tests {
         SessionContext::new(principal, Protocol::Sftp, IpAddr::V4(Ipv4Addr::LOCALHOST))
     }
 
+    fn session_with_claims(access_key: &str, claims: HashMap<String, Value>) -> SessionContext {
+        let identity = UserIdentity::new(rustfs_credentials::Credentials {
+            access_key: access_key.to_string(),
+            secret_key: "secret".to_string(),
+            claims: Some(claims),
+            ..Default::default()
+        });
+        let principal = ProtocolPrincipal::new(Arc::new(identity));
+        SessionContext::new(principal, Protocol::WebDav, IpAddr::V4(Ipv4Addr::LOCALHOST))
+    }
+
+    #[test]
+    fn policy_claims_preserve_authenticated_service_account_claims() {
+        let parent = "parent-user";
+        let mut stored_claims = HashMap::new();
+        stored_claims.insert("parent".to_string(), Value::String(parent.to_string()));
+        stored_claims.insert(IAM_POLICY_CLAIM_NAME_SA.to_string(), Value::String(INHERITED_POLICY_TYPE.to_string()));
+        let session = session_with_claims("service-account", stored_claims);
+
+        let claims = policy_claims_for_session(&session);
+
+        assert_eq!(claims.get("parent").and_then(Value::as_str), Some(parent));
+        assert_eq!(claims.get(IAM_POLICY_CLAIM_NAME_SA).and_then(Value::as_str), Some(INHERITED_POLICY_TYPE));
+        assert_eq!(claims.get("principal").and_then(Value::as_str), Some("service-account"));
+    }
+
+    #[test]
+    fn policy_claims_overwrite_untrusted_principal_claim() {
+        let session = session_with_claims(
+            "authenticated-service-account",
+            HashMap::from([("principal".to_string(), Value::String("forged-principal".to_string()))]),
+        );
+
+        let claims = policy_claims_for_session(&session);
+
+        assert_eq!(claims.get("principal").and_then(Value::as_str), Some("authenticated-service-account"));
+    }
+
     #[tokio::test]
     async fn with_test_auth_override_allow_returns_ok() {
         let session = test_session();

crates/protocols/src/common/session.rs

@@ -84,6 +84,11 @@ impl SessionContext {
     pub fn access_key(&self) -> &str {
         self.principal.access_key()
     }
+
+    /// Get the authenticated credentials for this session.
+    pub fn credentials(&self) -> &Credentials {
+        &self.principal.user_identity.credentials
+    }
 }
 
 /// Build a SessionContext suitable for driver-level unit tests. The

crates/protocols/src/ftps/driver.rs

@@ -129,14 +129,7 @@ where
         }
 
         let mut list_result = Vec::new();
-        match self
+        match self.storage.list_buckets(session_context.credentials()).await {
-            .storage
-            .list_buckets(
-                &session_context.principal.user_identity.credentials.access_key,
-                &session_context.principal.user_identity.credentials.secret_key,
-            )
-            .await
-        {
             Ok(output) => {
                 if let Some(buckets) = output.buckets {
                     for bucket in buckets {
@@ -190,15 +183,7 @@ where
                 Error::new(ErrorKind::PermanentFileNotAvailable, format!("Failed to build ListObjectsV2Input: {}", e))
             })?;
 
-            if let Ok(output) = self
+            if let Ok(output) = self.storage.list_objects_v2(list_input, session_context.credentials()).await {
-                .storage
-                .list_objects_v2(
-                    list_input,
-                    &session_context.principal.user_identity.credentials.access_key,
-                    &session_context.principal.user_identity.credentials.secret_key,
-                )
-                .await
-            {
                 // Delete all objects in this page
                 if let Some(objects) = output.contents {
                     for obj in objects {
@@ -209,12 +194,7 @@ where
 
                             let _ = self
                                 .storage
-                                .delete_object(
+                                .delete_object(bucket, &obj_key, session_context.credentials())
-                                    bucket,
-                                    &obj_key,
-                                    &session_context.principal.user_identity.credentials.access_key,
-                                    &session_context.principal.user_identity.credentials.secret_key,
-                                )
                                 .await;
                         }
                     }
@@ -231,15 +211,7 @@ where
         }
 
         // Then delete the bucket
-        match self
+        match self.storage.delete_bucket(bucket, session_context.credentials()).await {
-            .storage
-            .delete_bucket(
-                bucket,
-                &session_context.principal.user_identity.credentials.access_key,
-                &session_context.principal.user_identity.credentials.secret_key,
-            )
-            .await
-        {
             Ok(_) => Ok(()),
             Err(e) if e.to_string().contains("NoSuchBucket") => Ok(()),
             Err(e) => {
@@ -277,16 +249,7 @@ where
                 .await
                 .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
 
-            match self
+            match self.storage.head_object(&bucket, &key, session_context.credentials()).await {
-                .storage
-                .head_object(
-                    &bucket,
-                    &key,
-                    &session_context.principal.user_identity.credentials.access_key,
-                    &session_context.principal.user_identity.credentials.secret_key,
-                )
-                .await
-            {
                 Ok(output) => {
                     let size = output.content_length.unwrap_or(0) as u64;
                     let modified = output.last_modified.map(|dt| {
@@ -323,15 +286,7 @@ where
                 .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
 
             let bucket_clone = bucket.clone();
-            match self
+            match self.storage.head_bucket(&bucket, session_context.credentials()).await {
-                .storage
-                .head_bucket(
-                    &bucket,
-                    &session_context.principal.user_identity.credentials.access_key,
-                    &session_context.principal.user_identity.credentials.secret_key,
-                )
-                .await
-            {
                 Ok(_) => Ok(FtpsMetadata {
                     size: 0,
                     modified: Some(std::time::SystemTime::now()),
@@ -390,15 +345,7 @@ where
                 Error::new(ErrorKind::PermanentFileNotAvailable, format!("Failed to build ListObjectsV2Input: {}", e))
             })?;
 
-        match self
+        match self.storage.list_objects_v2(list_input, session_context.credentials()).await {
-            .storage
-            .list_objects_v2(
-                list_input,
-                &session_context.principal.user_identity.credentials.access_key,
-                &session_context.principal.user_identity.credentials.secret_key,
-            )
-            .await
-        {
             Ok(output) => {
                 let mut fileinfos = Vec::new();
 
@@ -515,8 +462,7 @@ where
             .get_object(
                 &bucket,
                 &key,
-                &session_context.principal.user_identity.credentials.access_key,
+                session_context.credentials(),
-                &session_context.principal.user_identity.credentials.secret_key,
                 Some(start_pos), // Pass start_pos for range request
             )
             .await
@@ -624,15 +570,7 @@ where
             .build()
             .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Failed to build PutObjectInput"))?;
 
-        match self
+        match self.storage.put_object(put_input, session_context.credentials()).await {
-            .storage
-            .put_object(
-                put_input,
-                &session_context.principal.user_identity.credentials.access_key,
-                &session_context.principal.user_identity.credentials.secret_key,
-            )
-            .await
-        {
             Ok(_output) => {
                 Ok(file_size as u64) // Return the size of the uploaded object
             }
@@ -681,16 +619,7 @@ where
                 .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
 
             // Delete file
-            match self
+            match self.storage.delete_object(&bucket, &key, session_context.credentials()).await {
-                .storage
-                .delete_object(
-                    &bucket,
-                    &key,
-                    &session_context.principal.user_identity.credentials.access_key,
-                    &session_context.principal.user_identity.credentials.secret_key,
-                )
-                .await
-            {
                 Ok(_) => Ok(()),
                 Err(e) => {
                     error!(
@@ -748,15 +677,7 @@ where
             .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
 
         // Create bucket for directory
-        match self
+        match self.storage.create_bucket(&bucket, session_context.credentials()).await {
-            .storage
-            .create_bucket(
-                &bucket,
-                &session_context.principal.user_identity.credentials.access_key,
-                &session_context.principal.user_identity.credentials.secret_key,
-            )
-            .await
-        {
             Ok(_) => {
                 debug!(
                     event = EVENT_FTPS_DIRECTORY_STATE,
@@ -856,15 +777,7 @@ where
             .map_err(|_| Error::new(ErrorKind::PermanentFileNotAvailable, "Access denied"))?;
 
         // Check if bucket exists
-        match self
+        match self.storage.head_bucket(&bucket, session_context.credentials()).await {
-            .storage
-            .head_bucket(
-                &bucket,
-                &session_context.principal.user_identity.credentials.access_key,
-                &session_context.principal.user_identity.credentials.secret_key,
-            )
-            .await
-        {
             Ok(_) => Ok(()),
             Err(e) => {
                 error!(

crates/protocols/src/sftp/attrs.rs

@@ -137,7 +137,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
             // on success. Size and mtime are not returned by HeadBucket.
             None => {
                 self.authorize(&S3Action::HeadBucket, &bucket, None).await?;
-                self.run_backend("head_bucket", self.storage.head_bucket(&bucket, self.access_key(), self.secret_key()))
+                self.run_backend("head_bucket", self.storage.head_bucket(&bucket, self.credentials()))
                     .await?;
                 Ok(s3_attrs_to_sftp(0, None, true))
             }
@@ -154,11 +154,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
             Some(object_key) => {
                 self.authorize(&S3Action::HeadObject, &bucket, Some(&object_key)).await?;
                 match self
-                    .run_backend_with_err(
+                    .run_backend_with_err("head_object", self.storage.head_object(&bucket, &object_key, self.credentials()))
-                        "head_object",
-                        self.storage
-                            .head_object(&bucket, &object_key, self.access_key(), self.secret_key()),
-                    )
                     .await?
                 {
                     Ok(out) => {
@@ -183,10 +179,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
                             .build()
                             .map_err(|e| s3_error_to_sftp("build_list_objects", e))?;
                         let out = self
-                            .run_backend(
+                            .run_backend("list_objects_v2", self.storage.list_objects_v2(input, self.credentials()))
-                                "list_objects_v2",
-                                self.storage.list_objects_v2(input, self.access_key(), self.secret_key()),
-                            )
                             .await?;
 
                         let has_contents = out.contents.map(|c| !c.is_empty()).unwrap_or(false);

crates/protocols/src/sftp/dir.rs

@@ -102,10 +102,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
         let input = builder.build().map_err(|e| s3_error_to_sftp("build_list_objects", e))?;
 
         let out = self
-            .run_backend(
+            .run_backend("list_objects_v2", self.storage.list_objects_v2(input, self.credentials()))
-                "list_objects_v2",
-                self.storage.list_objects_v2(input, self.access_key(), self.secret_key()),
-            )
             .await?;
 
         let mut entries = Vec::new();
@@ -196,10 +193,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
         // Issue list_objects_v2. On Err the destructive caller never
         // runs because validate_directory_empty returns the Err.
         let out = self
-            .run_backend(
+            .run_backend("list_objects_v2", self.storage.list_objects_v2(input, self.credentials()))
-                "list_objects_v2",
-                self.storage.list_objects_v2(input, self.access_key(), self.secret_key()),
-            )
             .await?;
 
         // Count content entries that are not the directory's own marker.
@@ -234,7 +228,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
         self.authorize(&S3Action::ListBuckets, "", None).await?;
 
         let out = self
-            .run_backend("list_buckets", self.storage.list_buckets(self.access_key(), self.secret_key()))
+            .run_backend("list_buckets", self.storage.list_buckets(self.credentials()))
             .await?;
 
         let mut entries = Vec::new();
@@ -280,7 +274,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
     /// MKDIR for a bucket-level path: authorise and issue CreateBucket.
     pub(super) async fn mkdir_bucket(&self, bucket: &str) -> Result<(), SftpError> {
         self.authorize(&S3Action::CreateBucket, bucket, None).await?;
-        self.run_backend("create_bucket", self.storage.create_bucket(bucket, self.access_key(), self.secret_key()))
+        self.run_backend("create_bucket", self.storage.create_bucket(bucket, self.credentials()))
             .await?;
         Ok(())
     }
@@ -302,7 +296,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
             .body(Some(streaming))
             .build()
             .map_err(|e| s3_error_to_sftp("build_put_object", e))?;
-        self.run_backend("put_object", self.storage.put_object(input, self.access_key(), self.secret_key()))
+        self.run_backend("put_object", self.storage.put_object(input, self.credentials()))
             .await?;
         Ok(())
     }
@@ -312,7 +306,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
     pub(super) async fn rmdir_bucket(&self, bucket: &str) -> Result<(), SftpError> {
         self.validate_directory_empty(bucket, "").await?;
         self.authorize(&S3Action::DeleteBucket, bucket, None).await?;
-        self.run_backend("delete_bucket", self.storage.delete_bucket(bucket, self.access_key(), self.secret_key()))
+        self.run_backend("delete_bucket", self.storage.delete_bucket(bucket, self.credentials()))
             .await?;
         Ok(())
     }
@@ -326,12 +320,8 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
 
         let marker_key = path::encode_dir_object(&prefix);
         self.authorize(&S3Action::DeleteObject, bucket, Some(&marker_key)).await?;
-        self.run_backend(
+        self.run_backend("delete_object", self.storage.delete_object(bucket, &marker_key, self.credentials()))
-            "delete_object",
+            .await?;
-            self.storage
-                .delete_object(bucket, &marker_key, self.access_key(), self.secret_key()),
-        )
-        .await?;
         Ok(())
     }
 
@@ -398,7 +388,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
                 if prefix.is_empty() { None } else { Some(prefix.as_str()) },
             )
             .await?;
-            self.run_backend("head_bucket", self.storage.head_bucket(&bucket, self.access_key(), self.secret_key()))
+            self.run_backend("head_bucket", self.storage.head_bucket(&bucket, self.credentials()))
                 .await?;
             DirCursor::Listing {
                 bucket,

crates/protocols/src/sftp/driver.rs

@@ -34,6 +34,7 @@ use crate::common::client::s3::StorageBackend;
 use crate::common::gateway::{AuthorizationError, S3Action, authorize_operation};
 use crate::common::session::SessionContext;
 use russh_sftp::protocol::{Attrs, Data, File, FileAttributes, Handle, Name, OpenFlags, Packet, Status, StatusCode, Version};
+use rustfs_credentials::Credentials;
 use rustfs_utils::MaskedAccessKey;
 use s3s::dto::{AbortMultipartUploadInput, CopyObjectInput, CopySource};
 use std::collections::HashMap;
@@ -165,16 +166,14 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
         super::read_cache::ReadCache::new(Arc::clone(&self.read_cache_in_use))
     }
 
-    /// Borrow the authenticated principal's S3 access key. Each StorageBackend
+    /// Borrow the authenticated principal's S3 access key for diagnostics.
-    /// call needs this alongside the secret key for signing.
     pub(super) fn access_key(&self) -> &str {
-        &self.session_context.principal.user_identity.credentials.access_key
+        &self.credentials().access_key
     }
 
-    /// Borrow the authenticated principal's S3 secret key. Used together with
+    /// Borrow the authenticated principal credentials for backend calls.
-    /// access_key for signing every backend call.
+    pub(super) fn credentials(&self) -> &Credentials {
-    pub(super) fn secret_key(&self) -> &str {
+        self.session_context.credentials()
-        &self.session_context.principal.user_identity.credentials.secret_key
     }
 
     /// Returns Err(PermissionDenied) when the driver is read-only,
@@ -787,12 +786,8 @@ impl<S: StorageBackend + Send + Sync + 'static> russh_sftp::server::Handler for
 
         self.authorize(&S3Action::DeleteObject, &bucket, Some(&object_key)).await?;
 
-        self.run_backend(
+        self.run_backend("delete_object", self.storage.delete_object(&bucket, &object_key, self.credentials()))
-            "delete_object",
+            .await?;
-            self.storage
-                .delete_object(&bucket, &object_key, self.access_key(), self.secret_key()),
-        )
-        .await?;
         Ok(ok_status(id))
     }
 
@@ -898,11 +893,7 @@ impl<S: StorageBackend + Send + Sync + 'static> russh_sftp::server::Handler for
         // single-shot vs multipart-copy branch below.
         self.authorize(&S3Action::HeadObject, &src_bucket, Some(&src_object)).await?;
         let head = self
-            .run_backend(
+            .run_backend("head_object", self.storage.head_object(&src_bucket, &src_object, self.credentials()))
-                "head_object",
-                self.storage
-                    .head_object(&src_bucket, &src_object, self.access_key(), self.secret_key()),
-            )
             .await?;
         let content_length = head.content_length.unwrap_or(0).max(0) as u64;
 
@@ -920,7 +911,7 @@ impl<S: StorageBackend + Send + Sync + 'static> russh_sftp::server::Handler for
                 .key(dst_object.clone())
                 .build()
                 .map_err(|e| s3_error_to_sftp("build_copy_object", e))?;
-            self.run_backend("copy_object", self.storage.copy_object(input, self.access_key(), self.secret_key()))
+            self.run_backend("copy_object", self.storage.copy_object(input, self.credentials()))
                 .await?;
         } else {
             self.multipart_copy(&src_bucket, &src_object, &dst_bucket, &dst_object, content_length)
@@ -932,12 +923,8 @@ impl<S: StorageBackend + Send + Sync + 'static> russh_sftp::server::Handler for
         // delete separately.
         self.authorize(&S3Action::DeleteObject, &src_bucket, Some(&src_object))
             .await?;
-        self.run_backend(
+        self.run_backend("delete_object", self.storage.delete_object(&src_bucket, &src_object, self.credentials()))
-            "delete_object",
+            .await?;
-            self.storage
-                .delete_object(&src_bucket, &src_object, self.access_key(), self.secret_key()),
-        )
-        .await?;
 
         Ok(ok_status(id))
     }
@@ -1029,14 +1016,12 @@ impl<S: StorageBackend + Send + Sync + 'static> russh_sftp::server::Handler for
 impl<S: StorageBackend + Send + Sync + 'static> Drop for SftpDriver<S> {
     fn drop(&mut self) {
         // Snapshot credentials, peer IP, and the per-call backend
-        // timeout before draining the handle table. self.access_key()
+        // timeout before draining the handle table. Borrowing
-        // and self.secret_key() borrow self.session_context immutably,
+        // self.session_context inside the loop would conflict with the
-        // which conflicts with the mutable borrow of self.handles
+        // mutable borrow of self.handles. The timeout is copied into each
-        // inside the loop. The timeout is copied into each spawned
+        // spawned abort task so the deadline applies uniformly to inline
-        // abort task so the deadline applies uniformly to inline calls
+        // calls and Drop-time aborts.
-        // and Drop-time aborts.
+        let credentials = self.session_context.credentials().clone();
-        let access_key = self.session_context.principal.user_identity.credentials.access_key.clone();
-        let secret_key = self.session_context.principal.user_identity.credentials.secret_key.clone();
         let peer = self.session_context.source_ip;
         let backend_op_timeout_secs = self.backend_op_timeout_secs;
 
@@ -1056,7 +1041,7 @@ impl<S: StorageBackend + Send + Sync + 'static> Drop for SftpDriver<S> {
                             key = %key,
                             upload_id = %upload_id,
                             peer = %peer,
-                            access_key = %access_key,
+                            access_key = %MaskedAccessKey(&credentials.access_key),
                             "skipped abort of orphaned multipart upload on session drop, principal lacks s3:AbortMultipartUpload, bucket lifecycle rules must reclaim parts",
                         );
                     }
@@ -1065,8 +1050,7 @@ impl<S: StorageBackend + Send + Sync + 'static> Drop for SftpDriver<S> {
             };
 
             let storage = Arc::clone(&self.storage);
-            let access_key = access_key.clone();
+            let credentials = credentials.clone();
-            let secret_key = secret_key.clone();
             let upload_id = upload_id_owned;
 
             // Cap the global abort fan-out so a burst of session
@@ -1122,7 +1106,7 @@ impl<S: StorageBackend + Send + Sync + 'static> Drop for SftpDriver<S> {
                     };
                     match tokio::time::timeout(
                         std::time::Duration::from_secs(backend_op_timeout_secs),
-                        storage.abort_multipart_upload(input, &access_key, &secret_key),
+                        storage.abort_multipart_upload(input, &credentials),
                     )
                     .await
                     {

crates/protocols/src/sftp/read.rs

@@ -46,11 +46,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
         // the body. These are cached on the handle so READ can detect EOF
         // and FSTAT can answer without another backend call.
         let head = self
-            .run_backend(
+            .run_backend("head_object", self.storage.head_object(&bucket, &object_key, self.credentials()))
-                "head_object",
-                self.storage
-                    .head_object(&bucket, &object_key, self.access_key(), self.secret_key()),
-            )
             .await?;
         let size = head.content_length.unwrap_or(0).max(0) as u64;
         let mtime = timestamp_to_mtime(head.last_modified);
@@ -166,7 +162,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
             .run_backend(
                 "get_object_range",
                 self.storage
-                    .get_object_range(bucket, key, self.access_key(), self.secret_key(), offset, fetch_len),
+                    .get_object_range(bucket, key, self.credentials(), offset, fetch_len),
             )
             .await?;
 

crates/protocols/src/sftp/write.rs

@@ -293,11 +293,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
             // not-found error means the key is free. Any other error is
             // propagated rather than misinterpreted as "does not exist".
             match self
-                .run_backend_with_err(
+                .run_backend_with_err("head_object", self.storage.head_object(&bucket, &object_key, self.credentials()))
-                    "head_object",
-                    self.storage
-                        .head_object(&bucket, &object_key, self.access_key(), self.secret_key()),
-                )
                 .await?
             {
                 Ok(_) => return Err(SftpError::code(StatusCode::Failure)),
@@ -385,7 +381,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
                 .map_err(|e| s3_error_to_sftp("build_put_object", e))?;
 
             let outcome = self
-                .run_backend_with_err("put_object", self.storage.put_object(input, self.access_key(), self.secret_key()))
+                .run_backend_with_err("put_object", self.storage.put_object(input, self.credentials()))
                 .await?;
 
             let backend_err = match outcome {
@@ -448,7 +444,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
             .map_err(|e| s3_error_to_sftp("build_upload_part", e))?;
 
         let out = self
-            .run_backend("upload_part", self.storage.upload_part(input, self.access_key(), self.secret_key()))
+            .run_backend("upload_part", self.storage.upload_part(input, self.credentials()))
             .await?;
 
         let e_tag = out.e_tag.ok_or_else(|| {
@@ -528,11 +524,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
             .map_err(|e| s3_error_to_sftp("build_create_multipart_upload", e))?;
 
         let out = self
-            .run_backend(
+            .run_backend("create_multipart_upload", self.storage.create_multipart_upload(input, self.credentials()))
-                "create_multipart_upload",
-                self.storage
-                    .create_multipart_upload(input, self.access_key(), self.secret_key()),
-            )
             .await?;
 
         let upload_id = out.upload_id.ok_or_else(|| {
@@ -585,8 +577,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
         let result = self
             .run_backend(
                 "complete_multipart_upload",
-                self.storage
+                self.storage.complete_multipart_upload(input, self.credentials()),
-                    .complete_multipart_upload(input, self.access_key(), self.secret_key()),
             )
             .await;
         result?;
@@ -852,12 +843,8 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
             .build()
             .map_err(|e| s3_error_to_sftp("build_abort_multipart_upload", e))?;
 
-        self.run_backend(
+        self.run_backend("abort_multipart_upload", self.storage.abort_multipart_upload(input, self.credentials()))
-            "abort_multipart_upload",
+            .await?;
-            self.storage
-                .abort_multipart_upload(input, self.access_key(), self.secret_key()),
-        )
-        .await?;
         Ok(())
     }
 
@@ -1066,10 +1053,7 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
                     .map_err(|e| s3_error_to_sftp("build_upload_part_copy", e))?;
 
                 let out = self
-                    .run_backend(
+                    .run_backend("upload_part_copy", self.storage.upload_part_copy(input, self.credentials()))
-                        "upload_part_copy",
-                        self.storage.upload_part_copy(input, self.access_key(), self.secret_key()),
-                    )
                     .await?;
 
                 let e_tag = out.copy_part_result.and_then(|r| r.e_tag).ok_or_else(|| {
@@ -1125,6 +1109,7 @@ mod tests {
     use crate::common::dummy_storage::{AbortCall, DummyBackend, DummyError};
     use crate::common::gateway::with_test_auth_override;
     use russh_sftp::protocol::{FileAttributes, OpenFlags, StatusCode};
+    use rustfs_credentials::Credentials;
     use s3s::dto::ETag;
     use std::sync::Arc;
     use std::time::Duration;
@@ -2324,9 +2309,10 @@ mod tests {
         let backend = Arc::new(DummyBackend::new());
         backend.queue_head_object_err(DummyError::AccessDenied("pinned".to_string()));
         let driver = build_driver(backend, TEST_PART_SIZE);
+        let credentials = Credentials::default();
 
         let result = driver
-            .run_backend_with_err("head_object", driver.storage.head_object("b", "k", "ak", "sk"))
+            .run_backend_with_err("head_object", driver.storage.head_object("b", "k", &credentials))
             .await;
 
         match result {

crates/protocols/src/webdav/driver.rs

@@ -22,6 +22,7 @@ use dav_server::fs::{
 };
 use futures_util::{FutureExt, StreamExt, stream};
 use percent_encoding::percent_decode_str;
+use rustfs_credentials::Credentials;
 use rustfs_utils::MaskedAccessKey;
 use rustfs_utils::path;
 use s3s::S3ErrorCode;
@@ -198,15 +199,7 @@ where
         let key = self.key.clone();
 
         async move {
-            match storage
+            match storage.head_object(&bucket, &key, session_context.credentials()).await {
-                .head_object(
-                    &bucket,
-                    &key,
-                    &session_context.principal.user_identity.credentials.access_key,
-                    &session_context.principal.user_identity.credentials.secret_key,
-                )
-                .await
-            {
                 Ok(output) => {
                     let size = output.content_length.unwrap_or(0) as u64;
                     let modified = output
@@ -288,14 +281,7 @@ where
         async move {
             let start_pos = *position.read().await;
             match storage
-                .get_object_range(
+                .get_object_range(&bucket, &key, session_context.credentials(), start_pos, count as u64)
-                    &bucket,
-                    &key,
-                    &session_context.principal.user_identity.credentials.access_key,
-                    &session_context.principal.user_identity.credentials.secret_key,
-                    start_pos,
-                    count as u64,
-                )
                 .await
             {
                 Ok(output) => {
@@ -407,14 +393,7 @@ where
                 .build()
                 .map_err(|_| FsError::GeneralFailure)?;
 
-            match storage
+            match storage.put_object(put_input, session_context.credentials()).await {
-                .put_object(
-                    put_input,
-                    &session_context.principal.user_identity.credentials.access_key,
-                    &session_context.principal.user_identity.credentials.secret_key,
-                )
-                .await
-            {
                 Ok(_) => {
                     debug!(
                         event = EVENT_WEBDAV_OBJECT_WRITE_STATE,
@@ -522,11 +501,8 @@ where
         self
     }
 
-    fn credentials(&self) -> (&str, &str) {
+    fn credentials(&self) -> &Credentials {
-        (
+        self.session_context.credentials()
-            &self.session_context.principal.user_identity.credentials.access_key,
-            &self.session_context.principal.user_identity.credentials.secret_key,
-        )
     }
 
     fn is_missing_head_object_error(error: &str) -> bool {
@@ -538,7 +514,7 @@ where
     }
 
     async fn prefix_has_entries(&self, bucket: &str, prefix: &str) -> FsResult<bool> {
-        let (access_key, secret_key) = self.credentials();
+        let credentials = self.credentials();
         let list_input = ListObjectsV2Input::builder()
             .bucket(bucket.to_string())
             .prefix(Some(prefix.to_string()))
@@ -546,32 +522,28 @@ where
             .build()
             .map_err(|_| FsError::GeneralFailure)?;
 
-        let output = self
+        let output = self.storage.list_objects_v2(list_input, credentials).await.map_err(|e| {
-            .storage
+            error!(
-            .list_objects_v2(list_input, access_key, secret_key)
+                event = EVENT_WEBDAV_LIST_FAILED,
-            .await
+                component = LOG_COMPONENT_PROTOCOLS,
-            .map_err(|e| {
+                subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
-                error!(
+                bucket = %bucket,
-                    event = EVENT_WEBDAV_LIST_FAILED,
+                prefix = %prefix,
-                    component = LOG_COMPONENT_PROTOCOLS,
+                error = %e,
-                    subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
+                "webdav list failed"
-                    bucket = %bucket,
+            );
-                    prefix = %prefix,
+            FsError::GeneralFailure
-                    error = %e,
+        })?;
-                    "webdav list failed"
-                );
-                FsError::GeneralFailure
-            })?;
 
         Ok(output.contents.map(|c| !c.is_empty()).unwrap_or(false)
             || output.common_prefixes.map(|c| !c.is_empty()).unwrap_or(false))
     }
 
     async fn copy_object_streaming(&self, src_bucket: &str, src_key: &str, dst_bucket: &str, dst_key: &str) -> FsResult<()> {
-        let (access_key, secret_key) = self.credentials();
+        let credentials = self.credentials();
         let get_output = self
             .storage
-            .get_object(src_bucket, src_key, access_key, secret_key, None)
+            .get_object(src_bucket, src_key, credentials, None)
             .await
             .map_err(|e| {
                 error!(
@@ -625,24 +597,21 @@ where
 
         let put_input = put_builder.build().map_err(|_| FsError::GeneralFailure)?;
 
-        self.storage
+        self.storage.put_object(put_input, credentials).await.map_err(|e| {
-            .put_object(put_input, access_key, secret_key)
+            error!(
-            .await
+                event = EVENT_WEBDAV_COPY_FAILED,
-            .map_err(|e| {
+                component = LOG_COMPONENT_PROTOCOLS,
-                error!(
+                subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
-                    event = EVENT_WEBDAV_COPY_FAILED,
+                state = "destination_write_failed",
-                    component = LOG_COMPONENT_PROTOCOLS,
+                src_bucket = %src_bucket,
-                    subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
+                src_object = %src_key,
-                    state = "destination_write_failed",
+                dst_bucket = %dst_bucket,
-                    src_bucket = %src_bucket,
+                dst_object = %dst_key,
-                    src_object = %src_key,
+                error = %e,
-                    dst_bucket = %dst_bucket,
+                "webdav copy failed"
-                    dst_object = %dst_key,
+            );
-                    error = %e,
+            FsError::GeneralFailure
-                    "webdav copy failed"
+        })?;
-                );
-                FsError::GeneralFailure
-            })?;
 
         Ok(())
     }
@@ -653,7 +622,7 @@ where
         dst_bucket: &str,
         rename_pairs: &[(String, String)],
     ) -> FsResult<()> {
-        let (access_key, secret_key) = self.credentials();
+        let credentials = self.credentials();
 
         for (src_obj_key, dst_obj_key) in rename_pairs {
             self.copy_object_streaming(src_bucket, src_obj_key, dst_bucket, dst_obj_key)
@@ -662,7 +631,7 @@ where
 
         for (src_obj_key, _) in rename_pairs {
             self.storage
-                .delete_object(src_bucket, src_obj_key, access_key, secret_key)
+                .delete_object(src_bucket, src_obj_key, credentials)
                 .await
                 .map_err(|e| {
                     error!(
@@ -683,7 +652,7 @@ where
     }
 
     async fn probe_head_object(&self, bucket: &str, key: &str) -> FsResult<HeadObjectProbe> {
-        let (access_key, secret_key) = self.credentials();
+        let credentials = self.credentials();
 
         if authorize_operation(&self.session_context, &S3Action::HeadObject, bucket, Some(key))
             .await
@@ -692,7 +661,7 @@ where
             return Ok(HeadObjectProbe::Forbidden);
         }
 
-        match self.storage.head_object(bucket, key, access_key, secret_key).await {
+        match self.storage.head_object(bucket, key, credentials).await {
             Ok(output) => Ok(HeadObjectProbe::Found(Box::new(output))),
             Err(e) => {
                 let err_msg = e.to_string();
@@ -816,8 +785,8 @@ where
     async fn list_buckets(&self) -> FsResult<Vec<WebDavDirEntry>> {
         match authorize_operation(&self.session_context, &S3Action::ListBuckets, "", None).await {
             Ok(()) => {
-                let (access_key, secret_key) = self.credentials();
+                let credentials = self.credentials();
-                return match self.storage.list_buckets(access_key, secret_key).await {
+                return match self.storage.list_buckets(credentials).await {
                     Ok(output) => Ok(Self::bucket_entries(output)),
                     Err(error) => {
                         error!(
@@ -825,7 +794,7 @@ where
                             component = LOG_COMPONENT_PROTOCOLS,
                             subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
                             error = %error,
-                            access_key = %MaskedAccessKey(access_key),
+                            access_key = %MaskedAccessKey(credentials.access_key.as_str()),
                             "webdav bucket list failed"
                         );
                         Err(FsError::GeneralFailure)
@@ -908,15 +877,7 @@ where
             .build()
             .map_err(|_| FsError::GeneralFailure)?;
 
-        match self
+        match self.storage.list_objects_v2(list_input, self.credentials()).await {
-            .storage
-            .list_objects_v2(
-                list_input,
-                &self.session_context.principal.user_identity.credentials.access_key,
-                &self.session_context.principal.user_identity.credentials.secret_key,
-            )
-            .await
-        {
             Ok(output) => {
                 let mut entries = Vec::new();
 
@@ -1054,15 +1015,7 @@ where
 
             let list_input = list_input.build().map_err(|_| FsError::GeneralFailure)?;
 
-            if let Ok(output) = self
+            if let Ok(output) = self.storage.list_objects_v2(list_input, self.credentials()).await {
-                .storage
-                .list_objects_v2(
-                    list_input,
-                    &self.session_context.principal.user_identity.credentials.access_key,
-                    &self.session_context.principal.user_identity.credentials.secret_key,
-                )
-                .await
-            {
                 // Delete all objects in this page
                 if let Some(objects) = output.contents {
                     for obj in objects {
@@ -1071,15 +1024,7 @@ where
                                 .await
                                 .map_err(|_| FsError::Forbidden)?;
 
-                            let _ = self
+                            let _ = self.storage.delete_object(bucket, &obj_key, self.credentials()).await;
-                                .storage
-                                .delete_object(
-                                    bucket,
-                                    &obj_key,
-                                    &self.session_context.principal.user_identity.credentials.access_key,
-                                    &self.session_context.principal.user_identity.credentials.secret_key,
-                                )
-                                .await;
                         }
                     }
                 }
@@ -1095,15 +1040,7 @@ where
         }
 
         // Then delete the bucket
-        match self
+        match self.storage.delete_bucket(bucket, self.credentials()).await {
-            .storage
-            .delete_bucket(
-                bucket,
-                &self.session_context.principal.user_identity.credentials.access_key,
-                &self.session_context.principal.user_identity.credentials.secret_key,
-            )
-            .await
-        {
             Ok(_) => Ok(()),
             Err(e) if e.to_string().contains("NoSuchBucket") => Ok(()),
             Err(e) => {
@@ -1250,15 +1187,7 @@ where
                     .await
                     .map_err(|_| FsError::Forbidden)?;
 
-                match self
+                match self.storage.head_bucket(&bucket, self.credentials()).await {
-                    .storage
-                    .head_bucket(
-                        &bucket,
-                        &self.session_context.principal.user_identity.credentials.access_key,
-                        &self.session_context.principal.user_identity.credentials.secret_key,
-                    )
-                    .await
-                {
                     Ok(_) => Ok(Box::new(WebDavMetaData {
                         size: 0,
                         modified: SystemTime::now(),
@@ -1318,15 +1247,7 @@ where
                     .build()
                     .map_err(|_| FsError::GeneralFailure)?;
 
-                match self
+                match self.storage.put_object(put_input, self.credentials()).await {
-                    .storage
-                    .put_object(
-                        put_input,
-                        &self.session_context.principal.user_identity.credentials.access_key,
-                        &self.session_context.principal.user_identity.credentials.secret_key,
-                    )
-                    .await
-                {
                     Ok(_) => {
                         debug!(
                             event = EVENT_WEBDAV_DIRECTORY_STATE,
@@ -1360,15 +1281,7 @@ where
                 .await
                 .map_err(|_| FsError::Forbidden)?;
 
-            match self
+            match self.storage.create_bucket(&bucket, self.credentials()).await {
-                .storage
-                .create_bucket(
-                    &bucket,
-                    &self.session_context.principal.user_identity.credentials.access_key,
-                    &self.session_context.principal.user_identity.credentials.secret_key,
-                )
-                .await
-            {
                 Ok(_) => {
                     debug!(
                         event = EVENT_WEBDAV_DIRECTORY_STATE,
@@ -1438,15 +1351,7 @@ where
 
                     let list_input = list_input.build().map_err(|_| FsError::GeneralFailure)?;
 
-                    if let Ok(output) = self
+                    if let Ok(output) = self.storage.list_objects_v2(list_input, self.credentials()).await {
-                        .storage
-                        .list_objects_v2(
-                            list_input,
-                            &self.session_context.principal.user_identity.credentials.access_key,
-                            &self.session_context.principal.user_identity.credentials.secret_key,
-                        )
-                        .await
-                    {
                         if let Some(objects) = output.contents {
                             for obj in objects {
                                 if let Some(obj_key) = obj.key {
@@ -1454,15 +1359,7 @@ where
                                         .await
                                         .map_err(|_| FsError::Forbidden)?;
 
-                                    let _ = self
+                                    let _ = self.storage.delete_object(&bucket, &obj_key, self.credentials()).await;
-                                        .storage
-                                        .delete_object(
-                                            &bucket,
-                                            &obj_key,
-                                            &self.session_context.principal.user_identity.credentials.access_key,
-                                            &self.session_context.principal.user_identity.credentials.secret_key,
-                                        )
-                                        .await;
                                 }
                             }
                         }
@@ -1479,12 +1376,7 @@ where
                 // Also delete the directory marker itself
                 let _ = self
                     .storage
-                    .delete_object(
+                    .delete_object(&bucket, &prefix_with_slash, self.credentials())
-                        &bucket,
-                        &prefix_with_slash,
-                        &self.session_context.principal.user_identity.credentials.access_key,
-                        &self.session_context.principal.user_identity.credentials.secret_key,
-                    )
                     .await;
 
                 return Ok(());
@@ -1515,16 +1407,7 @@ where
                 .await
                 .map_err(|_| FsError::Forbidden)?;
 
-            match self
+            match self.storage.delete_object(&bucket, &key, self.credentials()).await {
-                .storage
-                .delete_object(
-                    &bucket,
-                    &key,
-                    &self.session_context.principal.user_identity.credentials.access_key,
-                    &self.session_context.principal.user_identity.credentials.secret_key,
-                )
-                .await
-            {
                 Ok(_) => {
                     debug!(
                         event = EVENT_WEBDAV_OBJECT_DELETE_STATE,
@@ -1566,7 +1449,7 @@ where
 
             let src_key = src_key.ok_or(FsError::Forbidden)?;
             let dst_key = dst_key.ok_or(FsError::Forbidden)?;
-            let (access_key, secret_key) = self.credentials();
+            let credentials = self.credentials();
             let resolved_src = self.resolve_path(&src_bucket, &src_key).await?;
             let (src_prefix, include_src_marker) = match resolved_src {
                 ResolvedPath::File(_) => {
@@ -1584,7 +1467,7 @@ where
                         .await?;
 
                     self.storage
-                        .delete_object(&src_bucket, &src_key, access_key, secret_key)
+                        .delete_object(&src_bucket, &src_key, credentials)
                         .await
                         .map_err(|e| {
                             error!(
@@ -1656,25 +1539,21 @@ where
                 }
 
                 let list_input = list_builder.build().map_err(|_| FsError::GeneralFailure)?;
-                let output = self
+                let output = self.storage.list_objects_v2(list_input, credentials).await.map_err(|e| {
-                    .storage
+                    error!(
-                    .list_objects_v2(list_input, access_key, secret_key)
+                        event = EVENT_WEBDAV_RENAME_STATE,
-                    .await
+                        component = LOG_COMPONENT_PROTOCOLS,
-                    .map_err(|e| {
+                        subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
-                        error!(
+                        state = "directory_list_failed",
-                            event = EVENT_WEBDAV_RENAME_STATE,
+                        src_bucket = %src_bucket,
-                            component = LOG_COMPONENT_PROTOCOLS,
+                        src_prefix = %src_prefix,
-                            subsystem = LOG_SUBSYSTEM_WEBDAV_DRIVER,
+                        dst_bucket = %dst_bucket,
-                            state = "directory_list_failed",
+                        dst_prefix = %dst_prefix,
-                            src_bucket = %src_bucket,
+                        error = %e,
-                            src_prefix = %src_prefix,
+                        "WebDAV rename directory listing failed"
-                            dst_bucket = %dst_bucket,
+                    );
-                            dst_prefix = %dst_prefix,
+                    FsError::GeneralFailure
-                            error = %e,
+                })?;
-                            "WebDAV rename directory listing failed"
-                        );
-                        FsError::GeneralFailure
-                    })?;
 
                 let mut page_pairs: Vec<(String, String)> = Vec::new();
                 if let Some(objects) = output.contents {
@@ -1785,8 +1664,7 @@ mod tests {
             &self,
             _bucket: &str,
             _key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
             _start_pos: Option<u64>,
         ) -> Result<GetObjectOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
@@ -1796,20 +1674,14 @@ mod tests {
             &self,
             _bucket: &str,
             _key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
             _start_pos: u64,
             _length: u64,
         ) -> Result<GetObjectOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
 
-        async fn put_object(
+        async fn put_object(&self, _input: PutObjectInput, _credentials: &Credentials) -> Result<PutObjectOutput, Self::Error> {
-            &self,
-            _input: PutObjectInput,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<PutObjectOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
 
@@ -1817,8 +1689,7 @@ mod tests {
             &self,
             _bucket: &str,
             _key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<DeleteObjectOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
@@ -1827,57 +1698,39 @@ mod tests {
             &self,
             _bucket: &str,
             _key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<HeadObjectOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
 
-        async fn head_bucket(
+        async fn head_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result<HeadBucketOutput, Self::Error> {
-            &self,
-            _bucket: &str,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<HeadBucketOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
 
         async fn list_objects_v2(
             &self,
             _input: ListObjectsV2Input,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<ListObjectsV2Output, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
 
-        async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result<ListBucketsOutput, Self::Error> {
+        async fn list_buckets(&self, _credentials: &Credentials) -> Result<ListBucketsOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
 
-        async fn create_bucket(
+        async fn create_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result<CreateBucketOutput, Self::Error> {
-            &self,
-            _bucket: &str,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<CreateBucketOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
 
-        async fn delete_bucket(
+        async fn delete_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result<DeleteBucketOutput, Self::Error> {
-            &self,
-            _bucket: &str,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<DeleteBucketOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
 
         async fn copy_object(
             &self,
             _input: CopyObjectInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<CopyObjectOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
@@ -1885,8 +1738,7 @@ mod tests {
         async fn create_multipart_upload(
             &self,
             _input: CreateMultipartUploadInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<CreateMultipartUploadOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
@@ -1894,8 +1746,7 @@ mod tests {
         async fn upload_part(
             &self,
             _input: UploadPartInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<UploadPartOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
@@ -1903,8 +1754,7 @@ mod tests {
         async fn complete_multipart_upload(
             &self,
             _input: CompleteMultipartUploadInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<CompleteMultipartUploadOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
@@ -1912,8 +1762,7 @@ mod tests {
         async fn abort_multipart_upload(
             &self,
             _input: AbortMultipartUploadInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<AbortMultipartUploadOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
@@ -1921,8 +1770,7 @@ mod tests {
         async fn upload_part_copy(
             &self,
             _input: UploadPartCopyInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<UploadPartCopyOutput, Self::Error> {
             unreachable!("parse_path tests should not hit storage")
         }
@@ -2099,8 +1947,7 @@ mod tests {
             &self,
             bucket: &str,
             key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
             _start_pos: Option<u64>,
         ) -> Result<GetObjectOutput, Self::Error> {
             let data = self
@@ -2127,8 +1974,7 @@ mod tests {
             &self,
             _bucket: &str,
             _key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
             _start_pos: u64,
             _length: u64,
         ) -> Result<GetObjectOutput, Self::Error> {
@@ -2138,8 +1984,7 @@ mod tests {
         async fn put_object(
             &self,
             mut input: PutObjectInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<PutObjectOutput, Self::Error> {
             let bucket = input.bucket.clone();
             let key = input.key.clone();
@@ -2163,8 +2008,7 @@ mod tests {
             &self,
             bucket: &str,
             key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<DeleteObjectOutput, Self::Error> {
             let mut state = self.state.lock().expect("recording storage lock poisoned");
             state.delete_keys.push(key.to_string());
@@ -2179,26 +2023,19 @@ mod tests {
             &self,
             _bucket: &str,
             _key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<HeadObjectOutput, Self::Error> {
             unreachable!("head_object is not used in rename regression tests")
         }
 
-        async fn head_bucket(
+        async fn head_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result<HeadBucketOutput, Self::Error> {
-            &self,
-            _bucket: &str,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<HeadBucketOutput, Self::Error> {
             unreachable!("head_bucket is not used in rename regression tests")
         }
 
         async fn list_objects_v2(
             &self,
             input: ListObjectsV2Input,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<ListObjectsV2Output, Self::Error> {
             let prefix = input.prefix.unwrap_or_default();
             let mut keys: Vec<String> = self
@@ -2226,25 +2063,15 @@ mod tests {
             })
         }
 
-        async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result<ListBucketsOutput, Self::Error> {
+        async fn list_buckets(&self, _credentials: &Credentials) -> Result<ListBucketsOutput, Self::Error> {
             unreachable!("list_buckets is not used in rename regression tests")
         }
 
-        async fn create_bucket(
+        async fn create_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result<CreateBucketOutput, Self::Error> {
-            &self,
-            _bucket: &str,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<CreateBucketOutput, Self::Error> {
             unreachable!("create_bucket is not used in rename regression tests")
         }
 
-        async fn delete_bucket(
+        async fn delete_bucket(&self, bucket: &str, _credentials: &Credentials) -> Result<DeleteBucketOutput, Self::Error> {
-            &self,
-            bucket: &str,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<DeleteBucketOutput, Self::Error> {
             self.state
                 .lock()
                 .expect("recording storage lock poisoned")
@@ -2256,8 +2083,7 @@ mod tests {
         async fn copy_object(
             &self,
             _input: CopyObjectInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<CopyObjectOutput, Self::Error> {
             unreachable!("copy_object is not used in rename regression tests")
         }
@@ -2265,8 +2091,7 @@ mod tests {
         async fn create_multipart_upload(
             &self,
             _input: CreateMultipartUploadInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<CreateMultipartUploadOutput, Self::Error> {
             unreachable!("create_multipart_upload is not used in rename regression tests")
         }
@@ -2274,8 +2099,7 @@ mod tests {
         async fn upload_part(
             &self,
             _input: UploadPartInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<UploadPartOutput, Self::Error> {
             unreachable!("upload_part is not used in rename regression tests")
         }
@@ -2283,8 +2107,7 @@ mod tests {
         async fn complete_multipart_upload(
             &self,
             _input: CompleteMultipartUploadInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<CompleteMultipartUploadOutput, Self::Error> {
             unreachable!("complete_multipart_upload is not used in rename regression tests")
         }
@@ -2292,8 +2115,7 @@ mod tests {
         async fn abort_multipart_upload(
             &self,
             _input: AbortMultipartUploadInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<AbortMultipartUploadOutput, Self::Error> {
             unreachable!("abort_multipart_upload is not used in rename regression tests")
         }
@@ -2301,8 +2123,7 @@ mod tests {
         async fn upload_part_copy(
             &self,
             _input: UploadPartCopyInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<UploadPartCopyOutput, Self::Error> {
             unreachable!("upload_part_copy is not used in rename regression tests")
         }

crates/protocols/src/webdav/server.rs

@@ -687,6 +687,7 @@ mod tests {
     use futures_util::stream;
     use http_body_util::StreamBody;
     use hyper::body::Frame;
+    use rustfs_credentials::Credentials;
     use s3s::dto::*;
     use std::fmt::{Debug, Formatter};
     use std::net::{Ipv4Addr, SocketAddr};
@@ -715,8 +716,7 @@ mod tests {
             &self,
             _bucket: &str,
             _key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
             _start_pos: Option<u64>,
         ) -> Result<GetObjectOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
@@ -726,20 +726,14 @@ mod tests {
             &self,
             _bucket: &str,
             _key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
             _start_pos: u64,
             _length: u64,
         ) -> Result<GetObjectOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
 
-        async fn put_object(
+        async fn put_object(&self, _input: PutObjectInput, _credentials: &Credentials) -> Result<PutObjectOutput, Self::Error> {
-            &self,
-            _input: PutObjectInput,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<PutObjectOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
 
@@ -747,8 +741,7 @@ mod tests {
             &self,
             _bucket: &str,
             _key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<DeleteObjectOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
@@ -757,57 +750,39 @@ mod tests {
             &self,
             _bucket: &str,
             _key: &str,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<HeadObjectOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
 
-        async fn head_bucket(
+        async fn head_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result<HeadBucketOutput, Self::Error> {
-            &self,
-            _bucket: &str,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<HeadBucketOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
 
         async fn list_objects_v2(
             &self,
             _input: ListObjectsV2Input,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<ListObjectsV2Output, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
 
-        async fn list_buckets(&self, _access_key: &str, _secret_key: &str) -> Result<ListBucketsOutput, Self::Error> {
+        async fn list_buckets(&self, _credentials: &Credentials) -> Result<ListBucketsOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
 
-        async fn create_bucket(
+        async fn create_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result<CreateBucketOutput, Self::Error> {
-            &self,
-            _bucket: &str,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<CreateBucketOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
 
-        async fn delete_bucket(
+        async fn delete_bucket(&self, _bucket: &str, _credentials: &Credentials) -> Result<DeleteBucketOutput, Self::Error> {
-            &self,
-            _bucket: &str,
-            _access_key: &str,
-            _secret_key: &str,
-        ) -> Result<DeleteBucketOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
 
         async fn copy_object(
             &self,
             _input: CopyObjectInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<CopyObjectOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
@@ -815,8 +790,7 @@ mod tests {
         async fn create_multipart_upload(
             &self,
             _input: CreateMultipartUploadInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<CreateMultipartUploadOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
@@ -824,8 +798,7 @@ mod tests {
         async fn upload_part(
             &self,
             _input: UploadPartInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<UploadPartOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
@@ -833,8 +806,7 @@ mod tests {
         async fn complete_multipart_upload(
             &self,
             _input: CompleteMultipartUploadInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<CompleteMultipartUploadOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
@@ -842,8 +814,7 @@ mod tests {
         async fn abort_multipart_upload(
             &self,
             _input: AbortMultipartUploadInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<AbortMultipartUploadOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }
@@ -851,8 +822,7 @@ mod tests {
         async fn upload_part_copy(
             &self,
             _input: UploadPartCopyInput,
-            _access_key: &str,
+            _credentials: &Credentials,
-            _secret_key: &str,
         ) -> Result<UploadPartCopyOutput, Self::Error> {
             unreachable!("connection tests should not hit storage")
         }

crates/utils/src/http/metadata_compat.rs

@@ -88,6 +88,21 @@ pub const SUFFIX_TIER_SKIP_FV_ID: &str = "tier-skip-fvid";
 /// Per-target delete-marker version ids are stored one key per target ARN.
 pub const SUFFIX_REPLICATION_DELETE_MARKER_VERSION_ARN_PREFIX: &str = "replication-delete-marker-version-";
 
+// On-demand migration provenance. Written by the migration write-back onto
+// every pulled object so operators and later tooling can tell a migrated
+// object from a client write and audit where it came from. Internal keys, so
+// the existing internal-key filter keeps them out of client-visible metadata.
+/// Source of a pulled object as `<provider>:<bucket>`.
+pub const SUFFIX_ODM_SOURCE: &str = "odm-source";
+/// ETag the source reported for the pulled object, verbatim.
+pub const SUFFIX_ODM_SOURCE_ETAG: &str = "odm-source-etag";
+/// Source `Last-Modified` of the pulled object, RFC 3339.
+pub const SUFFIX_ODM_SOURCE_LAST_MODIFIED: &str = "odm-source-last-modified";
+/// Source version id of the pulled object; empty for an unversioned source.
+pub const SUFFIX_ODM_SOURCE_VERSION_ID: &str = "odm-source-version-id";
+/// When the object was pulled from the source, RFC 3339.
+pub const SUFFIX_ODM_PULLED_AT: &str = "odm-pulled-at";
+
 /// Case-insensitive (ASCII) check that `s` begins with `prefix`. Equivalent to
 /// `s.to_lowercase().starts_with(prefix)` when `prefix` is ASCII (as both internal prefixes are),
 /// but without allocating.
@@ -591,6 +606,80 @@ mod tests {
         assert!(!contains_key_bytes(&meta_sys, &long_suffix));
     }
 
+    const ODM_PROVENANCE_SUFFIXES: [&str; 5] = [
+        SUFFIX_ODM_SOURCE,
+        SUFFIX_ODM_SOURCE_ETAG,
+        SUFFIX_ODM_SOURCE_LAST_MODIFIED,
+        SUFFIX_ODM_SOURCE_VERSION_ID,
+        SUFFIX_ODM_PULLED_AT,
+    ];
+
+    #[test]
+    fn odm_provenance_suffixes_are_reserved_internal_keys_under_both_prefixes() {
+        for suffix in ODM_PROVENANCE_SUFFIXES {
+            for prefix in [RUSTFS_INTERNAL_PREFIX, MINIO_INTERNAL_PREFIX] {
+                let key = format!("{prefix}{suffix}");
+                assert!(is_internal_key(&key), "{key} must be an internal key");
+                assert!(has_internal_suffix(&key, suffix), "{key} must match its own suffix");
+                assert!(has_internal_suffix(&key.to_uppercase(), suffix), "{key} must match case-insensitively");
+                assert_eq!(strip_internal_prefix(&key).as_deref(), Some(suffix));
+            }
+            assert!(
+                !is_internal_key(&format!("x-amz-meta-{suffix}")),
+                "{suffix} must not leak as user metadata"
+            );
+            assert!(suffix.starts_with("odm-"), "{suffix} must stay in the odm- namespace");
+            assert!(
+                suffix.bytes().all(|b| b.is_ascii_lowercase() || b == b'-'),
+                "{suffix} must be lowercase-hyphenated"
+            );
+        }
+        let distinct: std::collections::HashSet<&str> = ODM_PROVENANCE_SUFFIXES.into_iter().collect();
+        assert_eq!(distinct.len(), ODM_PROVENANCE_SUFFIXES.len(), "provenance suffixes must be distinct");
+    }
+
+    #[test]
+    fn odm_provenance_values_round_trip_under_both_prefixes() {
+        let mut metadata = HashMap::new();
+        insert_str(&mut metadata, SUFFIX_ODM_SOURCE, "s3:legacy-bucket".to_string());
+        insert_str(&mut metadata, SUFFIX_ODM_SOURCE_ETAG, "0123456789abcdef0123456789abcdef-3".to_string());
+        insert_str(&mut metadata, SUFFIX_ODM_SOURCE_LAST_MODIFIED, "2026-01-02T03:04:05Z".to_string());
+        insert_str(&mut metadata, SUFFIX_ODM_SOURCE_VERSION_ID, String::new());
+        insert_str(&mut metadata, SUFFIX_ODM_PULLED_AT, "2026-09-02T00:00:00Z".to_string());
+
+        assert_eq!(
+            metadata.len(),
+            2 * ODM_PROVENANCE_SUFFIXES.len(),
+            "every marker is written under both prefixes"
+        );
+        for suffix in ODM_PROVENANCE_SUFFIXES {
+            assert!(metadata.contains_key(&internal_key_rustfs(suffix)), "missing RustFS key for {suffix}");
+            assert!(
+                metadata.contains_key(&format!("{MINIO_INTERNAL_PREFIX}{suffix}")),
+                "missing MinIO key for {suffix}"
+            );
+            assert!(contains_key_str(&metadata, suffix));
+        }
+        assert_eq!(get_str(&metadata, SUFFIX_ODM_SOURCE).as_deref(), Some("s3:legacy-bucket"));
+        assert_eq!(get_str(&metadata, SUFFIX_ODM_SOURCE_VERSION_ID).as_deref(), Some(""));
+        assert_eq!(
+            get_consistent_str(&metadata, SUFFIX_ODM_SOURCE_ETAG),
+            Some("0123456789abcdef0123456789abcdef-3")
+        );
+
+        // A MinIO-only reader must still find the marker.
+        let minio_only: HashMap<String, String> = metadata
+            .iter()
+            .filter(|(key, _)| starts_with_ignore_ascii_case(key, MINIO_INTERNAL_PREFIX))
+            .map(|(key, value)| (key.clone(), value.clone()))
+            .collect();
+        assert_eq!(get_str(&minio_only, SUFFIX_ODM_PULLED_AT).as_deref(), Some("2026-09-02T00:00:00Z"));
+
+        remove_str(&mut metadata, SUFFIX_ODM_SOURCE);
+        assert!(!contains_key_str(&metadata, SUFFIX_ODM_SOURCE));
+        assert!(contains_key_str(&metadata, SUFFIX_ODM_SOURCE_ETAG));
+    }
+
     #[test]
     fn target_delete_marker_versions_preserve_arn_case_and_report_conflicts() {
         let arn = "arn:rustfs:replication::Target:Bucket";

rustfs/src/app/object/internal_put.rs

@@ -0,0 +1,968 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Internal object write entry points for trusted in-process callers.
+//!
+//! A system write (on-demand migration write-back, future replays) must look
+//! like an ordinary client write: bucket default SSE, quota, versioning,
+//! Object Lock defaults, replication scheduling and creation events all apply.
+//! The single-object entry point runs the same [`DefaultObjectUsecase::put_object_core`]
+//! as the S3 PutObject handler; the multipart entry points mirror the S3
+//! multipart handlers' policy steps against the same storage contract.
+
+use super::*;
+
+use crate::app::object_data_cache::invalidate_object_data_cache_after_complete_multipart_success;
+use crate::app::storage_api::multipart_usecase::contract::multipart::{CompletePart, MultipartOperations as _};
+use crate::app::storage_api::object_usecase::compression::is_multipart_disk_compression_enabled;
+use crate::app::storage_api::object_usecase::io::WriteEncryption;
+use crate::app::storage_api::object_usecase::options::{
+    extract_metadata_from_mime, get_complete_multipart_upload_opts_with_replication_authorization,
+};
+use crate::app::storage_api::object_usecase::sse::{
+    EncryptionKeyKind, PrepareEncryptionRequest, mark_encrypted_multipart_metadata, sse_decryption, sse_prepare_encryption,
+};
+use crate::capacity::record_capacity_write;
+use crate::runtime_sources::NotifyInterface;
+use http::HeaderName;
+
+/// Inputs of an internal object write. Content and user metadata follow the
+/// S3 request shape so the shared write path treats them exactly like a
+/// client PUT: `content_headers` are the standard object headers
+/// (`Content-Type`, `Cache-Control`, `Content-Encoding`, `Content-Disposition`,
+/// `Content-Language`, `Expires`), `user_metadata` carries `x-amz-meta-*`
+/// entries with the prefix stripped, `tags` is the `x-amz-tagging` query
+/// string and `internal_metadata` holds `x-rustfs-internal-*` /
+/// `x-minio-internal-*` keys written verbatim.
+pub(crate) struct InternalPutContext {
+    pub(crate) bucket: String,
+    pub(crate) key: String,
+    /// Plaintext object length. The single-object path requires it, exactly
+    /// like S3 PutObject rejects an unknown `Content-Length`.
+    pub(crate) size: Option<u64>,
+    /// Lowercase hex MD5 the body must hash to; the write fails with
+    /// `BadDigest` otherwise and nothing is committed.
+    pub(crate) expected_md5_hex: Option<String>,
+    /// ETag to store instead of the computed one.
+    pub(crate) preserve_etag: Option<String>,
+    pub(crate) content_headers: HashMap<String, String>,
+    pub(crate) user_metadata: HashMap<String, String>,
+    pub(crate) tags: Option<String>,
+    pub(crate) internal_metadata: HashMap<String, String>,
+    /// Publish the `s3:ObjectCreated:*` event for the write.
+    pub(crate) emit_events: bool,
+    /// `userIdentity.principalId` of the creation event.
+    pub(crate) principal_id: &'static str,
+}
+
+const INTERNAL_PUT_METHOD_NAME: &str = "PUT";
+
+fn api_error_from_s3(err: S3Error) -> ApiError {
+    ApiError {
+        code: err.code().clone(),
+        message: err.message().unwrap_or_default().to_string(),
+        source: None,
+    }
+}
+
+fn not_initialized() -> ApiError {
+    ApiError {
+        code: S3ErrorCode::InternalError,
+        message: "Not init".to_string(),
+        source: None,
+    }
+}
+
+/// Build the header view of an internal write so header-driven policy
+/// (content-type detection, compressibility, standard metadata capture) runs
+/// unchanged.
+fn internal_put_headers(content_headers: &HashMap<String, String>) -> Result<HeaderMap, ApiError> {
+    let mut headers = HeaderMap::with_capacity(content_headers.len());
+    for (name, value) in content_headers {
+        let name = HeaderName::from_bytes(name.as_bytes())
+            .map_err(|err| ApiError::invalid_request(format!("invalid content header name {name:?}: {err}")))?;
+        let value = HeaderValue::from_str(value)
+            .map_err(|err| ApiError::invalid_request(format!("invalid content header value for {name}: {err}")))?;
+        headers.insert(name, value);
+    }
+    Ok(headers)
+}
+
+fn header_string(headers: &HeaderMap, name: &str) -> Option<String> {
+    headers.get(name).and_then(|value| value.to_str().ok()).map(ToOwned::to_owned)
+}
+
+fn internal_put_content_input(headers: &HeaderMap, tags: Option<String>) -> PutObjectContentInput {
+    PutObjectContentInput {
+        cache_control: header_string(headers, "cache-control"),
+        content_disposition: header_string(headers, "content-disposition"),
+        content_encoding: header_string(headers, "content-encoding"),
+        content_language: header_string(headers, "content-language"),
+        content_type: header_string(headers, "content-type"),
+        expires: header_string(headers, "expires"),
+        website_redirect_location: None,
+        tagging: tags,
+        storage_class: None,
+    }
+}
+
+async fn validate_internal_write_target(key: &str, bucket: &str, headers: &HeaderMap) -> Result<(), ApiError> {
+    validate_object_key(key, INTERNAL_PUT_METHOD_NAME).map_err(api_error_from_s3)?;
+    validate_table_catalog_object_mutation(bucket, key)
+        .await
+        .map_err(api_error_from_s3)?;
+    validate_archive_content_encoding(
+        key,
+        headers.get("content-type").and_then(|value| value.to_str().ok()),
+        headers.get("content-encoding").and_then(|value| value.to_str().ok()),
+    )
+    .map_err(api_error_from_s3)
+}
+
+fn internal_events_wanted() -> bool {
+    crate::module_switches::is_notify_module_enabled()
+        || rustfs_notify::notification_system().is_some_and(|system| system.has_live_listeners())
+}
+
+/// Creation event of an internal write, published on successful completion
+/// the way [`OperationHelper`] publishes it for an S3 request.
+pub(super) struct InternalPutObjectEvent {
+    builder: EventArgsBuilder,
+    notify: Arc<dyn NotifyInterface>,
+    request_context: request_context::RequestContext,
+}
+
+impl InternalPutObjectEvent {
+    /// `None` when neither the notify module nor a live listener wants events.
+    pub(super) fn new(
+        notify: Arc<dyn NotifyInterface>,
+        request_context: request_context::RequestContext,
+        event_name: EventName,
+        bucket: &str,
+        key: &str,
+        principal_id: &'static str,
+    ) -> Option<Self> {
+        if !internal_events_wanted() {
+            return None;
+        }
+        Some(Self {
+            builder: Self::builder(event_name, bucket, key, principal_id),
+            notify,
+            request_context,
+        })
+    }
+
+    fn builder(event_name: EventName, bucket: &str, key: &str, principal_id: &'static str) -> EventArgsBuilder {
+        // The object is a placeholder until `object()` supplies the committed
+        // ObjectInfo, matching the S3 helper.
+        let placeholder = ObjectInfo {
+            bucket: bucket.to_string(),
+            name: key.to_string(),
+            ..Default::default()
+        };
+        EventArgsBuilder::new(event_name, bucket.to_string(), convert_ecstore_object_info(placeholder))
+            .req_param("principalId", principal_id)
+    }
+
+    pub(super) fn object(self, obj_info: ObjectInfo) -> Self {
+        let Self {
+            builder,
+            notify,
+            request_context,
+        } = self;
+        Self {
+            builder: builder.object(convert_ecstore_object_info(obj_info)),
+            notify,
+            request_context,
+        }
+    }
+
+    pub(super) fn version_id(self, version_id: String) -> Self {
+        let Self {
+            builder,
+            notify,
+            request_context,
+        } = self;
+        Self {
+            builder: builder.version_id(version_id),
+            notify,
+            request_context,
+        }
+    }
+
+    /// Publish the event when `result` is a success; failures publish nothing.
+    pub(super) fn complete<T>(self, result: &S3Result<S3Response<T>>) {
+        let Ok(response) = result else {
+            return;
+        };
+        let Self {
+            builder,
+            notify,
+            request_context,
+        } = self;
+        let event_args = builder
+            .resp_elements(build_event_resp_elements(response, &request_context.request_id))
+            .build();
+        spawn_background_with_context(Some(request_context), async move {
+            notify.notify(event_args).await;
+        });
+    }
+}
+
+impl DefaultObjectUsecase {
+    /// Write one object through the ordinary PutObject path on behalf of a
+    /// trusted internal caller.
+    ///
+    /// The body is consumed exactly like a client request body: hashed
+    /// against `expected_md5_hex`, compressed and encrypted per bucket policy,
+    /// and committed under quota admission. On any failure nothing is left
+    /// behind.
+    pub(crate) async fn internal_put_object<B>(&self, ctx: InternalPutContext, body: B) -> Result<ObjectInfo, ApiError>
+    where
+        B: Stream<Item = io::Result<Bytes>> + Send + Sync + 'static,
+    {
+        let start_time = Instant::now();
+        let InternalPutContext {
+            bucket,
+            key,
+            size,
+            expected_md5_hex,
+            preserve_etag,
+            content_headers,
+            user_metadata,
+            tags,
+            internal_metadata,
+            emit_events,
+            principal_id,
+        } = ctx;
+        let Some(size) = size else {
+            return Err(ApiError::invalid_request("internal put requires a known object size"));
+        };
+        let size = i64::try_from(size).map_err(|_| ApiError::invalid_request("internal put size exceeds the supported range"))?;
+
+        let headers = internal_put_headers(&content_headers)?;
+        validate_internal_write_target(&key, &bucket, &headers).await?;
+
+        let write = PutObjectWriteRequest {
+            bucket,
+            key,
+            size,
+            quota_operation: QuotaOperation::PutObject,
+            ciphertext_passthrough: false,
+            inbound_replication_put: false,
+            headers: &headers,
+            query: None,
+            trailing_headers: None,
+            version_id: None,
+            sse: PutObjectSseInput {
+                server_side_encryption: None,
+                ssekms_key_id: None,
+                sse_customer_algorithm: None,
+                sse_customer_key: None,
+                sse_customer_key_md5: None,
+            },
+            user_metadata,
+            internal_metadata,
+            content: internal_put_content_input(&headers, tags),
+            object_lock: PutObjectLockInput {
+                legal_hold_status: None,
+                mode: None,
+                retain_until_date: None,
+            },
+            content_md5: expected_md5_hex.map(PutObjectContentMd5::Hex),
+            preserve_etag,
+            origin: PutObjectOrigin::Internal {
+                principal_id,
+                emit_events,
+            },
+        };
+        let committed = self
+            .put_object_core(write, StreamingBlob::wrap(body), start_time)
+            .await
+            .map_err(api_error_from_s3)?;
+
+        let obj_info = committed.obj_info.clone();
+        let result: S3Result<S3Response<()>> = Ok(S3Response::new(()));
+        committed.finish(&result);
+        Ok(obj_info)
+    }
+
+    /// Start a multipart upload for an internal write. The session carries the
+    /// same metadata, bucket default SSE session material, Object Lock
+    /// defaults and replication decision a client-initiated session would.
+    pub(crate) async fn internal_create_multipart_upload(&self, ctx: &InternalPutContext) -> Result<String, ApiError> {
+        let headers = internal_put_headers(&ctx.content_headers)?;
+        validate_internal_write_target(&ctx.key, &ctx.bucket, &headers).await?;
+        let store = self.object_store().ok_or_else(not_initialized)?;
+
+        let mut metadata = ctx.user_metadata.clone();
+        namespace_reserved_user_metadata(&mut metadata);
+        extract_metadata_from_mime(&headers, &mut metadata);
+        if let Some(tags) = ctx.tags.clone() {
+            metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags);
+        }
+
+        let object_lock_config_state = load_bucket_object_lock_config_state(&ctx.bucket)
+            .await
+            .map_err(api_error_from_s3)?;
+        apply_bucket_default_lock_retention(&ctx.bucket, &object_lock_config_state, &mut metadata, false)
+            .map_err(api_error_from_s3)?;
+
+        // Internal callers carry no credential; a bucket default of SSE-KMS is
+        // authorized as an internal write, like every other system write.
+        let prepared_material = sse_prepare_encryption(PrepareEncryptionRequest {
+            bucket: &ctx.bucket,
+            key: &ctx.key,
+            server_side_encryption: None,
+            ssekms_key_id: None,
+            ssekms_context: None,
+            sse_customer_algorithm: None,
+            sse_customer_key: None,
+            sse_customer_key_md5: None,
+            principal: None,
+        })
+        .await?;
+        if let Some(material) = prepared_material {
+            let mut encryption_metadata = encryption_material_to_metadata(&material)?;
+            if material.key_kind == EncryptionKeyKind::Object {
+                mark_encrypted_multipart_metadata(&mut encryption_metadata);
+            }
+            metadata.extend(encryption_metadata);
+        }
+
+        if is_multipart_disk_compression_enabled() && is_disk_compressible(&headers, &ctx.key) {
+            insert_str(
+                &mut metadata,
+                SUFFIX_COMPRESSION,
+                compression_metadata_value(CompressionAlgorithm::default()),
+            );
+        }
+        metadata.extend(ctx.internal_metadata.clone());
+
+        let mt2 = metadata.clone();
+        let mut opts = put_opts_with_replication_authorization(&ctx.bucket, &ctx.key, None, &headers, metadata, false)
+            .await
+            .map_err(ApiError::from)?;
+
+        let dsc = must_replicate_object(
+            &ctx.bucket,
+            &ctx.key,
+            &mt2,
+            "".to_string(),
+            opts.delete_marker_replication_status(),
+            opts.clone(),
+        )
+        .await;
+        if dsc.replicate_any() {
+            insert_str(&mut opts.user_defined, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
+            insert_str(
+                &mut opts.user_defined,
+                SUFFIX_REPLICATION_STATUS,
+                dsc.pending_status().unwrap_or_default(),
+            );
+        }
+
+        let current_opts = get_opts(&ctx.bucket, &ctx.key, opts.version_id.clone(), None, &headers)
+            .await
+            .map_err(ApiError::from)?;
+        match store.get_object_info(&ctx.bucket, &ctx.key, &current_opts).await {
+            Ok(existing_obj_info) => {
+                validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &opts)
+                    .map_err(api_error_from_s3)?;
+            }
+            Err(err) => {
+                if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
+                    return Err(ApiError::from(err));
+                }
+            }
+        }
+
+        let upload = store
+            .new_multipart_upload(&ctx.bucket, &ctx.key, &opts)
+            .await
+            .map_err(ApiError::from)?;
+        Ok(upload.upload_id)
+    }
+
+    /// Stage one part of an internal multipart upload. Compression and managed
+    /// SSE follow the session metadata recorded at creation; the staged part
+    /// is verified against `expected_md5_hex` when given.
+    pub(crate) async fn internal_upload_part<B>(
+        &self,
+        ctx: &InternalPutContext,
+        upload_id: &str,
+        part_number: usize,
+        size: u64,
+        expected_md5_hex: Option<String>,
+        body: B,
+    ) -> Result<CompletePart, ApiError>
+    where
+        B: Stream<Item = io::Result<Bytes>> + Send + Sync + Unpin + 'static,
+    {
+        let size =
+            i64::try_from(size).map_err(|_| ApiError::invalid_request("internal part size exceeds the supported range"))?;
+        let bucket = ctx.bucket.as_str();
+        let key = ctx.key.as_str();
+        let store = self.object_store().ok_or_else(not_initialized)?;
+        let mut opts = ObjectOptions::default();
+        let session = store
+            .get_multipart_info(bucket, key, upload_id, &opts)
+            .await
+            .map_err(ApiError::from)?;
+
+        let upload_part_admission = match get_concurrency_manager()
+            .admit_multipart_part(size)
+            .await
+            .map_err(|_| ApiError::other(io::Error::other("foreground write admission closed")))?
+        {
+            ForegroundWriteAdmission::Disabled => None,
+            ForegroundWriteAdmission::Admitted(permit) => {
+                counter!("rustfs.upload_part.foreground_admission.total", "result" => "admitted").increment(1);
+                Some(permit)
+            }
+            ForegroundWriteAdmission::Rejected => {
+                counter!("rustfs.upload_part.foreground_admission.total", "result" => "rejected").increment(1);
+                return Err(ApiError {
+                    code: S3ErrorCode::SlowDown,
+                    message: "foreground write concurrency limit reached, please reduce your request rate".to_string(),
+                    source: None,
+                });
+            }
+        };
+
+        let buffer_size = get_buffer_size_opt_in(size);
+        let body = tokio::io::BufReader::with_capacity(buffer_size, StreamReader::new(body));
+        let actual_size = size;
+        let mut write_plan = WritePlan::new();
+        let mut reader = if rustfs_utils::http::contains_key_str(&session.user_defined, SUFFIX_COMPRESSION) {
+            let hrd = HashReader::from_stream(body, size, actual_size, expected_md5_hex, None, false).map_err(ApiError::from)?;
+            write_plan = write_plan.with_compression(CompressionAlgorithm::default());
+            hrd
+        } else {
+            HashReader::from_stream(body, size, actual_size, expected_md5_hex, None, false).map_err(ApiError::from)?
+        };
+        opts.want_checksum = reader.checksum();
+
+        if session
+            .user_defined
+            .contains_key("x-amz-server-side-encryption-customer-algorithm")
+        {
+            return Err(ApiError::invalid_request("internal multipart writes cannot continue an SSE-C session"));
+        }
+        if session.user_defined.contains_key("x-amz-server-side-encryption") {
+            // Reuses the envelope prepared at creation; the session pins the key.
+            let managed_material = sse_decryption(DecryptionRequest {
+                bucket,
+                key,
+                metadata: &session.user_defined,
+                sse_customer_key: None,
+                sse_customer_key_md5: None,
+                principal: None,
+            })
+            .await?
+            .ok_or_else(|| ApiError::from(StorageError::other("Missing managed SSE session material")))?;
+            let managed_write = match managed_material.key_kind {
+                EncryptionKeyKind::Object => {
+                    WriteEncryption::multipart_object_key(managed_material.key_bytes, part_number as u32)
+                }
+                EncryptionKeyKind::Direct => {
+                    WriteEncryption::multipart(managed_material.key_bytes, managed_material.base_nonce, part_number)
+                }
+            };
+            write_plan = write_plan.with_encryption(managed_write);
+        }
+
+        reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;
+        let mut reader = PutObjReader::new(reader);
+
+        let _upload_part_admission = upload_part_admission;
+        let info = store
+            .put_object_part(bucket, key, upload_id, part_number, &mut reader, &opts)
+            .await
+            .map_err(ApiError::from)?;
+        drop(_upload_part_admission);
+
+        Ok(CompletePart {
+            part_num: info.part_num,
+            etag: info.etag,
+            ..Default::default()
+        })
+    }
+
+    /// Complete an internal multipart upload: versioning, quota admission,
+    /// Object Lock validation of the overwritten version, usage accounting,
+    /// immediate ILM transition, replication scheduling and the creation
+    /// event, as the S3 CompleteMultipartUpload handler performs them.
+    pub(crate) async fn internal_complete_multipart_upload(
+        &self,
+        ctx: &InternalPutContext,
+        upload_id: &str,
+        parts: Vec<CompletePart>,
+    ) -> Result<ObjectInfo, ApiError> {
+        let bucket = ctx.bucket.clone();
+        let key = ctx.key.clone();
+        if parts.is_empty() {
+            return Err(ApiError::invalid_request("You must specify at least one part"));
+        }
+        if parts.windows(2).any(|pair| pair[0].part_num >= pair[1].part_num) {
+            return Err(ApiError::invalid_request("multipart parts must be listed in ascending part number order"));
+        }
+        validate_table_catalog_object_mutation(&bucket, &key)
+            .await
+            .map_err(api_error_from_s3)?;
+        let store = self.object_store().ok_or_else(not_initialized)?;
+
+        let headers = HeaderMap::new();
+        let mut opts =
+            get_complete_multipart_upload_opts_with_replication_authorization(&headers, false).map_err(ApiError::from)?;
+        opts.preserve_etag = ctx.preserve_etag.clone();
+        let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
+        opts.versioned = versioned;
+        opts.version_suspended = BucketVersioningSys::prefix_suspended(&bucket, &key).await;
+        let capacity_scope_token = Uuid::new_v4();
+        opts.capacity_scope_token = Some(capacity_scope_token);
+
+        let current_opts =
+            internal_object_info_lookup_opts(get_opts(&bucket, &key, None, None, &headers).await.map_err(ApiError::from)?);
+        let object_lock_config_state = load_bucket_object_lock_config_state(&bucket)
+            .await
+            .map_err(api_error_from_s3)?;
+        let previous_current_sizes = match store.get_object_info(&bucket, &key, &current_opts).await {
+            Ok(existing_obj_info) => {
+                validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &current_opts)
+                    .map_err(api_error_from_s3)?;
+                let physical_size = existing_obj_info.size.max(0) as u64;
+                let logical_size = quota_object_size(&existing_obj_info);
+                Some((physical_size, logical_size))
+            }
+            Err(err) => {
+                if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
+                    return Err(ApiError::from(err));
+                }
+                None
+            }
+        };
+
+        let cache_adapter = self.object_data_cache();
+        let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await;
+
+        let quota_metadata_sys = self.bucket_metadata_sys();
+        let quota_tracking = quota_metadata_sys.is_some();
+        let mut quota_enabled = false;
+        if let Some(metadata_sys) = quota_metadata_sys {
+            let quota_checker = QuotaChecker::new(metadata_sys);
+            let check_result =
+                map_quota_check_outcome(&bucket, quota_checker.check_quota(&bucket, QuotaOperation::PutObject, 0).await)
+                    .map_err(api_error_from_s3)?;
+            quota_enabled = check_result.quota_limit.is_some();
+            apply_quota_admission(&mut opts, &check_result).map_err(api_error_from_s3)?;
+        }
+
+        let previous_current_size = match previous_current_sizes {
+            Some((_, Ok(logical_size))) if quota_enabled => Some(logical_size),
+            Some((_, Err(err))) if quota_enabled => return Err(ApiError::from(err)),
+            Some((physical_size, _)) => Some(physical_size),
+            None => None,
+        };
+
+        let event = ctx.emit_events.then(|| {
+            InternalPutObjectEvent::new(
+                current_notify_interface_for_context(self.context.as_deref()),
+                request_context::RequestContext::fallback(),
+                EventName::ObjectCreatedCompleteMultipartUpload,
+                &bucket,
+                &key,
+                ctx.principal_id,
+            )
+        });
+
+        // The spawned task owns the commit so a cancelled caller cannot leave
+        // the bookkeeping half done.
+        let complete_commit = spawn_traced_join({
+            let store = Arc::clone(&store);
+            let bucket = bucket.clone();
+            let key = key.clone();
+            let upload_id = upload_id.to_string();
+            let opts = opts.clone();
+            async move {
+                let obj_info = store
+                    .clone()
+                    .complete_multipart_upload(&bucket, &key, &upload_id, 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 quota_tracking {
+                    let committed_size = quota_accounting_object_size(&obj_info, quota_enabled).map_err(api_error_from_s3)?;
+                    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 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() {
+                    schedule_object_replication(obj_info.clone(), store, dsc).await;
+                }
+
+                rustfs_scanner::record_dirty_usage_bucket(&bucket);
+                Ok::<_, ApiError>(obj_info)
+            }
+        });
+        let obj_info = complete_commit.await.map_err(|err| {
+            ApiError::other(io::Error::other(format!("complete multipart upload commit owner task failed: {err}")))
+        })??;
+
+        if let Some(event) = event.flatten() {
+            let mut event = event.object(obj_info.clone());
+            if versioned && let Some(version_id) = obj_info.version_id {
+                event = event.version_id(version_id.to_string());
+            }
+            let result: S3Result<S3Response<()>> = Ok(S3Response::new(()));
+            event.complete(&result);
+        }
+        Ok(obj_info)
+    }
+
+    /// Discard an internal multipart upload and its staged parts.
+    pub(crate) async fn internal_abort_multipart_upload(&self, bucket: &str, key: &str, upload_id: &str) -> Result<(), ApiError> {
+        let store = self.object_store().ok_or_else(not_initialized)?;
+        store
+            .abort_multipart_upload(bucket, key, upload_id, &ObjectOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+        rustfs_scanner::record_dirty_usage_bucket(bucket);
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+    use rustfs_utils::http::{
+        MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX, SUFFIX_ODM_PULLED_AT, SUFFIX_ODM_SOURCE, SUFFIX_ODM_SOURCE_ETAG,
+        SUFFIX_ODM_SOURCE_LAST_MODIFIED, SUFFIX_ODM_SOURCE_VERSION_ID, contains_key_str, get_str,
+    };
+
+    const TEST_PRINCIPAL: &str = "rustfs-internal-put-test";
+
+    fn md5_hex(body: &[u8]) -> String {
+        hex_simd::encode_to_string(Md5::digest(body), hex_simd::AsciiCase::Lower)
+    }
+
+    fn body_stream(chunks: Vec<Bytes>) -> impl Stream<Item = io::Result<Bytes>> + Send + Sync + Unpin + 'static {
+        futures::stream::iter(chunks.into_iter().map(Ok))
+    }
+
+    fn provenance_metadata() -> HashMap<String, String> {
+        let mut internal_metadata = HashMap::new();
+        insert_str(&mut internal_metadata, SUFFIX_ODM_SOURCE, "s3:source-bucket".to_string());
+        insert_str(
+            &mut internal_metadata,
+            SUFFIX_ODM_SOURCE_ETAG,
+            "\"0123456789abcdef0123456789abcdef-3\"".to_string(),
+        );
+        insert_str(
+            &mut internal_metadata,
+            SUFFIX_ODM_SOURCE_LAST_MODIFIED,
+            "2026-01-02T03:04:05Z".to_string(),
+        );
+        insert_str(&mut internal_metadata, SUFFIX_ODM_SOURCE_VERSION_ID, String::new());
+        insert_str(&mut internal_metadata, SUFFIX_ODM_PULLED_AT, "2026-09-02T00:00:00Z".to_string());
+        internal_metadata
+    }
+
+    fn internal_context(bucket: &str, key: &str, body: &[u8]) -> InternalPutContext {
+        InternalPutContext {
+            bucket: bucket.to_string(),
+            key: key.to_string(),
+            size: Some(body.len() as u64),
+            expected_md5_hex: Some(md5_hex(body)),
+            preserve_etag: None,
+            content_headers: HashMap::from([
+                ("Content-Type".to_string(), "text/plain".to_string()),
+                ("Cache-Control".to_string(), "max-age=60".to_string()),
+            ]),
+            user_metadata: HashMap::from([("origin".to_string(), "unit-test".to_string())]),
+            tags: Some("team=storage".to_string()),
+            internal_metadata: provenance_metadata(),
+            emit_events: false,
+            principal_id: TEST_PRINCIPAL,
+        }
+    }
+
+    async fn internal_put_test_bucket(prefix: &str) -> (Arc<ECStore>, String) {
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
+        let bucket = format!("{prefix}-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create internal put test bucket");
+        (store, bucket)
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn internal_put_object_writes_through_the_shared_put_path() {
+        let (store, bucket) = internal_put_test_bucket("internal-put").await;
+        let body = b"internal write-back body".to_vec();
+        let ctx = internal_context(&bucket, "dir/object.txt", &body);
+
+        let obj_info = DefaultObjectUsecase::from_global()
+            .internal_put_object(ctx, body_stream(vec![Bytes::from(body.clone())]))
+            .await
+            .expect("internal put must succeed");
+
+        assert_eq!(obj_info.etag.as_deref(), Some(md5_hex(&body).as_str()));
+        assert_eq!(obj_info.size, body.len() as i64);
+
+        let stored = store
+            .get_object_info(&bucket, "dir/object.txt", &ObjectOptions::default())
+            .await
+            .expect("internal put must leave a readable object");
+        let metadata = &stored.user_defined;
+        assert_eq!(metadata.get("content-type").map(String::as_str), Some("text/plain"));
+        assert_eq!(metadata.get("cache-control").map(String::as_str), Some("max-age=60"));
+        assert_eq!(metadata.get("origin").map(String::as_str), Some("unit-test"));
+        assert_eq!(stored.user_tags.as_str(), "team=storage", "tags must be committed as object tags");
+        for suffix in [
+            SUFFIX_ODM_SOURCE,
+            SUFFIX_ODM_SOURCE_ETAG,
+            SUFFIX_ODM_SOURCE_LAST_MODIFIED,
+            SUFFIX_ODM_SOURCE_VERSION_ID,
+            SUFFIX_ODM_PULLED_AT,
+        ] {
+            assert!(
+                metadata.contains_key(&format!("{RUSTFS_INTERNAL_PREFIX}{suffix}")),
+                "missing RustFS provenance key {suffix}"
+            );
+            assert!(
+                metadata.contains_key(&format!("{MINIO_INTERNAL_PREFIX}{suffix}")),
+                "missing MinIO provenance key {suffix}"
+            );
+        }
+        assert_eq!(get_str(metadata, SUFFIX_ODM_SOURCE).as_deref(), Some("s3:source-bucket"));
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn internal_put_object_preserves_the_caller_etag() {
+        let (store, bucket) = internal_put_test_bucket("internal-put-etag").await;
+        let body = b"etag is preserved verbatim".to_vec();
+        let mut ctx = internal_context(&bucket, "preserved.bin", &body);
+        ctx.preserve_etag = Some("0123456789abcdef0123456789abcdef-3".to_string());
+
+        let obj_info = DefaultObjectUsecase::from_global()
+            .internal_put_object(ctx, body_stream(vec![Bytes::from(body.clone())]))
+            .await
+            .expect("internal put with a preserved ETag must succeed");
+        assert_eq!(obj_info.etag.as_deref(), Some("0123456789abcdef0123456789abcdef-3"));
+
+        let stored = store
+            .get_object_info(&bucket, "preserved.bin", &ObjectOptions::default())
+            .await
+            .expect("preserved-ETag object must be readable");
+        assert_eq!(stored.etag.as_deref(), Some("0123456789abcdef0123456789abcdef-3"));
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn internal_put_object_rejects_a_digest_mismatch_without_committing() {
+        let (store, bucket) = internal_put_test_bucket("internal-put-digest").await;
+        let body = b"body whose digest will not match".to_vec();
+        let mut ctx = internal_context(&bucket, "mismatch.bin", &body);
+        ctx.expected_md5_hex = Some(md5_hex(b"a different body"));
+
+        let err = DefaultObjectUsecase::from_global()
+            .internal_put_object(ctx, body_stream(vec![Bytes::from(body)]))
+            .await
+            .expect_err("digest mismatch must fail the internal put");
+        assert_eq!(err.code, S3ErrorCode::BadDigest, "unexpected error: {err}");
+
+        let lookup = store
+            .get_object_info(&bucket, "mismatch.bin", &ObjectOptions::default())
+            .await;
+        assert!(
+            lookup.as_ref().is_err_and(is_err_object_not_found),
+            "a rejected internal put must not leave an object behind"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn internal_put_object_requires_a_known_size() {
+        let (_store, bucket) = internal_put_test_bucket("internal-put-size").await;
+        let body = b"unknown length".to_vec();
+        let mut ctx = internal_context(&bucket, "unknown.bin", &body);
+        ctx.size = None;
+
+        let err = DefaultObjectUsecase::from_global()
+            .internal_put_object(ctx, body_stream(vec![Bytes::from(body)]))
+            .await
+            .expect_err("an unknown size must be rejected before the body is read");
+        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn internal_multipart_roundtrip_completes_and_abort_leaves_nothing() {
+        const FIRST_PART_SIZE: usize = 5 * 1024 * 1024;
+        let (store, bucket) = internal_put_test_bucket("internal-mpu").await;
+        let usecase = DefaultObjectUsecase::from_global();
+        let first_part = vec![0x41u8; FIRST_PART_SIZE];
+        let last_part = b"tail of the multipart object".to_vec();
+        let mut ctx = internal_context(&bucket, "multipart/object.bin", &[]);
+        ctx.size = None;
+        ctx.expected_md5_hex = None;
+        ctx.preserve_etag = Some("0123456789abcdef0123456789abcdef-2".to_string());
+
+        let upload_id = usecase
+            .internal_create_multipart_upload(&ctx)
+            .await
+            .expect("internal multipart create must succeed");
+        let part_one = usecase
+            .internal_upload_part(
+                &ctx,
+                &upload_id,
+                1,
+                first_part.len() as u64,
+                Some(md5_hex(&first_part)),
+                body_stream(vec![Bytes::from(first_part.clone())]),
+            )
+            .await
+            .expect("first internal part must stage");
+        let part_two = usecase
+            .internal_upload_part(
+                &ctx,
+                &upload_id,
+                2,
+                last_part.len() as u64,
+                Some(md5_hex(&last_part)),
+                body_stream(vec![Bytes::from(last_part.clone())]),
+            )
+            .await
+            .expect("last internal part must stage");
+        assert_eq!(part_one.part_num, 1);
+        assert_eq!(part_two.part_num, 2);
+
+        let obj_info = usecase
+            .internal_complete_multipart_upload(&ctx, &upload_id, vec![part_one, part_two])
+            .await
+            .expect("internal multipart complete must succeed");
+        assert_eq!(obj_info.size, (first_part.len() + last_part.len()) as i64);
+        assert_eq!(obj_info.parts.len(), 2);
+        assert_eq!(obj_info.etag.as_deref(), Some("0123456789abcdef0123456789abcdef-2"));
+        let stored = store
+            .get_object_info(&bucket, &ctx.key, &ObjectOptions::default())
+            .await
+            .expect("completed multipart object must be readable");
+        assert_eq!(stored.user_defined.get("content-type").map(String::as_str), Some("text/plain"));
+        assert_eq!(stored.user_defined.get("origin").map(String::as_str), Some("unit-test"));
+        assert!(contains_key_str(&stored.user_defined, SUFFIX_ODM_SOURCE));
+
+        let aborted_upload_id = usecase
+            .internal_create_multipart_upload(&ctx)
+            .await
+            .expect("second internal multipart create must succeed");
+        usecase
+            .internal_upload_part(
+                &ctx,
+                &aborted_upload_id,
+                1,
+                last_part.len() as u64,
+                None,
+                body_stream(vec![Bytes::from(last_part.clone())]),
+            )
+            .await
+            .expect("part of the aborted upload must stage");
+        usecase
+            .internal_abort_multipart_upload(&bucket, &ctx.key, &aborted_upload_id)
+            .await
+            .expect("internal abort must succeed");
+        let uploads = store
+            .list_multipart_uploads(&bucket, &ctx.key, None, None, None, 100)
+            .await
+            .expect("list multipart uploads after abort");
+        assert!(
+            uploads.uploads.iter().all(|upload| upload.upload_id != aborted_upload_id),
+            "aborted internal upload must not linger: {:?}",
+            uploads.uploads
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn internal_complete_multipart_upload_rejects_unordered_parts() {
+        let (_store, bucket) = internal_put_test_bucket("internal-mpu-order").await;
+        let ctx = internal_context(&bucket, "unordered.bin", &[]);
+        let parts = vec![
+            CompletePart {
+                part_num: 2,
+                ..Default::default()
+            },
+            CompletePart {
+                part_num: 1,
+                ..Default::default()
+            },
+        ];
+        let err = DefaultObjectUsecase::from_global()
+            .internal_complete_multipart_upload(&ctx, "upload", parts)
+            .await
+            .expect_err("unordered parts must be rejected before touching the store");
+        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
+    }
+
+    #[test]
+    fn internal_put_event_names_the_principal_and_creation_event() {
+        let event_args = InternalPutObjectEvent::builder(EventName::ObjectCreatedPut, "bucket", "key", TEST_PRINCIPAL).build();
+        assert_eq!(event_args.event_name, EventName::ObjectCreatedPut);
+        assert_eq!(event_args.bucket_name, "bucket");
+        assert_eq!(event_args.req_params.get("principalId").map(String::as_str), Some(TEST_PRINCIPAL));
+        assert!(!event_args.is_replication_request());
+    }
+
+    #[test]
+    fn internal_put_headers_normalize_names_and_reject_invalid_values() {
+        let headers = internal_put_headers(&HashMap::from([
+            ("Content-Type".to_string(), "application/json".to_string()),
+            ("Cache-Control".to_string(), "no-cache".to_string()),
+        ]))
+        .expect("valid content headers must build");
+        assert_eq!(headers.get("content-type").and_then(|v| v.to_str().ok()), Some("application/json"));
+        let content = internal_put_content_input(&headers, Some("a=b".to_string()));
+        assert_eq!(content.content_type.as_deref(), Some("application/json"));
+        assert_eq!(content.cache_control.as_deref(), Some("no-cache"));
+        assert_eq!(content.tagging.as_deref(), Some("a=b"));
+        assert!(content.storage_class.is_none());
+
+        let err = internal_put_headers(&HashMap::from([("Content-Type".to_string(), "bad\nvalue".to_string())]))
+            .expect_err("a header value with a control character must be rejected");
+        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
+    }
+}

rustfs/src/app/object/mod.rs

@@ -191,6 +191,10 @@ mod delete;
 mod extract;
 mod get;
 mod head;
+// Consumed by the on-demand migration write-back (rustfs/backlog#2153); until
+// that lands only tests construct the internal entry points.
+#[cfg_attr(not(test), expect(dead_code, reason = "wired by the on-demand migration write-back"))]
+mod internal_put;
 mod put;
 mod restore;
 mod shared;
@@ -202,6 +206,7 @@ pub(crate) use self::copy::*;
 pub(crate) use self::delete::*;
 pub(crate) use self::extract::*;
 pub(crate) use self::get::*;
+pub(crate) use self::internal_put::*;
 use self::put::*;
 pub(crate) use self::put::{guard_put_object_body_read_timeout, put_object_body_read_timeout};
 pub(crate) use self::shared::*;

rustfs/src/app/object/put.rs

@@ -895,6 +895,226 @@ fn is_post_object_sse_kms_requested(input: &PutObjectInput, headers: &HeaderMap)
     is_sse_kms_requested(input, headers)
 }
 
+/// Standard content headers and the tagging / storage-class values of a PUT
+/// that become object metadata through [`apply_put_request_metadata`].
+pub(super) struct PutObjectContentInput {
+    pub(super) cache_control: Option<CacheControl>,
+    pub(super) content_disposition: Option<ContentDisposition>,
+    pub(super) content_encoding: Option<ContentEncoding>,
+    pub(super) content_language: Option<ContentLanguage>,
+    pub(super) content_type: Option<ContentType>,
+    pub(super) expires: Option<String>,
+    pub(super) website_redirect_location: Option<WebsiteRedirectLocation>,
+    pub(super) tagging: Option<TaggingHeader>,
+    pub(super) storage_class: Option<StorageClass>,
+}
+
+/// Encryption inputs of a PUT after the S3 input/header merge.
+pub(super) struct PutObjectSseInput {
+    pub(super) server_side_encryption: Option<ServerSideEncryption>,
+    pub(super) ssekms_key_id: Option<SSEKMSKeyId>,
+    pub(super) sse_customer_algorithm: Option<SSECustomerAlgorithm>,
+    pub(super) sse_customer_key: Option<s3s::dto::SSECustomerKey>,
+    pub(super) sse_customer_key_md5: Option<SSECustomerKeyMD5>,
+}
+
+/// Explicit Object Lock values of a PUT; all `None` means the bucket default
+/// retention decides.
+pub(super) struct PutObjectLockInput {
+    pub(super) legal_hold_status: Option<ObjectLockLegalHoldStatus>,
+    pub(super) mode: Option<ObjectLockMode>,
+    pub(super) retain_until_date: Option<Timestamp>,
+}
+
+/// Expected body MD5 as the caller carries it. It is decoded at the same
+/// point of the write path for every origin so an invalid digest keeps its
+/// precedence relative to quota, admission and Object Lock errors.
+pub(super) enum PutObjectContentMd5 {
+    /// `Content-MD5` request header value.
+    Base64(String),
+    /// Lowercase hex digest, as an internal caller already holds it.
+    #[cfg_attr(not(test), expect(dead_code, reason = "constructed by the internal put entry point"))]
+    Hex(String),
+}
+
+/// Where a single-object write originates.
+pub(super) enum PutObjectOrigin<'a> {
+    /// The S3 PutObject/PostObject handler: request-bound identity, the
+    /// audit/notification chain and the bucket-generation guard installed by
+    /// the access layer all come from the request.
+    S3 {
+        req: &'a S3Request<PutObjectInput>,
+        event_name: EventName,
+    },
+    /// A trusted in-process caller writing on the server's behalf. There is no
+    /// request and no credential: managed-SSE authorization treats the write
+    /// as internal, and the creation event, when requested, names
+    /// `principal_id` instead of an access key.
+    #[cfg_attr(not(test), expect(dead_code, reason = "constructed by the internal put entry point"))]
+    Internal { principal_id: &'static str, emit_events: bool },
+}
+
+impl PutObjectOrigin<'_> {
+    fn replication_request_authorized(&self) -> bool {
+        match self {
+            Self::S3 { req, .. } => replication_request_authorized(req),
+            Self::Internal { .. } => false,
+        }
+    }
+
+    fn apply_bucket_generation_guard(&self, bucket: &str, opts: &mut ObjectOptions) -> S3Result<()> {
+        match self {
+            Self::S3 { req, .. } => apply_bucket_generation_guard(req, bucket, opts),
+            Self::Internal { .. } => Ok(()),
+        }
+    }
+
+    fn sse_principal(&self) -> Option<SseKmsPrincipal> {
+        match self {
+            Self::S3 { req, .. } => SseKmsPrincipal::from_request(req),
+            Self::Internal { .. } => None,
+        }
+    }
+}
+
+/// Every input the shared single-object write path needs, independent of
+/// whether an S3 request or an internal caller produced it.
+pub(super) struct PutObjectWriteRequest<'a> {
+    pub(super) bucket: String,
+    pub(super) key: String,
+    /// Authoritative plaintext length; never negative.
+    pub(super) size: i64,
+    pub(super) quota_operation: QuotaOperation,
+    /// Authorized SSE-C replication body that is already ciphertext.
+    pub(super) ciphertext_passthrough: bool,
+    pub(super) inbound_replication_put: bool,
+    /// Request headers, or the object's content headers for an internal write.
+    pub(super) headers: &'a HeaderMap,
+    pub(super) query: Option<&'a str>,
+    pub(super) trailing_headers: Option<s3s::TrailingHeaders>,
+    pub(super) version_id: Option<String>,
+    pub(super) sse: PutObjectSseInput,
+    /// User metadata keyed the way s3s delivers it (`x-amz-meta-` stripped).
+    pub(super) user_metadata: HashMap<String, String>,
+    /// Internal `x-rustfs-internal-*` / `x-minio-internal-*` keys written
+    /// verbatim onto the object; empty for S3 requests.
+    pub(super) internal_metadata: HashMap<String, String>,
+    pub(super) content: PutObjectContentInput,
+    pub(super) object_lock: PutObjectLockInput,
+    pub(super) content_md5: Option<PutObjectContentMd5>,
+    /// ETag to store instead of the computed one; `None` keeps the computed
+    /// (or replication-header-derived) value.
+    pub(super) preserve_etag: Option<String>,
+    pub(super) origin: PutObjectOrigin<'a>,
+}
+
+/// Audit/notification completion of a write, per origin.
+pub(super) enum PutObjectCompletion {
+    S3(OperationHelper),
+    Internal(Option<Box<InternalPutObjectEvent>>),
+}
+
+impl PutObjectCompletion {
+    fn object(self, obj_info: ObjectInfo) -> Self {
+        match self {
+            Self::S3(helper) => Self::S3(helper.object(obj_info)),
+            Self::Internal(event) => Self::Internal(event.map(|event| Box::new(event.object(obj_info)))),
+        }
+    }
+
+    fn version_id(self, version_id: String) -> Self {
+        match self {
+            Self::S3(helper) => Self::S3(helper.version_id(version_id)),
+            Self::Internal(event) => Self::Internal(event.map(|event| Box::new(event.version_id(version_id)))),
+        }
+    }
+
+    fn complete<T>(self, result: &S3Result<S3Response<T>>) -> Self {
+        match self {
+            Self::S3(helper) => Self::S3(helper.complete(result)),
+            Self::Internal(event) => {
+                if let Some(event) = event {
+                    event.complete(result);
+                }
+                Self::Internal(None)
+            }
+        }
+    }
+}
+
+/// Record the failed completion and hand the error back to the caller.
+fn fail_put_object(completion: PutObjectCompletion, err: S3Error) -> S3Error {
+    let result: S3Result<S3Response<()>> = Err(err);
+    let _ = completion.complete(&result);
+    match result {
+        Err(err) => err,
+        Ok(_) => unreachable!("failed PutObject completion carries an error"),
+    }
+}
+
+/// A committed single-object write awaiting its response-side completion.
+pub(super) struct PutObjectCommitted {
+    pub(super) obj_info: ObjectInfo,
+    pub(super) put_versioned: bool,
+    pub(super) effective_sse: Option<ServerSideEncryption>,
+    pub(super) effective_kms_key_id: Option<SSEKMSKeyId>,
+    pub(super) sse_customer_algorithm: Option<SSECustomerAlgorithm>,
+    pub(super) sse_customer_key_md5: Option<SSECustomerKeyMD5>,
+    pub(super) put_extra_checksum_headers: Vec<(&'static str, String)>,
+    completion: PutObjectCompletion,
+    put_request_guard: PutObjectGuard,
+    bucket: String,
+    key: String,
+    start_time: Instant,
+    size: i64,
+    use_zero_copy_eager_put_path: bool,
+    concurrent_put_requests: usize,
+    buffer_size: usize,
+}
+
+impl PutObjectCommitted {
+    /// Publish the audit entry / creation event for `result`, record the
+    /// request-level PutObject metrics and release the request guard.
+    pub(super) fn finish<T>(self, result: &S3Result<S3Response<T>>) {
+        let Self {
+            completion,
+            mut put_request_guard,
+            bucket,
+            key,
+            start_time,
+            size,
+            use_zero_copy_eager_put_path,
+            concurrent_put_requests,
+            buffer_size,
+            ..
+        } = self;
+        let _ = completion.complete(result);
+
+        // Record PutObject metrics via zero-copy-metrics
+        {
+            let duration_ms = start_time.elapsed().as_millis() as f64;
+            rustfs_io_metrics::record_put_object(
+                duration_ms,
+                size,
+                use_zero_copy_eager_put_path, // Track if zero-copy was enabled
+            );
+        }
+
+        debug!(
+            target: "rustfs::app::object_usecase",
+            component = "app",
+            subsystem = "object",
+            bucket = %bucket,
+            key = %key,
+            concurrent_put_requests,
+            buffer_size,
+            "PutObject request completed"
+        );
+
+        put_request_guard.finish_ok();
+    }
+}
+
 impl DefaultObjectUsecase {
     fn should_use_large_put_concurrency_tuning(size: i64) -> bool {
         size >= DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES
@@ -1055,7 +1275,7 @@ impl DefaultObjectUsecase {
         };
 
         // Resolve the authoritative decoded/plain object length (rejecting negative/unknown) before anything else consumes it.
-        let mut size = resolve_put_object_authoritative_size(&req.headers, content_length)?;
+        let size = resolve_put_object_authoritative_size(&req.headers, content_length)?;
 
         if let Some(limit) = max_content_length
             && u64::try_from(size).is_ok_and(|size| size > limit)
@@ -1063,6 +1283,140 @@ impl DefaultObjectUsecase {
             return Err(S3Error::new(S3ErrorCode::EntityTooLarge));
         }
 
+        let write = PutObjectWriteRequest {
+            bucket: bucket.clone(),
+            key,
+            size,
+            quota_operation,
+            ciphertext_passthrough,
+            inbound_replication_put,
+            headers: &req.headers,
+            query: req.uri.query(),
+            trailing_headers: req.trailing_headers.clone(),
+            version_id,
+            sse: PutObjectSseInput {
+                server_side_encryption,
+                ssekms_key_id,
+                sse_customer_algorithm,
+                sse_customer_key,
+                sse_customer_key_md5,
+            },
+            user_metadata: metadata.unwrap_or_default(),
+            internal_metadata: HashMap::new(),
+            content: PutObjectContentInput {
+                cache_control,
+                content_disposition,
+                content_encoding,
+                content_language,
+                content_type,
+                expires,
+                website_redirect_location,
+                tagging,
+                storage_class,
+            },
+            object_lock: PutObjectLockInput {
+                legal_hold_status: object_lock_legal_hold_status,
+                mode: object_lock_mode,
+                retain_until_date: object_lock_retain_until_date,
+            },
+            content_md5: content_md5.map(PutObjectContentMd5::Base64),
+            preserve_etag: None,
+            origin: PutObjectOrigin::S3 { req: &req, event_name },
+        };
+        let committed = self.put_object_core(write, body, start_time).await?;
+
+        let raw_version = committed.obj_info.version_id.map(|v| v.to_string());
+        let put_version = if committed.put_versioned { raw_version } else { None };
+
+        let e_tag = committed.obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
+
+        let expiration = resolve_put_object_expiration(&bucket, &committed.obj_info).await;
+
+        let mut checksums = PutObjectChecksums {
+            crc32: input.checksum_crc32,
+            crc32c: input.checksum_crc32c,
+            sha1: input.checksum_sha1,
+            sha256: input.checksum_sha256,
+            crc64nvme: input.checksum_crc64nvme,
+        };
+        apply_trailing_checksums(
+            input.checksum_algorithm.as_ref().map(|a| a.as_str()),
+            &req.trailing_headers,
+            &mut checksums,
+        );
+
+        let output = PutObjectOutput {
+            e_tag,
+            server_side_encryption: committed.effective_sse.clone(),
+            sse_customer_algorithm: committed.sse_customer_algorithm.clone(),
+            sse_customer_key_md5: committed.sse_customer_key_md5.clone(),
+            ssekms_key_id: committed.effective_kms_key_id.clone(),
+            expiration,
+            checksum_crc32: checksums.crc32,
+            checksum_crc32c: checksums.crc32c,
+            checksum_sha1: checksums.sha1,
+            checksum_sha256: checksums.sha256,
+            checksum_crc64nvme: checksums.crc64nvme,
+            version_id: put_version,
+            ..Default::default()
+        };
+
+        // For browser-based POST uploads (multipart/form-data), response status/body handling
+        // is decided by s3s PostObject serializer (success_action_status / redirect semantics).
+
+        let response_build_stage_start = put_stage_metrics_enabled.then(Instant::now);
+        let mut response = S3Response::new(output);
+        // Echo XXHash3/64/128 / SHA-512 checksums that s3s PutObjectOutput has no typed
+        // field for (#1256).
+        inject_additional_checksum_headers(&mut response.headers, &committed.put_extra_checksum_headers);
+        rustfs_io_metrics::record_put_object_stage_duration_from("app_response_build", response_build_stage_start);
+        let result = Ok(response);
+        committed.finish(&result);
+
+        result
+    }
+
+    /// The single-object write path shared by the S3 handler and internal
+    /// callers: quota admission, foreground write admission, bucket default
+    /// SSE, Object Lock defaults, put options, the hashing/compressing/
+    /// encrypting reader, the owned store commit, usage accounting,
+    /// replication scheduling and the creation-event setup. The caller shapes
+    /// the request before and builds its response after.
+    pub(super) async fn put_object_core(
+        &self,
+        write: PutObjectWriteRequest<'_>,
+        body: StreamingBlob,
+        start_time: Instant,
+    ) -> S3Result<PutObjectCommitted> {
+        let put_stage_metrics_enabled = rustfs_io_metrics::put_stage_metrics_enabled();
+        let PutObjectWriteRequest {
+            bucket,
+            key,
+            mut size,
+            quota_operation,
+            ciphertext_passthrough,
+            inbound_replication_put,
+            headers,
+            query,
+            trailing_headers,
+            version_id,
+            sse,
+            user_metadata,
+            internal_metadata,
+            content,
+            object_lock,
+            content_md5,
+            preserve_etag,
+            origin,
+        } = write;
+        let PutObjectSseInput {
+            server_side_encryption,
+            ssekms_key_id,
+            sse_customer_algorithm,
+            sse_customer_key,
+            sse_customer_key_md5,
+        } = sse;
+
         // The app check preserves the existing S3 error contract; the storage
         // commit path reserves the exact net logical growth under its locks.
         let quota_stage_start = put_stage_metrics_enabled.then(Instant::now);
@@ -1084,7 +1438,7 @@ impl DefaultObjectUsecase {
 
         let ingress_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let should_compress =
-            is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64 && !ciphertext_passthrough;
+            is_disk_compressible(headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64 && !ciphertext_passthrough;
 
         // Resolve the store through the request-bound server context
         // (backlog#1052 S6), not the process-global handle, so an embedded
@@ -1175,7 +1529,7 @@ impl DefaultObjectUsecase {
         let (put_path, zero_copy_eager_put_path_status, use_zero_copy_eager_put_path, use_empty_or_small_eager_put_path) =
             select_put_path_with_concurrency(
                 size,
-                &req.headers,
+                headers,
                 server_side_encryption_requested,
                 should_compress,
                 false,
@@ -1200,33 +1554,33 @@ impl DefaultObjectUsecase {
         validate_sse_headers_for_write(
             effective_sse.as_ref(),
             effective_kms_key_id.as_ref(),
-            extract_ssekms_context_from_headers(&req.headers)?.as_ref(),
+            extract_ssekms_context_from_headers(headers)?.as_ref(),
             sse_customer_algorithm.as_ref(),
             sse_customer_key.as_ref(),
             sse_customer_key_md5.as_ref(),
             true, // PutObject requires all three: algorithm, key, key_md5
         )?;
 
-        let mut metadata = metadata.unwrap_or_default();
+        let mut metadata = user_metadata;
-        let has_explicit_object_lock_retention = object_lock_mode.is_some()
+        let has_explicit_object_lock_retention = object_lock.mode.is_some()
-            || object_lock_retain_until_date.is_some()
+            || object_lock.retain_until_date.is_some()
-            || has_replication_retention_update(&req.headers, inbound_replication_put);
+            || has_replication_retention_update(headers, inbound_replication_put);
         let object_lock_config_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let object_lock_config_state = load_bucket_object_lock_config_state(&bucket).await?;
         rustfs_io_metrics::record_put_object_stage_duration_from("app_object_lock_config_lookup", object_lock_config_stage_start);
         apply_put_request_metadata(
             &mut metadata,
-            &req.headers,
+            headers,
             &key,
-            cache_control,
+            content.cache_control,
-            content_disposition,
+            content.content_disposition,
-            content_encoding,
+            content.content_encoding,
-            content_language,
+            content.content_language,
-            content_type,
+            content.content_type,
-            expires,
+            content.expires,
-            website_redirect_location,
+            content.website_redirect_location,
-            tagging,
+            content.tagging,
-            storage_class.clone(),
+            content.storage_class,
         )?;
         apply_bucket_default_lock_retention(
             &bucket,
@@ -1234,29 +1588,33 @@ impl DefaultObjectUsecase {
             &mut metadata,
             has_explicit_object_lock_retention,
         )?;
+        metadata.extend(internal_metadata);
 
         let put_opts_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let mut opts: ObjectOptions = put_opts_with_replication_authorization(
             &bucket,
             &key,
             version_id.clone(),
-            &req.headers,
+            headers,
             metadata.clone(),
-            replication_request_authorized(&req),
+            origin.replication_request_authorized(),
         )
         .await
         .map_err(ApiError::from)?;
+        if let Some(etag) = preserve_etag {
+            opts.preserve_etag = Some(etag);
+        }
         if let Some(quota_check) = quota_check.as_ref() {
             apply_quota_admission(&mut opts, quota_check)?;
         }
         rustfs_io_metrics::record_put_object_stage_duration_from("app_put_opts_build", put_opts_stage_start);
-        apply_bucket_generation_guard(&req, &bucket, &mut opts)?;
+        origin.apply_bucket_generation_guard(&bucket, &mut opts)?;
         apply_put_request_object_lock_opts(
             &bucket,
             &object_lock_config_state,
-            object_lock_legal_hold_status,
+            object_lock.legal_hold_status,
-            object_lock_mode,
+            object_lock.mode,
-            object_lock_retain_until_date,
+            object_lock.retain_until_date,
             &mut opts,
         )?;
         let eager_put_commit_cancellation =
@@ -1278,7 +1636,7 @@ impl DefaultObjectUsecase {
         let prelookup_stage_start = (prelookup_required && put_stage_metrics_enabled).then(Instant::now);
         let prelookup_previous_current_size: Option<Option<u64>> = if prelookup_required {
             let current_opts: ObjectOptions = internal_object_info_lookup_opts(
-                get_opts(&bucket, &key, version_id.clone(), None, &req.headers)
+                get_opts(&bucket, &key, version_id.clone(), None, headers)
                     .await
                     .map_err(ApiError::from)?,
             );
@@ -1315,16 +1673,18 @@ impl DefaultObjectUsecase {
             )?;
         }
 
-        let mut md5hex = if let Some(base64_md5) = content_md5 {
+        let mut md5hex = match content_md5 {
-            let md5 = base64_simd::STANDARD
+            Some(PutObjectContentMd5::Base64(base64_md5)) => {
-                .decode_to_vec(base64_md5.as_bytes())
+                let md5 = base64_simd::STANDARD
-                .map_err(|e| ApiError::from(StorageError::other(format!("Invalid content MD5: {e}"))))?;
+                    .decode_to_vec(base64_md5.as_bytes())
-            Some(hex_simd::encode_to_string(&md5, hex_simd::AsciiCase::Lower))
+                    .map_err(|e| ApiError::from(StorageError::other(format!("Invalid content MD5: {e}"))))?;
-        } else {
+                Some(hex_simd::encode_to_string(&md5, hex_simd::AsciiCase::Lower))
-            None
+            }
+            Some(PutObjectContentMd5::Hex(md5hex)) => Some(md5hex),
+            None => None,
         };
 
-        let mut sha256hex = get_content_sha256_with_query(&req.headers, req.uri.query());
+        let mut sha256hex = get_content_sha256_with_query(headers, query);
 
         let mut write_plan = WritePlan::new();
         // Additional-checksum (XXHash3/64/128, SHA-512) values to echo on the PutObject
@@ -1342,7 +1702,7 @@ impl DefaultObjectUsecase {
             let mut hrd =
                 HashReader::from_stream(body, size, size, md5hex.take(), sha256hex.take(), false).map_err(ApiError::from)?;
 
-            if let Err(err) = hrd.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
+            if let Err(err) = hrd.add_checksum_from_s3s(headers, trailing_headers.clone(), false) {
                 return Err(ApiError::from(err).into());
             }
 
@@ -1392,7 +1752,7 @@ impl DefaultObjectUsecase {
         };
 
         if size >= 0 {
-            if let Err(err) = reader.add_checksum_from_s3s(&req.headers, req.trailing_headers.clone(), false) {
+            if let Err(err) = reader.add_checksum_from_s3s(headers, trailing_headers.clone(), false) {
                 return Err(ApiError::from(err).into());
             }
 
@@ -1402,12 +1762,35 @@ impl DefaultObjectUsecase {
         rustfs_io_metrics::record_put_object_path(put_path);
         rustfs_io_metrics::record_put_object_stage_duration_from("ingress_prepare", ingress_stage_start);
 
-        let mut helper = OperationHelper::new(&req, event_name, S3Operation::PutObject);
+        let (mut completion, request_context) = match &origin {
-        let ssekms_context = extract_ssekms_context_from_headers(&req.headers)?;
+            PutObjectOrigin::S3 { req, event_name } => (
+                PutObjectCompletion::S3(OperationHelper::new(req, *event_name, S3Operation::PutObject)),
+                req.extensions.get::<request_context::RequestContext>().cloned(),
+            ),
+            PutObjectOrigin::Internal {
+                principal_id,
+                emit_events,
+            } => {
+                let principal_id = *principal_id;
+                let request_context = request_context::RequestContext::fallback();
+                let event = emit_events.then(|| {
+                    InternalPutObjectEvent::new(
+                        current_notify_interface_for_context(self.context.as_deref()),
+                        request_context.clone(),
+                        EventName::ObjectCreatedPut,
+                        &bucket,
+                        &key,
+                        principal_id,
+                    )
+                });
+                (PutObjectCompletion::Internal(event.flatten().map(Box::new)), Some(request_context))
+            }
+        };
+        let ssekms_context = extract_ssekms_context_from_headers(headers)?;
 
         // Apply encryption using unified SSE API.
         let encryption_stage_start = put_stage_metrics_enabled.then(Instant::now);
-        let write_principal = SseKmsPrincipal::from_request(&req);
+        let write_principal = origin.sse_principal();
         let encryption_request = EncryptionRequest {
             bucket: &bucket,
             key: &key,
@@ -1430,11 +1813,7 @@ impl DefaultObjectUsecase {
         } else {
             match sse_encryption(encryption_request).await {
                 Ok(material) => material,
-                Err(err) => {
+                Err(err) => return Err(fail_put_object(completion, err.into())),
-                    let result = Err(err.into());
-                    let _ = helper.complete(&result);
-                    return result;
-                }
             }
         };
 
@@ -1460,7 +1839,6 @@ impl DefaultObjectUsecase {
 
         let mt2 = metadata.clone();
         opts.user_defined.extend(metadata);
-        let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
         let request_id = request_context
             .as_ref()
             .map(|ctx| ctx.request_id.clone())
@@ -1660,100 +2038,41 @@ impl DefaultObjectUsecase {
         let PutObjectCommitResult { obj_info, put_versioned } = match put_commit_result {
             Ok(Ok(result)) => result,
             Ok(Err(err)) => {
-                let result: S3Result<S3Response<PutObjectOutput>> = Err(err);
                 put_request_guard.finish_err();
-                let _ = helper.complete(&result);
+                return Err(fail_put_object(completion, err));
-                return result;
             }
             Err(err) => {
-                let result: S3Result<S3Response<PutObjectOutput>> = Err(S3Error::with_message(
-                    S3ErrorCode::InternalError,
-                    format!("put object commit owner task failed: {err}"),
-                ));
                 put_request_guard.finish_err();
-                let _ = helper.complete(&result);
+                return Err(fail_put_object(
-                return result;
+                    completion,
+                    S3Error::with_message(S3ErrorCode::InternalError, format!("put object commit owner task failed: {err}")),
+                ));
             }
         };
 
-        let raw_version = obj_info.version_id.map(|v| v.to_string());
+        completion = completion.object(obj_info.clone());
-
+        if let Some(version_id) = obj_info.version_id {
-        helper = helper.object(obj_info.clone());
+            completion = completion.version_id(version_id.to_string());
-        if let Some(version_id) = &raw_version {
-            helper = helper.version_id(version_id.clone());
         }
 
-        let put_version = if put_versioned { raw_version } else { None };
+        Ok(PutObjectCommitted {
-
+            obj_info,
-        let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
+            put_versioned,
-
+            effective_sse,
-        let expiration = resolve_put_object_expiration(&bucket, &obj_info).await;
+            effective_kms_key_id,
-
+            sse_customer_algorithm,
-        let mut checksums = PutObjectChecksums {
+            sse_customer_key_md5,
-            crc32: input.checksum_crc32,
+            put_extra_checksum_headers,
-            crc32c: input.checksum_crc32c,
+            completion,
-            sha1: input.checksum_sha1,
+            put_request_guard,
-            sha256: input.checksum_sha256,
+            bucket,
-            crc64nvme: input.checksum_crc64nvme,
+            key,
-        };
+            start_time,
-        apply_trailing_checksums(
+            size,
-            input.checksum_algorithm.as_ref().map(|a| a.as_str()),
+            use_zero_copy_eager_put_path,
-            &req.trailing_headers,
-            &mut checksums,
-        );
-
-        let output = PutObjectOutput {
-            e_tag,
-            server_side_encryption: effective_sse,
-            sse_customer_algorithm: sse_customer_algorithm.clone(),
-            sse_customer_key_md5: sse_customer_key_md5.clone(),
-            ssekms_key_id: effective_kms_key_id,
-            expiration,
-            checksum_crc32: checksums.crc32,
-            checksum_crc32c: checksums.crc32c,
-            checksum_sha1: checksums.sha1,
-            checksum_sha256: checksums.sha256,
-            checksum_crc64nvme: checksums.crc64nvme,
-            version_id: put_version,
-            ..Default::default()
-        };
-
-        // For browser-based POST uploads (multipart/form-data), response status/body handling
-        // is decided by s3s PostObject serializer (success_action_status / redirect semantics).
-
-        let response_build_stage_start = put_stage_metrics_enabled.then(Instant::now);
-        let mut response = S3Response::new(output);
-        // Echo XXHash3/64/128 / SHA-512 checksums that s3s PutObjectOutput has no typed
-        // field for (#1256).
-        inject_additional_checksum_headers(&mut response.headers, &put_extra_checksum_headers);
-        rustfs_io_metrics::record_put_object_stage_duration_from("app_response_build", response_build_stage_start);
-        let result = Ok(response);
-        let _ = helper.complete(&result);
-
-        // Record PutObject metrics via zero-copy-metrics
-        {
-            let duration_ms = start_time.elapsed().as_millis() as f64;
-            rustfs_io_metrics::record_put_object(
-                duration_ms,
-                size,
-                use_zero_copy_eager_put_path, // Track if zero-copy was enabled
-            );
-        }
-
-        debug!(
-            target: "rustfs::app::object_usecase",
-            component = "app",
-            subsystem = "object",
-            bucket = %bucket,
-            key = %key,
             concurrent_put_requests,
             buffer_size,
-            "PutObject request completed"
+        })
-        );
-
-        put_request_guard.finish_ok();
-
-        result
     }
 }
 

rustfs/src/protocols/client.rs

@@ -163,8 +163,7 @@ fn build_object_uri(bucket: &str, key: &str, query: &[(&str, Option<&str>)]) ->
 struct RequestParams<'a> {
     bucket: Option<String>,
     object: Option<String>,
-    access_key: &'a str,
+    credentials: &'a rustfs_credentials::Credentials,
-    secret_key: &'a str,
 }
 
 /// Protocol storage client that implements the StorageBackend trait
@@ -181,39 +180,22 @@ impl ProtocolStorageClient {
     }
 
     /// Create a proper S3Request with ReqInfo extension for authorization
-    async fn create_request<T>(
+    fn create_request<T>(input: T, method: Method, uri: http::Uri, params: RequestParams<'_>) -> S3Result<S3Request<T>> {
-        &self,
-        input: T,
-        method: Method,
-        uri: http::Uri,
-        params: RequestParams<'_>,
-    ) -> S3Result<S3Request<T>> {
         let mut extensions = http::Extensions::default();
 
         let is_owner = if let Some(global_cred) = current_action_credentials() {
-            params.access_key == global_cred.access_key
+            params.credentials.access_key == global_cred.access_key
         } else {
             false
         };
 
         let credentials = Some(s3s::auth::Credentials {
-            access_key: params.access_key.to_string(),
+            access_key: params.credentials.access_key.clone(),
-            secret_key: params.secret_key.to_string().into(),
+            secret_key: params.credentials.secret_key.clone().into(),
         });
 
         extensions.insert(ReqInfo {
-            cred: Some(rustfs_credentials::Credentials {
+            cred: Some(params.credentials.clone()),
-                access_key: params.access_key.to_string(),
-                secret_key: params.secret_key.to_string(),
-                session_token: String::new(),
-                expiration: None,
-                status: String::new(),
-                parent_user: String::new(),
-                groups: None,
-                claims: None,
-                name: None,
-                description: None,
-            }),
             is_owner,
             bucket: params.bucket,
             object: params.object,
@@ -247,8 +229,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         &self,
         bucket: &str,
         key: &str,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
         start_pos: Option<u64>,
     ) -> Result<GetObjectOutput, Self::Error> {
         trace_protocol_request("get_object", Some(bucket), Some(key));
@@ -279,19 +260,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         })?;
 
         let uri = build_object_uri(bucket, key, &[])?;
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::GET,
-                Method::GET,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket.to_string()),
-                    bucket: Some(bucket.to_string()),
+                object: Some(key.to_string()),
-                    object: Some(key.to_string()),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.get_object(req).await {
             Ok(response) => Ok(response.output),
@@ -302,8 +280,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
     async fn put_object(
         &self,
         input: PutObjectInput,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
     ) -> Result<PutObjectOutput, Self::Error> {
         trace!(
             event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST,
@@ -330,19 +307,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
             }
         }
 
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::PUT,
-                Method::PUT,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket),
-                    bucket: Some(bucket),
+                object: Some(key),
-                    object: Some(key),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
         let req = S3Request { headers, ..req };
 
         match self.fs.put_object(req).await {
@@ -355,8 +329,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         &self,
         bucket: &str,
         key: &str,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
     ) -> Result<DeleteObjectOutput, Self::Error> {
         trace_protocol_request("delete_object", Some(bucket), Some(key));
 
@@ -369,19 +342,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
             })?;
 
         let uri = build_object_uri(bucket, key, &[])?;
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::DELETE,
-                Method::DELETE,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket.to_string()),
-                    bucket: Some(bucket.to_string()),
+                object: Some(key.to_string()),
-                    object: Some(key.to_string()),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.delete_object(req).await {
             Ok(response) => Ok(response.output),
@@ -393,8 +363,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         &self,
         bucket: &str,
         key: &str,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
     ) -> Result<HeadObjectOutput, Self::Error> {
         trace_protocol_request("head_object", Some(bucket), Some(key));
 
@@ -407,19 +376,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
             })?;
 
         let uri = build_object_uri(bucket, key, &[])?;
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::HEAD,
-                Method::HEAD,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket.to_string()),
-                    bucket: Some(bucket.to_string()),
+                object: Some(key.to_string()),
-                    object: Some(key.to_string()),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.head_object(req).await {
             Ok(response) => Ok(response.output),
@@ -427,7 +393,11 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         }
     }
 
-    async fn head_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<HeadBucketOutput, Self::Error> {
+    async fn head_bucket(
+        &self,
+        bucket: &str,
+        credentials: &rustfs_credentials::Credentials,
+    ) -> Result<HeadBucketOutput, Self::Error> {
         trace_protocol_request("head_bucket", Some(bucket), None);
 
         let input = HeadBucketInput::builder().bucket(bucket.to_string()).build().map_err(|e| {
@@ -435,19 +405,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         })?;
 
         let uri = build_bucket_uri(bucket, &[])?;
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::HEAD,
-                Method::HEAD,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket.to_string()),
-                    bucket: Some(bucket.to_string()),
+                object: None,
-                    object: None,
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.head_bucket(req).await {
             Ok(response) => Ok(response.output),
@@ -458,26 +425,22 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
     async fn list_objects_v2(
         &self,
         input: ListObjectsV2Input,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
     ) -> Result<ListObjectsV2Output, Self::Error> {
         trace_protocol_request("list_objects_v2", Some(&input.bucket), None);
 
         let bucket = input.bucket.clone();
         let uri = build_bucket_uri(&bucket, &[("list-type", Some("2"))])?;
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::GET,
-                Method::GET,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket),
-                    bucket: Some(bucket),
+                object: None,
-                    object: None,
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.list_objects_v2(req).await {
             Ok(response) => Ok(response.output),
@@ -485,13 +448,13 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         }
     }
 
-    async fn list_buckets(&self, access_key: &str, secret_key: &str) -> Result<ListBucketsOutput, Self::Error> {
+    async fn list_buckets(&self, credentials: &rustfs_credentials::Credentials) -> Result<ListBucketsOutput, Self::Error> {
         trace!(
             event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST,
             component = LOG_COMPONENT_PROTOCOLS,
             subsystem = LOG_SUBSYSTEM_STORAGE_CLIENT,
             operation = "list_buckets",
-            access_key = %MaskedAccessKey(access_key),
+            access_key = %MaskedAccessKey(&credentials.access_key),
             "Protocol storage client request"
         );
 
@@ -499,19 +462,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
             s3s::S3Error::with_message(s3s::S3ErrorCode::InvalidRequest, format!("Failed to build ListBucketsInput: {}", e))
         })?;
 
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::GET,
-                Method::GET,
+            http::Uri::from_static("/"),
-                http::Uri::from_static("/"),
+            RequestParams {
-                RequestParams {
+                bucket: None,
-                    bucket: None,
+                object: None,
-                    object: None,
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.list_buckets(req).await {
             Ok(response) => Ok(response.output),
@@ -542,7 +502,11 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         self.fs.list_buckets(request).await.map(|response| response.output)
     }
 
-    async fn create_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<CreateBucketOutput, Self::Error> {
+    async fn create_bucket(
+        &self,
+        bucket: &str,
+        credentials: &rustfs_credentials::Credentials,
+    ) -> Result<CreateBucketOutput, Self::Error> {
         trace_protocol_request("create_bucket", Some(bucket), None);
 
         let input = CreateBucketInput::builder().bucket(bucket.to_string()).build().map_err(|e| {
@@ -550,19 +514,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         })?;
 
         let uri = build_bucket_uri(bucket, &[])?;
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::PUT,
-                Method::PUT,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket.to_string()),
-                    bucket: Some(bucket.to_string()),
+                object: None,
-                    object: None,
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.create_bucket(req).await {
             Ok(response) => Ok(response.output),
@@ -574,8 +535,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         &self,
         bucket: &str,
         key: &str,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
         start_pos: u64,
         length: u64,
     ) -> Result<GetObjectOutput, Self::Error> {
@@ -607,19 +567,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
             })?;
 
         let uri = build_object_uri(bucket, key, &[])?;
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::GET,
-                Method::GET,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket.to_string()),
-                    bucket: Some(bucket.to_string()),
+                object: Some(key.to_string()),
-                    object: Some(key.to_string()),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.get_object(req).await {
             Ok(response) => Ok(response.output),
@@ -630,8 +587,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
     async fn copy_object(
         &self,
         input: CopyObjectInput,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
     ) -> Result<CopyObjectOutput, Self::Error> {
         trace!(
             event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST,
@@ -647,19 +603,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         let key = input.key.clone();
         let uri = build_object_uri(&bucket, &key, &[])?;
 
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::PUT,
-                Method::PUT,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket),
-                    bucket: Some(bucket),
+                object: Some(key),
-                    object: Some(key),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.copy_object(req).await {
             Ok(response) => Ok(response.output),
@@ -667,7 +620,11 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         }
     }
 
-    async fn delete_bucket(&self, bucket: &str, access_key: &str, secret_key: &str) -> Result<DeleteBucketOutput, Self::Error> {
+    async fn delete_bucket(
+        &self,
+        bucket: &str,
+        credentials: &rustfs_credentials::Credentials,
+    ) -> Result<DeleteBucketOutput, Self::Error> {
         trace_protocol_request("delete_bucket", Some(bucket), None);
 
         let input = DeleteBucketInput::builder().bucket(bucket.to_string()).build().map_err(|e| {
@@ -675,19 +632,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         })?;
 
         let uri = build_bucket_uri(bucket, &[])?;
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::DELETE,
-                Method::DELETE,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket.to_string()),
-                    bucket: Some(bucket.to_string()),
+                object: None,
-                    object: None,
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.delete_bucket(req).await {
             Ok(response) => Ok(response.output),
@@ -698,8 +652,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
     async fn create_multipart_upload(
         &self,
         input: CreateMultipartUploadInput,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
     ) -> Result<CreateMultipartUploadOutput, Self::Error> {
         trace!(
             event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST,
@@ -715,19 +668,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         let key = input.key.clone();
         let uri = build_object_uri(&bucket, &key, &[("uploads", None)])?;
 
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::POST,
-                Method::POST,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket),
-                    bucket: Some(bucket),
+                object: Some(key),
-                    object: Some(key),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.create_multipart_upload(req).await {
             Ok(response) => Ok(response.output),
@@ -738,8 +688,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
     async fn upload_part(
         &self,
         input: UploadPartInput,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
     ) -> Result<UploadPartOutput, Self::Error> {
         trace!(
             event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST,
@@ -786,19 +735,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
             }
         }
 
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::PUT,
-                Method::PUT,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket),
-                    bucket: Some(bucket),
+                object: Some(key),
-                    object: Some(key),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
         let req = S3Request { headers, ..req };
 
         match self.fs.upload_part(req).await {
@@ -810,8 +756,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
     async fn complete_multipart_upload(
         &self,
         input: CompleteMultipartUploadInput,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
     ) -> Result<CompleteMultipartUploadOutput, Self::Error> {
         trace!(
             event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST,
@@ -828,19 +773,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         let upload_id = input.upload_id.clone();
         let uri = build_object_uri(&bucket, &key, &[("uploadId", Some(upload_id.as_str()))])?;
 
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::POST,
-                Method::POST,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket),
-                    bucket: Some(bucket),
+                object: Some(key),
-                    object: Some(key),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.complete_multipart_upload(req).await {
             Ok(response) => Ok(response.output),
@@ -851,8 +793,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
     async fn abort_multipart_upload(
         &self,
         input: AbortMultipartUploadInput,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
     ) -> Result<AbortMultipartUploadOutput, Self::Error> {
         trace!(
             event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST,
@@ -870,19 +811,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
         let upload_id = input.upload_id.clone();
         let uri = build_object_uri(&bucket, &key, &[("uploadId", Some(upload_id.as_str()))])?;
 
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::DELETE,
-                Method::DELETE,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket),
-                    bucket: Some(bucket),
+                object: Some(key),
-                    object: Some(key),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.abort_multipart_upload(req).await {
             Ok(response) => Ok(response.output),
@@ -893,8 +831,7 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
     async fn upload_part_copy(
         &self,
         input: UploadPartCopyInput,
-        access_key: &str,
+        credentials: &rustfs_credentials::Credentials,
-        secret_key: &str,
     ) -> Result<UploadPartCopyOutput, Self::Error> {
         trace!(
             event = EVENT_PROTOCOL_STORAGE_CLIENT_REQUEST,
@@ -921,19 +858,16 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
             ],
         )?;
 
-        let req = self
+        let req = Self::create_request(
-            .create_request(
+            input,
-                input,
+            Method::PUT,
-                Method::PUT,
+            uri,
-                uri,
+            RequestParams {
-                RequestParams {
+                bucket: Some(bucket),
-                    bucket: Some(bucket),
+                object: Some(key),
-                    object: Some(key),
+                credentials,
-                    access_key,
+            },
-                    secret_key,
+        )?;
-                },
-            )
-            .await?;
 
         match self.fs.upload_part_copy(req).await {
             Ok(response) => Ok(response.output),
@@ -945,6 +879,47 @@ impl rustfs_protocols::common::client::s3::StorageBackend for ProtocolStorageCli
 #[cfg(test)]
 mod tests {
     use super::*;
+    use rustfs_credentials::{IAM_POLICY_CLAIM_NAME_SA, INHERITED_POLICY_TYPE};
+
+    #[test]
+    fn create_request_preserves_authenticated_service_account_identity() {
+        let claims = std::collections::HashMap::from([
+            ("parent".to_string(), serde_json::json!("alice")),
+            (IAM_POLICY_CLAIM_NAME_SA.to_string(), serde_json::json!(INHERITED_POLICY_TYPE)),
+        ]);
+        let credentials = rustfs_credentials::Credentials {
+            access_key: "service-account".to_string(),
+            secret_key: "secret".to_string(),
+            session_token: "signed-service-account-token".to_string(),
+            parent_user: "alice".to_string(),
+            groups: Some(vec!["developers".to_string()]),
+            claims: Some(claims.clone()),
+            ..Default::default()
+        };
+
+        let request = ProtocolStorageClient::create_request(
+            ListObjectsV2Input::default(),
+            Method::GET,
+            http::Uri::from_static("/bucket?list-type=2"),
+            RequestParams {
+                bucket: Some("bucket".to_string()),
+                object: None,
+                credentials: &credentials,
+            },
+        )
+        .expect("request should build");
+        let request_info = request.extensions.get::<ReqInfo>().expect("request info should be present");
+        let copied = request_info.cred.as_ref().expect("credentials should be present");
+
+        assert_eq!(copied.access_key, credentials.access_key);
+        assert_eq!(copied.secret_key, credentials.secret_key);
+        assert_eq!(copied.session_token, credentials.session_token);
+        assert_eq!(copied.parent_user, credentials.parent_user);
+        assert_eq!(copied.groups, credentials.groups);
+        assert_eq!(copied.claims, Some(claims));
+        assert!(copied.is_service_account());
+    }
+
     #[cfg(feature = "webdav")]
     #[test]
     fn request_extensions_preserve_authenticated_identity_and_source_ip() {

rustfs/src/storage/access.rs

@@ -4361,10 +4361,15 @@ mod tests {
         apply_bucket_generation_guard(&req, &bucket, &mut opts).expect("apply the RestoreObject authorization guard");
         assert_eq!(opts.expected_bucket_incarnation_id, Some(authorized_incarnation_id));
 
-        let err = crate::app::object_usecase::DefaultObjectUsecase::with_context(Some(app_context))
+        // The RestoreObject usecase future is large enough that, inlined into
-            .execute_restore_object(req)
+        // this test body, the test thread's 2 MiB stack sits within a few KiB
-            .await
+        // of overflowing on Linux; heap-pin it so unrelated growth in bucket
-            .expect_err("the old RestoreObject authorization must not reach the recreated bucket");
+        // metadata futures cannot tip the test over.
+        let err = Box::pin(
+            crate::app::object_usecase::DefaultObjectUsecase::with_context(Some(app_context)).execute_restore_object(req),
+        )
+        .await
+        .expect_err("the old RestoreObject authorization must not reach the recreated bucket");
         assert_eq!(err.code(), &S3ErrorCode::NoSuchBucket);
         store
             .delete_bucket(&bucket, &DeleteBucketOptions::default())