gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-07 04:25:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(storage): complete Snowball and decommission follow-ups (#7039)

Browse Source 
* feat(storage): complete Snowball and capacity follow-ups

* fix(ecstore): clarify V3 capacity gate guidance

* fix(ecstore): keep target contention retryable

* fix(ecstore): preserve typed target lock errors

* fix(ecstore): harden decommission recovery

* fix(ecstore): close decommission recovery races

* fix(ecstore): fail closed on multipart cleanup gaps

* fix(ecstore): model capacity mutation parameters

* fix(ecstore): settle checkpoint capacity retries
This commit is contained in:
cxymds
2026-09-03 11:58:59 +08:00
committed by GitHub
parent d011ec9952
commit df30dff1a7
19 changed files with 4817 additions and 449 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/app/object/extract.rs
+68 -11
View File
@@ -373,10 +373,10 @@ const EXTRACT_MAX_EFFECTIVE_PAX_HEADER_BYTES: usize = 8 * 1024;
const EXTRACT_MAX_EFFECTIVE_PAX_USER_METADATA_BYTES: usize = 2 * 1024;
const EXTRACT_MAX_EFFECTIVE_PAX_FIELDS: usize = 4096;
const EXTRACT_MAX_EXPANDED_PAX_METADATA_BYTES: u64 = 128 * 1024 * 1024;
const EXTRACT_SMALL_MEMBER_MAX_BYTES: usize = 64 * 1024;
const EXTRACT_DEFAULT_MAX_INFLIGHT: usize = 1;
const EXTRACT_SMALL_MEMBER_MAX_BYTES: usize = 128 * 1024;
const EXTRACT_DEFAULT_MAX_INFLIGHT: usize = 16;
const EXTRACT_BATCH_MAX_MEMBERS: usize = 16;
const EXTRACT_BATCH_MAX_STAGING_BYTES: usize = 2 * 1024 * 1024;
const EXTRACT_BATCH_MAX_STAGING_BYTES: usize = 3 * 1024 * 1024;
const EXTRACT_MEMBER_CONTEXT_OVERHEAD_BYTES: usize = 512;
const EXTRACT_METADATA_ENTRY_OVERHEAD_BYTES: usize = 64;
const ENV_RUSTFS_SNOWBALL_EXTRACT_MAX_INFLIGHT: &str = "RUSTFS_SNOWBALL_EXTRACT_MAX_INFLIGHT";
@@ -1864,7 +1864,34 @@ fn resolve_put_object_extract_options(headers: &HeaderMap) -> S3Result<PutObject
}
fn put_object_extract_limits() -> ArchiveLimits {
ArchiveLimits::default()
static LIMITS: OnceLock<ArchiveLimits> = OnceLock::new();
*LIMITS.get_or_init(|| {
normalize_put_object_extract_limits(
rustfs_utils::get_env_u64(
rustfs_config::ENV_SNOWBALL_MAX_ENTRY_BYTES,
rustfs_config::DEFAULT_SNOWBALL_MAX_ENTRY_BYTES,
),
rustfs_utils::get_env_u64(
rustfs_config::ENV_SNOWBALL_MAX_UNPACKED_BYTES,
rustfs_config::DEFAULT_SNOWBALL_MAX_UNPACKED_BYTES,
),
)
})
}
fn normalize_put_object_extract_limits(max_entry_bytes: u64, max_unpacked_bytes: u64) -> ArchiveLimits {
let defaults = ArchiveLimits::default();
let max_total_unpacked_size = max_unpacked_bytes.clamp(1, rustfs_config::MAX_SNOWBALL_UNPACKED_BYTES);
let max_entry_size = max_entry_bytes
.clamp(1, rustfs_config::MAX_SNOWBALL_ENTRY_BYTES)
.min(max_total_unpacked_size);
ArchiveLimits {
max_entry_size,
max_total_unpacked_size,
max_decoded_size: max_total_unpacked_size.saturating_add(max_entry_size),
..defaults
}
}
fn build_put_object_extract_archive<R>(decoder: R, limits: ArchiveLimits) -> Archive<R>
@@ -2175,12 +2202,13 @@ impl DefaultObjectUsecase {
.is_some_and(|result| result.uses_durable_reservations);
// Without ignore-errors, the legacy contract stops before attempting a
// later member after the first storage failure. Parallel commits cannot
// preserve that boundary, so concurrency requires both ignore-errors
// and an explicit max-inflight value above the serial default. Quota
// accounting can fail after storage commit, so quota-enabled imports
// also remain serial. An opted-in micro-batch is always drained; a
// fatal outcome stops later batches but cannot roll back peers that
// already committed in the current batch.
// preserve that boundary, so only ignore-errors requests use the
// configured micro-batch. Quota accounting can fail after storage
// commit, so quota-enabled imports also remain serial. Setting
// RUSTFS_SNOWBALL_EXTRACT_MAX_INFLIGHT=1 restores serial behavior. A
// micro-batch is always drained; a fatal outcome stops later batches
// but cannot roll back peers that already committed in the current
// batch.
let max_inflight = select_put_object_extract_max_inflight(
put_object_extract_max_inflight(),
extract_options.ignore_errors,
@@ -2856,7 +2884,7 @@ mod tests {
#[test]
fn snowball_max_inflight_has_a_serial_compatibility_floor_and_bounded_ceiling() {
assert_eq!(EXTRACT_DEFAULT_MAX_INFLIGHT, 1);
assert_eq!(EXTRACT_DEFAULT_MAX_INFLIGHT, EXTRACT_BATCH_MAX_MEMBERS);
assert_eq!(normalize_put_object_extract_max_inflight(0), 1);
assert_eq!(normalize_put_object_extract_max_inflight(1), 1);
assert_eq!(normalize_put_object_extract_max_inflight(usize::MAX), EXTRACT_BATCH_MAX_MEMBERS);
@@ -2877,6 +2905,35 @@ mod tests {
);
}
#[test]
fn snowball_archive_limits_preserve_defaults_and_clamp_operator_overrides() {
let defaults = ArchiveLimits::default();
assert_eq!(
normalize_put_object_extract_limits(
rustfs_config::DEFAULT_SNOWBALL_MAX_ENTRY_BYTES,
rustfs_config::DEFAULT_SNOWBALL_MAX_UNPACKED_BYTES,
),
defaults
);
let minimum = normalize_put_object_extract_limits(0, 0);
assert_eq!(minimum.max_entry_size, 1);
assert_eq!(minimum.max_total_unpacked_size, 1);
assert_eq!(minimum.max_decoded_size, 2);
let bounded = normalize_put_object_extract_limits(u64::MAX, u64::MAX);
assert_eq!(bounded.max_entry_size, rustfs_config::MAX_SNOWBALL_ENTRY_BYTES);
assert_eq!(bounded.max_total_unpacked_size, rustfs_config::MAX_SNOWBALL_UNPACKED_BYTES);
assert_eq!(
bounded.max_decoded_size,
rustfs_config::MAX_SNOWBALL_UNPACKED_BYTES + rustfs_config::MAX_SNOWBALL_ENTRY_BYTES
);
let entry_is_bounded_by_the_request_total = normalize_put_object_extract_limits(1024, 512);
assert_eq!(entry_is_bounded_by_the_request_total.max_entry_size, 512);
assert_eq!(entry_is_bounded_by_the_request_total.max_total_unpacked_size, 512);
}
#[test]
fn snowball_batch_state_flushes_on_duplicates_limits_and_serial_barriers() {
let mut state = ExtractBatchState::default();
rustfs/src/storage/rpc/node_service.rs
+4 -3
View File
@@ -181,10 +181,11 @@ fn remove_heal_control_replay(
static HEAL_CONTROL_REPLAY_CACHE: OnceLock<tokio::sync::Mutex<HashMap<String, Arc<HealControlReplayEntry>>>> = OnceLock::new();
static NODE_CAPABILITY_SERVER_EPOCH: LazyLock<Uuid> = LazyLock::new(Uuid::new_v4);
// v3 additionally promises the v6 tier-delete dispatch-manifest policy. The
// v3 additionally promises the v6 tier-delete dispatch-manifest policy; v4
// promises the sticky per-target decommission capacity fence. The
// existing periodic topology probe carries both capabilities so normal object
// operations do not add another peer RPC.
const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 3;
const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 4;
fn admit_heal_control_replay(
replay_cache: &mut HashMap<String, Arc<HealControlReplayEntry>>,
@@ -3770,7 +3771,7 @@ mod tests {
}
#[tokio::test]
async fn cross_pool_fence_probe_authenticates_supported_v3_state() {
async fn cross_pool_fence_probe_authenticates_supported_v4_state() {
let _ = rustfs_credentials::set_global_rpc_secret("cross-pool-fence-node-service-test-secret".to_string());
let endpoints = heal_control_test_endpoints_with_coordinator("node-0", true);
assert!(