Merge branch 'main' into houseme/fix-put-small-eager-1mib

crates/ecstore/src/config/com.rs

@@ -800,11 +800,22 @@ where
             if log_error {
                 error!("save_config_with_opts: err: {:?}, file: {}", err, file);
             }
-            Err(err)
+            Err(map_system_metadata_write_error(err, file))
         }
     }
 }
 
+/// A system metadata volume outage must remain retryable instead of being
+/// exposed as the user-facing bucket-not-found response.
+pub(crate) fn map_system_metadata_write_error(err: Error, file: &str) -> Error {
+    match err {
+        Error::BucketNotFound(_) | Error::VolumeNotFound => {
+            Error::InsufficientWriteQuorum(RUSTFS_META_BUCKET.to_string(), file.to_string())
+        }
+        other => other,
+    }
+}
+
 fn new_server_config() -> Config {
     Config::new()
 }
@@ -2796,14 +2807,14 @@ mod tests {
     use super::{
         SERVER_CONFIG_LOCK, ServerConfigSnapshot, apply_dynamic_config_for_sub_sys_with, build_scalar_config_object,
         config_task_join_error, configs_semantically_equal, decode_server_config_blob, encode_server_config_blob,
-        heal_config_descriptor, is_standard_object_server_config, lookup_configs, new_and_save_server_config, read_config,
+        heal_config_descriptor, is_standard_object_server_config, lookup_configs, map_system_metadata_write_error,
-        read_config_no_lock_preserve_empty_with_metadata, read_config_preserve_empty, read_config_with_metadata,
+        new_and_save_server_config, read_config, read_config_no_lock_preserve_empty_with_metadata, read_config_preserve_empty,
-        read_config_without_migrate, read_server_config_snapshot, save_server_config, save_server_config_snapshot,
+        read_config_with_metadata, read_config_without_migrate, read_server_config_snapshot, save_config_with_opts_inner,
-        save_server_config_snapshot_with_generation, server_config_transaction_lock_path, should_warn_ignored_scalar_section,
+        save_server_config, save_server_config_snapshot, save_server_config_snapshot_with_generation,
-        storage_class_kvs_mut,
+        server_config_transaction_lock_path, should_warn_ignored_scalar_section, storage_class_kvs_mut,
     };
     use crate::config::{audit, heal, notify, oidc, scanner};
-    use crate::disk::endpoint::Endpoint;
+    use crate::disk::{RUSTFS_META_BUCKET, endpoint::Endpoint};
     use crate::error::{Error, Result};
     use crate::layout::endpoints::SetupType;
     use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
@@ -2835,6 +2846,72 @@ mod tests {
         assert!(rendered.contains("panicked"));
         assert!(!rendered.contains("do-not-expose-payload"));
     }
+
+    #[test]
+    fn system_metadata_volume_failures_map_to_retryable_write_errors() {
+        for error in [Error::VolumeNotFound, Error::BucketNotFound(RUSTFS_META_BUCKET.to_string())] {
+            assert_eq!(
+                map_system_metadata_write_error(error, "buckets/example/.metadata.bin"),
+                Error::InsufficientWriteQuorum(RUSTFS_META_BUCKET.to_string(), "buckets/example/.metadata.bin".to_string())
+            );
+        }
+
+        let other = Error::other("metadata encoding failed");
+        assert_eq!(map_system_metadata_write_error(other.clone(), "buckets/example/.metadata.bin"), other);
+    }
+
+    #[derive(Debug, Default)]
+    struct MetadataWriteStore {
+        error: Option<Error>,
+    }
+
+    #[async_trait::async_trait]
+    impl crate::storage_api_contracts::object::ObjectIO for MetadataWriteStore {
+        type Error = Error;
+        type RangeSpec = HTTPRangeSpec;
+        type HeaderMap = HeaderMap;
+        type ObjectOptions = ObjectOptions;
+        type ObjectInfo = ObjectInfo;
+        type GetObjectReader = GetObjectReader;
+        type PutObjectReader = PutObjReader;
+
+        async fn get_object_reader(
+            &self,
+            _bucket: &str,
+            _object: &str,
+            _range: Option<Self::RangeSpec>,
+            _headers: Self::HeaderMap,
+            _opts: &Self::ObjectOptions,
+        ) -> core::result::Result<Self::GetObjectReader, Self::Error> {
+            Err(Error::FileNotFound)
+        }
+
+        async fn put_object(
+            &self,
+            _bucket: &str,
+            _object: &str,
+            _data: &mut Self::PutObjectReader,
+            _opts: &Self::ObjectOptions,
+        ) -> core::result::Result<Self::ObjectInfo, Self::Error> {
+            Err(self.error.clone().expect("test store error should be configured"))
+        }
+    }
+
+    #[tokio::test]
+    async fn save_config_preserves_retryable_system_volume_errors() {
+        let store = Arc::new(MetadataWriteStore {
+            error: Some(Error::BucketNotFound(RUSTFS_META_BUCKET.to_string())),
+        });
+        let error =
+            save_config_with_opts_inner(store, "buckets/example/.metadata.bin", Vec::new(), &ObjectOptions::default(), false)
+                .await
+                .expect_err("missing metadata volume must fail");
+
+        assert_eq!(
+            error,
+            Error::InsufficientWriteQuorum(RUSTFS_META_BUCKET.to_string(), "buckets/example/.metadata.bin".to_string())
+        );
+    }
     use rustfs_lock::client::LockClient;
     use rustfs_lock::client::local::LocalClient;
     use rustfs_lock::{LockError, LockInfo, LockResponse, LockStats};

crates/ecstore/src/data_usage/mod.rs

@@ -422,9 +422,7 @@ async fn save_data_usage_in_backend(
     if publication_epoch != expected_publication_epoch {
         return Err(Error::other("data usage publication epoch changed before save"));
     }
-    crate::config::com::save_config(store.clone(), &DATA_USAGE_OBJ_NAME_PATH, data)
+    crate::config::com::save_config(store.clone(), &DATA_USAGE_OBJ_NAME_PATH, data).await?;
-        .await
-        .map_err(Error::other)?;
     drop(publication_guard);
 
     cleanup_observed_data_usage_after_authoritative_save_with_publication(store.as_ref(), &data_usage_info, Some(store.as_ref()))
@@ -641,7 +639,7 @@ where
     {
         Ok(reader) => reader,
         Err(Error::FileNotFound | Error::ObjectNotFound(_, _) | Error::ConfigNotFound) => return Ok(None),
-        Err(err) => return Err(err),
+        Err(err) => return Err(map_data_usage_metadata_read_error(err, object)),
     };
     let revision = reader
         .object_info
@@ -656,6 +654,18 @@ where
     Ok(Some((data_usage_info, revision)))
 }
 
+/// A missing usage object is harmless during bucket creation, but a missing
+/// system metadata volume is a storage outage. Keep the latter retryable and
+/// distinguishable from the user bucket not existing.
+fn map_data_usage_metadata_read_error(err: Error, object: &str) -> Error {
+    match err {
+        Error::BucketNotFound(_) | Error::VolumeNotFound => {
+            Error::InsufficientReadQuorum(RUSTFS_META_BUCKET.to_string(), object.to_string())
+        }
+        other => other,
+    }
+}
+
 fn data_usage_contains_bucket(data_usage_info: &DataUsageInfo, bucket: &str) -> bool {
     data_usage_info.buckets_usage.contains_key(bucket) || data_usage_info.bucket_sizes.contains_key(bucket)
 }
@@ -912,7 +922,7 @@ where
             )
             .await;
         drop(publication_guard);
-        match save_result {
+        match save_result.map_err(|err| crate::config::com::map_system_metadata_write_error(err, object)) {
             Ok(_) => return Ok(()),
             Err(err) => {
                 if let Some((observed, observed_revision)) = load_data_usage_for_bucket_removal(store, object).await? {
@@ -2759,6 +2769,7 @@ mod tests {
     struct UsageCacheReadStore {
         transient_failures: Mutex<usize>,
         reads: Mutex<Vec<String>>,
+        terminal_error: Mutex<Option<Error>>,
     }
 
     impl UsageCacheReadStore {
@@ -2766,6 +2777,15 @@ mod tests {
             Self {
                 transient_failures: Mutex::new(n),
                 reads: Mutex::new(Vec::new()),
+                terminal_error: Mutex::new(None),
+            }
+        }
+
+        fn with_terminal_error(error: Error) -> Self {
+            Self {
+                transient_failures: Mutex::new(0),
+                reads: Mutex::new(Vec::new()),
+                terminal_error: Mutex::new(Some(error)),
             }
         }
 
@@ -2793,6 +2813,9 @@ mod tests {
             _opts: &Self::ObjectOptions,
         ) -> Result<Self::GetObjectReader, Self::Error> {
             self.reads.lock().await.push(object.to_string());
+            if let Some(error) = self.terminal_error.lock().await.clone() {
+                return Err(error);
+            }
             let mut remaining = self.transient_failures.lock().await;
             if *remaining > 0 {
                 *remaining -= 1;
@@ -2857,6 +2880,22 @@ mod tests {
         assert!(!is_data_usage_cache_absent(&Error::DiskNotFound));
     }
 
+    #[test]
+    fn data_usage_removal_maps_missing_system_volume_to_read_quorum() {
+        for error in [Error::VolumeNotFound, Error::BucketNotFound(RUSTFS_META_BUCKET.to_string())] {
+            assert_eq!(
+                map_data_usage_metadata_read_error(error, "bucket-metadata/.usage.json"),
+                Error::InsufficientReadQuorum(RUSTFS_META_BUCKET.to_string(), "bucket-metadata/.usage.json".to_string())
+            );
+        }
+
+        let missing_object = Error::ObjectNotFound(RUSTFS_META_BUCKET.to_string(), "bucket-metadata/.usage.json".to_string());
+        assert_eq!(
+            map_data_usage_metadata_read_error(missing_object.clone(), "bucket-metadata/.usage.json"),
+            missing_object
+        );
+    }
+
     #[tokio::test]
     async fn load_data_usage_cache_treats_absence_as_an_empty_cache_without_retrying() {
         let name = "usage-cache";
@@ -2872,6 +2911,22 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn data_usage_removal_surfaces_missing_system_volume_as_read_quorum() {
+        for cause in [Error::BucketNotFound(RUSTFS_META_BUCKET.to_string()), Error::VolumeNotFound] {
+            let store = UsageCacheReadStore::with_terminal_error(cause);
+
+            let error = load_data_usage_for_bucket_removal(&store, "bucket-metadata/.usage.json")
+                .await
+                .expect_err("missing system metadata volume must not be treated as an absent usage object");
+
+            assert_eq!(
+                error,
+                Error::InsufficientReadQuorum(RUSTFS_META_BUCKET.to_string(), "bucket-metadata/.usage.json".to_string())
+            );
+        }
+    }
+
     #[tokio::test]
     async fn load_data_usage_cache_retries_a_transient_failure() {
         let name = "usage-cache";

crates/ecstore/src/set_disk/mod.rs

@@ -2525,6 +2525,7 @@ fn record_get_object_reader_path_observation(
             GET_OBJECT_PATH_CODEC_STREAMING => 5,
             GET_OBJECT_PATH_REMOTE_TRANSITION => 6,
             GET_OBJECT_PATH_EMPTY => 7,
+            GET_OBJECT_PATH_LEGACY_DUPLEX => 8,
             _ => 255,
         },
         Ordering::Relaxed,

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

@@ -89,7 +89,10 @@ use tokio::io::AsyncWriteExt;
 const ENV_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE: &str = "RUSTFS_GET_MID_SIZE_STREAMING_ENABLE";
 const DEFAULT_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE: bool = true;
 const GET_MID_SIZE_STREAMING_MIN_SIZE: usize = 128 * 1024 + 1;
-const GET_MID_SIZE_STREAMING_MAX_SIZE: usize = 1024 * 1024;
+// Exclude 1 MiB from the bounded mid-size reader until it has a demonstrated
+// high-concurrency performance envelope; existing codec/legacy gates decide
+// which established reader handles the object.
+const GET_MID_SIZE_STREAMING_MAX_SIZE: usize = 512 * 1024;
 
 fn is_get_mid_size_streaming_enabled() -> bool {
     #[cfg(test)]
@@ -8408,14 +8411,14 @@ mod mid_size_streaming_gate_tests {
 
     #[test]
     #[serial]
-    fn mid_size_streaming_includes_one_mib_and_rejects_larger_objects() {
+    fn mid_size_streaming_stops_at_512kib_and_rejects_one_mib() {
-        let (object_info, fi) = plain_metadata(1024 * 1024);
+        let (object_info, fi) = plain_metadata(512 * 1024);
         assert_eq!(
             get_mid_size_streaming_object_size_with_flags(&None, &object_info, &fi, &ObjectOptions::default(), true, true, true),
-            Some(1024 * 1024)
+            Some(512 * 1024)
         );
 
-        let (large_info, large_fi) = plain_metadata(1024 * 1024 + 1);
+        let (large_info, large_fi) = plain_metadata(512 * 1024 + 1);
         assert_eq!(
             get_mid_size_streaming_object_size_with_flags(
                 &None,
@@ -8428,6 +8431,20 @@ mod mid_size_streaming_gate_tests {
             ),
             None
         );
+
+        let (one_mib_info, one_mib_fi) = plain_metadata(1024 * 1024);
+        assert_eq!(
+            get_mid_size_streaming_object_size_with_flags(
+                &None,
+                &one_mib_info,
+                &one_mib_fi,
+                &ObjectOptions::default(),
+                true,
+                true,
+                true
+            ),
+            None
+        );
     }
 
     #[test]
@@ -9801,6 +9818,57 @@ mod inline_put_commit_path_tests {
         .await;
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn get_object_reader_routes_one_mib_away_from_mid_size_reader() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "one-mib-legacy-reader";
+        let object = "object.bin";
+        let payload = vec![0x5a; 1024 * 1024];
+        make_bucket(&disk_stores, bucket).await;
+        let storage_class = temp_env::with_var(INLINE_BLOCK_ENV, Some("1KiB"), || lookup_config_for_pools(&KVS::new(), &[4]))
+            .expect("test storage class should resolve");
+        set_disks.set_test_storage_class_config(storage_class);
+
+        let mut writer = PutObjReader::from_vec(payload.clone());
+        temp_env::async_with_vars(
+            [
+                (ENV_RUSTFS_GET_MID_SIZE_STREAMING_ENABLE, Some("true")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
+                (crate::set_disk::ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("off")),
+                (rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some("true")),
+            ],
+            async {
+                set_disks
+                    .put_object(bucket, object, &mut writer, &ObjectOptions::default())
+                    .await
+                    .expect("1 MiB fixture should commit");
+
+                crate::set_disk::reset_test_get_object_reader_path();
+                let mut reader = set_disks
+                    .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                    .await
+                    .expect("1 MiB legacy GET should succeed");
+                let mut restored = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut restored)
+                    .await
+                    .expect("1 MiB legacy reader should stream");
+
+                assert_eq!(restored, payload);
+                assert_eq!(
+                    crate::set_disk::test_get_object_reader_path_id(),
+                    8,
+                    "1 MiB must bypass mid-size and use legacy duplex when codec rollout is off"
+                );
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     async fn repeated_gets_reuse_the_set_erasure_shell() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;