From 371f5e4ffc04f096a7508e942c7d4ec96705824d Mon Sep 17 00:00:00 2001 From: Hauser Zhi Date: Thu, 5 Nov 2026 05:59:05 +0000 Subject: [PATCH 1/2] fix(ecstore): streamline lock RPC and listing ownership (#9176) * fix(ecstore): preserve lock cleanup and listing trace context Own remote lock admission and dispatched RPC cleanup across caller cancellation. Reuse bounded late-result cleanup and retain the listing request span in metacache tasks. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * perf(ecstore): streamline lock RPC and listing fanout Poll RPCs in callers and transfer pending work before channel eviction. Share walk options across producers and borrow cached peer keys. Keep deadline classification deterministic. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: heihutu Co-authored-by: zhi22915 --- .../ecstore/src/cache_value/metacache_set.rs | 210 ++++++- .../ecstore/src/cluster/rpc/remote_locker.rs | 583 ++++++++++++++++-- 2 files changed, 713 insertions(+), 80 deletions(-) diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index 4a9aef323..4f87e071a 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -34,7 +34,7 @@ use tokio::sync::Mutex as TokioMutex; use tokio::task::JoinSet; use tokio::time::timeout; use tokio_util::sync::CancellationToken; -use tracing::{debug, error, warn}; +use tracing::{Instrument, debug, error, warn}; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_METACACHE: &str = "metacache"; @@ -363,10 +363,15 @@ async fn list_path_raw_inner( let producer_errs: Arc<[OnceLock]> = (0..opts.disks.len()).map(|_| OnceLock::new()).collect::>().into(); let cancel_rx = CancellationToken::new(); + let walk_options = Arc::new(walk_dir_options(&opts)); for (disk_idx, disk) in opts.disks.iter().enumerate() { let opdisk = disk.clone(); - let opts_clone = opts.clone(); + let walk_options = walk_options.clone(); + #[cfg(test)] + let test_reader_behavior = opts.test_reader_behaviors.get(disk_idx).cloned(); + #[cfg(test)] + let test_disk_io_context_observed = opts.test_disk_io_context_observed.clone(); let fallback_claim_tracker = fallback_claim_tracker.clone(); let fds_clone = fds.clone(); #[cfg(test)] @@ -376,17 +381,17 @@ async fn list_path_raw_inner( let disk_io_context_clone = disk_io_context.clone(); let (rd, wr) = tokio::io::duplex(64 * 1024); readers.push(MetacacheReader::new(rd)); - jobs.spawn(with_optional_disk_io_context(disk_io_context_clone, async move { + let producer = with_optional_disk_io_context(disk_io_context_clone, async move { let mut wr = PublishedBytesWriter::new(wr); #[cfg(test)] - if let Some(observer) = &opts_clone.test_disk_io_context_observed { + if let Some(observer) = &test_disk_io_context_observed { observer.store( crate::disk::current_disk_io_context().is_some_and(|context| context.class == DiskIoClass::Background), std::sync::atomic::Ordering::Release, ); } #[cfg(test)] - let test_primary_error = if let Some(behavior) = opts_clone.test_reader_behaviors.get(disk_idx).cloned() { + let test_primary_error = if let Some(behavior) = test_reader_behavior { match behavior { TestReaderBehavior::Eof => return Ok(()), TestReaderBehavior::Entries(entries) => { @@ -443,8 +448,6 @@ async fn list_path_raw_inner( None }; - let wakl_opts = walk_dir_options(&opts_clone); - let mut need_fallback = false; let mut last_err = None; #[cfg(test)] @@ -459,7 +462,7 @@ async fn list_path_raw_inner( if !need_fallback && let Some(disk) = opdisk { let primary_walk_started = std::time::Instant::now(); - match disk.walk_dir(wakl_opts, &mut wr).await { + match disk.walk_dir((*walk_options).clone(), &mut wr).await { Ok(_res) => { rustfs_io_metrics::record_stage_duration( "metacache_walk_dir_primary", @@ -483,8 +486,8 @@ async fn list_path_raw_inner( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, - bucket = %opts_clone.bucket, - path = %opts_clone.path, + bucket = %walk_options.bucket, + path = %walk_options.base_dir, disk_index = disk_idx, state = "walk_dir_missing_path", error = ?err, @@ -495,8 +498,8 @@ async fn list_path_raw_inner( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, - bucket = %opts_clone.bucket, - path = %opts_clone.path, + bucket = %walk_options.bucket, + path = %walk_options.base_dir, disk_index = disk_idx, state = "walk_dir_failed", error = ?err, @@ -598,8 +601,8 @@ async fn list_path_raw_inner( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, - bucket = %opts_clone.bucket, - path = %opts_clone.path, + bucket = %walk_options.bucket, + path = %walk_options.base_dir, disk_index = disk_idx, state = "fallback_disk_missing_for_path", error = ?err, @@ -610,8 +613,8 @@ async fn list_path_raw_inner( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, - bucket = %opts_clone.bucket, - path = %opts_clone.path, + bucket = %walk_options.bucket, + path = %walk_options.base_dir, disk_index = disk_idx, state = "fallback_disk_missing", error = ?err, @@ -626,7 +629,7 @@ async fn list_path_raw_inner( } let fallback_walk_started = std::time::Instant::now(); - match disk.as_ref().walk_dir(walk_dir_options(&opts_clone), &mut wr).await { + match disk.as_ref().walk_dir((*walk_options).clone(), &mut wr).await { Ok(_r) => { rustfs_io_metrics::record_stage_duration( "metacache_walk_dir_fallback", @@ -652,8 +655,8 @@ async fn list_path_raw_inner( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, - bucket = %opts_clone.bucket, - path = %opts_clone.path, + bucket = %walk_options.bucket, + path = %walk_options.base_dir, disk_index = disk_idx, state = "fallback_walk_dir_missing_path", error = ?err, @@ -664,8 +667,8 @@ async fn list_path_raw_inner( event = EVENT_METACACHE_LISTING, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_METACACHE, - bucket = %opts_clone.bucket, - path = %opts_clone.path, + bucket = %walk_options.bucket, + path = %walk_options.base_dir, disk_index = disk_idx, state = "fallback_walk_dir_failed", error = ?err, @@ -687,12 +690,13 @@ async fn list_path_raw_inner( // warn!("list_path_raw: while need_fallback done"); Ok(()) - })); + }); + jobs.spawn(producer.in_current_span()); } let revjob_rx = rx.clone(); let mut revjobs = JoinSet::new(); - revjobs.spawn(async move { + let merge = async move { // Consumer-side peek timeout, and a caveat worth understanding // (rustfs/backlog#1217). // @@ -1035,7 +1039,8 @@ async fn list_path_raw_inner( } } Ok(ListPathRawOutcome { limit_reached }) - }); + }; + revjobs.spawn(merge.in_current_span()); let merge_started = std::time::Instant::now(); let Some(revjob) = revjobs.join_next().await else { @@ -1249,6 +1254,163 @@ mod tests { ); } + #[tokio::test] + async fn list_path_raw_fanout_retains_only_linear_disk_references() { + let root = tempfile::tempdir().expect("create disk root"); + let endpoint = crate::disk::endpoint::Endpoint::try_from(root.path().to_str().expect("UTF-8 disk path")) + .expect("parse disk endpoint"); + let local = Arc::new( + crate::disk::local::LocalDisk::new(&endpoint, false) + .await + .expect("open local disk"), + ); + let wrapper = crate::disk::disk_store::LocalDiskWrapper::new(local, false); + let disk = Arc::new(crate::disk::Disk::Local(Box::new(wrapper))); + let baseline = Arc::strong_count(&disk); + + for primary_count in [1, 4, 16] { + for fallback_count in [0, 2] { + let entered: Vec<_> = (0..primary_count).map(|_| Arc::new(tokio::sync::Notify::new())).collect(); + let stopped: Vec<_> = (0..primary_count).map(|_| Arc::new(tokio::sync::Notify::new())).collect(); + let behaviors = entered + .iter() + .zip(&stopped) + .map(|(entered, stopped)| TestReaderBehavior::StallWithSignals { + entered: entered.clone(), + stopped: stopped.clone(), + }) + .collect(); + let cancel = CancellationToken::new(); + let listing = tokio::spawn(list_path_raw( + cancel.clone(), + ListPathRawOptions { + disks: vec![Some(disk.clone()); primary_count], + fallback_disks: vec![Some(disk.clone()); fallback_count], + min_disks: primary_count, + test_reader_behaviors: behaviors, + peek_timeout: Some(Duration::from_secs(60)), + ..Default::default() + }, + )); + for signal in &entered { + timeout(Duration::from_secs(2), signal.notified()) + .await + .expect("every producer must enter the real fanout before counting references"); + } + + // The merge owns the configured primary/fallback slots, each producer + // owns its primary, and the shared fallback queue owns each fallback. + let expected = baseline + 2 * (primary_count + fallback_count); + let retained = Arc::strong_count(&disk); + cancel.cancel(); + timeout(Duration::from_secs(2), listing) + .await + .expect("cancellation must stop the listing") + .expect("listing task must join") + .expect("external cancellation must finish cleanly"); + for signal in &stopped { + timeout(Duration::from_secs(2), signal.notified()) + .await + .expect("cancellation must drop every producer"); + } + assert_eq!(Arc::strong_count(&disk), baseline, "cancellation must release every fanout reference"); + assert_eq!( + retained, expected, + "fanout must retain linear references for {primary_count} primary and {fallback_count} fallback slots" + ); + } + } + } + + #[tokio::test(flavor = "current_thread")] + async fn list_path_raw_producer_and_merge_preserve_request_span() { + use tracing_subscriber::{Layer, layer::SubscriberExt}; + + struct ProducerSpan(Arc); + #[derive(Default)] + struct StateField(bool); + impl tracing::field::Visit for StateField { + fn record_debug(&mut self, _: &tracing::field::Field, _: &dyn std::fmt::Debug) {} + + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.0 |= field.name() == "state" && value == "walk_dir_missing_path"; + } + } + impl Layer for ProducerSpan + where + S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>, + { + fn on_event(&self, event: &tracing::Event<'_>, context: tracing_subscriber::layer::Context<'_, S>) { + let mut state = StateField::default(); + event.record(&mut state); + if state.0 + && context + .lookup_current() + .is_some_and(|span| span.name() == "flink_list_request") + { + self.0.store(true, std::sync::atomic::Ordering::Release); + } + } + } + + // The subscriber is thread-local; keep the spawned tasks on this runtime thread. + let producer_observed = Arc::new(AtomicBool::new(false)); + let subscriber = tracing_subscriber::registry().with(ProducerSpan(producer_observed.clone())); + let _subscriber = tracing::subscriber::set_default(subscriber); + let request = tracing::info_span!("flink_list_request", request_id = "list-span-regression"); + let root = tempfile::tempdir().expect("create disk root"); + let endpoint = crate::disk::endpoint::Endpoint::try_from(root.path().to_str().expect("UTF-8 disk path")) + .expect("parse disk endpoint"); + let local = Arc::new( + crate::disk::local::LocalDisk::new(&endpoint, false) + .await + .expect("open local disk"), + ); + let wrapper = crate::disk::disk_store::LocalDiskWrapper::new(local, false); + let disk = Arc::new(crate::disk::Disk::Local(Box::new(wrapper))); + let error = list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![Some(disk)], + bucket: "missing-bucket".to_owned(), + min_disks: 1, + ..Default::default() + }, + ) + .instrument(request.clone()) + .await + .expect_err("missing bucket must fail the real disk producer"); + assert_eq!(error, DiskError::VolumeNotFound); + assert!( + producer_observed.load(std::sync::atomic::Ordering::Acquire), + "disk error must retain the request span" + ); + + let merge_observed = Arc::new(AtomicBool::new(false)); + let merge_sink = merge_observed.clone(); + let request_id = request.id(); + list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None], + min_disks: 1, + test_reader_behaviors: vec![TestReaderBehavior::Entries(vec![fallback_test_entry()])], + agreed: Some(Box::new(move |_| { + merge_sink.store(tracing::Span::current().id() == request_id, std::sync::atomic::Ordering::Release); + Box::pin(async { Ok(()) }) + })), + ..Default::default() + }, + ) + .instrument(request) + .await + .expect("merge should deliver the test entry"); + assert!( + merge_observed.load(std::sync::atomic::Ordering::Acquire), + "merge callback must retain the request span" + ); + } + #[tokio::test] async fn scanner_joinset_walk_uses_the_local_disk_background_gate() { let root = tempfile::tempdir().expect("local disk root should be created"); diff --git a/crates/ecstore/src/cluster/rpc/remote_locker.rs b/crates/ecstore/src/cluster/rpc/remote_locker.rs index da8839d91..718ca34af 100644 --- a/crates/ecstore/src/cluster/rpc/remote_locker.rs +++ b/crates/ecstore/src/cluster/rpc/remote_locker.rs @@ -18,6 +18,7 @@ use crate::cluster::rpc::client::{ use crate::cluster::rpc::set_tonic_rolling_mutation_body_digest; use async_trait::async_trait; use bytes::Bytes; +use futures::{FutureExt, future::BoxFuture}; use rustfs_lock::{ LockClient, LockError, LockInfo, LockRequest, LockResponse, LockStats, LockStatus, LockType, Result, types::{LockId, LockMetadata, LockPriority}, @@ -32,16 +33,17 @@ use rustfs_protos::{ }; use std::collections::HashMap; use std::future::Future; +use std::panic::AssertUnwindSafe; use std::pin::Pin; use std::sync::{Mutex, OnceLock}; use std::time::Duration; -use tokio::task::JoinHandle; use tokio::time::{Instant, timeout}; use tonic::service::interceptor::InterceptedService; use tonic::{Request, Response}; use tracing::{debug, info, warn}; const EVENT_NAMESPACE_LOCK_RELEASE: &str = "namespace_lock_release"; +const EVENT_REMOTE_LOCK_RPC_CLEANUP: &str = "remote_lock_rpc_cleanup"; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_REMOTE_LOCK: &str = "remote_lock"; @@ -49,18 +51,38 @@ fn attach_lock_mutation_body_digest(req set_tonic_rolling_mutation_body_digest(request) } -/// Work to run if an RPC that already timed out for its caller completes later. +/// Work to run if an RPC completes after its caller stops waiting. type LateCompletion = Option Pin + Send>> + Send>>; type LateFailure = LateCompletion<()>; +type RpcFuture = BoxFuture<'static, std::thread::Result>>; -struct RpcCleanup { - late: LateCompletion, - late_failure: LateFailure, +/// Own admission before connecting as well as while an RPC is in flight. +struct LockRequestGuard<'a> { + client: &'a RemoteClient, + op: &'static str, admission: Option, } -impl RpcCleanup { - fn new(late: LateCompletion, late_failure: LateFailure, admission: Option) -> Self { +impl LockRequestGuard<'_> { + fn into_admission(mut self) -> Option { + self.admission.take() + } +} + +impl Drop for LockRequestGuard<'_> { + fn drop(&mut self) { + self.client.release_lock_request(self.op, self.admission.take()); + } +} + +struct RpcCleanup<'a, T> { + late: LateCompletion, + late_failure: LateFailure, + admission: Option>, +} + +impl<'a, T> RpcCleanup<'a, T> { + fn new(late: LateCompletion, late_failure: LateFailure, admission: Option>) -> Self { Self { late, late_failure, @@ -69,6 +91,25 @@ impl RpcCleanup { } } +/// A dropped caller hands its RPC and cleanup to the same bounded owner as a timeout. +struct InFlightRpc<'a, T: Send + 'static> { + client: &'a RemoteClient, + op: &'static str, + resource_summary: &'a str, + future: Option>, + cleanup: Option>, +} + +impl Drop for InFlightRpc<'_, T> { + fn drop(&mut self) { + if let Some(future) = self.future.take() + && let Some(cleanup) = self.cleanup.take() + { + self.client.detach_rpc(self.op, self.resource_summary, future, cleanup); + } + } +} + /// The liveness window is this many RPC deadlines: a peer that completed a /// lock RPC within it is slow, not gone, and keeps its channel on a timeout. const LOCK_RPC_LIVENESS_WINDOW_DEADLINES: u32 = 2; @@ -137,6 +178,9 @@ fn with_lock_peer_health(addr: &str, update: impl FnOnce(&mut LockPeerChannel let mut peers = lock_peer_channel_health() .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(health) = peers.get_mut(addr) { + return update(health); + } update(peers.entry(addr.to_string()).or_default()) } @@ -158,6 +202,15 @@ fn reset_lock_peer_health_for_test(addr: &str) { .remove(addr); } +#[cfg(test)] +type EvictionPause = (tokio::sync::oneshot::Sender<()>, tokio::sync::oneshot::Receiver<()>); + +#[cfg(test)] +fn eviction_pauses_for_test() -> &'static Mutex> { + static PAUSES: OnceLock>> = OnceLock::new(); + PAUSES.get_or_init(Mutex::default) +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum EvictionTrigger { /// The caller's deadline expired while the stream was still open. @@ -506,6 +559,14 @@ impl RemoteClient { admission } + fn admit_lock_request_guard(&self, op: &'static str) -> Option> { + self.admit_lock_request(op).map(|admission| LockRequestGuard { + client: self, + op, + admission: Some(admission), + }) + } + fn release_lock_request(&self, op: &'static str, admission: Option) { if is_request_breaker_operation(op) && admission.is_some() { with_lock_peer_health(&self.addr, |health| { @@ -587,6 +648,17 @@ impl RemoteClient { trigger: EvictionTrigger, deadline: Duration, ) { + #[cfg(test)] + let pause = eviction_pauses_for_test() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(&self.addr); + #[cfg(test)] + if let Some((entered, resume)) = pause { + let _ = entered.send(()); + let _ = resume.await; + } + let now = Instant::now(); let cooldown = Self::eviction_cooldown(); let liveness_window = Self::liveness_window(deadline); @@ -617,22 +689,26 @@ impl RemoteClient { ); } - /// Keep a timed-out RPC running instead of cancelling its stream. + /// Keep an abandoned RPC running instead of cancelling its stream. /// /// Dropping the future sends `RST_STREAM`; under load those resets pile up /// in the server's pending-accept queue until it answers `GOAWAY /// too_many_resets` and kills every stream on the connection. A detached /// stream ends on its own within the internode RPC timeout, the number per /// peer is bounded, and a lock granted after its caller gave up is released. - fn detach_timed_out_rpc( + fn detach_rpc( &self, op: &'static str, resource_summary: &str, - handle: JoinHandle>, - late: LateCompletion, - late_failure: LateFailure, - admission: Option, + future: RpcFuture, + cleanup: RpcCleanup<'_, T>, ) { + let RpcCleanup { + late, + late_failure, + admission, + } = cleanup; + let admission = admission.and_then(LockRequestGuard::into_admission); let limit = Self::detached_rpc_limit(); let admitted = with_lock_peer_health(&self.addr, |health| { if health.detached_rpcs >= limit { @@ -643,25 +719,29 @@ impl RemoteClient { } }); if !admitted { - handle.abort(); + drop(future); self.release_lock_request(op, admission); if let Some(late_failure) = late_failure { drop(tokio::spawn(late_failure(()))); } rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_detached(op, "aborted"); debug!( + event = EVENT_REMOTE_LOCK_RPC_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REMOTE_LOCK, + state = "detached_budget_exhausted", addr = %self.addr, op, resource_summary, limit, - "Cancelled timed-out remote lock RPC because the detached stream budget is exhausted" + "Abandoned remote lock RPC cancelled" ); return; } rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_detached(op, "detached"); let addr = self.addr.clone(); tokio::spawn(async move { - let outcome = handle.await; + let outcome = future.await; with_lock_peer_health(&addr, |health| { health.detached_rpcs = health.detached_rpcs.saturating_sub(1); if is_request_breaker_operation(op) && admission.is_some() { @@ -688,19 +768,31 @@ impl RemoteClient { } rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_late_completion(op, "error"); debug!( + event = EVENT_REMOTE_LOCK_RPC_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REMOTE_LOCK, + state = "late_rpc_failed", addr = %addr, op, tonic_code = ?status.code(), tonic_message = status.message(), - "Detached remote lock RPC failed after its caller timed out" + "Abandoned remote lock RPC failed" ); } - Err(join_error) => { + Err(_) => { if let Some(late_failure) = late_failure { late_failure(()).await; } rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_late_completion(op, "join_error"); - debug!(addr = %addr, op, error = %join_error, "Detached remote lock RPC task ended abnormally"); + debug!( + event = EVENT_REMOTE_LOCK_RPC_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REMOTE_LOCK, + state = "late_rpc_panicked", + addr = %addr, + op, + "Abandoned remote lock RPC panicked" + ); } } }); @@ -788,16 +880,28 @@ impl RemoteClient { self.release_late_release(); rustfs_io_metrics::lock_metrics::record_remote_lock_late_release(outcome); if outcome == "released" { - debug!(addr = %self.addr, count = original_count, "Released remote locks granted after their caller timed out"); + debug!( + event = EVENT_REMOTE_LOCK_RPC_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REMOTE_LOCK, + state = "late_grants_released", + addr = %self.addr, + count = original_count, + "Released remote locks granted after their caller stopped waiting" + ); } else { let (log_warning, suppressed_timeouts) = self.timeout_log_decision("late_release", Self::rpc_timeout()); if log_warning { warn!( + event = EVENT_REMOTE_LOCK_RPC_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REMOTE_LOCK, + state = "late_grants_unreleased", addr = %self.addr, count = original_count, outcome, suppressed_timeouts, - "Could not release every remote lock granted after its caller timed out; the server lease will expire it" + "Could not release every remote lock granted after its caller stopped waiting; the server lease will expire it" ); } else { debug!(addr = %self.addr, count = original_count, outcome, suppressed_timeouts, "Suppressed repeated late remote lock release warning"); @@ -811,25 +915,47 @@ impl RemoteClient { resource_summary: &str, deadline: Duration, future: Fut, - cleanup: RpcCleanup, + cleanup: RpcCleanup<'_, T>, ) -> std::result::Result where Fut: Future> + Send + 'static, T: Send + 'static, { - let RpcCleanup { - late, + let mut rpc = InFlightRpc { + client: self, + op, + resource_summary, + future: Some(AssertUnwindSafe(future).catch_unwind().boxed()), + cleanup: Some(cleanup), + }; + let outcome = match rpc.future.as_mut() { + Some(future) => tokio::select! { + // A response becoming ready on the deadline tick must not turn an + // expired caller budget into a transport error and channel eviction. + biased; + _ = tokio::time::sleep(deadline) => None, + outcome = future => Some(outcome), + }, + None => return Err(LockError::internal("remote lock RPC has no in-flight future")), + }; + if outcome.is_some() { + rpc.future.take(); + } + let Some(RpcCleanup { + late: _, late_failure, admission, - } = cleanup; - let mut handle = tokio::spawn(future); - match timeout(deadline, &mut handle).await { - Ok(Ok(Ok(response))) => { - self.record_rpc_success(op, admission); - self.release_lock_request(op, admission); + }) = rpc.cleanup.as_mut() + else { + return Err(LockError::internal("remote lock RPC has no cleanup owner")); + }; + let token = admission.as_ref().and_then(|guard| guard.admission); + let result = match outcome { + Some(Ok(Ok(response))) => { + self.record_rpc_success(op, token); Ok(response) } - Ok(Ok(Err(err))) => { + Some(Ok(Err(err))) => { let reason = err.to_string(); if err.code() == tonic::Code::DeadlineExceeded { rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_timeout(&self.addr, op); @@ -862,12 +988,11 @@ impl RemoteClient { "Suppressed repeated remote lock RPC deadline" ); } - self.maybe_evict_connection(op, &reason, resource_summary, EvictionTrigger::Timeout, deadline) - .await; - self.release_lock_request(op, admission); - if let Some(late_failure) = late_failure { + if let Some(late_failure) = late_failure.take() { drop(tokio::spawn(late_failure(()))); } + self.maybe_evict_connection(op, &reason, resource_summary, EvictionTrigger::Timeout, deadline) + .await; return Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), deadline)); } // Only evict (and re-dial) the cached channel when the failure is a genuine @@ -928,30 +1053,32 @@ impl RemoteClient { ); } if transport_failure { + // Ambiguous-grant cleanup owns its work before channel eviction can suspend. + if let Some(late_failure) = late_failure.take() { + drop(tokio::spawn(late_failure(()))); + } self.maybe_evict_connection(op, &reason, resource_summary, EvictionTrigger::Transport, deadline) .await; } - self.release_lock_request(op, admission); - if transport_failure && let Some(late_failure) = late_failure { - drop(tokio::spawn(late_failure(()))); - } Err(LockError::internal(format!("{op} RPC failed: {reason}"))) } - Ok(Err(join_error)) => { + Some(Err(_)) => { warn!( + event = EVENT_REMOTE_LOCK_RPC_CLEANUP, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REMOTE_LOCK, + state = "rpc_panicked", addr = %self.addr, op, resource_summary, - error = %join_error, - "Remote lock RPC task ended abnormally" + "Remote lock RPC panicked" ); - self.release_lock_request(op, admission); - if let Some(late_failure) = late_failure { + if let Some(late_failure) = late_failure.take() { drop(tokio::spawn(late_failure(()))); } - Err(LockError::internal(format!("{op} RPC task failed: {join_error}"))) + Err(LockError::internal(format!("{op} RPC future panicked"))) } - Err(_) => { + None => { let reason = format!("RPC timed out after {deadline:?}"); rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_timeout(&self.addr, op); self.record_request_timeout(op, deadline); @@ -983,12 +1110,17 @@ impl RemoteClient { "Suppressed repeated remote lock RPC timeout" ); } + // Continue draining the abandoned stream while channel eviction waits. + drop(rpc); self.maybe_evict_connection(op, &reason, resource_summary, EvictionTrigger::Timeout, deadline) .await; - self.detach_timed_out_rpc(op, resource_summary, handle, late, late_failure, admission); - Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), deadline)) + return Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), deadline)); } + }; + if rpc.future.is_none() { + rpc.cleanup.take(); } + result } fn rpc_timeout_failure_response(request: &LockRequest, err: &LockError) -> LockResponse { @@ -1079,7 +1211,7 @@ impl RemoteClient { #[async_trait] impl LockClient for RemoteClient { async fn acquire_lock(&self, request: &LockRequest) -> Result { - let admission = match self.admit_lock_request("lock") { + let admission = match self.admit_lock_request_guard("lock") { Some(admission) => admission, None => return Ok(Self::rpc_backoff_failure_response(request)), }; @@ -1090,7 +1222,6 @@ impl LockClient for RemoteClient { Ok(client) => client, Err(err) => { self.record_request_timeout("lock", rpc_timeout); - self.release_lock_request("lock", Some(admission)); return Err(err); } }; @@ -1098,14 +1229,12 @@ impl LockClient for RemoteClient { let args = match serde_json::to_string(&bounded_request) { Ok(args) => args, Err(err) => { - self.release_lock_request("lock", Some(admission)); return Err(LockError::internal(format!("Failed to serialize request: {err}"))); } }; let mut req = Request::new(GenerallyLockRequest { args }); req.set_timeout(rpc_timeout); if let Err(err) = attach_lock_mutation_body_digest(&mut req) { - self.release_lock_request("lock", Some(admission)); return Err(err.into()); } let late = self.late_release_hook(request.lock_id.clone()); @@ -1147,7 +1276,7 @@ impl LockClient for RemoteClient { return Ok(Vec::new()); } - let admission = match self.admit_lock_request("lock_batch") { + let admission = match self.admit_lock_request_guard("lock_batch") { Some(admission) => admission, None => return Ok(Self::rpc_backoff_failure_batch(requests)), }; @@ -1161,7 +1290,6 @@ impl LockClient for RemoteClient { Ok(client) => client, Err(err) => { self.record_request_timeout("lock_batch", rpc_timeout); - self.release_lock_request("lock_batch", Some(admission)); return Err(err); } }; @@ -1175,14 +1303,12 @@ impl LockClient for RemoteClient { { Ok(args) => args, Err(err) => { - self.release_lock_request("lock_batch", Some(admission)); return Err(err); } }; let mut req = Request::new(BatchGenerallyLockRequest { args }); req.set_timeout(rpc_timeout); if let Err(err) = attach_lock_mutation_body_digest(&mut req) { - self.release_lock_request("lock_batch", Some(admission)); return Err(err.into()); } let late = self.late_release_batch_hook(requests.iter().map(|request| request.lock_id.clone()).collect()); @@ -1710,6 +1836,351 @@ mod tests { assert!(acquired_lock_ids(&lock_ids, &[]).is_empty()); } + #[tokio::test] + async fn test_remote_client_cancelled_connection_releases_request_admission() { + ensure_test_rpc_secret(); + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind pending-handshake peer"); + let addr = format!("http://{}", listener.local_addr().expect("pending-handshake peer address")); + let client = RemoteClient::new(addr.clone()); + let request = test_lock_request(Duration::from_secs(30)); + + for batch in [false, true] { + reset_lock_peer_health_for_test(&addr); + with_lock_peer_health(&addr, |health| { + health.request_backoff_until = Some(Instant::now() - Duration::from_secs(1)); + }); + let mut acquisition = Box::pin(async { + if batch { + client.acquire_locks_batch(std::slice::from_ref(&request)).await.map(|_| ()) + } else { + client.acquire_lock(&request).await.map(|_| ()) + } + }); + std::future::poll_fn(|cx| { + assert!(acquisition.as_mut().poll(cx).is_pending(), "peer must wait for its HTTP/2 handshake"); + std::task::Poll::Ready(()) + }) + .await; + let pending = lock_peer_health_for_test(&addr); + assert_eq!(pending.request_in_flight, 1, "pending connection owns one request admission"); + assert!(pending.request_probe_generation.is_some(), "the request is the half-open probe"); + + drop(acquisition); + let cancelled = lock_peer_health_for_test(&addr); + assert_eq!(cancelled.request_in_flight, 0, "cancellation must return request admission"); + assert!(cancelled.request_probe_generation.is_none(), "cancellation must unblock the next probe"); + let probe = client.admit_lock_request("lock").expect("cancelled peer can be probed again"); + client.release_lock_request("lock", Some(probe)); + } + reset_lock_peer_health_for_test(&addr); + } + + #[tokio::test] + async fn test_remote_client_cancelled_rpc_runs_late_grant_cleanup() { + let addr = "http://cancelled-rpc-late-grant"; + reset_lock_peer_health_for_test(addr); + let client = RemoteClient::new(addr.to_string()); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (grant_tx, grant_rx) = tokio::sync::oneshot::channel(); + let (released_tx, released_rx) = tokio::sync::oneshot::channel(); + let caller = tokio::spawn(async move { + let late: LateCompletion = Some(Box::new(move |granted| { + Box::pin(async move { + released_tx.send(granted).expect("late grant observer remains alive"); + }) + })); + client + .execute_rpc( + "lock", + "bucket/object", + Duration::from_secs(30), + async move { + started_tx.send(()).expect("caller observes RPC dispatch"); + Ok(grant_rx.await.expect("test decides when peer grants lock")) + }, + RpcCleanup::new(late, None, None), + ) + .await + }); + timeout(Duration::from_secs(1), started_rx) + .await + .expect("RPC must dispatch") + .expect("dispatch signal"); + caller.abort(); + assert!(caller.await.expect_err("caller was cancelled").is_cancelled()); + assert_eq!( + lock_peer_health_for_test(addr).detached_rpcs, + 1, + "cancelled RPC uses bounded detached ownership" + ); + grant_tx.send(true).expect("cancellation must preserve the running RPC"); + assert!( + timeout(Duration::from_secs(1), released_rx) + .await + .expect("late grant must be cleaned up") + .expect("cleanup result") + ); + assert_eq!(lock_peer_health_for_test(addr).detached_rpcs, 0, "completed cleanup returns its slot"); + reset_lock_peer_health_for_test(addr); + } + + #[tokio::test] + async fn test_remote_client_cancelled_rpc_runs_ambiguous_failure_cleanup() { + let addr = "http://cancelled-rpc-late-failure"; + reset_lock_peer_health_for_test(addr); + let client = RemoteClient::new(addr.to_string()); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (failure_tx, failure_rx) = tokio::sync::oneshot::channel(); + let (released_tx, released_rx) = tokio::sync::oneshot::channel(); + let caller = tokio::spawn(async move { + let late_failure: LateFailure = Some(Box::new(move |()| { + Box::pin(async move { + released_tx.send(()).expect("ambiguous cleanup observer remains alive"); + }) + })); + client + .execute_rpc( + "lock_batch", + "bucket/object", + Duration::from_secs(30), + async move { + started_tx.send(()).expect("caller observes RPC dispatch"); + failure_rx.await.expect("test decides when transport fails"); + Err::<(), _>(tonic::Status::cancelled("transport failure after caller cancellation")) + }, + RpcCleanup::new(None, late_failure, None), + ) + .await + }); + timeout(Duration::from_secs(1), started_rx) + .await + .expect("RPC must dispatch") + .expect("dispatch signal"); + caller.abort(); + assert!(caller.await.expect_err("caller was cancelled").is_cancelled()); + failure_tx.send(()).expect("cancellation must preserve the running RPC"); + timeout(Duration::from_secs(1), released_rx) + .await + .expect("ambiguous grant must be cleaned up") + .expect("cleanup signal"); + assert_eq!(lock_peer_health_for_test(addr).detached_rpcs, 0, "failed detached RPC returns its slot"); + reset_lock_peer_health_for_test(addr); + } + + #[tokio::test] + async fn test_remote_client_cancelled_completed_failure_runs_cleanup() { + for (label, status) in [ + ("transport", tonic::Status::cancelled("ambiguous transport failure")), + ("deadline", tonic::Status::deadline_exceeded("ambiguous deadline failure")), + ] { + let addr = format!("http://cancelled-completed-{label}"); + reset_lock_peer_health_for_test(&addr); + let client = RemoteClient::new(addr.clone()); + let (evicting_tx, evicting_rx) = tokio::sync::oneshot::channel(); + let (_resume_tx, resume_rx) = tokio::sync::oneshot::channel(); + eviction_pauses_for_test() + .lock() + .expect("eviction pause registry") + .insert(addr.clone(), (evicting_tx, resume_rx)); + let (released_tx, released_rx) = tokio::sync::oneshot::channel(); + let caller = tokio::spawn(async move { + let late_failure: LateFailure = Some(Box::new(move |()| { + Box::pin(async move { + released_tx + .send(()) + .expect("completed-failure cleanup observer remains alive"); + }) + })); + client + .execute_rpc( + "lock", + "bucket/object", + Duration::from_secs(30), + async move { Err::<(), _>(status) }, + RpcCleanup::new(None, late_failure, None), + ) + .await + }); + timeout(Duration::from_secs(1), evicting_rx) + .await + .expect("completed error must reach channel eviction") + .expect("eviction signal"); + timeout(Duration::from_secs(1), released_rx) + .await + .expect("cleanup must transfer before channel eviction suspends") + .expect("cleanup signal"); + caller.abort(); + assert!(caller.await.expect_err("caller is cancelled during eviction").is_cancelled()); + assert_eq!( + lock_peer_health_for_test(&addr).detached_rpcs, + 0, + "completed RPC cannot consume a detached slot" + ); + reset_lock_peer_health_for_test(&addr); + } + } + + #[tokio::test(start_paused = true)] + async fn test_remote_client_deadline_wins_over_same_tick_transport_timeout() { + let addr = "http://same-tick-rpc-deadline"; + reset_lock_peer_health_for_test(addr); + with_lock_peer_health(addr, |health| health.last_success = Some(Instant::now())); + let client = RemoteClient::new(addr.to_string()); + let budget = Duration::from_millis(50); + let (released_tx, released_rx) = tokio::sync::oneshot::channel(); + let late_failure: LateFailure = Some(Box::new(move |()| { + Box::pin(async move { + released_tx.send(()).expect("same-tick cleanup observer remains alive"); + }) + })); + let result = client + .execute_rpc( + "lock", + "bucket/object", + budget, + async move { + tokio::time::sleep(budget).await; + Err::<(), _>(tonic::Status::cancelled("inner transport deadline expired")) + }, + RpcCleanup::new(None, late_failure, None), + ) + .await; + assert!( + matches!(result, Err(LockError::Timeout { .. })), + "expired caller budget must retain timeout classification" + ); + let health = lock_peer_health_for_test(addr); + assert_eq!(health.consecutive_timeouts, 1, "same-tick completion must follow timeout eviction policy"); + assert!( + health.last_eviction.is_none(), + "a recently serving peer keeps its channel after a caller timeout" + ); + timeout(Duration::from_secs(1), released_rx) + .await + .expect("same-tick ambiguous completion must trigger cleanup") + .expect("cleanup signal"); + assert_eq!( + lock_peer_health_for_test(addr).detached_rpcs, + 0, + "same-tick completion returns detached admission" + ); + reset_lock_peer_health_for_test(addr); + } + + #[tokio::test] + async fn test_remote_client_timeout_drains_late_grant_during_eviction() { + let addr = "http://timeout-late-grant-during-eviction"; + reset_lock_peer_health_for_test(addr); + let client = RemoteClient::new(addr.to_string()); + let (evicting_tx, evicting_rx) = tokio::sync::oneshot::channel(); + let (resume_tx, resume_rx) = tokio::sync::oneshot::channel(); + eviction_pauses_for_test() + .lock() + .expect("eviction pause registry") + .insert(addr.to_string(), (evicting_tx, resume_rx)); + let (grant_tx, grant_rx) = tokio::sync::oneshot::channel(); + let (released_tx, released_rx) = tokio::sync::oneshot::channel(); + let caller = tokio::spawn(async move { + let late: LateCompletion = Some(Box::new(move |granted| { + Box::pin(async move { + released_tx.send(granted).expect("late grant observer remains alive"); + }) + })); + client + .execute_rpc( + "lock", + "bucket/object", + Duration::from_millis(25), + async move { Ok(grant_rx.await.expect("test controls the late grant")) }, + RpcCleanup::new(late, None, None), + ) + .await + }); + timeout(Duration::from_secs(1), evicting_rx) + .await + .expect("timed-out RPC must reach channel eviction") + .expect("eviction signal"); + assert_eq!( + lock_peer_health_for_test(addr).detached_rpcs, + 1, + "eviction waits with a detached RPC owner" + ); + grant_tx + .send(true) + .expect("timed-out stream remains available for a late grant"); + assert!( + timeout(Duration::from_secs(1), released_rx) + .await + .expect("late grant must be cleaned up before eviction resumes") + .expect("late cleanup result") + ); + assert!(!caller.is_finished(), "channel eviction is still paused"); + assert_eq!( + lock_peer_health_for_test(addr).detached_rpcs, + 0, + "late completion returns detached admission" + ); + resume_tx.send(()).expect("resume channel eviction"); + let result = caller.await.expect("timed-out caller completes without panicking"); + assert!( + matches!(result, Err(LockError::Timeout { .. })), + "caller preserves its RPC timeout result" + ); + reset_lock_peer_health_for_test(addr); + } + + #[tokio::test] + async fn test_remote_client_success_polls_rpc_on_callers_task() { + tokio::spawn(async { + let client = RemoteClient::new("http://direct-poll-success".to_string()); + let caller_id = tokio::task::id(); + let rpc_id = client + .execute_rpc( + "ping", + "health-lock-online", + Duration::from_secs(1), + async { Ok(tokio::task::id()) }, + RpcCleanup::new(None, None, None), + ) + .await + .expect("successful RPC must return its result"); + assert_eq!(rpc_id, caller_id, "successful RPC must not allocate and schedule a separate task"); + }) + .await + .expect("caller task must complete"); + } + + #[tokio::test] + async fn test_remote_client_rpc_panic_isolated_and_cleans_up() { + let (released_tx, released_rx) = tokio::sync::oneshot::channel(); + let caller = tokio::spawn(async move { + let client = RemoteClient::new("http://direct-poll-panic".to_string()); + let late_failure: LateFailure = Some(Box::new(move |()| { + Box::pin(async move { + released_tx.send(()).expect("panic cleanup observer remains alive"); + }) + })); + client + .execute_rpc::<(), _>( + "lock", + "bucket/object", + Duration::from_secs(1), + async { panic!("test peer RPC panic") }, + RpcCleanup::new(None, late_failure, None), + ) + .await + }); + let result = caller.await.expect("RPC panic must not escape to its caller"); + assert!( + matches!(result, Err(LockError::Internal { .. })), + "RPC panic remains an internal lock failure" + ); + timeout(Duration::from_secs(1), released_rx) + .await + .expect("panic must trigger ambiguous-grant cleanup") + .expect("cleanup signal"); + } + #[tokio::test] #[serial_test::serial] async fn test_remote_client_timeout_keeps_channel_of_recently_serving_peer() { From 57df9181d0c41f0e913ac17a1bafdc0517509939 Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Thu, 5 Nov 2026 06:03:33 +0000 Subject: [PATCH 2/2] fix(audit): preserve OIDC identity in S3 request audits (#9178) --- rustfs/src/storage/helper.rs | 207 +++++++++++++++++++++++++++++++++++ 1 file changed, 207 insertions(+) diff --git a/rustfs/src/storage/helper.rs b/rustfs/src/storage/helper.rs index 1a8b62601..1ddd4f5b5 100644 --- a/rustfs/src/storage/helper.rs +++ b/rustfs/src/storage/helper.rs @@ -26,6 +26,7 @@ use rustfs_audit::{ entity::{ApiDetailsBuilder, AuditEntryBuilder}, global::AuditLogger, }; +use rustfs_iam::sys::{OIDC_ISSUER_CLAIM, is_rustfs_oidc_claims}; use rustfs_io_metrics::record_s3_op; use rustfs_notify::EventArgsBuilder; use rustfs_policy::policy::action::Action; @@ -133,6 +134,21 @@ fn request_audit_builder( .filter(|cred| !cred.access_key.is_empty()) { builder = builder.access_key(&cred.access_key); + if !cred.parent_user.is_empty() { + builder = builder.parent_user(&cred.parent_user); + } + if let Some(claims) = cred.claims.as_ref().filter(|claims| is_rustfs_oidc_claims(claims)) { + let identity_claims = ["sub", "iss", OIDC_ISSUER_CLAIM, "oidc_provider"] + .into_iter() + .filter_map(|key| { + claims + .get(key) + .filter(|value| value.as_str().is_some_and(|value| !value.is_empty())) + .map(|value| (key.to_string(), value.clone())) + }) + .collect(); + builder = builder.req_claims(identity_claims); + } } builder } @@ -1037,6 +1053,17 @@ mod tests { secret_key: "audit-secret-must-not-leak".to_string(), session_token: "audit-token-must-not-leak".to_string(), parent_user: parent_user.to_string(), + claims: match key { + "service-account" => Some(std::collections::HashMap::from([( + rustfs_credentials::IAM_POLICY_CLAIM_NAME_SA.to_string(), + serde_json::json!(rustfs_credentials::INHERITED_POLICY_TYPE), + )])), + "sts-session" => Some(std::collections::HashMap::from([( + "sessionPolicy".to_string(), + serde_json::json!("private-session-policy"), + )])), + _ => None, + }, ..Default::default() }), ..Default::default() @@ -1057,15 +1084,195 @@ mod tests { assert_eq!(entry.access_key.as_deref(), expected_access_key); let json = serde_json::to_value(&entry).expect("audit entry should serialize"); assert_eq!(json.get("accessKey").and_then(|key| key.as_str()), expected_access_key); + let expected_parent_user = (!parent_user.is_empty()).then_some(parent_user); + assert_eq!(entry.parent_user.as_deref(), expected_parent_user); + assert_eq!(json.get("parentUser").and_then(|value| value.as_str()), expected_parent_user); + assert!(!json.as_object().unwrap().contains_key("requestClaims")); let rendered = serde_json::to_string(&entry).expect("audit entry should serialize"); assert!(!rendered.contains("audit-secret-must-not-leak")); assert!(!rendered.contains("audit-token-must-not-leak")); + assert!(!rendered.contains("private-session-policy")); } } }, ); } + fn oidc_audit_credentials(issuer: Option<&str>) -> Credentials { + let mut claims = std::collections::HashMap::from([ + ("sub".to_string(), serde_json::json!(" subject-9170 ")), + ("iss".to_string(), serde_json::json!("rustfs-oidc")), + ("oidc_provider".to_string(), serde_json::json!("default")), + ("email".to_string(), serde_json::json!("private@example.test")), + ("preferred_username".to_string(), serde_json::json!("private-user")), + ("groups".to_string(), serde_json::json!(["private-group"])), + ("roles".to_string(), serde_json::json!(["private-role"])), + ("policy".to_string(), serde_json::json!("private-policy")), + ("sessionPolicy".to_string(), serde_json::json!("private-encoded-policy")), + ("sessionPolicy-extracted".to_string(), serde_json::json!("private-decoded-policy")), + ("custom".to_string(), serde_json::json!({"token": "private-custom-token"})), + ]); + if let Some(issuer) = issuer { + claims.insert(rustfs_iam::sys::OIDC_ISSUER_CLAIM.to_string(), serde_json::json!(issuer)); + } + Credentials { + access_key: "oidc-session".to_string(), + parent_user: "oidc-parent".to_string(), + secret_key: "audit-secret-must-not-leak".to_string(), + session_token: "audit-token-must-not-leak".to_string(), + claims: Some(claims), + ..Default::default() + } + } + + #[test] + fn operation_helper_preserves_oidc_audit_identity_without_sensitive_claims() { + with_vars( + [ + (rustfs_config::ENV_NOTIFY_ENABLE, Some("false")), + (rustfs_config::ENV_AUDIT_ENABLE, Some("true")), + ], + || { + refresh_notify_module_enabled(); + refresh_audit_module_enabled(); + + for issuer in [Some("https://idp-a.example.test"), Some("https://idp-b.example.test"), None] { + for service_account in [false, true] { + let mut credentials = oidc_audit_credentials(issuer); + if service_account { + credentials.access_key = "oidc-service-account".to_string(); + credentials.claims.as_mut().unwrap().insert( + rustfs_credentials::IAM_POLICY_CLAIM_NAME_SA.to_string(), + serde_json::json!(rustfs_credentials::INHERITED_POLICY_TYPE), + ); + } + let before = serde_json::to_value(&credentials).expect("credentials should serialize"); + let mut expected_claims = serde_json::json!({ + "sub": " subject-9170 ", + "iss": "rustfs-oidc", + "oidc_provider": "default" + }); + if let Some(issuer) = issuer { + expected_claims["x-rustfs-internal-oidc-issuer"] = serde_json::json!(issuer); + } + + for outcome in [None, Some(true), Some(false)] { + let mut req = build_request((), Method::GET, Uri::from_static("/audit-bucket/object.txt")); + req.extensions.insert(ReqInfo { + cred: Some(credentials.clone()), + ..Default::default() + }); + let mut helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject); + if let Some(success) = outcome { + let result = if success { + Ok(S3Response::new(GetObjectOutput::default())) + } else { + Err(S3Error::new(S3ErrorCode::NoSuchKey)) + }; + helper = helper.complete(&result); + } + let OperationHelper::Enabled(state) = &mut helper else { + panic!("audit helper should be enabled"); + }; + let entry = state.audit_builder.take().expect("audit builder should exist").build(); + let json = serde_json::to_value(&entry).expect("audit entry should serialize"); + assert_eq!(json["accessKey"], credentials.access_key); + assert_eq!(json["parentUser"], "oidc-parent"); + assert_eq!(json["requestClaims"], expected_claims); + let rendered = serde_json::to_string(&entry).expect("audit entry should serialize"); + assert!(!rendered.contains("private-")); + assert!(!rendered.contains("private@example.test")); + assert!(!rendered.contains("audit-secret-must-not-leak")); + assert!(!rendered.contains("audit-token-must-not-leak")); + assert_eq!( + serde_json::to_value(req.extensions.get::().unwrap().cred.as_ref().unwrap()).unwrap(), + before + ); + } + } + } + }, + ); + } + + #[test] + fn authorization_denial_preserves_oidc_audit_identity() { + use rustfs_policy::policy::action::{Action, S3Action}; + + let credentials = oidc_audit_credentials(Some("https://idp.example.test")); + let before = serde_json::to_value(&credentials).unwrap(); + let mut req = build_request((), Method::GET, Uri::from_static("/audit-bucket/object.txt")); + req.extensions.insert(ReqInfo { + cred: Some(credentials), + ..Default::default() + }); + let entry = super::authorization_denial_entry( + &req, + &Action::S3Action(S3Action::GetObjectAction), + "iam_explicit_deny", + super::DenialScope::Request, + ); + let json = serde_json::to_value(entry).expect("denial should serialize"); + assert_eq!(json["parentUser"], "oidc-parent"); + assert_eq!( + json["requestClaims"], + serde_json::json!({ + "sub": " subject-9170 ", + "iss": "rustfs-oidc", + "oidc_provider": "default", + "x-rustfs-internal-oidc-issuer": "https://idp.example.test" + }) + ); + assert_eq!(json["api"]["status_code"], 403); + assert_eq!(json["tags"]["denialReason"], "iam_explicit_deny"); + let rendered = json.to_string(); + assert!(!rendered.contains("private")); + assert!(!rendered.contains("audit-secret-must-not-leak")); + assert!(!rendered.contains("audit-token-must-not-leak")); + assert_eq!( + serde_json::to_value(req.extensions.get::().unwrap().cred.as_ref().unwrap()).unwrap(), + before + ); + } + + #[test] + fn request_audit_identity_omits_non_oidc_and_empty_claims() { + for (key, value, expected_oidc) in [ + (rustfs_iam::sys::OIDC_ISSUER_CLAIM, serde_json::json!(""), true), + (rustfs_iam::sys::OIDC_ISSUER_CLAIM, serde_json::json!(["https://idp.example.test"]), true), + ("iss", serde_json::json!("other-session"), false), + ("sub", serde_json::json!(""), false), + ("oidc_provider", serde_json::Value::Null, false), + ] { + let mut credentials = oidc_audit_credentials(None); + credentials.claims.as_mut().unwrap().insert(key.to_string(), value); + let mut req = build_request((), Method::GET, Uri::from_static("/audit-bucket/object.txt")); + req.extensions.insert(ReqInfo { + cred: Some(credentials), + ..Default::default() + }); + let entry = super::request_audit_builder( + &req, + req.extensions.get::(), + EventName::ObjectAccessedGet, + "GetObject", + rustfs_audit::ApiDetails::default(), + "audit-identity-boundary", + ) + .build(); + let json = serde_json::to_value(entry).expect("audit entry should serialize"); + assert_eq!(json["parentUser"], "oidc-parent"); + if expected_oidc { + assert_eq!( + json["requestClaims"], + serde_json::json!({"sub": " subject-9170 ", "iss": "rustfs-oidc", "oidc_provider": "default"}) + ); + } else { + assert!(!json.as_object().unwrap().contains_key("requestClaims")); + } + } + } + #[test] fn operation_helper_audits_verified_client_addresses() { use crate::shared_types::RemoteAddr;