gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-11-08 19:43:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: merge main into rejoin readiness branch

Browse Source 
Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser Zhi
2026-11-05 15:45:24 +08:00
parent 6161280c6b 57df9181d0
commit fcc577f784
3 changed files with 920 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/cache_value/metacache_set.rs
+186 -24
View File
@@ -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<DiskError>]> = (0..opts.disks.len()).map(|_| OnceLock::new()).collect::<Vec<_>>().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<AtomicBool>);
#[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<S> Layer<S> 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");
crates/ecstore/src/cluster/rpc/remote_locker.rs
+527 -56
View File
@@ -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<T: rustfs_protos::CanonicalMutationBody>(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<T> = Option<Box<dyn FnOnce(T) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send>>;
type LateFailure = LateCompletion<()>;
type RpcFuture<T> = BoxFuture<'static, std::thread::Result<std::result::Result<T, tonic::Status>>>;
struct RpcCleanup<T> {
late: LateCompletion<T>,
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<LockRequestAdmission>,
}
impl<T> RpcCleanup<T> {
fn new(late: LateCompletion<T>, late_failure: LateFailure, admission: Option<LockRequestAdmission>) -> Self {
impl LockRequestGuard<'_> {
fn into_admission(mut self) -> Option<LockRequestAdmission> {
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<T>,
late_failure: LateFailure,
admission: Option<LockRequestGuard<'a>>,
}
impl<'a, T> RpcCleanup<'a, T> {
fn new(late: LateCompletion<T>, late_failure: LateFailure, admission: Option<LockRequestGuard<'a>>) -> Self {
Self {
late,
late_failure,
@@ -69,6 +91,25 @@ impl<T> RpcCleanup<T> {
}
}
/// 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<RpcFuture<T>>,
cleanup: Option<RpcCleanup<'a, T>>,
}
impl<T: Send + 'static> 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<R>(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<HashMap<String, EvictionPause>> {
static PAUSES: OnceLock<Mutex<HashMap<String, EvictionPause>>> = 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<LockRequestGuard<'_>> {
self.admit_lock_request(op).map(|admission| LockRequestGuard {
client: self,
op,
admission: Some(admission),
})
}
fn release_lock_request(&self, op: &'static str, admission: Option<LockRequestAdmission>) {
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<T: Send + 'static>(
fn detach_rpc<T: Send + 'static>(
&self,
op: &'static str,
resource_summary: &str,
handle: JoinHandle<std::result::Result<T, tonic::Status>>,
late: LateCompletion<T>,
late_failure: LateFailure,
admission: Option<LockRequestAdmission>,
future: RpcFuture<T>,
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<T>,
cleanup: RpcCleanup<'_, T>,
) -> std::result::Result<T, LockError>
where
Fut: Future<Output = std::result::Result<T, tonic::Status>> + 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<LockResponse> {
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<bool> = 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<bool> = 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() {
rustfs/src/storage/helper.rs
+207
View File
@@ -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<T>(
.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::<ReqInfo>().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::<ReqInfo>().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::<ReqInfo>(),
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;