gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-01 09:48:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(logging): bound ECStore debug output (#6809)

Browse Source 
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 <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-29 12:51:37 +08:00
committed by GitHub
parent 346388b63c
commit c0155f0dfa
32 changed files with 244 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/reliant/tiering.rs
+14 -6
View File
@@ -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(
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+1 -1
View File
@@ -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) {
crates/ecstore/src/bucket/replication/replication_pool.rs
+11 -2
View File
@@ -1048,7 +1048,6 @@ pub fn resync_start_conflict_id(error: &EcstoreError) -> Option<&str> {
}
/// Main replication pool structure
#[derive(Debug)]
pub struct ReplicationPool<S: ReplicationStorage> {
// Atomic counters for active workers
active_workers: Arc<AtomicI32>,
@@ -1094,6 +1093,16 @@ pub struct ReplicationPool<S: ReplicationStorage> {
resyncer: Arc<ReplicationResyncer>,
}
impl<S: ReplicationStorage> std::fmt::Debug for ReplicationPool<S> {
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<S: ReplicationStorage> ReplicationPool<S> {
/// Creates a new replication pool with specified options
pub async fn new(opts: ReplicationPoolOpts, stats: Arc<ReplicationStats>, storage: Arc<S>) -> Arc<Self> {
@@ -2132,7 +2141,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// 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<Self>,
buckets: &[String],
crates/ecstore/src/core/sets.rs
+1 -1
View File
@@ -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<ObjectInfo> {
if opts.delete_prefix && !opts.delete_prefix_object {
self.delete_prefix(bucket, object, &opts).await?;
crates/ecstore/src/data_usage/mod.rs
+1 -1
View File
@@ -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?;
crates/ecstore/src/object_api/types.rs
+70 -1
View File
@@ -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<Arc<crate::store::ECStore>>,
}
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.
crates/ecstore/src/runtime/instance.rs
+2 -2
View File
@@ -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 {
crates/ecstore/src/services/tier/test_util.rs
+1 -1
View File
@@ -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"));
crates/ecstore/src/services/tier/tier.rs
+3 -3
View File
@@ -1651,7 +1651,7 @@ impl TierOperationLease {
) -> std::result::Result<Self, AdminError> {
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),
crates/ecstore/src/set_disk/mod.rs
+2 -2
View File
@@ -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();
}
crates/ecstore/src/set_disk/ops/multipart.rs
+1 -1
View File
@@ -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)
})
{
crates/ecstore/src/set_disk/ops/object.rs
+2 -2
View File
@@ -3976,7 +3976,7 @@ impl<R: AsyncRead + Unpin> AsyncRead for TransitionUploadReader<R> {
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<ObjectInfo> {
// Scanner cleanup carries the per-peer lease fence as transient
// request metadata. Consume it before any delete-prefix fanout so it
crates/ecstore/src/store/list_objects.rs
+1 -1
View File
@@ -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 {
crates/ecstore/src/store/mod.rs
+81 -3
View File
@@ -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<InstanceContext>) -> Arc<ECStore> {
crates/iam/src/lib.rs
+1 -1
View File
@@ -131,7 +131,7 @@ pub(crate) async fn notify_iam_load_user(access_key: &str, temp: bool) -> Vec<Ia
assert!(!probe.panic, "notification probe panic");
let should_fail = probe
.remaining_failures
.fetch_update(std::sync::atomic::Ordering::SeqCst, std::sync::atomic::Ordering::SeqCst, |remaining| {
.try_update(std::sync::atomic::Ordering::SeqCst, std::sync::atomic::Ordering::SeqCst, |remaining| {
remaining.checked_sub(1)
})
.is_ok();
crates/io-core/src/backpressure.rs
+1 -1
View File
@@ -227,7 +227,7 @@ impl BackpressureMonitor {
// usize::MAX, which would permanently reject all future acquisitions.
let prev = match self
.current
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1))
.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1))
{
Ok(prev) => prev,
Err(_) => {
crates/io-core/src/pool.rs
+2 -2
View File
@@ -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();
crates/io-metrics/src/process_lock_metrics.rs
+2 -2
View File
@@ -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)]
crates/lock/src/namespace/tests.rs
+2 -2
View File
@@ -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);
crates/object-data-cache/src/memory.rs
+1 -1
View File
@@ -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);
crates/object-data-cache/src/moka_backend.rs
+1 -1
View File
@@ -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),
crates/replication/src/stats.rs
+1 -1
View File
@@ -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)]
crates/s3select-api/src/query/session.rs
+1 -1
View File
@@ -231,7 +231,7 @@ impl QueryExecutionTrackerInner {
fn mark_timed_out(&self) -> Option<u8> {
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()
crates/scanner/src/lib.rs
+5 -5
View File
@@ -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)
}
crates/scanner/src/scanner/tests.rs
+1 -1
View File
@@ -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);
crates/scanner/src/scanner_io/dirty_usage.rs
+1 -1
View File
@@ -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))
}
crates/scanner/src/scanner_io/guards.rs
+2 -2
View File
@@ -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)
}
crates/targets/src/store.rs
+1 -1
View File
@@ -641,7 +641,7 @@ impl<T: Serialize + DeserializeOwned + Send + Sync> QueueStore<T> {
// 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
crates/targets/src/testkit.rs
+1 -1
View File
@@ -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()
}
rustfs/src/app/object_data_cache/cold_fill.rs
+3 -3
View File
@@ -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 {
rustfs/src/server/http.rs
+1 -1
View File
@@ -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)
};
scripts/check_logging_guardrails.sh
+26
View File
@@ -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