fix: align object version limit handling (#7415)

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/config/README.md

@@ -66,6 +66,11 @@ Current guidance:
 
 - `RUSTFS_BROWSER_REDIRECT_URL` sets the externally reachable browser origin used for OIDC callback, console success redirect, and logout fallback URLs. Configure it to the public scheme and authority without a path, for example `https://console.example.com`. In load-balancer deployments, keep OIDC authorize and callback requests on the same backend node because the in-flight OIDC `state` is local to the RustFS node.
 
+## S3 API environment variables
+
+- `RUSTFS_API_OBJECT_MAX_VERSIONS` caps the number of retained versions for a single object. It defaults to `9223372036854775807`, matching MinIO's practical-unlimited default. Set a positive integer to enforce a lower per-object metadata bound.
+- `MINIO_API_OBJECT_MAX_VERSIONS` is accepted as a compatibility alias when the canonical RustFS variable is not set.
+
 ## Distributed endpoint locality
 
 - `RUSTFS_LOCAL_ENDPOINT_HOST` identifies this server's host in a distributed `RUSTFS_VOLUMES` topology without resolving every peer during startup. Set it to exactly one host, without a scheme, port, or path. It is accepted only for orchestrated URL topologies and must match at least one endpoint on the RustFS server port; invalid or unmatched values fail startup. Leave it unset to retain DNS-based locality discovery.

crates/config/src/constants/api.rs

@@ -90,3 +90,15 @@ pub const ENV_API_MAX_CONNECTIONS: &str = "RUSTFS_API_MAX_CONNECTIONS";
 
 /// Default for `RUSTFS_API_MAX_CONNECTIONS` (`0` = unlimited).
 pub const DEFAULT_API_MAX_CONNECTIONS: usize = 0;
+
+/// Maximum retained versions per object.
+///
+/// The default follows MinIO and is effectively unlimited for practical
+/// deployments. Operators can lower it to bound per-object metadata growth.
+/// Environment variable: RUSTFS_API_OBJECT_MAX_VERSIONS
+/// MinIO-compatible alias: MINIO_API_OBJECT_MAX_VERSIONS
+/// Example: RUSTFS_API_OBJECT_MAX_VERSIONS=50000
+pub const ENV_API_OBJECT_MAX_VERSIONS: &str = "RUSTFS_API_OBJECT_MAX_VERSIONS";
+
+/// Default for `RUSTFS_API_OBJECT_MAX_VERSIONS`.
+pub const DEFAULT_API_OBJECT_MAX_VERSIONS: u64 = 9_223_372_036_854_775_807;

crates/ecstore/src/disk/error.rs

@@ -425,6 +425,7 @@ impl From<rustfs_filemeta::Error> for DiskError {
             rustfs_filemeta::Error::FileVersionNotFound => DiskError::FileVersionNotFound,
             rustfs_filemeta::Error::FileCorrupt => DiskError::FileCorrupt,
             rustfs_filemeta::Error::MethodNotAllowed => DiskError::MethodNotAllowed,
+            rustfs_filemeta::Error::MaxVersionsExceeded => DiskError::MaxVersionsExceeded,
             e => DiskError::other(e),
         }
     }

crates/ecstore/src/error/mod.rs

@@ -588,6 +588,7 @@ impl From<rustfs_filemeta::Error> for StorageError {
             rustfs_filemeta::Error::FileVersionNotFound => StorageError::FileVersionNotFound,
             rustfs_filemeta::Error::FileCorrupt => StorageError::FileCorrupt,
             rustfs_filemeta::Error::Unexpected => StorageError::Unexpected,
+            rustfs_filemeta::Error::MaxVersionsExceeded => StorageError::MaxVersionsExceeded,
             rustfs_filemeta::Error::Io(io_error) => io_error.into(),
             _ => StorageError::Io(std::io::Error::other(e)),
         }

crates/filemeta/src/error.rs

@@ -37,6 +37,9 @@ pub enum Error {
     #[error("Method not allowed")]
     MethodNotAllowed,
 
+    #[error("You've exceeded the limit on the number of versions you can create on this object")]
+    MaxVersionsExceeded,
+
     #[error("Unexpected error")]
     Unexpected,
 
@@ -86,6 +89,7 @@ impl PartialEq for Error {
             (Error::FileCorrupt, Error::FileCorrupt) => true,
             (Error::DoneForNow, Error::DoneForNow) => true,
             (Error::MethodNotAllowed, Error::MethodNotAllowed) => true,
+            (Error::MaxVersionsExceeded, Error::MaxVersionsExceeded) => true,
             (Error::FileNotFound, Error::FileNotFound) => true,
             (Error::FileVersionNotFound, Error::FileVersionNotFound) => true,
             (Error::VolumeNotFound, Error::VolumeNotFound) => true,
@@ -111,6 +115,7 @@ impl Clone for Error {
             Error::FileCorrupt => Error::FileCorrupt,
             Error::DoneForNow => Error::DoneForNow,
             Error::MethodNotAllowed => Error::MethodNotAllowed,
+            Error::MaxVersionsExceeded => Error::MaxVersionsExceeded,
             Error::VolumeNotFound => Error::VolumeNotFound,
             Error::Io(e) => Error::Io(std::io::Error::new(e.kind(), e.to_string())),
             Error::RmpSerdeDecode(s) => Error::RmpSerdeDecode(s.clone()),

crates/filemeta/src/filemeta.rs

@@ -34,11 +34,14 @@ use rustfs_utils::http::{
 };
 use s3s::header::X_AMZ_RESTORE;
 use serde::{Deserialize, Serialize};
+#[cfg(test)]
+use std::cell::Cell;
 use std::cmp::Ordering;
 use std::collections::BTreeMap;
 use std::convert::TryFrom;
 use std::hash::Hasher;
 use std::io::{Read, Write};
+use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
 use std::{collections::HashMap, io::Cursor};
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
@@ -67,8 +70,46 @@ const _XL_FLAG_INLINE_DATA: u8 = 1 << 2;
 const META_DATA_READ_DEFAULT: usize = 4 << 10;
 const MSGP_UINT32_SIZE: usize = 5;
 
-/// Max object versions per object, default is 10000
-const DEFAULT_OBJECT_MAX_VERSIONS: usize = 10000;
+/// Default max object versions per object, aligned with MinIO's default.
+pub const DEFAULT_OBJECT_MAX_VERSIONS: usize = if usize::BITS >= 64 {
+    9_223_372_036_854_775_807
+} else {
+    usize::MAX
+};
+
+static OBJECT_MAX_VERSIONS: AtomicUsize = AtomicUsize::new(DEFAULT_OBJECT_MAX_VERSIONS);
+
+#[cfg(test)]
+thread_local! {
+    static OBJECT_MAX_VERSIONS_OVERRIDE: Cell<Option<usize>> = const { Cell::new(None) };
+}
+
+#[inline]
+pub fn object_max_versions() -> usize {
+    #[cfg(test)]
+    if let Some(limit) = OBJECT_MAX_VERSIONS_OVERRIDE.with(Cell::get) {
+        return limit;
+    }
+
+    OBJECT_MAX_VERSIONS.load(AtomicOrdering::Relaxed)
+}
+
+pub fn set_object_max_versions(limit: usize) -> Result<()> {
+    if limit == 0 {
+        return Err(Error::other("object max versions must be greater than 0"));
+    }
+    OBJECT_MAX_VERSIONS.store(limit, AtomicOrdering::Relaxed);
+    Ok(())
+}
+
+#[cfg(test)]
+fn set_object_max_versions_override_for_test(limit: Option<usize>) -> Option<usize> {
+    OBJECT_MAX_VERSIONS_OVERRIDE.with(|override_limit| {
+        let previous = override_limit.get();
+        override_limit.set(limit);
+        previous
+    })
+}
 
 /// Returns the inline data map key for a version_id. "null" for null version.
 pub(crate) fn data_key_for_version(version_id: Option<Uuid>) -> String {
@@ -460,18 +501,6 @@ impl FileMeta {
             return Err(Error::other("file meta version invalid"));
         }
 
-        // check max versions limit
-        if self.versions.len() + 1 > DEFAULT_OBJECT_MAX_VERSIONS {
-            return Err(Error::other(
-                "You've exceeded the limit on the number of versions you can create on this object",
-            ));
-        }
-
-        if self.versions.is_empty() {
-            self.versions.push(FileMetaShallowVersion::try_from(version)?);
-            return Ok(());
-        }
-
         let vid = version.get_version_id();
         let vid_is_null = vid.is_none() || vid == Some(Uuid::nil());
         let existing_idx = if vid_is_null {
@@ -490,6 +519,15 @@ impl FileMeta {
             return self.set_idx(fidx, version);
         }
 
+        if self.versions.len() >= object_max_versions() {
+            return Err(Error::MaxVersionsExceeded);
+        }
+
+        if self.versions.is_empty() {
+            self.versions.push(FileMetaShallowVersion::try_from(version)?);
+            return Ok(());
+        }
+
         let new_shallow = FileMetaShallowVersion::try_from(version)?;
         let insert_pos = self
             .versions
@@ -1330,6 +1368,88 @@ mod test {
         }
     }
 
+    struct ObjectMaxVersionsRestore {
+        previous: Option<usize>,
+    }
+
+    impl Drop for ObjectMaxVersionsRestore {
+        fn drop(&mut self) {
+            set_object_max_versions_override_for_test(self.previous);
+        }
+    }
+
+    fn with_object_max_versions_for_test<R>(limit: usize, test: impl FnOnce() -> R) -> R {
+        let previous = set_object_max_versions_override_for_test(Some(limit));
+        let _restore = ObjectMaxVersionsRestore { previous };
+        test()
+    }
+
+    #[test]
+    fn add_version_filemata_rejects_new_version_above_configured_limit() {
+        with_object_max_versions_for_test(2, || {
+            let mut fm = FileMeta::new();
+            fm.add_version_filemata(valid_object_version(Uuid::from_u128(1), vec![10, 20]))
+                .expect("add first version within limit");
+            fm.add_version_filemata(valid_object_version(Uuid::from_u128(2), vec![10, 20]))
+                .expect("add second version at limit");
+
+            let err = fm
+                .add_version_filemata(valid_object_version(Uuid::from_u128(3), vec![10, 20]))
+                .expect_err("new version above limit must fail");
+
+            assert_eq!(err, Error::MaxVersionsExceeded);
+            assert_eq!(fm.versions.len(), 2, "failed insert must not mutate version list");
+        });
+    }
+
+    #[test]
+    fn add_version_filemata_allows_same_version_replacement_at_limit() {
+        with_object_max_versions_for_test(2, || {
+            let mut fm = FileMeta::new();
+            let target = Uuid::from_u128(10);
+            fm.add_version_filemata(valid_object_version(target, vec![10, 20]))
+                .expect("add target version");
+            fm.add_version_filemata(valid_object_version(Uuid::from_u128(20), vec![10, 20]))
+                .expect("add peer version at limit");
+
+            fm.add_version_filemata(valid_object_version(target, vec![30, 40]))
+                .expect("same version replacement at limit must succeed");
+
+            assert_eq!(fm.versions.len(), 2);
+            let replaced = fm
+                .versions
+                .iter()
+                .find(|version| version.header.version_id == Some(target))
+                .expect("target version must remain present")
+                .parse_version_meta()
+                .expect("parse replaced version");
+            assert_eq!(replaced.object.expect("object version").part_sizes, vec![30, 40]);
+        });
+    }
+
+    #[test]
+    fn add_version_allows_null_version_replacement_at_limit() {
+        with_object_max_versions_for_test(1, || {
+            let mut fm = FileMeta::new();
+            let mut first = FileInfo::new("object", 2, 2);
+            first.mod_time = Some(OffsetDateTime::now_utc());
+            first.version_id = None;
+            fm.add_version(first).expect("add initial null version");
+
+            let mut replacement = FileInfo::new("object", 2, 2);
+            replacement.mod_time = Some(OffsetDateTime::now_utc());
+            replacement.version_id = None;
+            replacement.size = 42;
+            fm.add_version(replacement)
+                .expect("null version replacement at limit must succeed");
+
+            assert_eq!(fm.versions.len(), 1);
+            assert_eq!(fm.versions[0].header.version_id, Some(Uuid::nil()));
+            let replaced = fm.versions[0].parse_version_meta().expect("parse null replacement");
+            assert_eq!(replaced.object.expect("object version").size, 42);
+        });
+    }
+
     #[test]
     fn add_version_filemata_uses_canonical_equal_time_order() {
         let mod_time = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("valid test timestamp");

crates/utils/src/envs.rs

@@ -125,6 +125,7 @@ const EXTERNAL_COMPATIBLE_SUFFIXES: &[&str] = &[
     "ACCESS_KEY",
     "ACCESS_KEY_FILE",
     "ADDRESS",
+    "API_OBJECT_MAX_VERSIONS",
     "API_XFF_HEADER",
     "AUDIT_WEBHOOK_AUTH_TOKEN",
     "AUDIT_WEBHOOK_CLIENT_CERT",
@@ -900,4 +901,15 @@ mod tests {
             });
         });
     }
+
+    #[test]
+    fn external_env_compat_includes_api_object_max_versions() {
+        let report =
+            build_external_env_compat_report_from_entries([("MINIO_API_OBJECT_MAX_VERSIONS".to_string(), "50000".to_string())]);
+
+        assert_eq!(
+            report.mapped_pairs,
+            vec![("MINIO_API_OBJECT_MAX_VERSIONS".to_string(), "RUSTFS_API_OBJECT_MAX_VERSIONS".to_string())]
+        );
+    }
 }

rustfs/src/error.rs

@@ -14,9 +14,13 @@
 
 use crate::storage_api::error::contract::{StorageErrorCode, range::HTTPRangeError};
 use crate::storage_api::error::{QuotaError, StorageError};
+use http::StatusCode;
 use rustfs_kms::KmsUnavailableError;
 use s3s::{S3Error, S3ErrorCode};
 
+const MAX_VERSIONS_EXCEEDED_CODE: &str = "MaxVersionsExceeded";
+const MAX_VERSIONS_EXCEEDED_MESSAGE: &str = "You've exceeded the limit on the number of versions you can create on this object";
+
 /// Marks a request body that exceeded a presigned upload size capability.
 ///
 /// This marker must survive the body-reader and storage layers so the client
@@ -284,6 +288,9 @@ impl ApiError {
             S3ErrorCode::EvaluatorBindingDoesNotExist => "A column name or a path provided does not exist in the SQL expression".to_string(),
             S3ErrorCode::InvalidColumnIndex => "The column index is invalid. Please check the service documentation and try again.".to_string(),
             S3ErrorCode::UnsupportedFunction => "Encountered an unsupported SQL function.".to_string(),
+            S3ErrorCode::Custom(code) if &**code == MAX_VERSIONS_EXCEEDED_CODE => {
+                MAX_VERSIONS_EXCEEDED_MESSAGE.to_string()
+            }
             _ => code.as_str().to_string(),
         }
     }
@@ -362,6 +369,9 @@ fn error_chain_s3s_body_stream_error(err: &(dyn std::error::Error + 'static)) ->
 impl From<ApiError> for S3Error {
     fn from(err: ApiError) -> Self {
         let mut s3e = S3Error::with_message(err.code, err.message);
+        if matches!(s3e.code(), S3ErrorCode::Custom(code) if &**code == MAX_VERSIONS_EXCEEDED_CODE) {
+            s3e.set_status_code(StatusCode::BAD_REQUEST);
+        }
         if let Some(source) = err.source {
             s3e.set_source(source);
         }
@@ -455,6 +465,7 @@ impl From<StorageError> for ApiError {
             | StorageError::InsufficientWriteQuorum(_, _) => S3ErrorCode::ServiceUnavailable,
             StorageError::NamespaceLockQuorumUnavailable { .. } => S3ErrorCode::ServiceUnavailable,
             StorageError::QuotaExceeded { .. } => S3ErrorCode::InvalidRequest,
+            StorageError::MaxVersionsExceeded => S3ErrorCode::Custom(MAX_VERSIONS_EXCEEDED_CODE.into()),
             StorageError::Lock(_) => S3ErrorCode::ServiceUnavailable,
             StorageError::DecommissionNotStarted => S3ErrorCode::InvalidRequest,
             StorageError::DecommissionAlreadyRunning => S3ErrorCode::InvalidRequest,
@@ -485,6 +496,8 @@ impl From<StorageError> for ApiError {
 
         let message = if matches!(&err, StorageError::QuotaExceeded { .. }) {
             err.to_string()
+        } else if matches!(&err, StorageError::MaxVersionsExceeded) {
+            ApiError::error_code_to_message(&code)
         } else if code == S3ErrorCode::InternalError && matches!(&err, StorageError::Io(_)) {
             ApiError::error_code_to_message(&code)
         } else if code == S3ErrorCode::InternalError {
@@ -1189,6 +1202,19 @@ mod tests {
         assert_eq!(api_error.message, "Bucket quota exceeded. Current usage: 5 bytes, limit: 10 bytes");
     }
 
+    #[test]
+    fn max_versions_exceeded_maps_to_minio_compatible_s3_error() {
+        let api_error: ApiError = StorageError::MaxVersionsExceeded.into();
+
+        assert_eq!(api_error.code, S3ErrorCode::Custom(MAX_VERSIONS_EXCEEDED_CODE.into()));
+        assert_eq!(api_error.message, MAX_VERSIONS_EXCEEDED_MESSAGE);
+
+        let s3_error: S3Error = api_error.into();
+        assert_eq!(s3_error.code(), &S3ErrorCode::Custom(MAX_VERSIONS_EXCEEDED_CODE.into()));
+        assert_eq!(s3_error.message(), Some(MAX_VERSIONS_EXCEEDED_MESSAGE));
+        assert_eq!(s3_error.status_code(), Some(StatusCode::BAD_REQUEST));
+    }
+
     #[test]
     fn test_api_error_to_s3_error_without_source() {
         let api_error = ApiError {

rustfs/src/startup_runtime.rs

@@ -17,12 +17,108 @@ use crate::{
     startup_runtime_hooks::{init_profiling_runtime, install_default_crypto_provider, log_startup_runtime_diagnostics},
     startup_tls_material::init_outbound_tls_material,
 };
-use std::io::Result;
+use rustfs_config::ENV_API_OBJECT_MAX_VERSIONS;
+use rustfs_utils::EnvParseOutcome;
+use std::io::{Error, Result};
 
 pub(crate) async fn init_startup_runtime_foundation(config: &Config) -> Result<()> {
     log_startup_runtime_diagnostics();
     init_profiling_runtime().await;
     rustfs_trusted_proxies::init();
     install_default_crypto_provider();
+    init_object_max_versions_config()?;
     init_outbound_tls_material(config).await
 }
+
+fn init_object_max_versions_config() -> Result<()> {
+    let limit = match rustfs_utils::get_env_parse_outcome::<u64>(ENV_API_OBJECT_MAX_VERSIONS) {
+        EnvParseOutcome::Absent => rustfs_filemeta::DEFAULT_OBJECT_MAX_VERSIONS,
+        EnvParseOutcome::Invalid => {
+            return Err(Error::other(format!(
+                "{ENV_API_OBJECT_MAX_VERSIONS} must be a positive integer no greater than {}",
+                usize::MAX
+            )));
+        }
+        EnvParseOutcome::Parsed(value) => object_max_versions_limit_from_u64(value)?,
+    };
+
+    rustfs_filemeta::set_object_max_versions(limit).map_err(Error::other)
+}
+
+fn object_max_versions_limit_from_u64(value: u64) -> Result<usize> {
+    if value == 0 {
+        return Err(Error::other(format!(
+            "{ENV_API_OBJECT_MAX_VERSIONS} must be a positive integer no greater than {}",
+            usize::MAX
+        )));
+    }
+
+    usize::try_from(value).map_err(|_| {
+        Error::other(format!(
+            "{ENV_API_OBJECT_MAX_VERSIONS} must be a positive integer no greater than {}",
+            usize::MAX
+        ))
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    struct ObjectMaxVersionsRestore {
+        previous: usize,
+    }
+
+    impl Drop for ObjectMaxVersionsRestore {
+        fn drop(&mut self) {
+            rustfs_filemeta::set_object_max_versions(self.previous).expect("restore object max versions limit after test");
+        }
+    }
+
+    fn with_object_max_versions_env<R>(rustfs_value: Option<&str>, minio_value: Option<&str>, test: impl FnOnce() -> R) -> R {
+        static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+        let _serial = LOCK.lock().expect("serialize object max versions env tests");
+        let previous = rustfs_filemeta::object_max_versions();
+        let _restore = ObjectMaxVersionsRestore { previous };
+
+        temp_env::with_vars(
+            [
+                (rustfs_config::ENV_API_OBJECT_MAX_VERSIONS, rustfs_value),
+                ("MINIO_API_OBJECT_MAX_VERSIONS", minio_value),
+            ],
+            test,
+        )
+    }
+
+    #[test]
+    fn object_max_versions_env_sets_filemeta_limit() {
+        with_object_max_versions_env(Some("3"), None, || {
+            init_object_max_versions_config().expect("valid object max versions env must initialize");
+            assert_eq!(rustfs_filemeta::object_max_versions(), 3);
+        });
+    }
+
+    #[test]
+    fn minio_object_max_versions_env_alias_sets_filemeta_limit() {
+        with_object_max_versions_env(None, Some("4"), || {
+            init_object_max_versions_config().expect("valid MinIO alias must initialize");
+            assert_eq!(rustfs_filemeta::object_max_versions(), 4);
+        });
+    }
+
+    #[test]
+    fn object_max_versions_env_rejects_zero() {
+        with_object_max_versions_env(Some("0"), None, || {
+            let err = init_object_max_versions_config().expect_err("zero object max versions must fail startup config");
+            assert!(err.to_string().contains(rustfs_config::ENV_API_OBJECT_MAX_VERSIONS));
+        });
+    }
+
+    #[test]
+    fn object_max_versions_env_rejects_malformed_value() {
+        with_object_max_versions_env(Some("not-a-number"), None, || {
+            let err = init_object_max_versions_config().expect_err("malformed object max versions must fail startup config");
+            assert!(err.to_string().contains(rustfs_config::ENV_API_OBJECT_MAX_VERSIONS));
+        });
+    }
+}