From 474fcf78fb320de0b12e66281056b679f65f7cd8 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 23:43:19 +0800 Subject: [PATCH] fix: align object version limit handling (#7415) Co-authored-by: zhi22915 --- crates/config/README.md | 5 + crates/config/src/constants/api.rs | 12 +++ crates/ecstore/src/disk/error.rs | 1 + crates/ecstore/src/error/mod.rs | 1 + crates/filemeta/src/error.rs | 5 + crates/filemeta/src/filemeta.rs | 148 ++++++++++++++++++++++++++--- crates/utils/src/envs.rs | 12 +++ rustfs/src/error.rs | 26 +++++ rustfs/src/startup_runtime.rs | 98 ++++++++++++++++++- 9 files changed, 293 insertions(+), 15 deletions(-) diff --git a/crates/config/README.md b/crates/config/README.md index 0e2142f53..525129606 100644 --- a/crates/config/README.md +++ b/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. diff --git a/crates/config/src/constants/api.rs b/crates/config/src/constants/api.rs index e6fb4be46..fedb49b95 100644 --- a/crates/config/src/constants/api.rs +++ b/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; diff --git a/crates/ecstore/src/disk/error.rs b/crates/ecstore/src/disk/error.rs index 7747cfa6d..c8db76400 100644 --- a/crates/ecstore/src/disk/error.rs +++ b/crates/ecstore/src/disk/error.rs @@ -425,6 +425,7 @@ impl From 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), } } diff --git a/crates/ecstore/src/error/mod.rs b/crates/ecstore/src/error/mod.rs index d64c194a5..281c70a7e 100644 --- a/crates/ecstore/src/error/mod.rs +++ b/crates/ecstore/src/error/mod.rs @@ -588,6 +588,7 @@ impl From 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)), } diff --git a/crates/filemeta/src/error.rs b/crates/filemeta/src/error.rs index 37f9cdcfe..e07fe6535 100644 --- a/crates/filemeta/src/error.rs +++ b/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()), diff --git a/crates/filemeta/src/filemeta.rs b/crates/filemeta/src/filemeta.rs index 2fc3f4320..05e8b5119 100644 --- a/crates/filemeta/src/filemeta.rs +++ b/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> = 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) -> Option { + 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) -> 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, + } + + impl Drop for ObjectMaxVersionsRestore { + fn drop(&mut self) { + set_object_max_versions_override_for_test(self.previous); + } + } + + fn with_object_max_versions_for_test(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"); diff --git a/crates/utils/src/envs.rs b/crates/utils/src/envs.rs index 5e572e5c0..03fe9b06d 100644 --- a/crates/utils/src/envs.rs +++ b/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())] + ); + } } diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index 2d1945970..773f7b869 100644 --- a/rustfs/src/error.rs +++ b/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 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 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 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 { diff --git a/rustfs/src/startup_runtime.rs b/rustfs/src/startup_runtime.rs index d0e6ad1d3..0d6947bdd 100644 --- a/rustfs/src/startup_runtime.rs +++ b/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::(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 { + 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(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)); + }); + } +}