Merge branch 'main' into cxymds/feat/pool-write-block-diagnostics

Signed-off-by: houseme <housemecn@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
+/// Default max object versions per object, aligned with MinIO's default.
-const DEFAULT_OBJECT_MAX_VERSIONS: usize = 10000;
+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/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs

@@ -37,6 +37,8 @@ fn segment_proof() -> SegmentInvalidationProof {
         key_format: envelope.key_format,
         baseline_scan_plan_digest: envelope.baseline_scan_plan_digest,
         process_epoch: envelope.process_epoch,
+        generation_start: envelope.generation_start,
+        generation_end: envelope.generation_end,
         durable_producer_identity: true,
         invalidation_domain: SegmentInvalidationDomain::LocalSingleSet,
         distributed_ec_invalidation: false,
@@ -126,6 +128,20 @@ fn segment_observation_trusted_proposal_requires_identity_and_complete_producer_
         Err(SegmentInvalidationError::InvalidProof)
     );
 
+    let mut wrong_generation_start = proof.clone();
+    wrong_generation_start.generation_start = wrong_generation_start.generation_start.saturating_sub(1);
+    assert_eq!(
+        admit_segment_invalidation(&envelope, &wrong_generation_start, ["hot/one"]),
+        Err(SegmentInvalidationError::InvalidProof)
+    );
+
+    let mut wrong_generation_end = proof.clone();
+    wrong_generation_end.generation_end = wrong_generation_end.generation_end.saturating_add(1);
+    assert_eq!(
+        admit_segment_invalidation(&envelope, &wrong_generation_end, ["hot/one"]),
+        Err(SegmentInvalidationError::InvalidProof)
+    );
+
     let mut no_durable_identity = proof.clone();
     no_durable_identity.durable_producer_identity = false;
     assert_eq!(

crates/scanner/src/segment_invalidation.rs

@@ -77,6 +77,8 @@ pub struct SegmentInvalidationProof {
     pub key_format: u16,
     pub baseline_scan_plan_digest: DataUsageScanPlanDigest,
     pub process_epoch: String,
+    pub generation_start: u64,
+    pub generation_end: u64,
     pub durable_producer_identity: bool,
     pub invalidation_domain: SegmentInvalidationDomain,
     pub distributed_ec_invalidation: bool,
@@ -108,11 +110,15 @@ fn validate_segment_invalidation_proof(
         || envelope.baseline_scan_plan_digest != proof.baseline_scan_plan_digest
         || envelope.process_epoch.is_empty()
         || envelope.process_epoch != proof.process_epoch
+        || envelope.generation_start != proof.generation_start
+        || envelope.generation_end != proof.generation_end
         || !proof.durable_producer_identity
         || !proof.cold_zero_walk_oracle
         || (proof.invalidation_domain == SegmentInvalidationDomain::DistributedEc && !proof.distributed_ec_invalidation)
         || envelope.generation_start == 0
         || envelope.generation_end < envelope.generation_start
+        || proof.generation_start == 0
+        || proof.generation_end < proof.generation_start
         || envelope.restart_gap
         || envelope.overflow
         || !SegmentInvalidationProducer::REQUIRED
@@ -190,6 +196,8 @@ mod tests {
             key_format: envelope.key_format,
             baseline_scan_plan_digest: envelope.baseline_scan_plan_digest,
             process_epoch: envelope.process_epoch,
+            generation_start: envelope.generation_start,
+            generation_end: envelope.generation_end,
             durable_producer_identity: true,
             invalidation_domain: SegmentInvalidationDomain::LocalSingleSet,
             distributed_ec_invalidation: false,
@@ -241,6 +249,20 @@ mod tests {
             Err(SegmentInvalidationError::InvalidProof)
         );
 
+        let mut wrong_generation_start = proof.clone();
+        wrong_generation_start.generation_start = wrong_generation_start.generation_start.saturating_sub(1);
+        assert_eq!(
+            admit_segment_invalidation(&envelope, &wrong_generation_start, ["hot/one"]),
+            Err(SegmentInvalidationError::InvalidProof)
+        );
+
+        let mut wrong_generation_end = proof.clone();
+        wrong_generation_end.generation_end = wrong_generation_end.generation_end.saturating_add(1);
+        assert_eq!(
+            admit_segment_invalidation(&envelope, &wrong_generation_end, ["hot/one"]),
+            Err(SegmentInvalidationError::InvalidProof)
+        );
+
         let mut no_durable_identity = proof.clone();
         no_durable_identity.durable_producer_identity = false;
         assert_eq!(

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())]
+        );
+    }
 }

docs/testing/scanner-checkpoint-fixture.md

@@ -64,7 +64,7 @@ The nested `segment_observation` fixture compares diagnostic on/off runs of the
 
 Entry/byte overflow and malformed keys reject the fixture proposal. Missing producers, process restarts, event gaps, and compacted child coverage remain **unverified production capabilities**, not simulated success cases in this fixture. Mainline bucket dirty generations and hashed metadata-cache invalidation stripes are not an exact, replayable object-key stream. The open [prefix reuse proposal #7208](https://github.com/rustfs/rustfs/pull/7208) is a separate candidate implementation; these tests neither import its hint map nor activate its skip path.
 
-The ECStore `segment_observation_equal_size_mutations_retire_metadata_generation` test uses the existing exact-key, test-only invalidation probe and actual owner operations. A same-length PUT must change the returned body and ETag while retiring the old generation; metadata-only PUT must change returned metadata and retire the old generation while size and ETag remain equal. Setup uses the existing full-fanout cache-priming helper; the observed mutations use normal owner locking. This is focused producer evidence, not an end-to-end connection between the owner probe and scanner range selection. The existing semantic mutation matrix covers additional owner entry points separately.
+The ECStore `segment_observation_equal_size_mutations_retire_metadata_generation` test uses the existing exact-key, test-only invalidation probe and actual owner operations. A same-length PUT must change the returned body and ETag while retiring the old generation; metadata-only PUT must change returned metadata and retire the old generation while size and ETag remain equal. Setup uses the existing full-fanout cache-priming helper; the observed mutations use normal owner locking. This is focused producer evidence, not an end-to-end connection between the owner probe and scanner range selection. The segment invalidation proof is bound to the same generation window as the observed envelope, so an old distributed or cold-walk proof cannot authorize a later mutation range. The existing semantic mutation matrix covers additional owner entry points separately.
 
 ```sh
 cargo test -p rustfs-scanner --lib segment_observation -- --list

rustfs/src/error.rs

@@ -14,9 +14,13 @@
 
 use crate::storage_api::error::contract::{StorageErrorCode, range::HTTPRangeError};
 use crate::storage_api::error::{PoolMetadataError, 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
@@ -308,6 +312,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(),
         }
     }
@@ -386,6 +393,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);
         }
@@ -479,6 +489,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,
@@ -509,6 +520,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 {
@@ -1226,6 +1239,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));
+        });
+    }
+}