From 68fb52b752772c7263a025b91f564233bd287db3 Mon Sep 17 00:00:00 2001 From: Hauser Zhi Date: Sun, 8 Nov 2026 14:34:01 +0800 Subject: [PATCH 1/7] chore(deps): update flake.lock (#9236) --- flake.lock | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/flake.lock b/flake.lock index ae3c441ba..56ed6dc5c 100644 --- a/flake.lock +++ b/flake.lock @@ -2,11 +2,11 @@ "nodes": { "nixpkgs": { "locked": { - "lastModified": 1793466162, - "narHash": "sha256-pt+peAEoOK3lLLdtbnQ1AGIBTZbdB6jrkgmurfcQwig=", + "lastModified": 1793977579, + "narHash": "sha256-WrSODdZFVIs/Hh2NzC0bY+ROByi0bATg8WfePX8dPj8=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "a0788123ada57ec92a79ab2581cda6d3c6cca75f", + "rev": "4fe25e12eb9f4c744966dd3de5a05246cf098601", "type": "github" }, "original": { @@ -29,11 +29,11 @@ ] }, "locked": { - "lastModified": 1793434469, - "narHash": "sha256-+sgVNI2c67di+XIeMaz5KSei4T65kX68s5Nl0E5Gzbc=", + "lastModified": 1794040562, + "narHash": "sha256-7eEiezSku3lbZPUOXW8we1a7+xd5axkCQdyI3rzOKjo=", "owner": "oxalica", "repo": "rust-overlay", - "rev": "6ecf534a7c16e7c969f26838fdd8c93597ae93f3", + "rev": "d3fe74269f2a6652437e21e78eb7e83be81f6982", "type": "github" }, "original": { From a6c3016493ee96c1f40a6dfdeca3ef34859ca9e8 Mon Sep 17 00:00:00 2001 From: Chris An Date: Sun, 8 Nov 2026 01:35:49 -0500 Subject: [PATCH 2/7] fix(cache): respect process cgroup memory limits (#9232) --- .../object-data-cache/src/runtime_memory.rs | 49 +++++++++++++++++-- 1 file changed, 45 insertions(+), 4 deletions(-) diff --git a/crates/object-data-cache/src/runtime_memory.rs b/crates/object-data-cache/src/runtime_memory.rs index 14c839258..c5f26f47a 100644 --- a/crates/object-data-cache/src/runtime_memory.rs +++ b/crates/object-data-cache/src/runtime_memory.rs @@ -66,11 +66,26 @@ pub fn select_effective_memory(host_total: u64, host_available: u64, cgroup: Opt /// Resolves the effective memory from an already-refreshed system handle. /// -/// `cgroup_limits()` is computed fresh on each call and is only implemented on -/// Linux (it returns `None` elsewhere), so non-Linux hosts always use host -/// values. +/// On Linux, the current process cgroup and its ancestors are read on each call, +/// including limits imposed by systemd services. System-wide cgroup limits are +/// used if process information is unavailable. Other platforms use host values. pub fn effective_memory_from_system(system: &System) -> EffectiveMemory { - let cgroup = system.cgroup_limits().map(|limits| (limits.total_memory, limits.free_memory)); + #[cfg(target_os = "linux")] + let limits = sysinfo::get_current_pid() + .ok() + .and_then(|pid| { + let mut processes = System::new().ok()?; + processes.refresh_processes_specifics( + sysinfo::ProcessesToUpdate::Some(&[pid]), + false, + sysinfo::ProcessRefreshKind::nothing().with_memory(), + ); + processes.process(pid)?.cgroup_limits() + }) + .or_else(|| system.cgroup_limits()); + #[cfg(not(target_os = "linux"))] + let limits = system.cgroup_limits(); + let cgroup = limits.map(|limits| (limits.total_memory, limits.free_memory)); select_effective_memory(system.total_memory(), system.available_memory(), cgroup) } @@ -89,6 +104,32 @@ mod tests { const GIB: u64 = 1024 * 1024 * 1024; + #[cfg(target_os = "linux")] + #[test] + fn effective_memory_uses_current_process_cgroup() { + use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System, get_current_pid}; + + let pid = get_current_pid().expect("current process ID"); + let mut system = System::new().expect("system memory handle"); + system.refresh_memory(); + let mut process_system = System::new().expect("process memory handle"); + process_system.refresh_processes_specifics( + ProcessesToUpdate::Some(&[pid]), + false, + ProcessRefreshKind::nothing().with_memory(), + ); + let limits = process_system.process(pid).expect("current process entry").cgroup_limits(); + let expected = select_effective_memory( + system.total_memory(), + system.available_memory(), + limits.map(|limits| (limits.total_memory, limits.free_memory)), + ); + let actual = super::effective_memory_from_system(&system); + assert_eq!(actual.basis, expected.basis); + assert_eq!(actual.total_bytes, expected.total_bytes); + assert!(actual.available_bytes <= actual.total_bytes); + } + #[test] fn select_effective_memory_prefers_constraining_cgroup() { let effective = select_effective_memory(64 * GIB, 40 * GIB, Some((2 * GIB, GIB))); From f10079f1a3b2aa7c741008d0951bc46278c3bc5a Mon Sep 17 00:00:00 2001 From: Chris An Date: Sun, 8 Nov 2026 01:35:58 -0500 Subject: [PATCH 3/7] fix(rpc): size and reserve replay cache for compact entries (#9233) --- crates/ecstore/src/cluster/rpc/http_auth.rs | 169 ++++++++++++++++++-- 1 file changed, 160 insertions(+), 9 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index aa42e623b..bdd979c97 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -127,9 +127,14 @@ const HTTP_PUT_FILE_CAPABILITY_AUTH_DOMAIN: &[u8] = b"rustfs-http-put-file-capab const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD"; const UNSIGNED_PAYLOAD_NONCE: &str = "unsigned"; const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes -const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(601); -const REPLAY_CACHE_RETENTION_SECS: usize = 601; -const REPLAY_CACHE_ENTRY_BYTES_ESTIMATE: u64 = 128; +// Timestamp validity and the wall-clock high-water mark still gate eviction, +// including future-dated signatures and replay attempts after clock rollback. +const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(301); +// Preserve capacity for the full validity window at maximum accepted clock skew. +const REPLAY_CACHE_CAPACITY_WINDOW_SECS: usize = 601; +// A reserved index avoids overlapping old/new tables during a rehash. Include +// index slack, compact expiry storage, and page/allocator bookkeeping. +const REPLAY_CACHE_ENTRY_BYTES_ESTIMATE: usize = 2 * std::mem::size_of::() + std::mem::size_of::() + 16; // Keep 16 CPU / 32 GiB field nodes at the 32M cap without requiring an env override. const REPLAY_CACHE_AUTO_MEMORY_PERCENT: u64 = 13; const REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU: usize = 4096; @@ -222,11 +227,11 @@ fn replay_cache_capacity_from_resources(cpu_count: usize, memory_limit_bytes: Op let cpu_count = cpu_count.max(1); let cpu_based_capacity = cpu_count .saturating_mul(REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU) - .saturating_mul(REPLAY_CACHE_RETENTION_SECS); + .saturating_mul(REPLAY_CACHE_CAPACITY_WINDOW_SECS); let memory_based_capacity = memory_limit_bytes .map(|bytes| { let budget = bytes.saturating_mul(REPLAY_CACHE_AUTO_MEMORY_PERCENT) / 100; - saturating_usize_from_u64(budget / REPLAY_CACHE_ENTRY_BYTES_ESTIMATE) + saturating_usize_from_u64(budget / u64::try_from(REPLAY_CACHE_ENTRY_BYTES_ESTIMATE).unwrap_or(u64::MAX)) }) .unwrap_or(REPLAY_CACHE_AUTO_MAX_CAPACITY); let capacity = memory_based_capacity @@ -497,6 +502,12 @@ fn publish_nonce_cache_metrics(metrics: Option>) { } impl RpcNonceCache { + fn reserve_index(&mut self, capacity: usize) -> std::io::Result<()> { + self.nonces + .try_reserve(capacity) + .map_err(|error| std::io::Error::new(std::io::ErrorKind::OutOfMemory, error)) + } + fn remove_expired(&mut self, now: Instant, wall_time: i64) -> usize { let mut removed = 0; while let Some(nonce) = self.expirations.pop_expired(now, wall_time) { @@ -537,6 +548,11 @@ impl RpcNonceCache { }), ); } + if self.nonces.is_empty() + && let Err(error) = self.reserve_index(record.capacity.min(REPLAY_CACHE_AUTO_MAX_CAPACITY)) + { + return (Err(error), Some(metrics)); + } self.nonces.insert(record.nonce); self.expirations .push_back(record.expires_at, record.signed_at.saturating_add(SIGNATURE_VALID_DURATION), record.nonce); @@ -1699,6 +1715,9 @@ pub fn verify_tonic_rpc_signature_with_bootstrap( } pub fn tonic_rpc_auth_failure_reason(error: &std::io::Error) -> &'static str { + if error.kind() == std::io::ErrorKind::OutOfMemory { + return "replay_cache_allocation"; + } match error.to_string().as_str() { "Missing RPC audience" => "missing_audience", "Invalid RPC request path" => "invalid_request_path", @@ -3138,9 +3157,9 @@ mod tests { assert_eq!(decision.source, ReplayCacheCapacitySource::Auto); assert_eq!(decision.memory_basis, Some(MemoryBasis::Host)); - assert_eq!(decision.memory_based_capacity, 17_448_304); + assert_eq!(decision.memory_based_capacity, 31_019_208); assert_eq!(decision.cpu_based_capacity, 19_693_568); - assert_eq!(decision.capacity, 17_448_304); + assert_eq!(decision.capacity, 19_693_568); } #[test] @@ -3150,13 +3169,13 @@ mod tests { replay_cache_capacity_decision(rustfs_utils::EnvParseOutcome::Absent, 16, Some(32 * gib), Some(MemoryBasis::Host)); assert_eq!(decision.source, ReplayCacheCapacitySource::Auto); - assert_eq!(decision.memory_based_capacity, 34_896_609); + assert_eq!(decision.memory_based_capacity, 62_038_416); assert_eq!(decision.cpu_based_capacity, 39_387_136); assert_eq!(decision.capacity, REPLAY_CACHE_AUTO_MAX_CAPACITY); let observed_field_node = replay_cache_capacity_decision(rustfs_utils::EnvParseOutcome::Absent, 16, Some(31 * gib), Some(MemoryBasis::Host)); - assert_eq!(observed_field_node.memory_based_capacity, 33_806_090); + assert_eq!(observed_field_node.memory_based_capacity, 60_099_715); assert_eq!(observed_field_node.cpu_based_capacity, 39_387_136); assert_eq!(observed_field_node.capacity, REPLAY_CACHE_AUTO_MAX_CAPACITY); } @@ -3229,6 +3248,138 @@ mod tests { } } + #[test] + fn nonce_cache_releases_expired_timestamp_capacity_without_allowing_clock_rollback_replay() { + let now = Instant::now(); + let expiry = now + REPLAY_CACHE_RETENTION; + let later = now + Duration::from_secs(302); + let nonce = Uuid::new_v4(); + let replacement = Uuid::new_v4(); + let mut cache = RpcNonceCache::default(); + cache + .check_and_record(test_nonce_record(nonce, 1_000, now, 1_000, expiry, 1)) + .0 + .expect("record initial nonce"); + + let boundary = cache + .check_and_record(test_nonce_record(nonce, 1_000, later, 1_300, later + REPLAY_CACHE_RETENTION, 1)) + .0 + .expect_err("a signature at the inclusive freshness boundary must remain protected"); + assert_eq!(boundary.to_string(), "RPC request replay detected"); + cache + .check_and_record(test_nonce_record(replacement, 1_302, later, 1_302, later + REPLAY_CACHE_RETENTION, 1)) + .0 + .expect("an expired signature must release bounded cache capacity"); + assert!(!cache.nonces.contains(&nonce)); + assert!(cache.nonces.contains(&replacement)); + + let rollback = cache + .check_and_record(test_nonce_record(nonce, 1_000, later, 1_000, later + REPLAY_CACHE_RETENTION, 1)) + .0 + .expect_err("clock rollback must not resurrect an evicted signature"); + assert_eq!(rollback.to_string(), "RPC request timestamp expired after clock regression"); + } + + #[test] + fn nonce_cache_keeps_future_timestamp_reserved_until_its_full_validity_window_closes() { + let now = Instant::now(); + let expiry = now + REPLAY_CACHE_RETENTION; + let later = now + Duration::from_secs(302); + let nonce = Uuid::new_v4(); + let mut cache = RpcNonceCache::default(); + cache + .check_and_record(test_nonce_record(nonce, 1_300, now, 1_000, expiry, 1)) + .0 + .expect("record nonce signed at maximum allowed forward skew"); + let replay = cache + .check_and_record(test_nonce_record(nonce, 1_300, later, 1_302, later + REPLAY_CACHE_RETENTION, 1)) + .0 + .expect_err("monotonic expiry must not remove a still-valid future signature"); + assert_eq!(replay.to_string(), "RPC request replay detected"); + let full = cache + .check_and_record(test_nonce_record(Uuid::new_v4(), 1_302, later, 1_302, later + REPLAY_CACHE_RETENTION, 1)) + .0 + .expect_err("a still-valid nonce must continue consuming capacity"); + assert_eq!(full.to_string(), "RPC replay cache capacity exceeded"); + let expired = now + Duration::from_secs(602); + cache + .check_and_record(test_nonce_record( + Uuid::new_v4(), + 1_602, + expired, + 1_602, + expired + REPLAY_CACHE_RETENTION, + 1, + )) + .0 + .expect("release capacity only once future timestamp validity has closed"); + } + + #[test] + fn nonce_index_rejects_unallocatable_capacity_with_typed_error() { + let mut cache = RpcNonceCache::default(); + let error = cache + .reserve_index(usize::MAX) + .expect_err("unallocatable capacity must fail without a panic"); + assert_eq!(error.kind(), std::io::ErrorKind::OutOfMemory); + assert!( + error + .get_ref() + .is_some_and(|source| source.is::()) + ); + assert_eq!(tonic_rpc_auth_failure_reason(&error), "replay_cache_allocation"); + assert!(cache.nonces.is_empty()); + assert!(cache.expirations.pages.is_empty()); + } + + #[test] + fn nonce_cache_reserves_the_bounded_index_before_sustained_traffic() { + let now = Instant::now(); + let expiry = now + REPLAY_CACHE_RETENTION; + let mut cache = RpcNonceCache::default(); + let capacity = 4096; + cache + .check_and_record(test_nonce_record(Uuid::from_u128(1), 1000, now, 1000, expiry, capacity)) + .0 + .expect("record first nonce"); + let allocated = cache.nonces.capacity(); + assert!(allocated >= capacity, "the hash index must not rehash while serving the bounded window"); + for n in 2..=capacity { + cache + .check_and_record(test_nonce_record( + Uuid::from_u128(u128::try_from(n).expect("nonce index")), + 1000, + now, + 1000, + expiry, + capacity, + )) + .0 + .expect("record within bounded capacity"); + } + assert_eq!(cache.nonces.capacity(), allocated); + let full = cache + .check_and_record(test_nonce_record(Uuid::from_u128(9999), 1000, now, 1000, expiry, capacity)) + .0 + .expect_err("preallocation must not relax the logical capacity fence"); + assert_eq!(full.to_string(), "RPC replay cache capacity exceeded"); + assert_eq!(cache.nonces.len(), capacity); + } + + #[test] + fn replay_cache_compact_storage_fits_two_cpu_window_in_2800_mib() { + let decision = replay_cache_capacity_decision( + rustfs_utils::EnvParseOutcome::Absent, + 2, + Some(2800 * 1024 * 1024), + Some(MemoryBasis::Cgroup), + ); + assert_eq!(decision.cpu_based_capacity, 4_923_392); + assert_eq!(decision.capacity, decision.cpu_based_capacity); + assert!(decision.memory_based_capacity >= decision.capacity); + assert_eq!(decision.source, ReplayCacheCapacitySource::Auto); + } + #[test] fn nonce_cache_expires_by_monotonic_deadline_and_fails_closed_at_capacity() { let now = Instant::now(); From 907410b2f549bdbd440e2dbee85f59093dcde710 Mon Sep 17 00:00:00 2001 From: Chris An Date: Sun, 8 Nov 2026 01:36:12 -0500 Subject: [PATCH 4/7] fix(select): bound Parquet allocations before decoding (#9231) --- Cargo.lock | 2 + crates/s3select-api/Cargo.toml | 2 + crates/s3select-api/src/lib.rs | 32 + crates/s3select-api/src/parquet_budget.rs | 1007 +++++++++++++++++ .../s3select-query/src/dispatcher/manager.rs | 3 +- .../src/dispatcher/parquet_table.rs | 319 +++++- .../table_catalog/maintenance/compaction.rs | 327 ++---- 7 files changed, 1416 insertions(+), 276 deletions(-) create mode 100644 crates/s3select-api/src/parquet_budget.rs diff --git a/Cargo.lock b/Cargo.lock index 7544aa301..1c4a77e12 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10759,6 +10759,7 @@ dependencies = [ "arc-swap", "async-compression", "async-trait", + "brotli 9.1.1", "bytes", "chrono", "crc-fast", @@ -10768,6 +10769,7 @@ dependencies = [ "futures-core", "hotpath", "http 1.5.0", + "lz4", "memchr", "metrics", "parking_lot 0.13.0", diff --git a/crates/s3select-api/Cargo.toml b/crates/s3select-api/Cargo.toml index 02a5195f2..3a6ad6c00 100644 --- a/crates/s3select-api/Cargo.toml +++ b/crates/s3select-api/Cargo.toml @@ -73,6 +73,8 @@ rustfs-storage-api.workspace = true futures = { workspace = true } futures-core = { workspace = true } flate2.workspace = true +brotli.workspace = true +lz4.workspace = true http.workspace = true memchr.workspace = true s3s = { workspace = true, features = ["minio"] } diff --git a/crates/s3select-api/src/lib.rs b/crates/s3select-api/src/lib.rs index cebec28a6..0e3ad523e 100644 --- a/crates/s3select-api/src/lib.rs +++ b/crates/s3select-api/src/lib.rs @@ -27,6 +27,7 @@ mod csv_input; mod input_stream; mod metrics; pub mod object_store; +pub mod parquet_budget; pub mod query; pub mod server; mod storage_api; @@ -218,6 +219,14 @@ fn classify_select_error_source(err: &(dyn StdError + 'static)) -> Option