fix(oidc): recover providers after startup discovery failures (#8475)

* fix(oidc): recover providers after startup discovery failures

* fix(oidc): report recovery failures with throttled diagnostics
This commit is contained in:
GatewayJ
2026-10-12 03:40:07 -07:00
committed by GitHub
parent dd90bb7488
commit 5228463a1c
10 changed files with 874 additions and 75 deletions
+110 -29
View File
@@ -13,7 +13,7 @@
// limitations under the License.
use super::{
config::{OidcProviderConfig, OidcProviderValidationResult, SourcedOidcProviderConfig},
transport::{OidcHttpError, ReqwestHttpClient, oidc_extra_root_certs},
transport::{DiscoveryRequest, OidcHttpError, ReqwestHttpClient, oidc_extra_root_certs, oidc_http_error_diagnostics},
};
use crate::{EVENT_OIDC_DIAGNOSTICS, LOG_COMPONENT_IAM, LOG_SUBSYSTEM_OIDC};
use openidconnect::core::{CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm};
@@ -25,7 +25,7 @@ use std::collections::HashMap;
use std::sync::RwLock;
use std::time::{Duration as StdDuration, Instant};
use tokio::time::sleep;
use tracing::warn;
use tracing::{debug, warn};
use url::Url;
pub(super) const OIDC_JWKS_REFRESH_INTERVAL: StdDuration = StdDuration::from_secs(24 * 60 * 60);
@@ -34,6 +34,55 @@ pub(super) const OIDC_DISCOVERY_TRANSPORT_RETRY_DELAY: StdDuration = StdDuration
pub(super) const OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY: &str = "JWKS request blocked by outbound policy";
pub(super) const OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY: &str = "OIDC provider discovery blocked by outbound policy";
/// Preserve recovery eligibility before discovery errors become public diagnostic strings.
#[derive(Debug, thiserror::Error)]
#[error("{message}")]
pub(super) struct DiscoveryFailure {
message: String,
pub(super) retryable: bool,
pub(super) diagnostics: (&'static str, String),
}
impl From<String> for DiscoveryFailure {
fn from(message: String) -> Self {
Self {
message,
retryable: false,
diagnostics: ("validation", String::new()),
}
}
}
fn retryable_discovery_status(status: http::StatusCode) -> bool {
status.is_server_error() || matches!(status, http::StatusCode::REQUEST_TIMEOUT | http::StatusCode::TOO_MANY_REQUESTS)
}
fn retryable_discovery_transport(error: &OidcHttpError) -> bool {
matches!(error, OidcHttpError::Reqwest(error) if !error.is_builder())
}
fn discovery_failure(error: DiscoveryError<OidcHttpError>, context: &str, blocked_context: &str) -> DiscoveryFailure {
let retryable = match &error {
DiscoveryError::Request(error) => retryable_discovery_transport(error),
DiscoveryError::Response(status, _, _) => retryable_discovery_status(*status),
_ => false,
};
let diagnostics = match &error {
DiscoveryError::Request(error) => oidc_http_error_diagnostics(error),
DiscoveryError::Response(status, _, _) => ("http_status", status.as_u16().to_string()),
_ => ("validation", String::new()),
};
let message = match error {
DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason)) => format!("{blocked_context}: {reason}"),
error => format!("{context}: {error}"),
};
DiscoveryFailure {
message,
retryable,
diagnostics,
}
}
// ---- Internal provider state ----
/// Discovered OIDC provider metadata.
@@ -180,6 +229,7 @@ impl<'c> AsyncHttpClient<'c> for JwksAcceptClient<'_> {
http::HeaderValue::from_static("application/json, application/jwk-set+json"),
);
}
request.extensions_mut().insert(DiscoveryRequest);
self.inner.call(request)
}
}
@@ -246,7 +296,17 @@ impl ProviderRuntime {
provider_id: &str,
config: &OidcProviderConfig,
) -> Result<ProviderState, String> {
let state = discover_provider(config, &self.http_client).await?;
self.recover_provider_state(provider_id, config)
.await
.map_err(|err| err.to_string())
}
pub(super) async fn recover_provider_state(
&self,
provider_id: &str,
config: &OidcProviderConfig,
) -> Result<ProviderState, DiscoveryFailure> {
let state = discover_provider_for_recovery(config, &self.http_client).await?;
let mut map = self.states.write().map_err(|e| {
let msg = e.to_string();
format!("provider state lock poisoned: {msg}")
@@ -301,6 +361,15 @@ pub(super) async fn discover_provider(
config: &OidcProviderConfig,
http_client: &ReqwestHttpClient,
) -> Result<ProviderState, String> {
discover_provider_for_recovery(config, http_client)
.await
.map_err(|err| err.to_string())
}
pub(super) async fn discover_provider_for_recovery(
config: &OidcProviderConfig,
http_client: &ReqwestHttpClient,
) -> Result<ProviderState, DiscoveryFailure> {
if let Some(issuer) = config.issuer.as_deref().filter(|issuer| !issuer.trim().is_empty()) {
return discover_provider_from_config_url(config, issuer, http_client).await;
}
@@ -310,6 +379,8 @@ pub(super) async fn discover_provider(
let base_issuer = normalize_config_url(&config.config_url)?;
let candidates = issuer_candidates(&base_issuer);
let mut last_errors = Vec::new();
let mut retryable = false;
let mut diagnostics = ("validation", String::new());
for candidate_issuer in candidates.iter() {
let issuer_url = IssuerUrl::new(candidate_issuer.clone()).map_err(|e| format!("invalid issuer URL: {e}"))?;
@@ -332,26 +403,21 @@ pub(super) async fn discover_provider(
metadata: DiscoveredProviderMetadata::Console(Box::new(metadata)),
discovered_at: Instant::now(),
})
.map_err(|err| match err {
DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason)) => {
format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}")
}
err => format!("discovery failed: {err}"),
})
.map_err(|err| discovery_failure(err, "discovery failed", OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY))
};
match discovered {
Ok(state) => return Ok(state),
Err(error)
if error.starts_with(OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY)
|| error.starts_with(OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY) =>
if error.message.starts_with(OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY)
|| error.message.starts_with(OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY) =>
{
return Err(error);
}
Err(error) => {
let is_transient_transport = error.contains("Request failed");
let is_transient_transport = error.message.contains("Request failed");
let should_retry = is_transient_transport && attempt + 1 < OIDC_DISCOVERY_TRANSPORT_RETRIES;
if should_retry {
warn!(
debug!(
event = EVENT_OIDC_DIAGNOSTICS,
component = LOG_COMPONENT_IAM,
subsystem = LOG_SUBSYSTEM_OIDC,
@@ -368,8 +434,12 @@ pub(super) async fn discover_provider(
continue;
}
if error.retryable || !retryable {
diagnostics = error.diagnostics.clone();
}
retryable |= error.retryable;
last_errors.push(format!("issuer '{candidate_issuer}': {error}"));
warn!(
debug!(
event = EVENT_OIDC_DIAGNOSTICS,
component = LOG_COMPONENT_IAM,
subsystem = LOG_SUBSYSTEM_OIDC,
@@ -388,11 +458,11 @@ pub(super) async fn discover_provider(
}
}
Err(format!(
"discovery failed for all issuer variants {:?}: {}",
candidates,
last_errors.join("; ")
))
Err(DiscoveryFailure {
message: format!("discovery failed for all issuer variants {:?}: {}", candidates, last_errors.join("; ")),
retryable,
diagnostics,
})
}
pub(super) async fn validate_oidc_provider_config(config: &OidcProviderConfig) -> Result<OidcProviderValidationResult, String> {
@@ -429,7 +499,7 @@ async fn discover_provider_from_config_url(
config: &OidcProviderConfig,
issuer: &str,
http_client: &ReqwestHttpClient,
) -> Result<ProviderState, String> {
) -> Result<ProviderState, DiscoveryFailure> {
let issuer_url = IssuerUrl::new(issuer.trim().to_string()).map_err(|e| format!("invalid issuer URL: {e}"))?;
let explicit_issuer = config.issuer.as_deref().is_some_and(|issuer| !issuer.trim().is_empty());
let discovery_url = if explicit_issuer {
@@ -440,6 +510,7 @@ async fn discover_provider_from_config_url(
.map_err(|err| err.to_string())?
};
let request = http::Request::builder()
.extension(DiscoveryRequest)
.uri(discovery_url.to_string())
.method(http::Method::GET)
.header(http::header::ACCEPT, "application/json")
@@ -449,12 +520,22 @@ async fn discover_provider_from_config_url(
let response = match http_client.call(request).await {
Ok(response) => response,
Err(OidcHttpError::ForbiddenOutbound(reason)) => {
return Err(format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}"));
return Err(format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}").into());
}
Err(err) => {
return Err(DiscoveryFailure {
retryable: retryable_discovery_transport(&err),
diagnostics: oidc_http_error_diagnostics(&err),
message: format!("discovery request failed: Request failed: {err}"),
});
}
Err(err) => return Err(format!("discovery request failed: Request failed: {err}")),
};
if response.status() != http::StatusCode::OK {
return Err(format!("discovery failed: HTTP status code {} at {}", response.status(), discovery_url));
return Err(DiscoveryFailure {
retryable: retryable_discovery_status(response.status()),
diagnostics: ("http_status", response.status().as_u16().to_string()),
message: format!("discovery failed: HTTP status code {} at {}", response.status(), discovery_url),
});
}
if !explicit_issuer
@@ -466,7 +547,9 @@ async fn discover_provider_from_config_url(
.is_some_and(|essence| essence.eq_ignore_ascii_case("application/json"))
})
{
return Err("Unexpected response Content-Type: expected application/json".into());
return Err("Unexpected response Content-Type: expected application/json"
.to_string()
.into());
}
let provider_metadata = DiscoveredProviderMetadata::parse(response.body(), config.hide_from_ui)
@@ -476,7 +559,8 @@ async fn discover_provider_from_config_url(
"unexpected issuer URI `{}` (expected `{}`)",
provider_metadata.issuer().as_str(),
issuer_url.as_str()
));
)
.into());
}
let jwks_url = if explicit_issuer {
@@ -494,10 +578,7 @@ async fn discover_provider_from_config_url(
.await
{
Ok(jwks) => jwks,
Err(DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason))) => {
return Err(format!("{OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY}: {reason}"));
}
Err(err) => return Err(format!("failed to fetch JWKS: {err}")),
Err(err) => return Err(discovery_failure(err, "failed to fetch JWKS", OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY)),
};
Ok(ProviderState {
+1
View File
@@ -299,6 +299,7 @@ async fn workload_discovery_verification_and_rotation() {
configs: HashMap::from([(config.id.clone(), test_sourced_config(config.clone()))]),
provider_runtime: ProviderRuntime::new(http_client, HashMap::from([(config.id.clone(), state)])),
state_store: OidcStateStore::new(),
pending_discovery: Default::default(),
};
assert!(sys.has_providers());
assert!(sys.provider_configs().all(|provider| provider.hide_from_ui));
+489
View File
@@ -0,0 +1,489 @@
// Copyright 2026 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use super::*;
use crate::oidc::test_support::{build_mocked_oidc_provider_config, oidc_es256_key_and_jwk, test_sourced_config};
use jsonwebtoken::{Algorithm, EncodingKey, Header};
use rustfs_utils::egress::OutboundPolicy;
use std::sync::atomic::{AtomicU16, AtomicUsize, Ordering};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio_util::task::AbortOnDropHandle;
struct RecoveryIdp {
base: String,
config: OidcProviderConfig,
key: EncodingKey,
discovery_status: Arc<AtomicU16>,
jwks_status: Arc<AtomicU16>,
requests: Arc<AtomicUsize>,
cancelled_requests: Arc<AtomicUsize>,
_server: AbortOnDropHandle<()>,
}
impl RecoveryIdp {
async fn start(id: &str, workload: bool, explicit_issuer: bool) -> Self {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind recovery IdP");
let base = format!("http://{}", listener.local_addr().expect("recovery IdP address"));
let mut config = build_mocked_oidc_provider_config(id, &base);
config.hide_from_ui = workload;
config.claim_name = "groups".into();
config.claim_prefix = "team/".into();
config.issuer = explicit_issuer.then(|| format!("https://issuer.example.com/{id}"));
let (key, jwk) = oidc_es256_key_and_jwk(id);
let mut document = serde_json::json!({
"issuer": config.issuer.as_deref().unwrap_or(&base),
"authorization_endpoint": format!("{base}/authorize"),
"token_endpoint": format!("{base}/token"),
"jwks_uri": format!("{base}/jwks"),
"response_types_supported": ["code"],
"subject_types_supported": ["public"],
"id_token_signing_alg_values_supported": ["ES256"],
});
if workload {
let fields = document.as_object_mut().expect("discovery document");
fields.remove("authorization_endpoint");
fields.remove("token_endpoint");
}
let document = document.to_string();
let jwks = serde_json::json!({"keys": [jwk]}).to_string();
let discovery_status = Arc::new(AtomicU16::new(503));
let jwks_status = Arc::new(AtomicU16::new(200));
let requests = Arc::new(AtomicUsize::new(0));
let cancelled_requests = Arc::new(AtomicUsize::new(0));
let server_discovery_status = discovery_status.clone();
let server_jwks_status = jwks_status.clone();
let server_requests = requests.clone();
let server_cancelled = cancelled_requests.clone();
let server = tokio::spawn(async move {
loop {
let (mut stream, _) = listener.accept().await.expect("accept discovery request");
let mut request = Vec::new();
while !request.ends_with(b"\r\n\r\n") {
request.push(stream.read_u8().await.expect("read discovery request"));
assert!(request.len() < 8192);
}
let headers = String::from_utf8(request).expect("HTTP request headers");
let path = headers
.lines()
.next()
.expect("request line")
.split_whitespace()
.nth(1)
.expect("path");
let (status, body) = if path.ends_with("/.well-known/openid-configuration") {
(server_discovery_status.load(Ordering::SeqCst), document.as_str())
} else {
assert_eq!(path, "/jwks");
(server_jwks_status.load(Ordering::SeqCst), jwks.as_str())
};
server_requests.fetch_add(1, Ordering::SeqCst);
if status == 0 {
continue; // Simulate a transport failure before a response is available.
}
if status == 1 {
let mut byte = [0];
assert_eq!(stream.read(&mut byte).await.expect("client closes cancelled discovery"), 0);
server_cancelled.fetch_add(1, Ordering::SeqCst);
continue;
}
let body = if status == 200 { body } else { "" };
let response = format!(
"HTTP/1.1 {status} Test\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
stream.write_all(response.as_bytes()).await.expect("write discovery response");
}
});
Self {
base,
config,
key,
discovery_status,
jwks_status,
requests,
cancelled_requests,
_server: AbortOnDropHandle::new(server),
}
}
fn token(&self, key: &EncodingKey, audience: &str) -> String {
let mut header = Header::new(Algorithm::ES256);
header.kid = Some(self.config.id.clone());
let now = time::OffsetDateTime::now_utc().unix_timestamp();
jsonwebtoken::encode(
&header,
&serde_json::json!({
"iss": self.config.issuer.as_deref().unwrap_or(&self.base),
"sub": "recovered-user", "aud": audience, "iat": now, "exp": now + 300,
"groups": ["readonly"],
}),
key,
)
.expect("sign recovery test token")
}
}
async fn initialize(idps: &[&RecoveryIdp]) -> Arc<OidcSys> {
let origins = idps.iter().map(|idp| idp.base.as_str()).collect::<Vec<_>>().join(",");
let client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&origins).expect("allow test IdPs"));
let snapshot = OidcConfigSnapshot::new(idps.iter().map(|idp| test_sourced_config(idp.config.clone())).collect());
Arc::new(
OidcSys::new_with_config_snapshot(client, snapshot)
.await
.expect("initialize OIDC with unavailable IdPs"),
)
}
async fn wait_until(mut condition: impl FnMut() -> bool) {
tokio::time::timeout(Duration::from_secs(5), async {
while !condition() {
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("recovery condition should be reached");
}
#[tokio::test]
async fn web_identity_recovers_transport_failure_and_preserves_verification() {
for explicit in [false, true] {
let idp = RecoveryIdp::start("workload", true, explicit).await;
idp.discovery_status.store(0, Ordering::SeqCst);
let sys = initialize(&[&idp]).await;
let adapter = StandardOidcAdapter::new(sys.clone());
let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
assert!(sys.has_providers());
assert!(sys.list_visible_providers().is_empty());
let token = idp.token(&idp.key, &idp.config.client_id);
assert!(adapter.verify_identity(&token).await.is_err());
idp.discovery_status.store(200, Ordering::SeqCst);
tokio::time::timeout(Duration::from_secs(5), sys.recover_failed_providers())
.await
.expect("recover workload discovery");
let identity = adapter
.verify_identity(&token)
.await
.expect("verify recovered workload token");
let authorization = mapper.map(identity);
assert_eq!(authorization.provider_id, "workload");
assert_eq!(authorization.claims.sub, "recovered-user");
assert_eq!(authorization.policies, vec!["team/readonly"]);
assert!(
sys.authorize_url("workload", "https://console.example.com/callback", None)
.await
.is_err()
);
let (wrong_key, _) = oidc_es256_key_and_jwk("wrong");
assert!(
adapter
.verify_identity(&idp.token(&wrong_key, &idp.config.client_id))
.await
.is_err()
);
assert!(adapter.verify_identity(&idp.token(&idp.key, "wrong-audience")).await.is_err());
}
}
#[tokio::test]
async fn recovery_waits_for_jwks_and_claims_pending_providers_once() {
let idp = RecoveryIdp::start("console", false, true).await;
idp.discovery_status.store(200, Ordering::SeqCst);
idp.jwks_status.store(503, Ordering::SeqCst);
let sys = initialize(&[&idp]).await;
assert!(sys.provider_runtime.get_provider_state("console").is_err());
let requests = idp.requests.load(Ordering::SeqCst);
idp.jwks_status.store(200, Ordering::SeqCst);
tokio::time::timeout(
Duration::from_secs(5),
futures::future::join_all((0..16).map(|_| sys.recover_failed_providers())),
)
.await
.expect("concurrent recovery calls complete");
assert!(sys.provider_runtime.get_provider_state("console").is_ok());
assert_eq!(
idp.requests.load(Ordering::SeqCst),
requests + 2,
"only one discovery/JWKS round should run"
);
}
#[tokio::test]
async fn recovery_backoff_is_independent_of_login_traffic() {
let idp = RecoveryIdp::start("console", false, true).await;
let sys = initialize(&[&idp]).await;
let requests = idp.requests.load(Ordering::SeqCst);
let recovery_sys = sys.clone();
let recovery = AbortOnDropHandle::new(tokio::spawn(async move { recovery_sys.recover_failed_providers().await }));
wait_until(|| idp.requests.load(Ordering::SeqCst) > requests).await;
assert_eq!(idp.requests.load(Ordering::SeqCst), requests + 1);
for _ in 0..32 {
assert!(
sys.authorize_url("console", "https://console.example.com/callback", None)
.await
.is_err()
);
}
tokio::time::sleep(Duration::from_millis(300)).await;
assert_eq!(
idp.requests.load(Ordering::SeqCst),
requests + 1,
"login attempts must not bypass discovery backoff"
);
idp.discovery_status.store(200, Ordering::SeqCst);
tokio::time::timeout(Duration::from_secs(5), recovery)
.await
.expect("recovery deadline")
.expect("recovery task");
assert!(
sys.authorize_url("console", "https://console.example.com/callback", None)
.await
.is_ok()
);
}
#[tokio::test]
async fn unavailable_provider_does_not_block_another_provider() {
let unavailable = RecoveryIdp::start("unavailable", false, true).await;
let available = RecoveryIdp::start("available", false, true).await;
let sys = initialize(&[&unavailable, &available]).await;
available.discovery_status.store(200, Ordering::SeqCst);
let recovery_sys = sys.clone();
let recovery = AbortOnDropHandle::new(tokio::spawn(async move { recovery_sys.recover_failed_providers().await }));
wait_until(|| sys.provider_runtime.get_provider_state("available").is_ok()).await;
assert!(sys.provider_runtime.get_provider_state("unavailable").is_err());
assert!(
sys.authorize_url("available", "https://console.example.com/callback", None)
.await
.is_ok()
);
drop(recovery);
}
#[tokio::test]
async fn recovery_preserves_ready_provider_and_inflight_sessions() {
let ready = RecoveryIdp::start("ready", false, true).await;
let recovering = RecoveryIdp::start("recovering", false, true).await;
ready.discovery_status.store(200, Ordering::SeqCst);
let sys = initialize(&[&ready, &recovering]).await;
let auth_url = sys
.authorize_url("ready", "https://console.example.com/callback", Some("/buckets".into()))
.await
.expect("begin login with ready provider");
let auth: HashMap<String, String> = url::Url::parse(&auth_url)
.expect("authorization URL")
.query_pairs()
.into_owned()
.collect();
let logout = sys
.create_logout_token("ready", "verified-id-token")
.await
.expect("retain logout continuation");
let discovered_at = sys
.provider_runtime
.get_provider_state("ready")
.expect("ready state")
.discovered_at;
let ready_requests = ready.requests.load(Ordering::SeqCst);
recovering.discovery_status.store(200, Ordering::SeqCst);
tokio::time::timeout(Duration::from_secs(5), sys.recover_failed_providers())
.await
.expect("recover second provider");
assert_eq!(ready.requests.load(Ordering::SeqCst), ready_requests);
assert_eq!(
sys.provider_runtime
.get_provider_state("ready")
.expect("ready state remains")
.discovered_at,
discovered_at
);
let session = sys
.state_store
.take(&auth["state"])
.await
.expect("in-flight login remains available");
assert_eq!(session.provider_id, "ready");
assert_eq!(session.nonce, auth["nonce"]);
assert_eq!(session.redirect_after.as_deref(), Some("/buckets"));
assert_eq!(
PkceCodeChallenge::from_code_verifier_sha256(&PkceCodeVerifier::new(session.pkce_verifier)).as_str(),
auth["code_challenge"]
);
assert_eq!(
sys.state_store
.take_logout(&logout)
.await
.expect("logout continuation remains available")
.provider_id,
"ready"
);
}
#[tokio::test]
async fn startup_schedules_only_recoverable_http_errors() {
for explicit in [false, true] {
let idp = RecoveryIdp::start("status", false, explicit).await;
for (status, retryable) in [
(408, true),
(429, true),
(500, true),
(503, true),
(400, false),
(401, false),
(403, false),
(404, false),
] {
idp.discovery_status.store(status, Ordering::SeqCst);
let sys = initialize(&[&idp]).await;
assert_eq!(
sys.pending_discovery.lock().await.len(),
usize::from(retryable),
"explicit={explicit}, status={status}"
);
assert!(sys.provider_runtime.get_provider_state("status").is_err());
}
}
}
#[tokio::test]
async fn recovery_bounds_concurrency_and_cancels_inflight_requests() {
let mut idps = Vec::new();
for index in 0..6 {
idps.push(RecoveryIdp::start(&format!("provider-{index}"), false, true).await);
}
let refs: Vec<_> = idps.iter().collect();
let sys = initialize(&refs).await;
let initial_requests: usize = idps.iter().map(|idp| idp.requests.load(Ordering::SeqCst)).sum();
for idp in &idps {
idp.discovery_status.store(1, Ordering::SeqCst);
}
let recovery_sys = sys.clone();
let recovery = AbortOnDropHandle::new(tokio::spawn(async move { recovery_sys.recover_failed_providers().await }));
let requests = || idps.iter().map(|idp| idp.requests.load(Ordering::SeqCst)).sum::<usize>();
wait_until(|| requests() >= initial_requests + 4).await;
tokio::time::sleep(Duration::from_millis(300)).await;
assert_eq!(requests(), initial_requests + 4, "at most four discovery rounds may run concurrently");
drop(recovery);
wait_until(|| {
idps.iter()
.map(|idp| idp.cancelled_requests.load(Ordering::SeqCst))
.sum::<usize>()
== 4
})
.await;
tokio::time::sleep(Duration::from_millis(100)).await;
assert_eq!(requests(), initial_requests + 4, "cancelled recovery must not start queued work");
assert!(
idps.iter()
.all(|idp| sys.provider_runtime.get_provider_state(&idp.config.id).is_err())
);
}
#[tokio::test]
async fn disabled_and_rejected_providers_do_not_schedule_recovery() {
let mut disabled = RecoveryIdp::start("disabled", false, true).await;
disabled.config.enabled = false;
let sys = initialize(&[&disabled]).await;
sys.recover_failed_providers().await;
assert!(!sys.has_providers());
assert_eq!(disabled.requests.load(Ordering::SeqCst), 0);
let blocked = RecoveryIdp::start("blocked", false, true).await;
let sys = OidcSys::new_with_config_snapshot(
ReqwestHttpClient::with_policy(OutboundPolicy::default()),
OidcConfigSnapshot::new(vec![test_sourced_config(blocked.config.clone())]),
)
.await
.expect("policy rejection remains non-fatal for startup");
assert!(sys.pending_discovery.lock().await.is_empty());
sys.recover_failed_providers().await;
assert_eq!(blocked.requests.load(Ordering::SeqCst), 0);
assert!(sys.provider_runtime.get_provider_state("blocked").is_err());
}
#[tokio::test]
async fn recovery_stops_on_issuer_validation_failure() {
let mut idp = RecoveryIdp::start("console", false, true).await;
idp.config.issuer = Some("https://unexpected.example.com".into());
let sys = initialize(&[&idp]).await;
idp.discovery_status.store(200, Ordering::SeqCst);
tokio::time::timeout(Duration::from_secs(5), sys.recover_failed_providers())
.await
.expect("validation failure stops recovery");
assert_eq!(
idp.requests.load(Ordering::SeqCst),
2,
"issuer rejection must not fetch JWKS or schedule another attempt"
);
assert!(sys.provider_runtime.get_provider_state("console").is_err());
assert!(
sys.authorize_url("console", "https://console.example.com/callback", None)
.await
.is_err()
);
}
#[tokio::test]
async fn recovery_logs_failure_reasons_without_repeating_warnings() {
use std::io::{Read, Seek};
for (explicit, status, expected_kind) in [
(false, 503, "http_status"),
(true, 503, "http_status"),
(true, 0, "request"),
(true, 1, "timeout"),
] {
let idp = RecoveryIdp::start("logging", false, explicit).await;
let sys = initialize(&[&idp]).await;
let requests = idp.requests.load(Ordering::SeqCst);
idp.discovery_status.store(status, Ordering::SeqCst);
let mut output = tempfile::tempfile().expect("create captured log file");
let subscriber = tracing_subscriber::fmt()
.without_time()
.with_ansi(false)
.with_max_level(tracing::Level::WARN)
.with_writer(output.try_clone().expect("clone captured log file"))
.finish();
let _subscriber = tracing::subscriber::set_default(subscriber);
tracing::callsite::rebuild_interest_cache();
tokio::time::timeout(Duration::from_secs(25), async {
tokio::join!(sys.recover_failed_providers(), async {
// Two failing rounds exercise suppression; a timeout is already bounded by the HTTP client.
while if status == 1 {
idp.cancelled_requests.load(Ordering::SeqCst) == 0
} else {
idp.requests.load(Ordering::SeqCst) < requests + if explicit { 2 } else { 4 }
} {
tokio::time::sleep(Duration::from_millis(10)).await;
}
idp.discovery_status.store(200, Ordering::SeqCst);
});
})
.await
.expect("recover after recording failure diagnostics");
output.rewind().expect("rewind captured log file");
let mut logs = String::new();
output.read_to_string(&mut logs).expect("read captured log file");
assert_eq!(logs.lines().count(), 1, "explicit={explicit}, status={status}: {logs}");
assert!(logs.contains("provider_discovery_recovery_pending"), "{logs}");
assert!(logs.contains(&format!("error_kind=\"{expected_kind}\"")), "{logs}");
if status == 503 {
assert!(logs.contains("error_detail=503"), "{logs}");
}
assert!(logs.contains("provider_id=logging"), "{logs}");
assert!(logs.contains("suppressed=0"), "{logs}");
}
}
+101 -4
View File
@@ -16,7 +16,7 @@ use super::{
OidcConfigQuery, OidcConfigSnapshot, OidcProviderConfig, OidcProviderSummary, OidcSiteReplicationSnapshot,
SourcedOidcProviderConfig, load_oidc_config_snapshot,
},
provider::{ProviderRuntime, discover_provider},
provider::{ProviderRuntime, discover_provider_for_recovery},
state::{OidcAuthSession, OidcLogoutSession, OidcStateStore},
transport::{OidcExtraRootCaProvider, ReqwestHttpClient, inspect_token_response_body, oidc_http_error_diagnostics},
};
@@ -37,11 +37,19 @@ use openidconnect::{
PkceCodeVerifier, PostLogoutRedirectUrl, RedirectUrl, RequestTokenError, Scope,
};
use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive};
use rustfs_utils::logging::LogThrottle;
use serde::{Deserialize, Serialize};
use std::borrow::Cow;
use std::collections::HashMap;
use std::sync::Arc;
use tracing::{debug, error, warn};
use std::time::Duration;
use tokio::sync::{Mutex, Semaphore};
use tracing::{debug, error, info, warn};
const OIDC_DISCOVERY_RECOVERY_CONCURRENCY: usize = 4;
const OIDC_DISCOVERY_RECOVERY_MAX_DELAY_MS: u64 = 30_000;
// A round can try two issuer variants three times, each fetching discovery and JWKS.
const OIDC_DISCOVERY_RECOVERY_TIMEOUT: Duration = Duration::from_secs(125);
/// Claims extracted from an OIDC ID token.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -60,6 +68,7 @@ pub struct OidcSys {
pub(super) configs: HashMap<String, SourcedOidcProviderConfig>,
pub(super) provider_runtime: ProviderRuntime,
pub(super) state_store: OidcStateStore,
pub(super) pending_discovery: Mutex<Vec<String>>,
}
pub(super) fn trusted_aud(other_audiences: &[String], audience: &Audience) -> bool {
@@ -81,8 +90,16 @@ impl OidcSys {
async fn new_with_http_client(http_client: ReqwestHttpClient) -> Result<Self, String> {
let server_config = crate::server_config::current_server_config();
let parsed_configs = load_oidc_config_snapshot(server_config.as_ref());
Self::new_with_config_snapshot(http_client, parsed_configs).await
}
async fn new_with_config_snapshot(
http_client: ReqwestHttpClient,
parsed_configs: OidcConfigSnapshot,
) -> Result<Self, String> {
let mut configs = HashMap::new();
let mut provider_states = HashMap::new();
let mut pending_discovery = Vec::new();
for sourced_config in parsed_configs.into_providers() {
let config = &sourced_config.config;
@@ -91,13 +108,15 @@ impl OidcSys {
continue;
}
match discover_provider(config, &http_client).await {
match discover_provider_for_recovery(config, &http_client).await {
Ok(state) => {
debug!(provider = %config.id, "OIDC provider discovered");
provider_states.insert(config.id.clone(), state);
configs.insert(config.id.clone(), sourced_config);
}
Err(e) => {
if e.retryable {
pending_discovery.push(config.id.clone());
}
error!(
event = EVENT_OIDC_DIAGNOSTICS,
component = LOG_COMPONENT_IAM,
@@ -114,21 +133,95 @@ impl OidcSys {
);
}
}
// Configuration also supplies redirect and authorization rules before recovery.
configs.insert(config.id.clone(), sourced_config);
}
Ok(Self {
configs,
provider_runtime: ProviderRuntime::new(http_client, provider_states),
state_store: OidcStateStore::new(),
pending_discovery: Mutex::new(pending_discovery),
})
}
/// Recover providers that failed startup discovery due to a transient error.
/// The server owns and cancels this future; dropping it cancels all outstanding requests.
/// Pending providers are claimed once so concurrent callers cannot duplicate recovery.
pub async fn recover_failed_providers(&self) {
let pending = std::mem::take(&mut *self.pending_discovery.lock().await);
let permits = Semaphore::new(OIDC_DISCOVERY_RECOVERY_CONCURRENCY);
futures::future::join_all(pending.iter().filter_map(|id| self.get_provider_config(id)).map(|config| {
let permits = &permits;
async move {
let mut delay_ms = 1_000_u64;
let failure_log = LogThrottle::new(60_000);
loop {
let sleep_ms = rand::random_range(delay_ms * 3 / 4..=delay_ms);
tokio::time::sleep(Duration::from_millis(sleep_ms)).await;
let Ok(permit) = permits.acquire().await else {
return;
};
let result = tokio::time::timeout(
OIDC_DISCOVERY_RECOVERY_TIMEOUT,
self.provider_runtime.recover_provider_state(&config.id, config),
)
.await;
drop(permit);
let (error_kind, error_detail) = match result {
Ok(Ok(_)) => {
info!(
event = EVENT_OIDC_DIAGNOSTICS,
component = LOG_COMPONENT_IAM,
subsystem = LOG_SUBSYSTEM_OIDC,
result = "provider_discovery_recovered",
provider_id = %config.id,
"oidc provider discovery recovered"
);
return;
}
Ok(Err(error)) if !error.retryable => {
error!(
event = EVENT_OIDC_DIAGNOSTICS,
component = LOG_COMPONENT_IAM,
subsystem = LOG_SUBSYSTEM_OIDC,
result = "provider_discovery_recovery_stopped",
provider_id = %config.id,
error = %error,
"oidc provider discovery requires configuration correction"
);
return;
}
Ok(Err(error)) => error.diagnostics,
Err(_) => ("attempt_timeout", OIDC_DISCOVERY_RECOVERY_TIMEOUT.as_secs().to_string()),
};
if let Some(suppressed) = failure_log.claim() {
warn!(
event = EVENT_OIDC_DIAGNOSTICS,
component = LOG_COMPONENT_IAM,
subsystem = LOG_SUBSYSTEM_OIDC,
result = "provider_discovery_recovery_pending",
provider_id = %config.id,
error_kind,
error_detail = %error_detail,
suppressed,
"oidc provider discovery remains unavailable"
);
}
delay_ms = delay_ms.saturating_mul(2).min(OIDC_DISCOVERY_RECOVERY_MAX_DELAY_MS);
}
}
}))
.await;
}
/// Create an OidcSys with no providers (useful for when OIDC is not configured).
pub fn empty() -> Result<Self, String> {
Ok(Self {
configs: HashMap::new(),
provider_runtime: ProviderRuntime::new(ReqwestHttpClient::new()?, HashMap::new()),
state_store: OidcStateStore::new(),
pending_discovery: Mutex::default(),
})
}
@@ -895,3 +988,7 @@ impl FederatedProviderQuery for StandardOidcAdapter {
#[cfg(test)]
#[path = "runtime_tests.rs"]
mod tests;
#[cfg(test)]
#[path = "recovery_tests.rs"]
mod recovery_tests;
+55 -13
View File
@@ -175,6 +175,7 @@ async fn web_identity_verification_refreshes_rotated_jwks() {
configs: HashMap::from([(config.id.clone(), test_sourced_config(config.clone()))]),
provider_runtime: ProviderRuntime::new(http_client, HashMap::from([(config.id.clone(), state)])),
state_store: OidcStateStore::new(),
pending_discovery: Default::default(),
};
let now = std::time::SystemTime::now()
@@ -208,6 +209,15 @@ async fn web_identity_verification_refreshes_rotated_jwks() {
}
#[tokio::test]
async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
assert_complete_console_login(false).await;
}
#[tokio::test]
async fn console_login_recovers_startup_discovery_with_authorization_rules() {
assert_complete_console_login(true).await;
}
async fn assert_complete_console_login(initially_unavailable: bool) {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
for hidden in [false, true] {
@@ -221,6 +231,8 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
};
let mut config = build_mocked_oidc_provider_config("console", &format!("{base}/.well-known/openid-configuration"));
config.hide_from_ui = hidden;
config.claim_name = "groups".into();
config.claim_prefix = "team/".into();
config.issuer = explicit.then(|| issuer.clone());
config.client_secret = Some(Nonce::new_random().secret().clone());
let server_config = config.clone();
@@ -230,6 +242,23 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
let (auth_tx, auth_rx) = tokio::sync::oneshot::channel::<HashMap<String, String>>();
let server = tokio::spawn(async move {
let mut auth_rx = Some(auth_rx);
let failures = if initially_unavailable {
if explicit { 1 } else { 2 }
} else {
0
};
for _ in 0..failures {
let (mut stream, _) = listener.accept().await.expect("startup discovery request");
let mut request = Vec::new();
while !request.ends_with(b"\r\n\r\n") {
request.push(stream.read_u8().await.expect("discovery request headers"));
assert!(request.len() < 8192);
}
stream
.write_all(b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
.await
.expect("startup discovery failure response");
}
for expected_path in ["/.well-known/openid-configuration", "/jwks", "/token"] {
let (mut stream, _) = tokio::time::timeout(StdDuration::from_secs(10), listener.accept())
.await
@@ -309,12 +338,21 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
}
});
let http_client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).unwrap());
let discovered = discover_provider(&config, &http_client).await.unwrap();
let sys = OidcSys {
configs: HashMap::from([(config.id.clone(), test_sourced_config(config))]),
provider_runtime: ProviderRuntime::new(http_client, HashMap::from([("console".into(), discovered)])),
state_store: OidcStateStore::new(),
};
let sys = Arc::new(
OidcSys::new_with_config_snapshot(http_client, OidcConfigSnapshot::new(vec![test_sourced_config(config)]))
.await
.expect("OIDC initialization remains available during an IdP outage"),
);
let adapter = StandardOidcAdapter::new(sys.clone());
let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules());
assert_eq!(sys.list_visible_providers().len(), usize::from(!hidden));
if initially_unavailable {
assert!(sys.provider_runtime.get_provider_state("console").is_err());
assert!(sys.authorize_url("console", redirect, None).await.is_err());
tokio::time::timeout(StdDuration::from_secs(10), sys.recover_failed_providers())
.await
.expect("discovery should recover after the IdP becomes available");
}
let auth_url = sys.authorize_url("console", redirect, Some("/buckets".into())).await.unwrap();
let auth_url = Url::parse(&auth_url).unwrap();
assert_eq!(auth_url.as_str().split('?').next(), Some(format!("{base}/authorize").as_str()));
@@ -328,15 +366,18 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
assert!(auth["scope"].split_whitespace().any(|scope| scope == "openid"));
let state = auth["state"].clone();
auth_tx.send(auth).unwrap();
let (claims, provider, session, _) = sys
.exchange_code(&state, "test-authorization-code", redirect)
let exchange = adapter
.exchange_identity(&state, "test-authorization-code", redirect)
.await
.unwrap_or_else(|err| panic!("hidden={hidden}, explicit={explicit}: {err}"));
assert_eq!(provider, "console");
assert_eq!(claims.sub, "existing-user");
assert_eq!(claims.email, "user@example.com");
assert_eq!(claims.groups, vec!["readwrite"]);
assert_eq!(session.redirect_after.as_deref(), Some("/buckets"));
let (identity, redirect_after, _) = exchange.into_parts();
let authorization = mapper.map(identity);
assert_eq!(authorization.provider_id, "console");
assert_eq!(authorization.claims.sub, "existing-user");
assert_eq!(authorization.claims.email, "user@example.com");
assert_eq!(authorization.claims.groups, vec!["readwrite"]);
assert_eq!(authorization.policies, vec!["team/readwrite"]);
assert_eq!(redirect_after.as_deref(), Some("/buckets"));
assert!(matches!(sys.exchange_code(&state, "test-authorization-code", redirect).await,
Err(error) if error == "invalid or expired OIDC state"));
server.await.unwrap();
@@ -470,6 +511,7 @@ async fn oidc_token_exchange_reports_forbidden_token_endpoint() {
)]),
),
state_store,
pending_discovery: Default::default(),
};
let error = match sys
+1
View File
@@ -35,6 +35,7 @@ pub fn make_test_sys(configs: Vec<OidcProviderConfig>) -> OidcSys {
HashMap::new(),
),
state_store: OidcStateStore::new(),
pending_discovery: Default::default(),
}
}
+32 -23
View File
@@ -312,6 +312,10 @@ struct CachedOidcExtraRootCerts {
certs: Vec<Certificate>,
}
/// Discovery callers report failures with provider context and recovery throttling.
#[derive(Clone)]
pub(super) struct DiscoveryRequest;
/// HTTP client adapter bridging reqwest 0.13 to the `openidconnect` `AsyncHttpClient` trait.
///
/// A fresh client is built for every request so the destination is re-validated and the
@@ -555,6 +559,7 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
Box::pin(async move {
let started_at = Instant::now();
let (parts, body) = request.into_parts();
let discovery_request = parts.extensions.get::<DiscoveryRequest>().is_some();
let method = parts.method.clone();
let uri = parts.uri.to_string();
if tracing::enabled!(tracing::Level::DEBUG) {
@@ -593,17 +598,19 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
let response = response.map_err(|err| {
let error = oidc_http_error_from_reqwest(&url, err);
error!(
event = EVENT_OIDC_HTTP,
component = LOG_COMPONENT_IAM,
subsystem = LOG_SUBSYSTEM_OIDC,
result = "request_failed",
method = %method,
uri = %uri,
elapsed_ms,
error = %error,
"oidc outbound http"
);
if !discovery_request {
error!(
event = EVENT_OIDC_HTTP,
component = LOG_COMPONENT_IAM,
subsystem = LOG_SUBSYSTEM_OIDC,
result = "request_failed",
method = %method,
uri = %uri,
elapsed_ms,
error = %error,
"oidc outbound http"
);
}
error
})?;
@@ -612,18 +619,20 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient {
let body_bytes = read_bounded_response_body(response, MAX_OIDC_RESPONSE_SIZE)
.await
.map_err(|err| {
error!(
event = EVENT_OIDC_HTTP,
component = LOG_COMPONENT_IAM,
subsystem = LOG_SUBSYSTEM_OIDC,
result = "response_body_failed",
method = %method,
uri = %uri,
status = status.as_u16(),
elapsed_ms,
error = %err,
"oidc outbound http"
);
if !discovery_request {
error!(
event = EVENT_OIDC_HTTP,
component = LOG_COMPONENT_IAM,
subsystem = LOG_SUBSYSTEM_OIDC,
result = "response_body_failed",
method = %method,
uri = %uri,
status = status.as_u16(),
elapsed_ms,
error = %err,
"oidc outbound http"
);
}
err
})?;
if tracing::enabled!(tracing::Level::DEBUG) {
+68 -4
View File
@@ -19,10 +19,12 @@ use rustfs_iam::{
};
use std::{
collections::hash_map::DefaultHasher,
future::Future,
hash::{Hash, Hasher},
io::{Error, Result},
sync::Arc,
};
use tokio_util::{sync::CancellationToken, task::AbortOnDropHandle};
use tracing::{error, info, warn};
const LOG_COMPONENT_MAIN: &str = "main";
@@ -31,7 +33,7 @@ const EVENT_KEYSTONE_AUTH_INITIALIZED: &str = "keystone_auth_initialized";
const EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED: &str = "keystone_auth_initialization_failed";
const EVENT_OIDC_INITIALIZATION_FAILED: &str = "oidc_initialization_failed";
pub(crate) async fn init_auth_integrations() -> Result<()> {
pub(crate) async fn init_auth_integrations(shutdown: CancellationToken) -> Result<Option<AbortOnDropHandle<()>>> {
let keystone_config = rustfs_keystone::KeystoneConfig::from_env().map_err(Error::other)?;
if keystone_config.enable {
match crate::auth_keystone::init_keystone_auth(keystone_config).await {
@@ -55,8 +57,13 @@ pub(crate) async fn init_auth_integrations() -> Result<()> {
match build_oidc_sys_with_extra_root_ca_provider(oidc_extra_root_ca_provider()).await {
Ok(oidc) => {
let (service, oidc_config_query) = standard_oidc_runtime(oidc);
crate::runtime_sources::publish_federated_identity_runtime(service, oidc_config_query);
let (service, oidc_config_query) = standard_oidc_runtime(oidc.clone());
if crate::runtime_sources::publish_federated_identity_runtime(service, oidc_config_query) {
return Ok(Some(spawn_oidc_discovery_recovery(
async move { oidc.recover_failed_providers().await },
shutdown,
)));
}
}
Err(e) => {
warn!(
@@ -69,7 +76,20 @@ pub(crate) async fn init_auth_integrations() -> Result<()> {
}
}
Ok(())
Ok(None)
}
fn spawn_oidc_discovery_recovery(
recovery: impl Future<Output = ()> + Send + 'static,
shutdown: CancellationToken,
) -> AbortOnDropHandle<()> {
AbortOnDropHandle::new(tokio::spawn(async move {
tokio::select! {
biased;
() = shutdown.cancelled() => {}
() = recovery => {}
}
}))
}
fn standard_oidc_runtime(oidc: Arc<OidcSys>) -> (Arc<FederatedIdentityService>, Arc<dyn OidcConfigQuery>) {
@@ -133,4 +153,48 @@ mod tests {
drop(oidc_config_query);
assert_eq!(Arc::strong_count(&retained), 1, "dropping the runtime should release its OIDC Arc");
}
#[tokio::test]
async fn oidc_recovery_shutdown_releases_pending_work() {
let shutdown = CancellationToken::new();
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
let (released_tx, released_rx) = tokio::sync::oneshot::channel::<()>();
let recovery = spawn_oidc_discovery_recovery(
async move {
let _release = released_tx;
let _ = started_tx.send(());
std::future::pending::<()>().await;
},
shutdown.child_token(),
);
started_rx.await.expect("recovery started");
shutdown.cancel();
tokio::time::timeout(std::time::Duration::from_secs(5), recovery)
.await
.expect("recovery observes shutdown")
.expect("recovery joins cleanly");
assert!(released_rx.await.is_err(), "shutdown drops in-flight recovery work");
}
#[tokio::test]
async fn oidc_recovery_owner_drop_releases_pending_work() {
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
let (released_tx, released_rx) = tokio::sync::oneshot::channel::<()>();
let recovery = spawn_oidc_discovery_recovery(
async move {
let _release = released_tx;
let _ = started_tx.send(());
std::future::pending::<()>().await;
},
CancellationToken::new(),
);
started_rx.await.expect("recovery started");
drop(recovery);
assert!(
tokio::time::timeout(std::time::Duration::from_secs(5), released_rx)
.await
.expect("owner drop aborts recovery")
.is_err()
);
}
}
+13
View File
@@ -130,6 +130,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
readiness,
} = lifecycle;
let StartupServiceRuntime {
oidc_recovery,
optional_runtimes,
heartbeat,
inventory,
@@ -166,6 +167,18 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
shutdown_token,
)
.await;
if let Some(recovery) = oidc_recovery
&& let Err(error) = recovery.await
{
tracing::warn!(
event = EVENT_SERVER_SHUTDOWN_STATE,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
result = "oidc_discovery_recovery_join_failed",
error = %error,
"OIDC discovery recovery task failed to join"
);
}
shutdown_connect_runtimes(heartbeat, inventory).await;
if let Some(runtime) = local_trace_capture {
runtime.shutdown().await;
+4 -2
View File
@@ -36,13 +36,14 @@ use crate::{
};
use rustfs_common::GlobalReadiness;
use std::{collections::BTreeSet, io::Result, sync::Arc};
use tokio_util::sync::CancellationToken;
use tokio_util::{sync::CancellationToken, task::AbortOnDropHandle};
const EVENT_DRIVE_UNAVAILABLE: &str = "drive_unavailable";
const LOG_COMPONENT_CONNECT: &str = "connect";
const LOG_SUBSYSTEM_INVENTORY: &str = "inventory";
pub(crate) struct StartupServiceRuntime {
pub(crate) oidc_recovery: Option<AbortOnDropHandle<()>>,
pub(crate) optional_runtimes: OptionalRuntimeServices,
pub(crate) heartbeat: Option<HeartbeatRuntime>,
pub(crate) inventory: Option<InventoryRuntime>,
@@ -111,7 +112,7 @@ pub(crate) async fn init_startup_runtime_services(
// self-pointing replication rules until the next restart. The task waits for IAM and
// bucket metadata itself, and its first pass runs off this path.
spawn_site_replication_reconcile_task(ctx.clone());
init_auth_integrations().await?;
let oidc_recovery = init_auth_integrations(ctx.child_token()).await?;
init_notification_runtime(endpoint_pools, buckets).await?;
let enable_scanner = init_background_service_runtime(store.clone()).await?;
init_observability_runtime(store.clone(), ctx.clone()).await;
@@ -119,6 +120,7 @@ pub(crate) async fn init_startup_runtime_services(
let inventory = start_inventory_runtime(heartbeat_config, heartbeat_nodes, inventory_drives, store, &ctx)?;
Ok(StartupServiceRuntime {
oidc_recovery,
optional_runtimes,
heartbeat,
inventory,