Merge branch 'main' into houseme/feat/mrf-staging-delivery

Cargo.lock

@@ -9503,6 +9503,7 @@ dependencies = [
  "clap",
  "const-str",
  "datafusion",
+ "faster-hex",
  "flatbuffers",
  "flate2",
  "futures",
@@ -9527,6 +9528,7 @@ dependencies = [
  "metrics",
  "metrics-util",
  "mime_guess",
+ "moka",
  "opentelemetry",
  "opentelemetry_sdk",
  "p256 0.14.0",
@@ -9619,6 +9621,7 @@ dependencies = [
  "urlencoding",
  "uuid",
  "x509-parser",
+ "xxhash-rust",
  "zeroize",
  "zip",
  "zstd 0.14.0",

crates/ecstore/src/bucket/metadata_sys.rs

@@ -1043,11 +1043,8 @@ pub async fn get_on_demand_migration_config(bucket: &str) -> Result<Option<(Vec<
 }
 
 /// Resolve opaque configuration from the store's own metadata system.
-pub async fn get_on_demand_migration_config_in(
-    ctx: &crate::runtime::instance::InstanceContext,
-    bucket: &str,
-) -> Result<Option<(Vec<u8>, OffsetDateTime)>> {
-    let sys = bucket_metadata_sys_of(ctx)?;
+pub async fn get_on_demand_migration_config_in(api: &ECStore, bucket: &str) -> Result<Option<(Vec<u8>, OffsetDateTime)>> {
+    let sys = bucket_metadata_sys_of(&api.ctx)?;
     let lock = sys.read().await;
     lock.get_on_demand_migration_config(bucket).await
 }

crates/ecstore/src/store/init.rs

@@ -353,6 +353,11 @@ async fn resume_rebalance_after_init(store: Arc<ECStore>, rx: CancellationToken)
 }
 
 impl ECStore {
+    /// Shutdown token owned by this store instance.
+    pub fn background_cancel_token(&self) -> Option<CancellationToken> {
+        self.ctx.background_cancel_token()
+    }
+
     /// Validate topology and process storage-class overrides before any disk is opened.
     pub fn validate_startup_storage_class(endpoint_pools: &EndpointServerPools) -> Result<()> {
         let drive_counts = startup_pool_drive_counts(endpoint_pools);

crates/madmin/src/on_demand_migration.rs

@@ -17,7 +17,7 @@
 //! Wire types for `PUT`/`GET`/`DELETE /v3/on-demand-migration/{bucket}`,
 //! `GET .../status`, `POST .../backfill?op=start|cancel` and
 //! `GET .../backfill` (ODM-12), mirroring the server's config model
-//! (`crates/ecstore/src/bucket/on_demand_migration/config.rs`) and handler
+//! (`rustfs/src/on_demand_migration/config.rs`) and handler
 //! responses (`rustfs/src/admin/handlers/on_demand_migration.rs`). The SDK
 //! owns its own copies, madmin-go style; the fixtures under
 //! `fixtures/on_demand_migration/` are the contract both sides pin

rustfs/Cargo.toml

@@ -276,6 +276,7 @@ rcgen = { workspace = true }
 # Async Runtime and Networking
 async-trait = { workspace = true }
 axum.workspace = true
+faster-hex.workspace = true
 futures.workspace = true
 futures-lite.workspace = true
 futures-util.workspace = true
@@ -291,6 +292,8 @@ tokio-rustls = { workspace = true, default-features = false, features = ["loggin
 aws-smithy-runtime-api = { workspace = true, features = ["http-1x"] }
 aws-smithy-types = { workspace = true }
 google-cloud-auth = { workspace = true, optional = true }
+moka = { workspace = true, features = ["sync"] }
+xxhash-rust = { workspace = true, features = ["xxh3"] }
 aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
 tokio-stream.workspace = true
 tokio-util = { workspace = true, features = ["io", "compat", "time"] }

rustfs/src/app/bucket_list_through.rs

@@ -443,7 +443,9 @@ mod tests {
     use super::*;
     use crate::app::bucket_usecase::DefaultBucketUsecase;
     use crate::app::gating_test_env::{run_large_stack_test, shared_gating_ecstore};
-    use crate::app::storage_api::bucket_usecase::s3::{ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response};
+    use crate::app::storage_api::bucket_usecase::s3::{
+        ListObjectsInput, ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response, XmlSerialize, XmlSerializer,
+    };
     use crate::app::storage_api::test::StoragePutObjReader;
     use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::app::storage_api::test::contract::object::ObjectIO as _;
@@ -451,7 +453,6 @@ mod tests {
         FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig,
         SourceCredentials, TlsConfig,
     };
-    use s3s::dto::ListObjectsInput;
     use std::time::Duration;
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
@@ -864,7 +865,7 @@ mod tests {
                             assert_eq!(output.next_marker.as_deref(), (index == 0).then_some(expected_key));
 
                             let mut xml = Vec::new();
-                            s3s::xml::Serialize::serialize(&output, &mut s3s::xml::Serializer::new(&mut xml))
+                            XmlSerialize::serialize(&output, &mut XmlSerializer::new(&mut xml))
                                 .expect("serialize the real v1 response");
                             assert!(!xml.contains(&0), "XML 1.0 forbids NUL in NextMarker");
                             let mut reader = quick_xml::Reader::from_reader(xml.as_slice());

rustfs/src/app/storage_api.rs

@@ -27,12 +27,16 @@ pub(crate) fn EndpointServerPools(
 /// S3 wire types for app-layer modules, funneled here so new files stay off
 /// the direct s3s surface (s3s footprint ratchet, `scripts/check_s3s_footprint.sh`).
 pub(crate) mod s3 {
+    #[cfg(test)]
+    pub(crate) use s3s::dto::ListObjectsInput;
     #[cfg(test)]
     pub(crate) use s3s::dto::{
         BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ListObjectsV2Input,
         ListObjectsV2Output, ReplicationConfiguration, ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus,
         ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, VersioningConfiguration,
     };
+    #[cfg(test)]
+    pub(crate) use s3s::xml::{Serialize as XmlSerialize, Serializer as XmlSerializer};
     pub(crate) use s3s::{S3Error, S3ErrorCode, S3Result};
     #[cfg(test)]
     pub(crate) use s3s::{S3Request, S3Response};

rustfs/src/on_demand_migration/backfill.rs

@@ -471,7 +471,7 @@ impl BackfillContext for BucketBackfillContext {
 
     async fn config_updated_at(&self) -> Result<Option<OffsetDateTime>, StorageError> {
         Ok(
-            super::config::decode_stored_config(get_on_demand_migration_config_in(&self.api.ctx, self.state.bucket()).await?)?
+            super::config::decode_stored_config(get_on_demand_migration_config_in(&self.api, self.state.bucket()).await?)?
                 .map(|(_, updated_at)| updated_at),
         )
     }
@@ -1472,7 +1472,7 @@ impl Job {
 /// Spawns [`run_backfill_recovery_loop`] on the store's shutdown token;
 /// `false` (nothing spawned) when the store has no background token.
 pub fn spawn_backfill_recovery_loop(runner: Arc<BackfillRunner>) -> bool {
-    let Some(cancel) = runner.api.ctx.background_cancel_token() else {
+    let Some(cancel) = runner.api.background_cancel_token() else {
         return false;
     };
     tokio::spawn(run_backfill_recovery_loop(runner, cancel));