From c0155f0dfa9276806965a1efc2465907f5c02fdd Mon Sep 17 00:00:00 2001 From: houseme Date: Sat, 29 Aug 2026 12:51:37 +0800 Subject: [PATCH] fix(logging): bound ECStore debug output (#6809) Also replace deprecated Atomic::fetch_update calls with try_update so the current Rust toolchain keeps lint and CI jobs warning-clean. Co-authored-by: heihutu --- crates/e2e_test/src/reliant/tiering.rs | 20 +++-- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 2 +- .../bucket/replication/replication_pool.rs | 13 ++- crates/ecstore/src/core/sets.rs | 2 +- crates/ecstore/src/data_usage/mod.rs | 2 +- crates/ecstore/src/object_api/types.rs | 71 +++++++++++++++- crates/ecstore/src/runtime/instance.rs | 4 +- crates/ecstore/src/services/tier/test_util.rs | 2 +- crates/ecstore/src/services/tier/tier.rs | 6 +- crates/ecstore/src/set_disk/mod.rs | 4 +- crates/ecstore/src/set_disk/ops/multipart.rs | 2 +- crates/ecstore/src/set_disk/ops/object.rs | 4 +- crates/ecstore/src/store/list_objects.rs | 2 +- crates/ecstore/src/store/mod.rs | 84 ++++++++++++++++++- crates/iam/src/lib.rs | 2 +- crates/io-core/src/backpressure.rs | 2 +- crates/io-core/src/pool.rs | 4 +- crates/io-metrics/src/process_lock_metrics.rs | 4 +- crates/lock/src/namespace/tests.rs | 4 +- crates/object-data-cache/src/memory.rs | 2 +- crates/object-data-cache/src/moka_backend.rs | 2 +- crates/replication/src/stats.rs | 2 +- crates/s3select-api/src/query/session.rs | 2 +- crates/scanner/src/lib.rs | 10 +-- crates/scanner/src/scanner/tests.rs | 2 +- crates/scanner/src/scanner_io/dirty_usage.rs | 2 +- crates/scanner/src/scanner_io/guards.rs | 4 +- crates/targets/src/store.rs | 2 +- crates/targets/src/testkit.rs | 2 +- rustfs/src/app/object_data_cache/cold_fill.rs | 6 +- rustfs/src/server/http.rs | 2 +- scripts/check_logging_guardrails.sh | 26 ++++++ 32 files changed, 244 insertions(+), 54 deletions(-) diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index 216072883..4f97f724e 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -210,19 +210,27 @@ async fn add_rustfs_tier(hot: &RustFSTestEnvironment, cold: &RustFSTestEnvironme } } -async fn remove_rustfs_tier_force(hot: &RustFSTestEnvironment) -> TestResult { - let path = format!("/rustfs/admin/v3/tier/{TIER_NAME}?force=true"); +fn clear_tiers_confirmation_token(now: OffsetDateTime) -> String { + let mut rand = "AGD1R25GI3I1GJGUGJFD7FBS4DFAASDF".to_string(); + rand.insert_str(3, &now.day().to_string()); + rand.insert_str(17, &now.month().to_string()); + rand.insert_str(23, &now.year().to_string()); + rand +} + +async fn clear_rustfs_tiers_force(hot: &RustFSTestEnvironment) -> TestResult { let deadline = Instant::now() + StdDuration::from_secs(30); loop { - let (status, resp) = - signed_admin_request(&hot.url, Method::DELETE, &path, None, &hot.access_key, &hot.secret_key).await?; + let rand = clear_tiers_confirmation_token(OffsetDateTime::now_utc()); + let path = format!("/rustfs/admin/v3/tier/clear?rand={rand}&force=true"); + let (status, resp) = signed_admin_request(&hot.url, Method::POST, &path, None, &hot.access_key, &hot.secret_key).await?; if status.is_success() { return Ok(()); } if (!resp.contains("TierNameBackendInUse") && !resp.contains(TIER_MUTATION_RECOVERY_CHANGED)) || Instant::now() >= deadline { - return Err(format!("RemoveTier(RustFS) failed: status={status}, body={resp}").into()); + return Err(format!("ClearTier(RustFS) failed: status={status}, body={resp}").into()); } // Tier mutation cleanup and startup recovery are asynchronous. tokio::time::sleep(StdDuration::from_millis(100)).await; @@ -1708,7 +1716,7 @@ async fn test_manual_transition_async_tier_failure_reports_terminal_partial() -> 0, ) .await?; - remove_rustfs_tier_force(&hot).await?; + clear_rustfs_tiers_force(&hot).await?; let due_mtime = OffsetDateTime::now_utc() - time::Duration::hours(25); put_backdated_single_part_object( diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index d0e5cec70..aa67a9bd6 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -490,7 +490,7 @@ impl ExpiryStats { } fn add_nonnegative(counter: &AtomicI64, delta: i64) { - let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_add(delta).max(0))); + let _ = counter.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_add(delta).max(0))); } fn increment_missed_expiry_tasks(&self) { diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 8807168f8..a6d5cf857 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -1048,7 +1048,6 @@ pub fn resync_start_conflict_id(error: &EcstoreError) -> Option<&str> { } /// Main replication pool structure -#[derive(Debug)] pub struct ReplicationPool { // Atomic counters for active workers active_workers: Arc, @@ -1094,6 +1093,16 @@ pub struct ReplicationPool { resyncer: Arc, } +impl std::fmt::Debug for ReplicationPool { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ReplicationPool") + .field("active_workers", &self.active_workers.load(Ordering::Relaxed)) + .field("active_lrg_workers", &self.active_lrg_workers.load(Ordering::Relaxed)) + .field("active_mrf_workers", &self.active_mrf_workers.load(Ordering::Relaxed)) + .finish_non_exhaustive() + } +} + impl ReplicationPool { /// Creates a new replication pool with specified options pub async fn new(opts: ReplicationPoolOpts, stats: Arc, storage: Arc) -> Arc { @@ -2132,7 +2141,7 @@ impl ReplicationPool { } /// Load bucket replication resync statuses into memory - #[instrument(skip(_cancellation_token))] + #[instrument(skip(self, buckets, _cancellation_token), fields(bucket_count = buckets.len()))] async fn load_resync( self: Arc, buckets: &[String], diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index 0a814865f..9f8fe3dc3 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -781,7 +781,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for Sets { .await } - #[tracing::instrument(skip(self))] + #[tracing::instrument(skip(self, opts))] async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { if opts.delete_prefix && !opts.delete_prefix_object { self.delete_prefix(bucket, object, &opts).await?; diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index 4ec3371ac..719fb8f27 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -578,7 +578,7 @@ pub(crate) async fn prepare_bucket_usage_for_namespace_change( guard: Option<&rustfs_lock::NamespaceLockGuard>, ) -> Result<(), Error> { ensure_bucket_namespace_guard(guard, bucket, "data usage cache cleanup")?; - let _ = USAGE_MEMORY_GENERATION.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| Some(current.saturating_add(1))); + let _ = USAGE_MEMORY_GENERATION.try_update(Ordering::AcqRel, Ordering::Acquire, |current| Some(current.saturating_add(1))); live_bucket_usage_cache().invalidate(bucket).await; clear_bucket_usage_memory(bucket, guard).await?; diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 9ea4b8324..be761a8c7 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -347,7 +347,7 @@ impl QuotaAdmission { } } -#[derive(Debug, Default, Clone)] +#[derive(Default, Clone)] pub struct ObjectOptions { // Use the maximum parity (N/2), used when saving server configuration files pub max_parity: bool, @@ -451,6 +451,75 @@ pub struct ObjectOptions { pub tier_delete_journal_api: Option>, } +impl std::fmt::Debug for ObjectOptions { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ObjectOptions") + .field("max_parity", &self.max_parity) + .field("mod_time", &self.mod_time) + .field("part_number", &self.part_number) + .field("delete_prefix", &self.delete_prefix) + .field("delete_prefix_object", &self.delete_prefix_object) + .field("version_id", &self.version_id.is_some()) + .field("lifecycle_delete_all", &self.lifecycle_delete_all.is_some()) + .field("lifecycle_delete_all_journal", &self.lifecycle_delete_all_journal.is_some()) + .field("expected_current_version_id", &self.expected_current_version_id.is_some()) + .field("expected_bucket_incarnation_id", &self.expected_bucket_incarnation_id) + .field("no_lock", &self.no_lock) + .field("metadata_cache_safe", &self.metadata_cache_safe) + .field("versioned", &self.versioned) + .field("version_suspended", &self.version_suspended) + .field("incl_free_versions", &self.incl_free_versions) + .field("skip_decommissioned", &self.skip_decommissioned) + .field("skip_rebalancing", &self.skip_rebalancing) + .field("skip_free_version", &self.skip_free_version) + .field("put_object_cancellation", &self.put_object_cancellation.is_some()) + .field("data_movement", &self.data_movement) + .field("raw_data_movement_read", &self.raw_data_movement_read) + .field("include_part_checksums", &self.include_part_checksums) + .field("src_pool_idx", &self.src_pool_idx) + .field("user_defined_count", &self.user_defined.len()) + .field("preserve_etag", &self.preserve_etag.is_some()) + .field("metadata_chg", &self.metadata_chg) + .field("http_preconditions", &self.http_preconditions.is_some()) + .field("delete_replication", &self.delete_replication.is_some()) + .field("delete_replication_config_snapshot", &self.delete_replication_config_snapshot) + .field("namespace_lock_fence", &self.namespace_lock_fence.is_some()) + .field("bucket_lifecycle_lock_fence", &self.bucket_lifecycle_lock_fence.is_some()) + .field("replication_request", &self.replication_request) + .field("proxy_request", &self.proxy_request) + .field("proxy_header_set", &self.proxy_header_set) + .field("replication_tagging_timestamp", &self.replication_tagging_timestamp) + .field("replication_retention_timestamp", &self.replication_retention_timestamp) + .field("replication_legalhold_timestamp", &self.replication_legalhold_timestamp) + .field("preserve_ciphertext", &self.preserve_ciphertext) + .field("delete_marker", &self.delete_marker) + .field("synthetic_version_id", &self.synthetic_version_id) + .field( + "transition", + &(self.data_movement + || !self.transition.status.is_empty() + || !self.transition.tier.is_empty() + || self.transition.expected_data_dir.is_some()), + ) + .field("expiration", &self.expiration) + .field( + "lifecycle_audit_event", + &(!self.lifecycle_audit_event.event.rule_id.is_empty() + || !self.lifecycle_audit_event.event.storage_class.is_empty()), + ) + .field("eval_metadata_count", &self.eval_metadata.as_ref().map(HashMap::len)) + .field("object_lock_retention", &self.object_lock_retention.is_some()) + .field("object_lock_delete", &self.object_lock_delete) + .field("object_lock_config_snapshot", &self.object_lock_config_snapshot.is_some()) + .field("want_checksum", &self.want_checksum) + .field("skip_verify_bitrot", &self.skip_verify_bitrot) + .field("capacity_scope_token", &self.capacity_scope_token) + .field("quota_admission", &self.quota_admission) + .field("tier_delete_journal_api", &self.tier_delete_journal_api.is_some()) + .finish() + } +} + /// Transient scanner-only carrier for target-side publication lease tokens. /// SetDisks consumes and removes this key before constructing durable /// FileInfo metadata; it must never appear in an S3-visible object. diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index 2c99e88c1..673de7b92 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -391,7 +391,7 @@ impl InstanceContext { let previous = self.data_movement_operation_epoch.load(Ordering::Acquire); let _ = self .data_movement_operation_epoch - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |epoch| Some(epoch.saturating_add(1))); + .try_update(Ordering::AcqRel, Ordering::Acquire, |epoch| Some(epoch.saturating_add(1))); let result = self.data_movement_operation_epoch.load(Ordering::Acquire); if result == u64::MAX { self.data_movement_operation_epoch_exhausted.store(true, Ordering::Release); @@ -412,7 +412,7 @@ impl InstanceContext { } let updated = self .data_movement_generation - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |generation| generation.checked_add(1)); + .try_update(Ordering::AcqRel, Ordering::Acquire, |generation| generation.checked_add(1)); match updated { Ok(previous) => { let Some(generation) = previous.checked_add(1) else { diff --git a/crates/ecstore/src/services/tier/test_util.rs b/crates/ecstore/src/services/tier/test_util.rs index 3dc112057..f2c4ddeb6 100644 --- a/crates/ecstore/src/services/tier/test_util.rs +++ b/crates/ecstore/src/services/tier/test_util.rs @@ -625,7 +625,7 @@ impl WarmBackend for MockWarmBackend { let reject_once = self .inner .reject_non_empty_remote_version_validations - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| remaining.checked_sub(1)) + .try_update(Ordering::AcqRel, Ordering::Acquire, |remaining| remaining.checked_sub(1)) .is_ok(); if reject_once || self.inner.reject_non_empty_remote_versions.load(Ordering::Acquire) { return Err(std::io::Error::other("mock warm backend requires an unversioned remote object")); diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index 047a51934..15a69bb54 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/crates/ecstore/src/services/tier/tier.rs @@ -1651,7 +1651,7 @@ impl TierOperationLease { ) -> std::result::Result { inner .active_leases - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |active| active.checked_add(1)) + .try_update(Ordering::AcqRel, Ordering::Acquire, |active| active.checked_add(1)) .map_err(|_| { let mut err = ERR_TIER_INVALID_CONFIG.clone(); err.message = "Remote tier operation lease capacity exhausted".to_string(); @@ -1670,7 +1670,7 @@ impl Drop for TierOperationLease { let result = self .inner .active_leases - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |active| active.checked_sub(1)); + .try_update(Ordering::AcqRel, Ordering::Acquire, |active| active.checked_sub(1)); match result { Ok(1) => self.inner.drained.notify_one(), Ok(_) => {} @@ -11909,7 +11909,7 @@ mod tests { let should_pause = match barrier .matches_before_pause - .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) + .try_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) { Ok(_) => false, Err(_) => barrier.armed.swap(false, Ordering::SeqCst), diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index c49958849..7fd95925a 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -3649,7 +3649,7 @@ impl SetDisks { let hash_bytes = hash.to_le_bytes(); let index = usize::from(u16::from_le_bytes([hash_bytes[0], hash_bytes[1]]) % GET_OBJECT_METADATA_CACHE_FENCE_SHARDS); let generation = &self.get_object_metadata_cache_generations[index]; - let previous = match generation.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| current.checked_add(1)) { + let previous = match generation.try_update(Ordering::AcqRel, Ordering::Acquire, |current| current.checked_add(1)) { Ok(previous) | Err(previous) => previous, }; let previous = GetObjectMetadataCacheGeneration { @@ -3666,7 +3666,7 @@ impl SetDisks { fn invalidate_all_get_object_metadata_cache(&self) { for generation in self.get_object_metadata_cache_generations.iter() { - let _ = generation.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| current.checked_add(1)); + let _ = generation.try_update(Ordering::AcqRel, Ordering::Acquire, |current| current.checked_add(1)); } self.get_object_metadata_cache.invalidate_all(); } diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 52e905877..6a9545199 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -395,7 +395,7 @@ async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartComm } }; if let Some(barrier) = barrier - && let Ok(previous) = barrier.arrivals.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| { + && let Ok(previous) = barrier.arrivals.try_update(Ordering::AcqRel, Ordering::Acquire, |current| { (current < barrier.expected_arrivals).then_some(current + 1) }) { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 699243632..7e0d86b19 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -3976,7 +3976,7 @@ impl AsyncRead for TransitionUploadReader { let read = u64::try_from(read).map_err(|_| std::io::Error::other("transition upload read count exceeds u64::MAX"))?; self.consumed - .fetch_update(Ordering::Release, Ordering::Relaxed, |consumed| consumed.checked_add(read)) + .try_update(Ordering::Release, Ordering::Relaxed, |consumed| consumed.checked_add(read)) .map_err(|_| std::io::Error::other("transition upload read count overflow"))?; Poll::Ready(Ok(())) } @@ -7049,7 +7049,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { (del_objects, del_errs, accounting) } - #[tracing::instrument(skip(self))] + #[tracing::instrument(skip(self, opts))] async fn delete_object(&self, bucket: &str, object: &str, mut opts: ObjectOptions) -> Result { // Scanner cleanup carries the per-peer lease fence as transient // request metadata. Consume it before any delete-prefix fanout so it diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index f5e0b323a..281ffc017 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -659,7 +659,7 @@ pub(crate) fn observe_scanner_namespace_mutations(bucket: &str, delta: u64) { } let _ = SCANNER_NAMESPACE_MUTATION_GENERATION - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| Some(current.saturating_add(delta))); + .try_update(Ordering::AcqRel, Ordering::Acquire, |current| Some(current.saturating_add(delta))); } pub(crate) async fn observe_list_objects_mutation(store: &ECStore, bucket: &str) -> u64 { diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 2f3d87c7a..0c874690f 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -275,11 +275,13 @@ pub struct ECStore { impl std::fmt::Debug for ECStore { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let disk_slot_count: usize = self.disk_map.values().map(Vec::len).sum(); + f.debug_struct("ECStore") .field("id", &self.id) - .field("disk_map", &self.disk_map) - .field("pools", &self.pools) - .field("pool_meta", &self.pool_meta) + .field("disk_map_pool_count", &self.disk_map.len()) + .field("disk_slot_count", &disk_slot_count) + .field("pool_count", &self.pools.len()) .finish_non_exhaustive() } } @@ -1154,6 +1156,7 @@ mod tests { use super::*; use crate::core::pools::{PoolDecommissionInfo, PoolStatus}; use crate::layout::endpoints::{Endpoints, PoolEndpoints, SetupType}; + use crate::object_api::ObjectOptions; use crate::runtime::global::reset_local_disk_test_state; use crate::runtime::sources::{clear_local_disk_id_map_for_test, local_disk_path_by_id}; use crate::store::init_format::{connect_load_init_formats, init_disks}; @@ -1170,6 +1173,81 @@ mod tests { assert!(infos.iter().all(|info| info.is_none())); } + #[test] + fn ecstore_debug_is_bounded_summary() { + let endpoint_pools = EndpointServerPools::default(); + let ctx = Arc::new(InstanceContext::new()); + let store = ECStore { + id: uuid::Uuid::new_v4(), + disk_map: [(0, vec![None, None, None, None])].into_iter().collect(), + pools: Vec::new(), + peer_sys: crate::cluster::rpc::S3PeerSys::new_with_instance_ctx(&endpoint_pools, ctx.clone()), + pool_meta: RwLock::new(PoolMeta::default()), + rebalance_meta: RwLock::new(None), + decommission_cancelers: RwLock::new(Vec::new()), + start_gate: Mutex::new(()), + pool_meta_save_gate: Mutex::default(), + ctx, + bucket_fence_registry: Arc::default(), + }; + + let rendered = format!("{store:?}"); + + assert!(rendered.len() < 256, "ECStore Debug should stay bounded: {rendered}"); + assert!(rendered.contains("disk_map_pool_count")); + assert!(rendered.contains("disk_slot_count")); + assert!(!rendered.contains("disk_map:")); + assert!(!rendered.contains("pools:")); + assert!(!rendered.contains("pool_meta")); + assert!(!rendered.contains("format.json")); + assert!(!rendered.contains("TimedActionSlot")); + assert!(!rendered.contains("DiskHealthTracker")); + } + + #[test] + fn object_options_debug_does_not_expand_tier_store_handle() { + let store = build_store_with_ctx(Arc::new(InstanceContext::new())); + let mut opts = ObjectOptions { + version_id: Some("large-version-id".repeat(1024)), + expected_current_version_id: Some("large-expected-version-id".repeat(1024)), + preserve_etag: Some("large-etag".repeat(1024)), + http_preconditions: Some(crate::storage_api_contracts::object::HTTPPreconditions { + if_match: Some("large-if-match".repeat(1024)), + if_none_match: Some("large-if-none-match".repeat(1024)), + ..Default::default() + }), + tier_delete_journal_api: Some(store), + ..Default::default() + }; + opts.user_defined.insert("large-user-metadata".to_owned(), "x".repeat(8192)); + opts.eval_metadata = Some([("large-eval-metadata".to_owned(), "y".repeat(8192))].into_iter().collect()); + opts.transition.status = "large-transition-status".repeat(1024); + opts.transition.tier = "large-transition-tier".repeat(1024); + opts.lifecycle_audit_event.event.rule_id = "large-rule-id".repeat(1024); + opts.lifecycle_audit_event.event.storage_class = "large-storage-class".repeat(1024); + + let rendered = format!("{opts:?}"); + + assert!(rendered.len() < 4096, "ObjectOptions Debug should stay bounded: {rendered}"); + assert!(rendered.contains("tier_delete_journal_api: true")); + assert!(rendered.contains("user_defined_count: 1")); + assert!(rendered.contains("eval_metadata_count: Some(1)")); + assert!(!rendered.contains("ECStore {")); + assert!(!rendered.contains("disk_map")); + assert!(!rendered.contains("large-version-id")); + assert!(!rendered.contains("large-expected-version-id")); + assert!(!rendered.contains("large-etag")); + assert!(!rendered.contains("large-if-match")); + assert!(!rendered.contains("large-transition")); + assert!(!rendered.contains("large-rule-id")); + assert!(!rendered.contains("large-storage-class")); + assert!(!rendered.contains("large-user-metadata")); + assert!(!rendered.contains("large-eval-metadata")); + assert!(!rendered.contains("format.json")); + assert!(!rendered.contains("TimedActionSlot")); + assert!(!rendered.contains("DiskHealthTracker")); + } + // Build a minimal ECStore carrying an explicit instance context. Empty // pools/disks are sufficient: the Phase 5 accessors read only `self.ctx`. fn build_store_with_ctx(ctx: Arc) -> Arc { diff --git a/crates/iam/src/lib.rs b/crates/iam/src/lib.rs index 239cd9c24..0525bff7a 100644 --- a/crates/iam/src/lib.rs +++ b/crates/iam/src/lib.rs @@ -131,7 +131,7 @@ pub(crate) async fn notify_iam_load_user(access_key: &str, temp: bool) -> Vec prev, Err(_) => { diff --git a/crates/io-core/src/pool.rs b/crates/io-core/src/pool.rs index 0167ac499..733f29b8a 100644 --- a/crates/io-core/src/pool.rs +++ b/crates/io-core/src/pool.rs @@ -448,7 +448,7 @@ impl PoolTier { if let Some(buffer) = buffer { let released_bytes = buffer.capacity() as u64; self.tier_current_allocated_bytes - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| { Some(current.saturating_sub(released_bytes)) }) .ok(); @@ -457,7 +457,7 @@ impl PoolTier { { metrics .current_allocated_bytes - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| { Some(current.saturating_sub(released_bytes)) }) .ok(); diff --git a/crates/io-metrics/src/process_lock_metrics.rs b/crates/io-metrics/src/process_lock_metrics.rs index e60386b36..3f71eb724 100644 --- a/crates/io-metrics/src/process_lock_metrics.rs +++ b/crates/io-metrics/src/process_lock_metrics.rs @@ -69,12 +69,12 @@ pub fn record_write_lock_held_acquire() { #[inline(always)] pub fn record_read_lock_held_release() { - let _ = READ_LOCKS_HELD.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| Some(value.saturating_sub(1))); + let _ = READ_LOCKS_HELD.try_update(Ordering::Relaxed, Ordering::Relaxed, |value| Some(value.saturating_sub(1))); } #[inline(always)] pub fn record_write_lock_held_release() { - let _ = WRITE_LOCKS_HELD.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| Some(value.saturating_sub(1))); + let _ = WRITE_LOCKS_HELD.try_update(Ordering::Relaxed, Ordering::Relaxed, |value| Some(value.saturating_sub(1))); } #[inline(always)] diff --git a/crates/lock/src/namespace/tests.rs b/crates/lock/src/namespace/tests.rs index 3316828f7..009d47cad 100644 --- a/crates/lock/src/namespace/tests.rs +++ b/crates/lock/src/namespace/tests.rs @@ -181,7 +181,7 @@ impl crate::client::LockClient for FlakyAcquireClient { self.acquire_attempts.fetch_add(1, Ordering::SeqCst); if self .failed_acquires_remaining - .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) + .try_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) .is_ok() { return Ok(LockResponse::failure("Lock acquisition timeout", request.acquire_timeout)); @@ -254,7 +254,7 @@ impl crate::client::LockClient for FlakyReleaseClient { self.release_attempts.fetch_add(1, Ordering::SeqCst); if self .failed_releases_remaining - .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) + .try_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) .is_ok() { return Ok(false); diff --git a/crates/object-data-cache/src/memory.rs b/crates/object-data-cache/src/memory.rs index cacc3db40..0b614358a 100644 --- a/crates/object-data-cache/src/memory.rs +++ b/crates/object-data-cache/src/memory.rs @@ -192,7 +192,7 @@ impl MemorySnapshotCell { } if self .pending_release - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |pending| pending.checked_add(bytes)) + .try_update(Ordering::AcqRel, Ordering::Acquire, |pending| pending.checked_add(bytes)) .is_err() { self.release_accounting_failed.store(true, Ordering::Release); diff --git a/crates/object-data-cache/src/moka_backend.rs b/crates/object-data-cache/src/moka_backend.rs index 81abaf719..81ed15df9 100644 --- a/crates/object-data-cache/src/moka_backend.rs +++ b/crates/object-data-cache/src/moka_backend.rs @@ -590,7 +590,7 @@ impl MokaBackend { // the synchronization, so relaxed ordering is sufficient here. let generation = match self .next_generation - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_add(1)) + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_add(1)) { Ok(previous) => previous + 1, Err(_) => return leader.finish(ObjectDataCacheFillResult::SkippedIdentityOverflow), diff --git a/crates/replication/src/stats.rs b/crates/replication/src/stats.rs index eab14930d..9f606bd23 100644 --- a/crates/replication/src/stats.rs +++ b/crates/replication/src/stats.rs @@ -265,7 +265,7 @@ fn saturating_atomic_sub(value: &AtomicI64, delta: i64) { return; } - let _ = value.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(delta).max(0))); + let _ = value.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(delta).max(0))); } #[derive(Debug, Clone, Default, Serialize, Deserialize)] diff --git a/crates/s3select-api/src/query/session.rs b/crates/s3select-api/src/query/session.rs index 40cdf4696..37c9e482e 100644 --- a/crates/s3select-api/src/query/session.rs +++ b/crates/s3select-api/src/query/session.rs @@ -231,7 +231,7 @@ impl QueryExecutionTrackerInner { fn mark_timed_out(&self) -> Option { self.state - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |state| { + .try_update(Ordering::AcqRel, Ordering::Acquire, |state| { (state < EXECUTION_FINISHED).then_some(EXECUTION_TIMED_OUT) }) .ok() diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index d089454f4..c84588b3d 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -176,7 +176,7 @@ impl Default for ForegroundReadGuard { impl Drop for ForegroundReadGuard { fn drop(&mut self) { let _ = - SCANNER_FOREGROUND_STREAM_READS.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)); + SCANNER_FOREGROUND_STREAM_READS.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)); } } @@ -206,7 +206,7 @@ impl ScannerRuntimeGuard { impl Drop for ScannerRuntimeGuard { fn drop(&mut self) { - let _ = SCANNER_RUNTIME_INSTANCES.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)); + let _ = SCANNER_RUNTIME_INSTANCES.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)); } } @@ -217,8 +217,8 @@ fn reset_scanner_runtime_instances_for_test() { impl Drop for ScannerActivityGuard { fn drop(&mut self) { - let _ = SCANNER_ACTIVE_WORK_UNITS - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(1))); + let _ = + SCANNER_ACTIVE_WORK_UNITS.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(1))); } } @@ -499,7 +499,7 @@ pub(crate) async fn runtime_tier_registry() -> TierRegistrySnapshot { fn next_tier_registry_generation() -> u64 { TIER_REGISTRY_GENERATION - .fetch_update(Ordering::AcqRel, Ordering::Relaxed, |current| Some(current.saturating_add(1))) + .try_update(Ordering::AcqRel, Ordering::Relaxed, |current| Some(current.saturating_add(1))) .unwrap_or(u64::MAX) } diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 96629b4d3..c427e6035 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -2481,7 +2481,7 @@ impl crate::ScannerConfigObjectDelete for MemoryConfigStore { } if self .block_publication_after_admissions - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| remaining.checked_sub(1)) + .try_update(Ordering::AcqRel, Ordering::Acquire, |remaining| remaining.checked_sub(1)) == Ok(1) { self.publication_admission_blocked.store(true, Ordering::Release); diff --git a/crates/scanner/src/scanner_io/dirty_usage.rs b/crates/scanner/src/scanner_io/dirty_usage.rs index e4e4955db..85bd2d806 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -45,7 +45,7 @@ pub(super) fn usize_to_u64_saturated(value: usize) -> u64 { pub(super) fn advance_generation(generation: &AtomicU64) -> u64 { generation - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| Some(current.saturating_add(1))) + .try_update(Ordering::AcqRel, Ordering::Acquire, |current| Some(current.saturating_add(1))) .map_or_else(|current| current, |previous| previous.saturating_add(1)) } diff --git a/crates/scanner/src/scanner_io/guards.rs b/crates/scanner/src/scanner_io/guards.rs index 8e638a3c0..52a3db927 100644 --- a/crates/scanner/src/scanner_io/guards.rs +++ b/crates/scanner/src/scanner_io/guards.rs @@ -195,14 +195,14 @@ impl Drop for DiskBucketScanGaugeReset { pub(super) fn decrement_atomic_usize(counter: &AtomicUsize) -> usize { counter - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(1))) + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(1))) .map(|previous| previous.saturating_sub(1)) .unwrap_or_else(|current| current) } pub(super) fn increment_atomic_usize(counter: &AtomicUsize) -> usize { counter - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_add(1))) + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_add(1))) .map(|previous| previous.saturating_add(1)) .unwrap_or_else(|current| current) } diff --git a/crates/targets/src/store.rs b/crates/targets/src/store.rs index e890e5517..4a19295ae 100644 --- a/crates/targets/src/store.rs +++ b/crates/targets/src/store.rs @@ -641,7 +641,7 @@ impl QueueStore { // The closure always returns Some, so the update never fails and the Result is discarded. let _ = self .failed_count - .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |current| Some(current.saturating_sub(1))); + .try_update(Ordering::SeqCst, Ordering::SeqCst, |current| Some(current.saturating_sub(1))); } /// Maps a per-entry stat outcome inside the ordered failed scan. A NotFound error means the file diff --git a/crates/targets/src/testkit.rs b/crates/targets/src/testkit.rs index c23eacad2..6fd21293a 100644 --- a/crates/targets/src/testkit.rs +++ b/crates/targets/src/testkit.rs @@ -53,7 +53,7 @@ impl Drop for HealthDropGuard { /// A budget of `usize::MAX` behaves as "always fail" for any realistic call count. fn consume_failure_budget(budget: &AtomicUsize) -> bool { budget - .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) + .try_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| remaining.checked_sub(1)) .is_ok() } diff --git a/rustfs/src/app/object_data_cache/cold_fill.rs b/rustfs/src/app/object_data_cache/cold_fill.rs index b54ab7014..327c0f5bc 100644 --- a/rustfs/src/app/object_data_cache/cold_fill.rs +++ b/rustfs/src/app/object_data_cache/cold_fill.rs @@ -226,7 +226,7 @@ impl ColdFillCoordinator { if self .active_sessions - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| { (current < MAX_ACTIVE_SESSIONS).then_some(current + 1) }) .is_err() @@ -238,7 +238,7 @@ impl ColdFillCoordinator { let session_id = match self .next_session_id - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_add(1)) + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_add(1)) { Ok(session_id) => session_id, Err(_) => { @@ -298,7 +298,7 @@ impl ColdFillCoordinator { fn reserve_global_waiter(&self) -> bool { let result = self .global_waiters - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| { + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| { (current < MAX_GLOBAL_WAITERS).then_some(current + 1) }); if let Ok(previous) = result { diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index a3020664e..400120b3c 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -425,7 +425,7 @@ fn record_active_http_requests(delta: i64) { } else { let decrement = (-delta) as u64; ACTIVE_HTTP_REQUESTS - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(decrement))) + .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(decrement))) .unwrap_or_else(|current| current) .saturating_sub(decrement) }; diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh index 364e8468a..9a3c33f9a 100755 --- a/scripts/check_logging_guardrails.sh +++ b/scripts/check_logging_guardrails.sh @@ -69,6 +69,11 @@ checked_files=( "crates/targets/src/target/webhook.rs" "crates/ecstore/src/store/peer.rs" "crates/ecstore/src/store/init.rs" + "crates/ecstore/src/store/mod.rs" + "crates/ecstore/src/object_api/types.rs" + "crates/ecstore/src/core/sets.rs" + "crates/ecstore/src/set_disk/ops/object.rs" + "crates/ecstore/src/bucket/replication/replication_pool.rs" "crates/s3-client/src/transition_api.rs" "crates/ecstore/src/services/tier/tier.rs" "crates/heal/src/heal/manager.rs" @@ -742,6 +747,27 @@ require_patterns "crates/obs/src/telemetry/rolling.rs" \ 'RollingAppender: Failed to rotate log file after' \ 'RollingAppender: failed to rotate log file' +for raw_ecstore_debug_field in \ + '.field("disk_map",' \ + '.field("pools",' \ + '.field("pool_meta",'; do + if rg -n -F -- "$raw_ecstore_debug_field" crates/ecstore/src/store/mod.rs >/dev/null; then + echo "❌ logging guardrail violation: ECStore Debug must stay bounded and must not render disk_map, pools, or pool_meta" >&2 + exit 1 + fi +done + +for raw_object_options_debug_field in \ + '.field("tier_delete_journal_api", &self.tier_delete_journal_api)' \ + '.field("user_defined", &self.user_defined)' \ + '.field("eval_metadata", &self.eval_metadata)' \ + '.field("http_preconditions", &self.http_preconditions)'; do + if rg -n -F -- "$raw_object_options_debug_field" crates/ecstore/src/object_api/types.rs >/dev/null; then + echo "❌ logging guardrail violation: ObjectOptions Debug must summarize large or request-derived fields" >&2 + exit 1 + fi +done + if rg -n -F -- 'warn!(name = %MaskedAccessKey(name), user_type = ?user_type, "IAM user identity missing")' crates/iam/src/store/object.rs >/dev/null; then echo "❌ logging guardrail violation: missing IAM identity is an expected debug event, not a warning" >&2 exit 1