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() { 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;