Merge remote-tracking branch 'origin/main' into fix/replication-target-version-ledger

Cargo.lock

@@ -9999,6 +9999,7 @@ dependencies = [
  "rustls-pki-types",
  "serde",
  "serde_json",
+ "serde_with",
  "serial_test",
  "sha1 0.11.0",
  "sha2 0.11.0",

Cargo.toml

@@ -191,6 +191,7 @@ rmp = { version = "0.8.15" }
 rmp-serde = { version = "1.3.1" }
 serde = { version = "1.0.229" }
 serde_ignored = { version = "0.1" }
+serde_with = { version = "3", default-features = false, features = ["macros", "std"] }
 serde_json = { version = "1.0.151" }
 serde_urlencoded = "0.7.1"
 

crates/ecstore/src/core/pools.rs

@@ -5137,6 +5137,23 @@ where
     }
 }
 
+fn select_pool_meta_replicas_for_read_probe<R>(
+    write_state: &PoolMetaWriteState,
+    replicas: Vec<R>,
+    operation: &str,
+) -> Result<PoolMetaSelection>
+where
+    R: Into<PoolMetaReplicaRead>,
+{
+    // Read-only planning probes must fail the current request on unsafe pool
+    // metadata, but they must not permanently poison the shared writer gate.
+    let mut probe_state = write_state.clone();
+    let selection = select_pool_meta_replicas_observing(&mut probe_state, replicas)?;
+    probe_state.observe_replicas(selection.replica_state);
+    probe_state.ensure_write_safe(operation)?;
+    Ok(selection)
+}
+
 async fn load_pool_meta_replicas<S>(pools: Vec<Arc<S>>, no_lock: bool) -> Result<PoolMetaSelection>
 where
     S: EcstoreObjectIO,
@@ -5156,6 +5173,19 @@ where
     select_pool_meta_replicas_observing(write_state, replicas)
 }
 
+async fn load_pool_meta_replicas_for_read_probe<S>(
+    pools: Vec<Arc<S>>,
+    no_lock: bool,
+    write_state: &PoolMetaWriteState,
+    operation: &str,
+) -> Result<PoolMetaSelection>
+where
+    S: EcstoreObjectIO,
+{
+    let replicas = read_pool_meta_replicas(pools, no_lock).await;
+    select_pool_meta_replicas_for_read_probe(write_state, replicas, operation)
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct PersistedPoolMetaV3 {
@@ -8997,9 +9027,7 @@ impl ECStore {
         })?;
         let pool_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
         let pool_meta_guard = pool_meta_lock.get_read_lock(get_lock_acquire_timeout()).await?;
-        let selection = load_pool_meta_replicas_observing(self.pools.clone(), true, write_state).await?;
-        write_state.observe_replicas(selection.replica_state);
-        write_state.ensure_write_safe(operation)?;
+        let selection = load_pool_meta_replicas_for_read_probe(self.pools.clone(), true, write_state, operation).await?;
         Ok((pool_meta_guard, selection.meta))
     }
 
@@ -17952,6 +17980,41 @@ mod tests {
         );
     }
 
+    #[test]
+    fn pool_meta_read_probe_does_not_latch_writer_state() {
+        let write_state = PoolMetaWriteState::default();
+        select_pool_meta_replicas_for_read_probe(
+            &write_state,
+            vec![PoolMetaReplica::Unreadable("transient read failure".to_string())],
+            "capacity probe",
+        )
+        .expect_err("an unreadable probe replica must fail the current admission");
+
+        write_state
+            .ensure_write_safe("ordinary object write")
+            .expect("a read-only capacity probe must not permanently latch the pool metadata writer");
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn pool_meta_read_guard_does_not_latch_after_unreadable_replica() {
+        let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
+        for set in &store.pools[1].disk_set {
+            let mut disks = set.disks.write().await;
+            let disk_count = disks.len();
+            *disks = vec![None; disk_count];
+        }
+
+        let mut write_state = store.pool_meta_save_gate.lock().await;
+        store
+            .acquire_pool_meta_read_guard(&mut write_state, "capacity probe")
+            .await
+            .expect_err("an unreadable metadata replica must reject this probe");
+        write_state
+            .ensure_write_safe("ordinary object write")
+            .expect("a failed read-only probe must remain retryable");
+    }
+
     #[test]
     fn pool_meta_write_state_blocks_on_any_recovery_required_selection() {
         fn assert_selection_blocks(replicas: Vec<PoolMetaReplica>) {

crates/heal/tests/mrf_pipeline_test.rs

@@ -29,7 +29,7 @@ use rustfs_heal::heal::{
     storage::{ECStoreHealStorage, HealStorageAPI},
 };
 use serial_test::serial;
-use std::{path::Path, sync::Arc, time::Duration};
+use std::{path::Path, process::Command, sync::Arc, time::Duration};
 
 mod storage_api;
 
@@ -40,10 +40,15 @@ const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin";
 const SCOPED_JOURNAL_REL: &str = "buckets/.heal/mrf/journal-scoped.bin";
 
 async fn heal_env() -> (Vec<std::path::PathBuf>, Arc<dyn HealStorageAPI>) {
-    let env = rustfs_test_utils::TestECStoreEnv::builder()
-        .prefix("rustfs_heal_mrf_test")
-        .build()
-        .await;
+    heal_env_at(None).await
+}
+
+async fn heal_env_at(base_dir: Option<&Path>) -> (Vec<std::path::PathBuf>, Arc<dyn HealStorageAPI>) {
+    let mut builder = rustfs_test_utils::TestECStoreEnv::builder().prefix("rustfs_heal_mrf_test");
+    if let Some(base_dir) = base_dir {
+        builder = builder.base_dir(base_dir);
+    }
+    let env = builder.build().await;
     let heal_storage: Arc<dyn HealStorageAPI> = Arc::new(ECStoreHealStorage::new(env.ecstore.clone()));
     (env.disk_paths, heal_storage)
 }
@@ -117,6 +122,12 @@ fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) {
     write_journal_path_to_disks(disk_paths, JOURNAL_REL, data);
 }
 
+fn journal_exists_on_all_disks(disk_paths: &[std::path::PathBuf], relative_path: &str) -> bool {
+    disk_paths
+        .iter()
+        .all(|path| Path::new(path).join(META_BUCKET).join(relative_path).exists())
+}
+
 async fn wait_until<F, Fut>(deadline: Duration, mut probe: F) -> bool
 where
     F: FnMut() -> Fut,
@@ -313,3 +324,69 @@ async fn journal_replay_retains_file_when_manager_is_full() {
         "the anchor remains until a successor snapshot can safely replace it"
     );
 }
+
+#[test]
+fn mrf_journal_child_process_fixture() {
+    let Ok(root) = std::env::var("RUSTFS_MRF_REPLAY_CHILD_ROOT") else {
+        return;
+    };
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("child runtime should build");
+    runtime.block_on(async {
+        let (disk_paths, _storage) = heal_env_at(Some(Path::new(&root))).await;
+        let mut journal = journal_record(1, "child-restart-bucket", "first-object", None, 0);
+        journal.extend(journal_record(1, "child-restart-bucket", "second-object", None, 0));
+        write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &journal);
+        write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &journal);
+        assert!(
+            journal_exists_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL),
+            "child process must publish the authoritative MRF journal before exiting"
+        );
+    });
+    std::process::exit(77);
+}
+
+/// A journal published by a different OS process must remain a durable anchor
+/// when the restarted process can only admit a prefix of the replayed intents.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn journal_replay_retains_child_process_anchor_when_manager_is_full() {
+    let temp_dir = tempfile::tempdir().expect("child process MRF root");
+    let status = Command::new(std::env::current_exe().expect("test binary path"))
+        .arg("mrf_journal_child_process_fixture")
+        .arg("--exact")
+        .arg("--nocapture")
+        .env("RUSTFS_MRF_REPLAY_CHILD_ROOT", temp_dir.path())
+        .status()
+        .expect("child MRF fixture should start");
+    assert_eq!(status.code(), Some(77), "child process did not reach the MRF journal boundary");
+
+    let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await;
+    assert!(
+        journal_exists_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL),
+        "restarted process must see the authoritative MRF journal left by the child"
+    );
+
+    let restarted = Arc::new(HealManager::new(
+        storage,
+        Some(HealConfig {
+            queue_size: 1,
+            heal_interval: Duration::from_secs(3600),
+            enable_auto_heal: false,
+            ..Default::default()
+        }),
+    ));
+    let replayed = mrf_queue::replay_journal_once(&restarted).await;
+    assert_eq!(replayed, 2, "the restarted process must decode the complete child journal");
+    assert_eq!(
+        restarted.operations_snapshot().await.queued_by_source.mrf,
+        1,
+        "bounded admission may accept only the prefix, but must not lose the replayed tail"
+    );
+    assert!(
+        journal_exists_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL),
+        "replay must retain the child-published journal until a successor snapshot can replace it"
+    );
+}

crates/iam/Cargo.toml

@@ -86,6 +86,7 @@ rustfs-ecstore = { workspace = true }
 rustfs-storage-api = { workspace = true }
 rustfs-policy.workspace = true
 serde_json = { workspace = true, features = ["raw_value"] }
+serde_with = { workspace = true }
 async-trait.workspace = true
 thiserror.workspace = true
 arc-swap = { workspace = true }

crates/iam/src/oidc.rs

@@ -19,11 +19,13 @@
 //! and ID token verification.
 
 use crate::oidc_state::{OidcAuthSession, OidcLogoutSession, OidcStateStore};
-use openidconnect::core::{CoreAuthenticationFlow, CoreClient, CoreIdToken, CoreJsonWebKeySet};
+use openidconnect::core::{
+    CoreAuthenticationFlow, CoreClient, CoreIdToken, CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm,
+};
 use openidconnect::{
     AsyncHttpClient, Audience, AuthType, AuthorizationCode, ClientId, ClientSecret, CsrfToken, DiscoveryError, IssuerUrl,
     JsonWebKeySetUrl, LogoutRequest, Nonce, PkceCodeChallenge, PkceCodeVerifier, PostLogoutRedirectUrl,
-    ProviderMetadataWithLogout, RedirectUrl, RequestTokenError, Scope,
+    ProviderMetadataWithLogout, RedirectUrl, RequestTokenError, Scope, TokenUrl,
 };
 use reqwest::{Certificate, Client};
 use rustfs_config::oidc::*;
@@ -52,6 +54,7 @@ const EVENT_OIDC_HTTP: &str = "oidc_http";
 const OIDC_JWKS_REFRESH_INTERVAL: StdDuration = StdDuration::from_secs(24 * 60 * 60);
 const OIDC_DISCOVERY_TRANSPORT_RETRIES: usize = 3;
 const OIDC_DISCOVERY_TRANSPORT_RETRY_DELAY: StdDuration = StdDuration::from_millis(50);
+const OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY: &str = "JWKS request blocked by outbound policy";
 const OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY: &str = "OIDC provider discovery blocked by outbound policy";
 const OIDC_HTTP_REQUEST_TIMEOUT: StdDuration = StdDuration::from_secs(10);
 const OIDC_HTTP_CONNECT_TIMEOUT: StdDuration = StdDuration::from_secs(3);
@@ -753,7 +756,7 @@ pub struct SourcedOidcProviderConfig {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct OidcProviderValidationResult {
     pub issuer: String,
-    pub authorization_endpoint: String,
+    pub authorization_endpoint: Option<String>,
     pub token_endpoint: Option<String>,
 }
 
@@ -783,10 +786,142 @@ pub struct OidcClaims {
 /// on-the-fly from metadata when needed.
 #[derive(Clone)]
 struct ProviderState {
-    metadata: ProviderMetadataWithLogout,
+    metadata: DiscoveredProviderMetadata,
     discovered_at: Instant,
 }
 
+// Workload issuers do not implement the browser authorization flow. Keep their
+// verification metadata separate rather than inventing an authorization URL.
+#[serde_with::serde_as]
+#[derive(Clone, Deserialize)]
+struct WorkloadProviderMetadata {
+    issuer: IssuerUrl,
+    jwks_uri: JsonWebKeySetUrl,
+    token_endpoint: Option<TokenUrl>,
+    #[serde_as(as = "serde_with::VecSkipError<_>")]
+    id_token_signing_alg_values_supported: Vec<CoreJwsSigningAlgorithm>,
+    #[serde(skip)]
+    jwks: CoreJsonWebKeySet,
+    // Discovery is extensible; report unsupported fields without logging values.
+    #[serde(flatten)]
+    additional_fields: HashMap<String, serde_json::Value>,
+}
+
+#[derive(Clone)]
+enum DiscoveredProviderMetadata {
+    Console(Box<ProviderMetadataWithLogout>),
+    Workload(Box<WorkloadProviderMetadata>),
+}
+
+impl DiscoveredProviderMetadata {
+    fn parse(body: &[u8], hide_from_ui: bool) -> Result<Self, String> {
+        let document: serde_json::Value = serde_json::from_slice(body).map_err(|err| err.to_string())?;
+        if hide_from_ui
+            && document
+                .as_object()
+                .is_some_and(|fields| !fields.contains_key("authorization_endpoint"))
+        {
+            let mut metadata: WorkloadProviderMetadata = serde_json::from_slice(body).map_err(|err| err.to_string())?;
+            if !metadata.additional_fields.is_empty() {
+                warn!(
+                    event = EVENT_OIDC_DIAGNOSTICS,
+                    component = LOG_COMPONENT_IAM,
+                    subsystem = LOG_SUBSYSTEM_OIDC,
+                    result = "workload_discovery_additional_fields",
+                    field_count = metadata.additional_fields.len(),
+                    "workload discovery contains additional fields"
+                );
+                metadata.additional_fields.clear();
+            }
+            Ok(Self::Workload(Box::new(metadata)))
+        } else {
+            serde_json::from_slice(body)
+                .map(|metadata| Self::Console(Box::new(metadata)))
+                .map_err(|err| err.to_string())
+        }
+    }
+
+    fn console(&self) -> Result<&ProviderMetadataWithLogout, String> {
+        match self {
+            Self::Console(metadata) => Ok(metadata),
+            Self::Workload(_) => Err("OIDC provider has no authorization endpoint; only web identity is supported".into()),
+        }
+    }
+
+    fn issuer(&self) -> &IssuerUrl {
+        match self {
+            Self::Console(metadata) => metadata.issuer(),
+            Self::Workload(metadata) => &metadata.issuer,
+        }
+    }
+
+    fn jwks_uri(&self) -> &JsonWebKeySetUrl {
+        match self {
+            Self::Console(metadata) => metadata.jwks_uri(),
+            Self::Workload(metadata) => &metadata.jwks_uri,
+        }
+    }
+
+    fn set_jwks(self, jwks: CoreJsonWebKeySet) -> Self {
+        match self {
+            Self::Console(metadata) => Self::Console(Box::new(metadata.set_jwks(jwks))),
+            Self::Workload(mut metadata) => {
+                metadata.jwks = jwks;
+                Self::Workload(metadata)
+            }
+        }
+    }
+
+    fn authorization_endpoint(&self) -> Option<String> {
+        match self {
+            Self::Console(metadata) => Some(metadata.authorization_endpoint().to_string()),
+            Self::Workload(_) => None,
+        }
+    }
+
+    fn token_endpoint(&self) -> Option<&TokenUrl> {
+        match self {
+            Self::Console(metadata) => metadata.token_endpoint(),
+            Self::Workload(metadata) => metadata.token_endpoint.as_ref(),
+        }
+    }
+
+    fn verifier(&self, config: &OidcProviderConfig) -> CoreIdTokenVerifier<'static> {
+        let client_id = ClientId::new(config.client_id.clone());
+        let secret = config.client_secret.as_ref().map(|secret| ClientSecret::new(secret.clone()));
+        let (issuer, jwks, algorithms) = match self {
+            Self::Console(metadata) => (metadata.issuer(), metadata.jwks(), metadata.id_token_signing_alg_values_supported()),
+            Self::Workload(metadata) => (&metadata.issuer, &metadata.jwks, &metadata.id_token_signing_alg_values_supported),
+        };
+        let verifier = match secret {
+            Some(secret) => CoreIdTokenVerifier::new_confidential_client(client_id, secret, issuer.clone(), jwks.clone()),
+            None => CoreIdTokenVerifier::new_public_client(client_id, issuer.clone(), jwks.clone()),
+        };
+        verifier.set_allowed_algs(algorithms.clone())
+    }
+}
+
+// This adapter is used only for discovery/JWKS fetches, never token exchange.
+struct JwksAcceptClient<'a> {
+    inner: &'a ReqwestHttpClient,
+    discovery_url: Option<Url>,
+}
+
+impl<'c> AsyncHttpClient<'c> for JwksAcceptClient<'_> {
+    type Error = OidcHttpError;
+    type Future = <ReqwestHttpClient as AsyncHttpClient<'c>>::Future;
+
+    fn call(&'c self, mut request: http::Request<Vec<u8>>) -> Self::Future {
+        if !self.discovery_url.as_ref().is_some_and(|url| request.uri() == url.as_str()) {
+            request.headers_mut().insert(
+                http::header::ACCEPT,
+                http::HeaderValue::from_static("application/json, application/jwk-set+json"),
+            );
+        }
+        self.inner.call(request)
+    }
+}
+
 impl ProviderState {
     fn is_stale(&self) -> bool {
         self.discovered_at.elapsed() >= OIDC_JWKS_REFRESH_INTERVAL
@@ -932,7 +1067,7 @@ impl OidcSys {
         let redirect = RedirectUrl::new(redirect_uri.to_string()).map_err(|e| format!("invalid redirect URI: {e}"))?;
 
         let client = CoreClient::from_provider_metadata(
-            state.metadata.clone(),
+            state.metadata.console()?.clone(),
             ClientId::new(config.client_id.clone()),
             config.client_secret.as_ref().map(|s| ClientSecret::new(s.clone())),
         )
@@ -994,7 +1129,7 @@ impl OidcSys {
 
         // Construct CoreClient on-the-fly with JWKS from discovery
         let client = CoreClient::from_provider_metadata(
-            provider_state.metadata.clone(),
+            provider_state.metadata.console()?.clone(),
             ClientId::new(config.client_id.clone()),
             config.client_secret.as_ref().map(|s| ClientSecret::new(s.clone())),
         )
@@ -1230,7 +1365,7 @@ impl OidcSys {
             );
 
             let client = CoreClient::from_provider_metadata(
-                refreshed_state.metadata,
+                refreshed_state.metadata.console()?.clone(),
                 ClientId::new(config.client_id.clone()),
                 config.client_secret.as_ref().map(|s| ClientSecret::new(s.clone())),
             )
@@ -1320,7 +1455,7 @@ impl OidcSys {
             .get(&session.provider_id)
             .ok_or_else(|| format!("unknown OIDC provider: {}", session.provider_id))?;
         let state = self.ensure_provider_state(&session.provider_id, config).await?;
-        let Some(end_session_endpoint) = state.metadata.additional_metadata().end_session_endpoint.clone() else {
+        let Some(end_session_endpoint) = state.metadata.console()?.additional_metadata().end_session_endpoint.clone() else {
             return Ok(None);
         };
 
@@ -1460,14 +1595,6 @@ impl OidcSys {
 
         state = self.ensure_provider_state_if_stale(&provider_id, &config, &state).await?;
 
-        // Reconstruct CoreClient from provider metadata
-        let client = CoreClient::from_provider_metadata(
-            state.metadata.clone(),
-            ClientId::new(config.client_id.clone()),
-            config.client_secret.as_ref().map(|s| ClientSecret::new(s.clone())),
-        )
-        .set_auth_type(AuthType::RequestBody);
-
         // Parse raw JWT string into CoreIdToken
         let id_token: CoreIdToken = jwt
             .parse()
@@ -1475,8 +1602,9 @@ impl OidcSys {
 
         // Verify the token (signature, issuer, audience, expiry) — skip nonce
         // (nonce is only required for the authorization code flow)
-        let verifier = client
-            .id_token_verifier()
+        let verifier = state
+            .metadata
+            .verifier(&config)
             .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
         if let Err(e) = id_token.claims(&verifier, |_: Option<&Nonce>| Ok(())) {
             state = self
@@ -1486,14 +1614,9 @@ impl OidcSys {
                     format!("ID token verification failed: {e}; failed to refresh provider metadata: {refresh_err}")
                 })?;
 
-            let client = CoreClient::from_provider_metadata(
-                state.metadata,
-                ClientId::new(config.client_id.clone()),
-                config.client_secret.as_ref().map(|s| ClientSecret::new(s.clone())),
-            )
-            .set_auth_type(AuthType::RequestBody);
-            let verifier = client
-                .id_token_verifier()
+            let verifier = state
+                .metadata
+                .verifier(&config)
                 .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
             id_token
                 .claims(&verifier, |_: Option<&Nonce>| Ok(()))
@@ -1868,18 +1991,39 @@ impl OidcSys {
             let issuer_url = IssuerUrl::new(candidate_issuer.clone()).map_err(|e| format!("invalid issuer URL: {e}"))?;
 
             for attempt in 0..OIDC_DISCOVERY_TRANSPORT_RETRIES {
-                match ProviderMetadataWithLogout::discover_async(issuer_url.clone(), http_client).await {
-                    Ok(metadata) => {
-                        return Ok(ProviderState {
-                            metadata,
+                let discovered = if config.hide_from_ui {
+                    Self::discover_provider_from_config_url(config, candidate_issuer, http_client).await
+                } else {
+                    let client = JwksAcceptClient {
+                        inner: http_client,
+                        discovery_url: Some(
+                            issuer_url
+                                .join(".well-known/openid-configuration")
+                                .map_err(|err| err.to_string())?,
+                        ),
+                    };
+                    ProviderMetadataWithLogout::discover_async(issuer_url.clone(), &client)
+                        .await
+                        .map(|metadata| ProviderState {
+                            metadata: DiscoveredProviderMetadata::Console(Box::new(metadata)),
                             discovered_at: Instant::now(),
-                        });
-                    }
-                    Err(DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason))) => {
-                        return Err(format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}"));
+                        })
+                        .map_err(|err| match err {
+                            DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason)) => {
+                                format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}")
+                            }
+                            err => format!("discovery failed: {err}"),
+                        })
+                };
+                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) =>
+                    {
+                        return Err(error);
                     }
                     Err(error) => {
-                        let error = format!("discovery failed: {error}");
                         let is_transient_transport = error.contains("Request failed");
                         let should_retry = is_transient_transport && attempt + 1 < OIDC_DISCOVERY_TRANSPORT_RETRIES;
                         if should_retry {
@@ -1933,7 +2077,14 @@ impl OidcSys {
         http_client: &ReqwestHttpClient,
     ) -> Result<ProviderState, String> {
         let issuer_url = IssuerUrl::new(issuer.trim().to_string()).map_err(|e| format!("invalid issuer URL: {e}"))?;
-        let discovery_url = discovery_url_from_config_url(&config.config_url)?;
+        let explicit_issuer = config.issuer.as_deref().is_some_and(|issuer| !issuer.trim().is_empty());
+        let discovery_url = if explicit_issuer {
+            discovery_url_from_config_url(&config.config_url)?
+        } else {
+            issuer_url
+                .join(".well-known/openid-configuration")
+                .map_err(|err| err.to_string())?
+        };
         let request = http::Request::builder()
             .uri(discovery_url.to_string())
             .method(http::Method::GET)
@@ -1946,13 +2097,25 @@ impl OidcSys {
             Err(OidcHttpError::ForbiddenOutbound(reason)) => {
                 return Err(format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}"));
             }
-            Err(err) => return Err(format!("discovery 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));
         }
 
-        let provider_metadata = serde_json::from_slice::<ProviderMetadataWithLogout>(response.body())
+        if !explicit_issuer
+            && let Some(content_type) = response.headers().get(http::header::CONTENT_TYPE)
+            && !content_type.to_str().ok().is_some_and(|value| {
+                value
+                    .split(';')
+                    .next()
+                    .is_some_and(|essence| essence.eq_ignore_ascii_case("application/json"))
+            })
+        {
+            return Err("Unexpected response Content-Type: expected application/json".into());
+        }
+
+        let provider_metadata = DiscoveredProviderMetadata::parse(response.body(), config.hide_from_ui)
             .map_err(|err| format!("failed to parse discovery response: {err}"))?;
         if provider_metadata.issuer() != &issuer_url {
             return Err(format!(
@@ -1962,11 +2125,23 @@ impl OidcSys {
             ));
         }
 
-        let jwks_url = jwks_url_from_config_url(&config.config_url, &issuer_url, provider_metadata.jwks_uri())?;
-        let jwks = match CoreJsonWebKeySet::fetch_async(&jwks_url, http_client).await {
+        let jwks_url = if explicit_issuer {
+            jwks_url_from_config_url(&config.config_url, &issuer_url, provider_metadata.jwks_uri())?
+        } else {
+            provider_metadata.jwks_uri().clone()
+        };
+        let jwks = match CoreJsonWebKeySet::fetch_async(
+            &jwks_url,
+            &JwksAcceptClient {
+                inner: http_client,
+                discovery_url: None,
+            },
+        )
+        .await
+        {
             Ok(jwks) => jwks,
             Err(DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason))) => {
-                return Err(format!("JWKS request blocked by outbound policy: {reason}"));
+                return Err(format!("{OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY}: {reason}"));
             }
             Err(err) => return Err(format!("failed to fetch JWKS: {err}")),
         };
@@ -2038,7 +2213,7 @@ pub async fn validate_oidc_provider_config_with_extra_root_ca(
 
     Ok(OidcProviderValidationResult {
         issuer: state.metadata.issuer().to_string(),
-        authorization_endpoint: state.metadata.authorization_endpoint().to_string(),
+        authorization_endpoint: state.metadata.authorization_endpoint(),
         token_endpoint: state.metadata.token_endpoint().map(ToString::to_string),
     })
 }
@@ -2598,7 +2773,7 @@ mod tests {
         }
     }
 
-    fn read_mock_oidc_request_path(stream: &mut impl std::io::Read) -> String {
+    fn read_mock_oidc_request(stream: &mut impl std::io::Read) -> String {
         let mut request_bytes = Vec::new();
         let mut buffer = [0u8; 4096];
         loop {
@@ -2615,8 +2790,11 @@ mod tests {
                 break;
             }
         }
-        let request = String::from_utf8_lossy(&request_bytes);
-        request
+        String::from_utf8_lossy(&request_bytes).into_owned()
+    }
+
+    fn read_mock_oidc_request_path(stream: &mut impl std::io::Read) -> String {
+        read_mock_oidc_request(stream)
             .lines()
             .next()
             .unwrap_or("")
@@ -2627,7 +2805,7 @@ mod tests {
     }
 
     fn mock_oidc_response(path: &str, discovery_body: &str, expected_jwks_path: &str, jwks_body: &str) -> String {
-        let (status, body) = if path.contains("/.well-known/openid-configuration") {
+        let (status, body) = if path.ends_with("/.well-known/openid-configuration") {
             (200, discovery_body)
         } else if path == expected_jwks_path {
             (200, jwks_body)
@@ -2646,6 +2824,7 @@ mod tests {
         build_discovery_issuer: F,
         max_requests: usize,
         signing_alg: &'static str,
+        workload: bool,
         jwks_response: J,
     ) -> Option<(String, std::thread::JoinHandle<()>)>
     where
@@ -2670,7 +2849,7 @@ mod tests {
         };
         let base = format!("http://{}", listener.local_addr().expect("listener local address should be available"));
         let (discovery_issuer, discovery_jwks_uri, expected_jwks_path) = build_discovery_issuer(&base);
-        let discovery_body = serde_json::json!({
+        let mut discovery_document = serde_json::json!({
             "issuer": discovery_issuer,
             "authorization_endpoint": format!("{base}/authorize"),
             "token_endpoint": format!("{base}/token"),
@@ -2679,8 +2858,14 @@ mod tests {
             "response_modes_supported": ["query"],
             "subject_types_supported": ["public"],
             "id_token_signing_alg_values_supported": [signing_alg],
-        })
-        .to_string();
+        });
+        if workload {
+            let fields = discovery_document.as_object_mut().expect("mock metadata is an object");
+            fields.remove("authorization_endpoint");
+            fields.remove("token_endpoint");
+            fields.insert("response_types_supported".into(), serde_json::json!(["id_token"]));
+        }
+        let discovery_body = discovery_document.to_string();
         let (ready_tx, ready_rx) = mpsc::channel();
 
         let handle = std::thread::spawn(move || {
@@ -2722,12 +2907,41 @@ mod tests {
                     .set_read_timeout(Some(Duration::from_secs(1)))
                     .expect("failed to set discovery mock read timeout");
 
-                let path = read_mock_oidc_request_path(&mut stream);
+                let request = read_mock_oidc_request(&mut stream);
+                let path = request
+                    .lines()
+                    .next()
+                    .and_then(|line| line.split_whitespace().nth(1))
+                    .unwrap_or("");
                 let jwks_body = jwks_response(jwks_fetches);
                 if path == expected_jwks_path {
                     jwks_fetches += 1;
                 }
-                let response = mock_oidc_response(&path, &discovery_body, &expected_jwks_path, &jwks_body);
+                let mut response = mock_oidc_response(path, &discovery_body, &expected_jwks_path, &jwks_body);
+                if path.contains("/.well-known/openid-configuration") {
+                    assert!(
+                        request
+                            .lines()
+                            .filter_map(|line| line.split_once(':'))
+                            .any(|(name, value)| { name.eq_ignore_ascii_case("accept") && value.trim() == "application/json" }),
+                        "discovery Accept must remain unchanged"
+                    );
+                }
+                if path == expected_jwks_path {
+                    let expected_type = if workload {
+                        "application/jwk-set+json"
+                    } else {
+                        "application/json"
+                    };
+                    let accepts_type = request.lines().filter_map(|line| line.split_once(':')).any(|(name, value)| {
+                        name.eq_ignore_ascii_case("accept") && value.split(',').any(|item| item.trim() == expected_type)
+                    });
+                    if !accepts_type {
+                        response = "HTTP/1.1 406 Not Acceptable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".into();
+                    } else if workload {
+                        response = response.replace("Content-Type: application/json", "Content-Type: application/jwk-set+json");
+                    }
+                }
                 let _ = stream.write_all(response.as_bytes());
                 let _ = stream.flush();
                 let _ = stream.shutdown(Shutdown::Both);
@@ -2752,7 +2966,7 @@ mod tests {
     where
         F: Fn(&str) -> (String, String, String) + Send + 'static,
     {
-        start_mock_oidc_discovery_server_with_jwks(build_discovery_issuer, max_requests, "RS256", |_| {
+        start_mock_oidc_discovery_server_with_jwks(build_discovery_issuer, max_requests, "RS256", false, |_| {
             r#"{"keys":[]}"#.to_string()
         })
     }
@@ -2779,6 +2993,7 @@ mod tests {
             |base| (base.to_string(), format!("{base}/jwks"), "/jwks".to_string()),
             4,
             "ES256",
+            false,
             move |fetch| {
                 if fetch == 0 {
                     initial_jwks.clone()
@@ -2833,6 +3048,391 @@ mod tests {
         handle.join().expect("rotating JWKS mock server should exit cleanly");
     }
 
+    #[tokio::test]
+    async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+        for hidden in [false, true] {
+            for explicit in [false, true] {
+                let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+                let base = format!("http://{}", listener.local_addr().unwrap());
+                let issuer = if explicit {
+                    "https://issuer.example.com".to_string()
+                } else {
+                    base.clone()
+                };
+                let mut config =
+                    build_mocked_oidc_provider_config("console", &format!("{base}/.well-known/openid-configuration"));
+                config.hide_from_ui = hidden;
+                config.issuer = explicit.then(|| issuer.clone());
+                config.client_secret = Some(Nonce::new_random().secret().clone());
+                let server_config = config.clone();
+                let server_base = base.clone();
+                let redirect = "https://console.example.com/oauth_callback";
+                let (key, jwk) = oidc_es256_key_and_jwk("console");
+                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);
+                    for expected_path in ["/.well-known/openid-configuration", "/jwks", "/token"] {
+                        let (mut stream, _) = tokio::time::timeout(StdDuration::from_secs(10), listener.accept())
+                            .await
+                            .unwrap()
+                            .unwrap();
+                        let mut bytes = Vec::new();
+                        let header_end = loop {
+                            bytes.push(stream.read_u8().await.unwrap());
+                            assert!(bytes.len() < 8192);
+                            if bytes.ends_with(b"\r\n\r\n") {
+                                break bytes.len();
+                            }
+                        };
+                        let headers = String::from_utf8(bytes).unwrap();
+                        let request_line = headers.lines().next().unwrap();
+                        assert_eq!(request_line.split_whitespace().nth(1), Some(expected_path));
+                        let headers_map: HashMap<_, _> = headers
+                            .lines()
+                            .skip(1)
+                            .filter_map(|line| line.split_once(':'))
+                            .map(|(name, value)| (name.to_ascii_lowercase(), value.trim().to_string()))
+                            .collect();
+                        let body = match expected_path {
+                            "/.well-known/openid-configuration" => {
+                                assert!(request_line.starts_with("GET "));
+                                assert_eq!(headers_map["accept"], "application/json");
+                                serde_json::json!({
+                                    "issuer": issuer, "authorization_endpoint": format!("{server_base}/authorize"),
+                                    "token_endpoint": format!("{server_base}/token"), "jwks_uri": format!("{server_base}/jwks"),
+                                    "response_types_supported": ["code"], "subject_types_supported": ["public"],
+                                    "id_token_signing_alg_values_supported": ["ES256"]
+                                })
+                            }
+                            "/jwks" => {
+                                assert!(request_line.starts_with("GET "));
+                                assert!(headers_map["accept"].contains("application/json"));
+                                serde_json::json!({"keys": [jwk]})
+                            }
+                            "/token" => {
+                                assert!(request_line.starts_with("POST "));
+                                assert_eq!(headers_map["accept"], "application/json");
+                                assert!(headers_map["content-type"].starts_with("application/x-www-form-urlencoded"));
+                                let length: usize = headers_map["content-length"].parse().unwrap();
+                                assert!(header_end + length < 16384);
+                                let mut body = vec![0; length];
+                                stream.read_exact(&mut body).await.unwrap();
+                                let form: HashMap<String, String> = url::form_urlencoded::parse(&body).into_owned().collect();
+                                assert_eq!(form["grant_type"], "authorization_code");
+                                assert_eq!(form["code"], "test-authorization-code");
+                                assert_eq!(form["client_id"], server_config.client_id);
+                                assert_eq!(Some(&form["client_secret"]), server_config.client_secret.as_ref());
+                                assert_eq!(form["redirect_uri"], redirect);
+                                let auth = auth_rx.take().unwrap().await.unwrap();
+                                let challenge = PkceCodeChallenge::from_code_verifier_sha256(&PkceCodeVerifier::new(form["code_verifier"].clone()));
+                                assert_eq!(challenge.as_str(), auth["code_challenge"]);
+                                let now = time::OffsetDateTime::now_utc().unix_timestamp();
+                                let mut header = Header::new(Algorithm::ES256);
+                                header.kid = Some("console".into());
+                                let token = jsonwebtoken::encode(&header, &serde_json::json!({
+                                    "iss": issuer, "sub": "existing-user", "aud": server_config.client_id,
+                                    "iat": now, "exp": now + 300, "nonce": auth["nonce"],
+                                    "email": "user@example.com", "groups": ["readwrite"]
+                                }), &key).unwrap();
+                                serde_json::json!({"access_token": "test-access-token", "token_type": "Bearer", "id_token": token})
+                            }
+                            _ => unreachable!(),
+                        }.to_string();
+                        stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap();
+                    }
+                });
+                let http_client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).unwrap());
+                let discovered = OidcSys::discover_provider(&config, &http_client).await.unwrap();
+                let sys = OidcSys {
+                    configs: HashMap::from([(config.id.clone(), config)]),
+                    provider_states: RwLock::new(HashMap::from([("console".into(), discovered)])),
+                    state_store: OidcStateStore::new(),
+                    http_client,
+                };
+                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()));
+                let auth: HashMap<String, String> = auth_url.query_pairs().into_owned().collect();
+                assert_eq!(auth["response_type"], "code");
+                assert_eq!(auth["client_id"], "rustfs-oidc-test");
+                assert!(!auth["nonce"].is_empty());
+                assert!(!auth["state"].is_empty());
+                assert_eq!(auth["redirect_uri"], redirect);
+                assert_eq!(auth["code_challenge_method"], "S256");
+                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)
+                    .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"));
+                assert!(matches!(sys.exchange_code(&state, "test-authorization-code", redirect).await,
+                    Err(error) if error == "invalid or expired OIDC state"));
+                server.await.unwrap();
+            }
+        }
+    }
+
+    #[test]
+    fn workload_metadata_requires_hidden_provider_and_valid_verification_fields() {
+        let document = serde_json::json!({
+            "issuer": "https://issuer.example.com",
+            "jwks_uri": "https://issuer.example.com/jwks",
+            "response_types_supported": ["id_token"],
+            "subject_types_supported": ["public"],
+            "id_token_signing_alg_values_supported": ["ES256"],
+        });
+        let parse =
+            |value: &serde_json::Value, hidden| DiscoveredProviderMetadata::parse(&serde_json::to_vec(value).unwrap(), hidden);
+        let metadata = parse(&document, true).expect("hidden workload metadata should parse");
+        assert!(metadata.authorization_endpoint().is_none());
+        assert!(metadata.console().err().unwrap().contains("only web identity"));
+        assert!(parse(&document, false).err().unwrap().contains("authorization_endpoint"));
+        for field in ["issuer", "jwks_uri", "id_token_signing_alg_values_supported"] {
+            let mut invalid = document.clone();
+            invalid.as_object_mut().unwrap().remove(field);
+            assert!(parse(&invalid, true).err().unwrap().contains(field), "missing {field}");
+        }
+        for endpoint in [serde_json::Value::Null, serde_json::json!(""), serde_json::json!("not a URL")] {
+            let mut invalid = document.clone();
+            invalid["authorization_endpoint"] = endpoint;
+            assert!(parse(&invalid, true).is_err(), "invalid endpoint must not select workload metadata");
+        }
+        let mut complete = document;
+        complete["authorization_endpoint"] = serde_json::json!("https://issuer.example.com/authorize");
+        for hidden in [true, false] {
+            let metadata = parse(&complete, hidden).expect("full providers keep the existing parser");
+            assert!(metadata.console().is_ok());
+            assert!(metadata.token_endpoint().is_none(), "token endpoint remains optional");
+        }
+    }
+
+    #[test]
+    fn workload_metadata_rejects_duplicate_fields() {
+        for hidden in [false, true] {
+            let document = format!(
+                r#"{{"issuer":"https://wrong.example.com","issuer":"https://issuer.example.com",{}"jwks_uri":"https://issuer.example.com/jwks","response_types_supported":["id_token"],"subject_types_supported":["public"],"id_token_signing_alg_values_supported":["ES256"]}}"#,
+                if hidden {
+                    ""
+                } else {
+                    r#""authorization_endpoint":"https://issuer.example.com/authorize","#
+                },
+            );
+            let error = DiscoveredProviderMetadata::parse(document.as_bytes(), hidden)
+                .err()
+                .expect("duplicate issuer must fail");
+            assert!(error.contains("duplicate field"), "{error}");
+        }
+    }
+
+    #[test]
+    fn workload_verifier_preserves_algorithm_secret_and_audience_policy() {
+        let secret = Nonce::new_random().secret().to_string();
+        let mut config = build_mocked_oidc_provider_config("workload", "https://issuer.example.com");
+        config.client_secret = Some(secret.clone());
+        config.other_audiences = vec!["additional-audience".into()];
+        let now = time::OffsetDateTime::now_utc().unix_timestamp();
+        let payload = serde_json::json!({
+            "iss": config.config_url, "sub": "repo:example/project:ref:refs/heads/main",
+            "aud": [config.client_id, "additional-audience"], "iat": now, "exp": now + 300,
+        });
+        let signed =
+            jsonwebtoken::encode(&Header::new(Algorithm::HS256), &payload, &EncodingKey::from_secret(secret.as_bytes())).unwrap();
+        let token: CoreIdToken = signed.parse().unwrap();
+        for workload in [false, true] {
+            for (algorithms, accepted) in [
+                (serde_json::json!(["HS256", "unsupported-future-algorithm"]), true),
+                (serde_json::json!(["ES256"]), false),
+                (serde_json::json!(["unsupported-future-algorithm"]), false),
+            ] {
+                let mut document = serde_json::json!({
+                    "issuer": config.config_url, "jwks_uri": "https://issuer.example.com/jwks",
+                    "response_types_supported": ["id_token"], "subject_types_supported": ["public"],
+                    "id_token_signing_alg_values_supported": algorithms,
+                });
+                if !workload {
+                    document["authorization_endpoint"] = serde_json::json!("https://issuer.example.com/authorize");
+                }
+                let metadata = DiscoveredProviderMetadata::parse(&serde_json::to_vec(&document).unwrap(), true).unwrap();
+                let verifier = metadata
+                    .verifier(&config)
+                    .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
+                assert_eq!(
+                    token.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
+                    accepted,
+                    "workload={workload}, algorithms={algorithms}"
+                );
+                if accepted {
+                    assert!(
+                        token.claims(&metadata.verifier(&config), |_: Option<&Nonce>| Ok(())).is_err(),
+                        "additional audiences require explicit trust"
+                    );
+                    let mut wrong_secret = config.clone();
+                    wrong_secret.client_secret = Some(Nonce::new_random().secret().to_string());
+                    let verifier = metadata
+                        .verifier(&wrong_secret)
+                        .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
+                    assert!(
+                        token.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_err(),
+                        "incorrect client secret must fail"
+                    );
+                }
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn workload_discovery_stops_after_forbidden_jwks() {
+        let requests = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let seen = Arc::clone(&requests);
+        let (base, handle) = start_mock_oidc_discovery_server_with_jwks(
+            |base| (base.to_string(), "http://192.168.65.254:8080/jwks".into(), "/jwks".into()),
+            2,
+            "ES256",
+            true,
+            move |_| {
+                seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                serde_json::json!({"keys": []}).to_string()
+            },
+        )
+        .expect("workload discovery mock must bind");
+        let mut config = build_mocked_oidc_provider_config("workload", &base);
+        config.hide_from_ui = true;
+        let client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).unwrap());
+        let error = OidcSys::discover_provider(&config, &client)
+            .await
+            .err()
+            .expect("private JWKS must be blocked");
+        assert!(error.starts_with(OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY), "{error}");
+        handle.join().unwrap();
+        assert_eq!(
+            requests.load(std::sync::atomic::Ordering::SeqCst),
+            1,
+            "a policy denial must not retry discovery"
+        );
+    }
+
+    #[tokio::test]
+    async fn workload_config_validation_reports_absent_console_endpoints() {
+        let (base, handle) = start_mock_oidc_discovery_server_with_jwks(
+            |base| (base.to_string(), format!("{base}/jwks"), "/jwks".into()),
+            2,
+            "ES256",
+            true,
+            |_| serde_json::json!({"keys": []}).to_string(),
+        )
+        .expect("workload discovery mock must bind");
+        let mut config = build_mocked_oidc_provider_config("workload", &base);
+        config.hide_from_ui = true;
+        // Inferred issuers keep the library's discovery URL construction.
+        config.config_url = format!("{base}/.well-known/openid-configuration?ignored=1#ignored");
+        let result = validate_mocked_oidc_provider_config(&config)
+            .await
+            .expect("hidden workload configuration should validate");
+        assert_eq!(result.issuer, base);
+        assert!(result.authorization_endpoint.is_none());
+        assert!(result.token_endpoint.is_none());
+        handle.join().unwrap();
+    }
+
+    #[tokio::test]
+    async fn workload_discovery_verification_and_rotation() {
+        for explicit_issuer in [false, true] {
+            let (_, initial_jwk) = oidc_es256_key_and_jwk("initial");
+            let (key, rotated_jwk) = oidc_es256_key_and_jwk("rotated");
+            let initial_jwks = serde_json::json!({"keys": [initial_jwk]}).to_string();
+            let rotated_jwks = serde_json::json!({"keys": [rotated_jwk]}).to_string();
+            let (base, handle) = start_mock_oidc_discovery_server_with_jwks(
+                |base| (base.to_string(), format!("{base}/jwks"), "/jwks".into()),
+                4,
+                "ES256",
+                true,
+                move |fetch| {
+                    if fetch == 0 {
+                        initial_jwks.clone()
+                    } else {
+                        rotated_jwks.clone()
+                    }
+                },
+            )
+            .expect("workload discovery mock must bind");
+            let mut config = build_mocked_oidc_provider_config("workload", &base);
+            config.hide_from_ui = true;
+            if explicit_issuer {
+                config.issuer = Some(base.clone());
+            }
+            let http_client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).unwrap());
+            let state = OidcSys::discover_provider(&config, &http_client)
+                .await
+                .expect("workload discovery must succeed");
+            assert!(state.metadata.authorization_endpoint().is_none());
+            let sys = OidcSys {
+                configs: HashMap::from([(config.id.clone(), config.clone())]),
+                provider_states: RwLock::new(HashMap::from([(config.id.clone(), state)])),
+                state_store: OidcStateStore::new(),
+                http_client,
+            };
+            assert!(sys.has_providers());
+            assert!(sys.list_visible_providers().is_empty());
+            let error = sys
+                .authorize_url(&config.id, "https://console.example.com/callback", None)
+                .await
+                .unwrap_err();
+            assert!(error.contains("only web identity"));
+            let now = time::OffsetDateTime::now_utc().unix_timestamp();
+            let payload = serde_json::json!({
+                "iss": base, "sub": "system:serviceaccount:default:reader", "aud": [config.client_id],
+                "iat": now, "exp": now + 300, "groups": ["readonly"],
+                "kubernetes.io": {"namespace": "default", "serviceaccount": {"name": "reader"}},
+            });
+            let mut header = Header::new(Algorithm::ES256);
+            header.kid = Some("rotated".into());
+            let token = jsonwebtoken::encode(&header, &payload, &key).unwrap();
+            let (claims, provider) = sys
+                .verify_web_identity_token(&token)
+                .await
+                .expect("rotation must retain workload discovery support");
+            assert_eq!(provider, config.id);
+            assert_eq!(claims.sub, "system:serviceaccount:default:reader");
+            assert_eq!(claims.groups, ["readonly"]);
+            // Repeat after the mock exits: the verified snapshot must be cached.
+            handle.join().unwrap();
+            assert!(sys.verify_web_identity_token(&token).await.is_ok());
+            for (field, value, expected) in [
+                ("iss", serde_json::json!("https://wrong.example.com"), "issuer"),
+                ("aud", serde_json::json!("wrong-audience"), "audience"),
+                ("exp", serde_json::json!(now - 60), "expired"),
+            ] {
+                let mut invalid = payload.clone();
+                invalid[field] = value;
+                let token = jsonwebtoken::encode(&header, &invalid, &key).unwrap();
+                let error = sys
+                    .verify_web_identity_token(&token)
+                    .await
+                    .expect_err("invalid workload token must fail");
+                assert!(error.to_lowercase().contains(expected), "{field}: {error}");
+            }
+            let (wrong_key, _) = oidc_es256_key_and_jwk("wrong");
+            let invalid = jsonwebtoken::encode(&header, &payload, &wrong_key).unwrap();
+            let error = sys
+                .verify_web_identity_token(&invalid)
+                .await
+                .expect_err("wrong signature must fail");
+            assert!(error.to_lowercase().contains("signature"), "{error}");
+            assert!(
+                sys.verify_web_identity_token(&token).await.is_ok(),
+                "failed refresh must preserve the cached keys"
+            );
+        }
+    }
+
     fn start_mock_oidc_tls_discovery_server<F>(
         build_discovery_issuer: F,
         max_requests: usize,
@@ -2958,7 +3558,7 @@ mod tests {
 
         Ok(OidcProviderValidationResult {
             issuer: state.metadata.issuer().to_string(),
-            authorization_endpoint: state.metadata.authorization_endpoint().to_string(),
+            authorization_endpoint: state.metadata.authorization_endpoint(),
             token_endpoint: state.metadata.token_endpoint().map(ToString::to_string),
         })
     }
@@ -3558,7 +4158,7 @@ mod tests {
             provider_states: RwLock::new(HashMap::from([(
                 provider_id.to_string(),
                 ProviderState {
-                    metadata,
+                    metadata: DiscoveredProviderMetadata::Console(Box::new(metadata)),
                     discovered_at: Instant::now(),
                 },
             )])),

crates/scanner/src/scanner.rs

@@ -70,12 +70,13 @@ use crate::storage_api::scan::{
     SCANNER_ACTIVITY_PROTOCOL_VERSION,
 };
 use crate::{
-    ECStore, EcstoreError, RUSTFS_META_BUCKET, SCANNER_PUBLICATION_EPOCH_CHANGED, ScannerLifecycleConfigExt as _,
-    ScannerReplicationConfigExt as _, delete_config_with_publication_admission_for_epoch, get_lifecycle_config,
-    get_replication_config, invalidate_admin_data_usage_snapshot_cache, invalidate_data_usage_snapshot_cache, read_config,
-    replace_bucket_usage_memory_from_info, save_config, save_config_shared_with_preconditions_and_lease_fence_and_scope,
-    save_config_with_preconditions, save_config_with_publication_admission_for_epoch, scanner_publication_admission_for_epoch,
-    scanner_publication_epoch, scanner_publication_epoch_changed,
+    DiskError, ECStore, EcstoreError, ListPathRawOptions, RUSTFS_META_BUCKET, SCANNER_PUBLICATION_EPOCH_CHANGED,
+    ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _, delete_config_with_publication_admission_for_epoch,
+    get_lifecycle_config, get_replication_config, invalidate_admin_data_usage_snapshot_cache,
+    invalidate_data_usage_snapshot_cache, list_path_raw, read_config, replace_bucket_usage_memory_from_info, save_config,
+    save_config_shared_with_preconditions_and_lease_fence_and_scope, save_config_with_preconditions,
+    save_config_with_publication_admission_for_epoch, scanner_publication_admission_for_epoch, scanner_publication_epoch,
+    scanner_publication_epoch_changed,
 };
 
 const LOG_COMPONENT_SCANNER: &str = "scanner";
@@ -947,6 +948,22 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
     init_data_scanner_with_storage(ctx, storeapi).await;
 }
 
+async fn run_scanner_usage_recovery_intents_for_startup(
+    ctx: CancellationToken,
+    storeapi: Arc<ECStore>,
+) -> Result<usize, ScannerError> {
+    let intent_ids = scanner_usage_recovery_intents_for_startup(&ctx, storeapi.clone()).await?;
+    let mut attempted = 0usize;
+    for intent_id in intent_ids {
+        if ctx.is_cancelled() {
+            break;
+        }
+        run_scanner_usage_recovery_intent(ctx.child_token(), storeapi.clone(), intent_id).await?;
+        attempted = attempted.saturating_add(1);
+    }
+    Ok(attempted)
+}
+
 /// Start normal scanning when enabled, or one resume-only cleanup attempt.
 /// The disabled branch returns a finite task for the startup owner to join;
 /// it never enables ordinary namespace scanning or accepts a new reset intent.
@@ -956,10 +973,32 @@ pub async fn init_scanner_with_recovery(
     enabled: bool,
 ) -> Option<tokio::task::JoinHandle<()>> {
     if enabled {
+        if let Err(error) = run_scanner_usage_recovery_intents_for_startup(ctx.clone(), storeapi.clone()).await {
+            warn!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                state = "recovery_intent_startup_discovery_failed",
+                error = %error,
+                "Scanner recovery intent startup discovery failed"
+            );
+        }
         init_data_scanner(ctx, storeapi).await;
         return None;
     }
     Some(tokio::spawn(async move {
+        if let Err(error) = run_scanner_usage_recovery_intents_for_startup(ctx.clone(), storeapi.clone()).await {
+            warn!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                state = "recovery_intent_startup_discovery_failed",
+                error = %error,
+                "Disabled scanner recovery intent startup discovery failed"
+            );
+        }
         if let Err(error) = resume_scanner_cycle_cleanup(ctx, storeapi).await {
             warn!(
                 target: "rustfs::scanner",
@@ -2216,11 +2255,14 @@ where
         false
     } else if let Some(notification_system) = storeapi.scanner_notification_system() {
         let acknowledgement_count = remote_dirty_usage_acknowledgements.len();
+        let acknowledgement_proof = remote_dirty_usage_acknowledgements.clone();
         let acknowledgements = remote_dirty_usage_acknowledgements.into_iter().map(Into::into).collect();
         remote_dirty_usage_acknowledgement_pending(
             cycle_info.current,
             acknowledgement_count,
+            &acknowledgement_proof,
             notification_system.acknowledge_scanner_dirty_usage(acknowledgements),
+            || probe_scanner_activity(storeapi.as_ref(), true),
         )
         .await
     } else {

crates/scanner/src/scanner/activity.rs

@@ -15,6 +15,7 @@
 use super::*;
 use crate::storage_api::ScannerStorage;
 use crate::storage_api::scan::SCANNER_ACTIVITY_V6_PROTOCOL_VERSION;
+use std::collections::HashSet;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(super) enum ScannerCycleWakeReason {
@@ -51,18 +52,25 @@ pub(crate) fn scanner_cycle_outcome_with_pending_maintenance(
     }
 }
 
-pub(super) async fn remote_dirty_usage_acknowledgement_pending<F, E>(
+pub(super) async fn remote_dirty_usage_acknowledgement_pending<F, E, C, CF>(
     cycle: u64,
     acknowledgement_count: usize,
+    acknowledgements: &[ScannerDirtyUsageAcknowledgement],
     acknowledgement: F,
+    confirm_after_error: C,
 ) -> bool
 where
     F: Future<Output = Result<bool, E>>,
     E: std::fmt::Display,
+    C: FnOnce() -> CF,
+    CF: Future<Output = Result<ScannerActivitySnapshot, String>>,
 {
     match acknowledgement.await {
         Ok(dirty_usage_pending) => dirty_usage_pending,
         Err(err) => {
+            if remote_dirty_usage_acknowledgement_loss_reconciled(acknowledgements, confirm_after_error().await) {
+                return false;
+            }
             warn!(
                 target: "rustfs::scanner",
                 event = EVENT_SCANNER_PERSIST_STATE,
@@ -79,6 +87,29 @@ where
     }
 }
 
+pub(super) fn remote_dirty_usage_acknowledgement_loss_reconciled(
+    acknowledgements: &[ScannerDirtyUsageAcknowledgement],
+    activity_after_error: Result<ScannerActivitySnapshot, String>,
+) -> bool {
+    if acknowledgements.is_empty() {
+        return false;
+    }
+    let Ok(activity_after_error) = activity_after_error else {
+        return false;
+    };
+    if !scanner_activity_allows_usage_publication(&activity_after_error) {
+        return false;
+    }
+    let mut acknowledged_hosts = HashSet::with_capacity(acknowledgements.len());
+    acknowledgements.iter().all(|acknowledgement| {
+        if !acknowledged_hosts.insert(acknowledgement.host.as_str()) {
+            return false;
+        }
+        scanner_activity_dirty_usage_state_for_host(&activity_after_error, &acknowledgement.host)
+            .is_some_and(|(instance_id, _generation, pending)| instance_id == acknowledgement.instance_id && !pending)
+    })
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(super) struct ScannerCleanIdleBackoff {
     pub(super) interval_multiplier: u32,

crates/scanner/src/scanner/cycle_state.rs

@@ -18,6 +18,8 @@ use crate::data_usage_define::{
     DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_RECOVERY_PATH, usage_floor_primary_read_error_allows_backup,
 };
 use crate::storage_api::owner::ObjectIO as _;
+use rustfs_filemeta::MetaCacheEntry;
+use std::collections::BTreeSet;
 use std::sync::atomic::AtomicU64;
 use tokio::io::AsyncReadExt as _;
 
@@ -42,6 +44,9 @@ const SCANNER_RECOVERY_INTENT_STATE_RUNNING: &str = "running";
 const SCANNER_RECOVERY_INTENT_STATE_COMPLETED: &str = "completed";
 const SCANNER_RECOVERY_INTENT_STATE_FAILED: &str = "failed";
 pub const SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD: &str = "scanner-usage-full-rebuild";
+const MAX_SCANNER_RECOVERY_INTENT_STARTUP_CANDIDATES: usize = 4096;
+const SCANNER_RECOVERY_INTENT_STARTUP_PAGE_SIZE: usize = 128;
+const MAX_SCANNER_RECOVERY_INTENT_STARTUP_REPLAY: usize = 64;
 
 #[cfg(test)]
 pub(super) mod cleanup_io_fault {
@@ -465,6 +470,14 @@ fn scanner_recovery_intent_path(intent_id: &str) -> Result<String, ScannerError>
     Ok(format!("{SCANNER_RECOVERY_INTENT_PREFIX}/{intent_id}.json"))
 }
 
+fn scanner_recovery_intent_id_from_entry_name(entry_name: &str) -> Option<String> {
+    let relative = entry_name
+        .strip_prefix(SCANNER_RECOVERY_INTENT_PREFIX)
+        .and_then(|name| name.strip_prefix('/'))?;
+    let intent_id = relative.strip_suffix(".json")?;
+    (!intent_id.contains('/') && is_canonical_sha256(intent_id)).then(|| intent_id.to_string())
+}
+
 pub fn scanner_recovery_actor_sha256(actor: &str) -> String {
     sha256_hex(&[b"scanner-recovery-actor-v1", actor.as_bytes()])
 }
@@ -623,6 +636,117 @@ pub async fn get_scanner_usage_recovery_intent(
     read_recovery_intent_record(storeapi, &path).await
 }
 
+fn scanner_recovery_intent_is_replayable(record: &ScannerRecoveryIntentRecord) -> bool {
+    matches!(
+        record.state.as_str(),
+        SCANNER_RECOVERY_INTENT_STATE_ACCEPTED | SCANNER_RECOVERY_INTENT_STATE_RUNNING
+    )
+}
+
+pub(super) async fn scanner_usage_recovery_intents_for_startup(
+    ctx: &CancellationToken,
+    storeapi: Arc<ECStore>,
+) -> Result<Vec<String>, ScannerError> {
+    let discovered = Arc::new(StdMutex::new(BTreeSet::<String>::new()));
+    for set in storeapi.all_set_disks() {
+        if ctx.is_cancelled() {
+            break;
+        }
+        let disks = set.get_local_disks().await;
+        if disks.is_empty() {
+            continue;
+        }
+        let read_quorum = disks.len().saturating_sub(set.default_parity_count).clamp(1, disks.len());
+        let mut forward_to = None;
+        loop {
+            let page_entries = Arc::new(StdMutex::new(Vec::<String>::new()));
+            let page_entries_for_set = page_entries.clone();
+            let list_result = list_path_raw(
+                ctx.child_token(),
+                ListPathRawOptions {
+                    disks: disks.clone(),
+                    bucket: RUSTFS_META_BUCKET.to_string(),
+                    path: SCANNER_RECOVERY_INTENT_PREFIX.to_string(),
+                    recursive: true,
+                    skip_hidden_prefix_check: true,
+                    forward_to: forward_to.clone(),
+                    min_disks: read_quorum,
+                    report_not_found: true,
+                    per_disk_limit: i32::try_from(SCANNER_RECOVERY_INTENT_STARTUP_PAGE_SIZE).unwrap_or(i32::MAX),
+                    agreed: Some(Box::new(move |entry: MetaCacheEntry| {
+                        let page_entries = page_entries_for_set.clone();
+                        Box::pin(async move {
+                            if scanner_recovery_intent_id_from_entry_name(&entry.name).is_some() {
+                                page_entries
+                                    .lock()
+                                    .unwrap_or_else(|poisoned| poisoned.into_inner())
+                                    .push(entry.name);
+                            }
+                        })
+                    })),
+                    ..Default::default()
+                },
+            )
+            .await;
+            match list_result {
+                Ok(()) => {}
+                Err(DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound) => break,
+                Err(err) => {
+                    return Err(ScannerError::Other(format!("failed to list scanner recovery intents for startup: {err}")));
+                }
+            }
+
+            let page = {
+                let mut guard = page_entries.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+                guard.sort();
+                guard.dedup();
+                std::mem::take(&mut *guard)
+            };
+            if page.is_empty() {
+                break;
+            }
+
+            {
+                let mut all = discovered.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+                for entry_name in &page {
+                    if let Some(intent_id) = scanner_recovery_intent_id_from_entry_name(entry_name) {
+                        all.insert(intent_id);
+                    }
+                    if all.len() >= MAX_SCANNER_RECOVERY_INTENT_STARTUP_CANDIDATES {
+                        break;
+                    }
+                }
+                if all.len() >= MAX_SCANNER_RECOVERY_INTENT_STARTUP_CANDIDATES {
+                    break;
+                }
+            }
+
+            if page.len() < SCANNER_RECOVERY_INTENT_STARTUP_PAGE_SIZE {
+                break;
+            }
+            forward_to = page.last().cloned();
+        }
+    }
+
+    let intent_ids = {
+        let guard = discovered.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        guard.iter().cloned().collect::<Vec<_>>()
+    };
+    let mut replayable = Vec::new();
+    for intent_id in intent_ids {
+        let Some(record) = get_scanner_usage_recovery_intent(storeapi.clone(), &intent_id).await? else {
+            continue;
+        };
+        if scanner_recovery_intent_is_replayable(&record) {
+            replayable.push(record.intent_id);
+            if replayable.len() >= MAX_SCANNER_RECOVERY_INTENT_STARTUP_REPLAY {
+                break;
+            }
+        }
+    }
+    Ok(replayable)
+}
+
 pub async fn accept_scanner_usage_recovery_intent(
     storeapi: Arc<impl ScannerObjectIO>,
     request: ScannerRecoveryIntentRequest,

crates/scanner/src/scanner/tests.rs

@@ -7498,13 +7498,28 @@ fn finalizing_post_scan_observation_advances_partially_without_dirty_ack() {
 
 #[tokio::test]
 async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
-    let pending = remote_dirty_usage_acknowledgement_pending(7, 1, std::future::ready(Ok::<bool, std::io::Error>(true))).await;
+    let acknowledgements = Vec::new();
+    let pending = remote_dirty_usage_acknowledgement_pending(
+        7,
+        1,
+        &acknowledgements,
+        std::future::ready(Ok::<bool, std::io::Error>(true)),
+        || async { Ok(BTreeMap::new()) },
+    )
+    .await;
     assert_eq!(
         scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, pending),
         ScannerCycleOutcome::CompletedWithPendingMaintenance
     );
 
-    let cleared = remote_dirty_usage_acknowledgement_pending(7, 1, std::future::ready(Ok::<bool, std::io::Error>(false))).await;
+    let cleared = remote_dirty_usage_acknowledgement_pending(
+        7,
+        1,
+        &acknowledgements,
+        std::future::ready(Ok::<bool, std::io::Error>(false)),
+        || async { Ok(BTreeMap::new()) },
+    )
+    .await;
     assert_eq!(
         scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, cleared),
         ScannerCycleOutcome::Completed
@@ -7513,7 +7528,9 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
     let failed = remote_dirty_usage_acknowledgement_pending(
         7,
         1,
+        &acknowledgements,
         std::future::ready(Err::<bool, _>(std::io::Error::other("injected acknowledgement failure"))),
+        || async { Err("confirmation probe failed".to_string()) },
     )
     .await;
     assert_eq!(
@@ -7522,6 +7539,76 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
     );
 }
 
+#[tokio::test]
+async fn scanner_cycle_confirms_lost_remote_ack_from_activity_snapshot() {
+    let acknowledgement = ScannerDirtyUsageAcknowledgement {
+        host: "node-2".to_string(),
+        instance_id: "epoch-a".to_string(),
+        kind: ScannerDirtyUsageAcknowledgementKind::Generation(5),
+    };
+    let cleared_activity = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
+    let response_lost = remote_dirty_usage_acknowledgement_pending(
+        8,
+        1,
+        std::slice::from_ref(&acknowledgement),
+        std::future::ready(Err::<bool, _>(std::io::Error::other("response lost after peer ack"))),
+        || async { Ok(cleared_activity) },
+    )
+    .await;
+    assert_eq!(
+        scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, response_lost),
+        ScannerCycleOutcome::Completed,
+        "a same-instance activity confirmation with no dirty work closes the uncertain ACK"
+    );
+
+    let duplicate_acknowledgements = vec![acknowledgement.clone(), acknowledgement.clone()];
+    let duplicate_target = remote_dirty_usage_acknowledgement_pending(
+        8,
+        duplicate_acknowledgements.len(),
+        &duplicate_acknowledgements,
+        std::future::ready(Err::<bool, _>(std::io::Error::other("duplicate target rejected before peer ack"))),
+        || async { Ok(BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))])) },
+    )
+    .await;
+    assert_eq!(
+        scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, duplicate_target),
+        ScannerCycleOutcome::CompletedWithPendingMaintenance,
+        "a request rejected before peer delivery cannot be recovered by a clean activity snapshot"
+    );
+
+    let restarted_activity = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-b", 7, 3))]);
+    let peer_restarted = remote_dirty_usage_acknowledgement_pending(
+        8,
+        1,
+        std::slice::from_ref(&acknowledgement),
+        std::future::ready(Err::<bool, _>(std::io::Error::other("response lost before restart was observed"))),
+        || async { Ok(restarted_activity) },
+    )
+    .await;
+    assert_eq!(
+        scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, peer_restarted),
+        ScannerCycleOutcome::CompletedWithPendingMaintenance,
+        "a new peer instance cannot confirm whether the old ACK reached durable dirty state"
+    );
+
+    let mut written_activity = scanner_node_activity("epoch-a", 7, 3);
+    written_activity.dirty_usage_generation = 6;
+    written_activity.dirty_usage_pending = true;
+    let concurrent_write = remote_dirty_usage_acknowledgement_pending(
+        8,
+        1,
+        &[acknowledgement],
+        std::future::ready(Err::<bool, _>(std::io::Error::other("response lost before concurrent write"))),
+        || async { Ok(BTreeMap::from([("node-2".to_string(), written_activity)])) },
+    )
+    .await;
+    assert_eq!(
+        scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, concurrent_write),
+        ScannerCycleOutcome::CompletedWithPendingMaintenance,
+        "new dirty usage on the same peer must keep maintenance pending"
+    );
+}
+
 #[test]
 #[serial]
 fn finalizing_an_already_durable_enum_without_proof_keeps_dirty_pending() {

crates/scanner/src/scanner/tests/recovery_control.rs

@@ -106,6 +106,20 @@ fn recovery_intent_request(key: &str, actor: &str) -> ScannerRecoveryIntentReque
     }
 }
 
+fn synthetic_recovery_intent_record(intent_id: String, state: &str) -> ScannerRecoveryIntentRecord {
+    ScannerRecoveryIntentRecord {
+        schema_version: 1,
+        intent_id,
+        action: SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD.to_string(),
+        mode: "full-rebuild".to_string(),
+        state: state.to_string(),
+        actor_sha256: "a".repeat(64),
+        idempotency_key_sha256: "b".repeat(64),
+        request_sha256: "c".repeat(64),
+        accepted_at_unix_secs: 1,
+    }
+}
+
 async fn assert_reset_fences(store: &Arc<ECStore>) {
     let data = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
         .await
@@ -311,6 +325,162 @@ async fn scanner_recovery_intent_executor_persists_failed_progress() {
     assert_eq!(failed.intent_id, record.intent_id);
 }
 
+#[tokio::test]
+#[serial]
+async fn scanner_recovery_intent_startup_discovers_only_non_terminal_records() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    let accepted = match accept_scanner_usage_recovery_intent(
+        store.clone(),
+        recovery_intent_request("intent-key-0001-startup-accepted", "operator-a"),
+    )
+    .await
+    .expect("accepted startup intent should persist")
+    {
+        ScannerRecoveryIntentAcceptResult::Accepted { record } => record,
+        other => panic!("first request must create accepted startup intent: {other:?}"),
+    };
+    let mut running = match accept_scanner_usage_recovery_intent(
+        store.clone(),
+        recovery_intent_request("intent-key-0001-startup-running", "operator-a"),
+    )
+    .await
+    .expect("running startup intent should persist")
+    {
+        ScannerRecoveryIntentAcceptResult::Accepted { record } => record,
+        other => panic!("first request must create running startup intent: {other:?}"),
+    };
+    running.state = "running".to_string();
+    let running_path = format!(".usage.v2.recovery-intents/{}.json", running.intent_id);
+    save_config(
+        store.clone(),
+        &running_path,
+        serde_json::to_vec(&running).expect("running intent should encode"),
+    )
+    .await
+    .expect("running intent override should persist");
+
+    for (key, state) in [
+        ("intent-key-0001-startup-completed", "completed"),
+        ("intent-key-0001-startup-failed", "failed"),
+    ] {
+        let mut terminal = match accept_scanner_usage_recovery_intent(store.clone(), recovery_intent_request(key, "operator-a"))
+            .await
+            .expect("terminal startup intent should persist")
+        {
+            ScannerRecoveryIntentAcceptResult::Accepted { record } => record,
+            other => panic!("first request must create terminal startup intent: {other:?}"),
+        };
+        terminal.state = state.to_string();
+        let path = format!(".usage.v2.recovery-intents/{}.json", terminal.intent_id);
+        save_config(
+            store.clone(),
+            &path,
+            serde_json::to_vec(&terminal).expect("terminal intent should encode"),
+        )
+        .await
+        .expect("terminal intent override should persist");
+    }
+    save_config(store.clone(), ".usage.v2.recovery-intents/not-a-sha.json", b"{}".to_vec())
+        .await
+        .expect("foreign startup key should persist");
+
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    let replayable = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted)
+        .await
+        .expect("startup discovery should tolerate terminal and foreign records");
+    let replayable = replayable.into_iter().collect::<std::collections::BTreeSet<_>>();
+    assert_eq!(
+        replayable,
+        [accepted.intent_id, running.intent_id].into_iter().collect(),
+        "startup discovery must only re-drive accepted/running durable intents"
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_recovery_intent_startup_pages_past_terminal_records() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    for index in 0..70 {
+        let intent_id = format!("{index:064x}");
+        let path = format!(".usage.v2.recovery-intents/{intent_id}.json");
+        let terminal = synthetic_recovery_intent_record(intent_id, "completed");
+        save_config(
+            store.clone(),
+            &path,
+            serde_json::to_vec(&terminal).expect("terminal paging fixture should encode"),
+        )
+        .await
+        .expect("terminal paging fixture should persist");
+    }
+    let pending_id = "f".repeat(64);
+    let pending = synthetic_recovery_intent_record(pending_id.clone(), "accepted");
+    save_config(
+        store.clone(),
+        &format!(".usage.v2.recovery-intents/{pending_id}.json"),
+        serde_json::to_vec(&pending).expect("pending paging fixture should encode"),
+    )
+    .await
+    .expect("pending paging fixture should persist");
+
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    let replayable = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted)
+        .await
+        .expect("startup discovery should page past terminal records");
+    assert_eq!(replayable, vec![pending_id]);
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_recovery_intent_startup_rejects_corrupt_pending_record() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    let record = match accept_scanner_usage_recovery_intent(
+        store.clone(),
+        recovery_intent_request("intent-key-0001-startup-corrupt", "operator-a"),
+    )
+    .await
+    .expect("corrupt startup fixture intent should persist")
+    {
+        ScannerRecoveryIntentAcceptResult::Accepted { record } => record,
+        other => panic!("first request must create corrupt startup fixture: {other:?}"),
+    };
+    let path = format!(".usage.v2.recovery-intents/{}.json", record.intent_id);
+    save_config(store.clone(), &path, b"{corrupt".to_vec())
+        .await
+        .expect("corrupt intent payload should persist");
+
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    let error = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted)
+        .await
+        .expect_err("startup discovery must not silently drop corrupt pending intent records");
+    assert!(error.to_string().contains("scanner recovery intent is invalid"), "{error}");
+}
+
+#[tokio::test]
+#[serial]
+async fn scanner_recovery_intent_disabled_startup_executes_persisted_non_terminal_intent() {
+    let (_dir, store) = setup_scanner_cycle_store().await;
+    let record = match accept_scanner_usage_recovery_intent(
+        store.clone(),
+        recovery_intent_request("intent-key-0001-startup-exec", "operator-a"),
+    )
+    .await
+    .expect("startup execution intent should persist")
+    {
+        ScannerRecoveryIntentAcceptResult::Accepted { record } => record,
+        other => panic!("first request must create startup execution intent: {other:?}"),
+    };
+
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    run_disabled_startup(CancellationToken::new(), restarted.clone()).await;
+
+    let completed = get_scanner_usage_recovery_intent(restarted, &record.intent_id)
+        .await
+        .expect("startup-executed intent should read")
+        .expect("startup-executed intent should remain durable");
+    assert_eq!(completed.state, "completed");
+    assert_eq!(completed.intent_id, record.intent_id);
+}
+
 #[tokio::test]
 #[serial]
 async fn scanner_recovery_intent_rejects_same_namespace_conflict() {

crates/scanner/src/scanner_folder.rs

@@ -21,8 +21,8 @@ use std::time::{Duration, Instant, SystemTime};
 use crate::ReplTargetSizeSummary;
 use crate::data_usage_define::{
     DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap,
-    DataUsageScanCheckpoint, DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt,
-    SizeReconciliationEntry, SizeSummary, hash_path,
+    DataUsageRawEnumerationCursor, DataUsageScanCheckpoint, DataUsageScanCheckpointReason, PendingScannerHeal,
+    PendingScannerHealKind, ScannerSizeSummaryExt, SizeReconciliationEntry, SizeSummary, hash_path,
 };
 use crate::error::ScannerError;
 use crate::runtime_config::{
@@ -55,6 +55,7 @@ use rustfs_scanner_metrics::metrics::{
     UpdateCurrentPathFn, current_path_updater, global_metrics,
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
+use sha2::{Digest, Sha256};
 use time::OffsetDateTime;
 use tokio::select;
 use tokio::sync::mpsc;
@@ -733,6 +734,7 @@ pub struct FolderScanner {
     coverage_frontier: Option<String>,
     resume_frontier: Option<String>,
     coverage_gap: bool,
+    raw_enumeration_progress: Vec<RawEnumerationProgress>,
     pending_heal_sync_deferred: bool,
     pending_heal_batch_dirty: bool,
     #[cfg(test)]
@@ -744,6 +746,50 @@ pub struct FolderScanner {
     list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
 }
 
+struct RawEnumerationProgress {
+    parent: String,
+    last_entry: Option<String>,
+    entries_seen: u64,
+    digest: Sha256,
+}
+
+impl RawEnumerationProgress {
+    fn new(parent: &str) -> Self {
+        let mut digest = Sha256::new();
+        update_raw_enumeration_digest(&mut digest, b"parent", parent.as_bytes());
+        Self {
+            parent: parent.to_string(),
+            last_entry: None,
+            entries_seen: 0,
+            digest,
+        }
+    }
+
+    fn record_entry(&mut self, entry: &str) {
+        update_raw_enumeration_digest(&mut self.digest, b"entry", entry.as_bytes());
+        self.last_entry = Some(entry.to_string());
+        self.entries_seen = self.entries_seen.saturating_add(1);
+    }
+
+    fn into_cursor(self) -> Option<DataUsageRawEnumerationCursor> {
+        if self.entries_seen == 0 {
+            return None;
+        }
+        Some(DataUsageRawEnumerationCursor::new(
+            self.parent,
+            self.last_entry,
+            self.entries_seen,
+            self.digest.finalize().into(),
+        ))
+    }
+}
+
+fn update_raw_enumeration_digest(digest: &mut Sha256, label: &[u8], value: &[u8]) {
+    digest.update(label);
+    digest.update(u64::try_from(value.len()).unwrap_or(u64::MAX).to_le_bytes());
+    digest.update(value);
+}
+
 fn size_reconciliation_entry_bytes(entry: &SizeReconciliationEntry) -> usize {
     entry.key.len()
         + entry.bucket.len()
@@ -999,6 +1045,41 @@ impl FolderScanner {
         }
     }
 
+    fn record_raw_enumeration_entry(&mut self, parent: &str, entry: &str) {
+        if self.old_cache.info.scan_progress.is_none() {
+            return;
+        }
+        if let Some(position) = self
+            .raw_enumeration_progress
+            .iter()
+            .position(|progress| progress.parent == parent)
+        {
+            self.raw_enumeration_progress.truncate(position + 1);
+        } else {
+            self.raw_enumeration_progress.push(RawEnumerationProgress::new(parent));
+        }
+        if let Some(progress) = self.raw_enumeration_progress.last_mut() {
+            progress.record_entry(entry);
+        }
+    }
+
+    fn finish_raw_enumeration_parent(&mut self, parent: &str) {
+        self.raw_enumeration_progress.retain(|progress| {
+            progress.parent != parent
+                && !progress
+                    .parent
+                    .strip_prefix(parent)
+                    .is_some_and(|suffix| suffix.starts_with(SLASH_SEPARATOR))
+        });
+    }
+
+    fn take_raw_enumeration_cursor(&mut self) -> Option<DataUsageRawEnumerationCursor> {
+        self.raw_enumeration_progress
+            .drain(..)
+            .next()
+            .and_then(RawEnumerationProgress::into_cursor)
+    }
+
     fn carry_forward_old_children(&mut self, parent_hash: &DataUsageHash, entry: &mut DataUsageEntry) {
         if entry.compacted {
             // Compacted entries store child totals directly; child links would be flattened twice.
@@ -1329,11 +1410,15 @@ impl FolderScanner {
             };
             let mut pending_entry_progress = 0_u64;
             let mut last_entry_progress = Instant::now();
+            let mut raw_enumeration_complete = false;
 
             loop {
                 let entry = match dir_reader.next_entry().await {
                     Ok(Some(entry)) => entry,
-                    Ok(None) => break,
+                    Ok(None) => {
+                        raw_enumeration_complete = true;
+                        break;
+                    }
                     Err(e) if e.kind() == ErrorKind::NotFound => {
                         debug!(
                             target: "rustfs::scanner::folder",
@@ -1345,6 +1430,7 @@ impl FolderScanner {
                             error = %e,
                             "Scanner folder state updated"
                         );
+                        raw_enumeration_complete = true;
                         break;
                     }
                     Err(e) if e.kind() == ErrorKind::NotADirectory => {
@@ -1358,6 +1444,7 @@ impl FolderScanner {
                             error = %e,
                             "Scanner folder state updated"
                         );
+                        raw_enumeration_complete = true;
                         break;
                     }
                     Err(e) => return Err(ScannerError::Io(e)),
@@ -1376,6 +1463,7 @@ impl FolderScanner {
                 if file_name.is_empty() || file_name == "." || file_name == ".." {
                     continue;
                 }
+                self.record_raw_enumeration_entry(&folder.name, &file_name);
                 let is_storage_format_entry = file_name == STORAGE_FORMAT_FILE;
 
                 let file_path = entry.path().to_string_lossy().to_string();
@@ -1686,6 +1774,9 @@ impl FolderScanner {
                 }
             }
             self.budget.record_entries_visited(pending_entry_progress);
+            if raw_enumeration_complete {
+                self.finish_raw_enumeration_parent(&folder.name);
+            }
 
             let mut found_erasure_data_directory = false;
             if self.is_erasure_mode && !found_object_metadata {
@@ -2533,6 +2624,7 @@ pub(crate) async fn scan_data_folder_scoped(
         coverage_gap: false,
         pending_heal_sync_deferred: false,
         pending_heal_batch_dirty: false,
+        raw_enumeration_progress: Vec::new(),
         #[cfg(test)]
         pending_heal_sync_count: 0,
         pending_size_reconciliation_keys: HashSet::new(),
@@ -2593,6 +2685,7 @@ pub(crate) async fn scan_data_folder_scoped(
             let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some();
             new_cache.info.scan_resume_after = None;
             new_cache.info.scan_checkpoint = None;
+            new_cache.info.scan_raw_enumeration_cursor = None;
             new_cache.info.scan_coverage_receipt = None;
             if had_scan_checkpoint {
                 global_metrics().record_scanner_checkpoint_cleared();
@@ -2610,6 +2703,9 @@ pub(crate) async fn scan_data_folder_scoped(
                 let root_hash = hash_path(&cache.info.name);
                 let root_has_progress = data_usage_root_has_progress(&root);
                 let pending_heals_changed = scanner.pending_heals_changed;
+                let raw_enumeration_cursor = scanner.take_raw_enumeration_cursor();
+                let carry_forward_cache =
+                    (raw_enumeration_cursor.is_some() && !root_has_progress).then(|| scanner.old_cache.cache.clone());
                 if root_has_progress {
                     scanner.carry_forward_old_children(&root_hash, &mut root);
                 }
@@ -2617,8 +2713,19 @@ pub(crate) async fn scan_data_folder_scoped(
                 let new_cache = scanner.as_mut_new_cache();
                 if root_has_progress {
                     new_cache.replace_hashed(&root_hash, &None, &root);
+                } else if let Some(cache) = carry_forward_cache {
+                    new_cache.cache = cache;
                 }
-                if partial_cache_is_useful(&root, pending_heals_changed) || !new_cache.info.size_reconciliation.is_empty() {
+                if raw_enumeration_cursor.is_some() {
+                    new_cache.info.scan_raw_enumeration_cursor = raw_enumeration_cursor;
+                    new_cache.info.scan_checkpoint = None;
+                    new_cache.info.scan_resume_after = None;
+                    new_cache.info.scan_coverage_receipt = None;
+                }
+                if partial_cache_is_useful(&root, pending_heals_changed)
+                    || new_cache.info.scan_raw_enumeration_cursor.is_some()
+                    || !new_cache.info.size_reconciliation.is_empty()
+                {
                     if new_cache.root().is_some() {
                         new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
                     }

crates/scanner/src/scanner_folder/tests.rs

@@ -353,6 +353,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         coverage_frontier: None,
         resume_frontier: None,
         coverage_gap: false,
+        raw_enumeration_progress: Vec::new(),
         pending_heal_sync_deferred: false,
         pending_heal_batch_dirty: false,
         pending_heal_sync_count: 0,
@@ -2637,6 +2638,93 @@ async fn test_scan_data_folder_returns_partial_cache_on_budget_cancel() {
     assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Directories));
 }
 
+#[tokio::test]
+#[serial]
+async fn test_scan_data_folder_returns_raw_cursor_on_enumeration_cancel_without_root_progress() {
+    let (scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard {
+        temp_dir: Some(temp_dir.clone()),
+    };
+
+    let bucket_dir = temp_dir.join("bucket");
+    tokio::fs::create_dir_all(&bucket_dir)
+        .await
+        .expect("failed to create bucket directory");
+    for entry in ["entry-a", "entry-b", "entry-c"] {
+        tokio::fs::write(bucket_dir.join(entry), b"data")
+            .await
+            .expect("failed to create raw directory entry");
+    }
+
+    let plan = crate::data_usage_define::DataUsageScanPlanDigest([11; 32]);
+    let source = crate::data_usage_define::DataUsageCacheSource::new(1, 0);
+    let identity = crate::data_usage_define::DataUsageScanIdentity {
+        version: 1,
+        bucket_incarnation: Uuid::from_u128(7),
+        set_layout: crate::data_usage_define::DataUsageScanPlanDigest([12; 32]),
+        publication_epoch: 3,
+        tier_registry_generation: 0,
+        scan_mode: HealScanMode::Normal,
+    };
+    let mut cache = DataUsageCache {
+        info: crate::data_usage_define::DataUsageCacheInfo {
+            name: "bucket".to_string(),
+            next_cycle: 7,
+            ..Default::default()
+        },
+        ..Default::default()
+    };
+    assert_eq!(
+        cache.prepare_bucket_checkpoint("bucket", 7, 3, source, plan, identity),
+        crate::data_usage_define::DataUsageCachePrepareOutcome::Reset
+    );
+
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default());
+    let _raw_entry_budget = enumeration_restart::install_raw_entry_budget(scanner.local_disk.path(), 1);
+
+    let result = scan_data_folder(
+        budget.token(),
+        budget.clone(),
+        vec![scanner.local_disk.clone()],
+        scanner.local_disk.clone(),
+        cache,
+        None,
+        HealScanMode::Normal,
+        SCANNER_SLEEPER.clone(),
+    )
+    .await;
+
+    let partial_cache = match result {
+        Err(ScannerError::PartialCache(partial_cache)) => partial_cache,
+        other => panic!("expected raw enumeration partial cache after cancellation, got {other:?}"),
+    };
+
+    assert!(
+        partial_cache
+            .root()
+            .is_none_or(|root| root.objects == 0 && root.versions == 0 && root.size == 0),
+        "raw cursor writer must not invent object progress"
+    );
+    assert!(partial_cache.info.last_update.is_some());
+    assert_eq!(partial_cache.info.next_cycle, 7);
+    assert!(!partial_cache.info.snapshot_complete);
+    assert!(partial_cache.info.scan_checkpoint.is_none());
+    assert!(partial_cache.info.scan_resume_after.is_none());
+
+    let raw_cursor = partial_cache
+        .info
+        .scan_raw_enumeration_cursor
+        .as_ref()
+        .expect("raw enumeration cancellation should persist a cursor");
+    assert_eq!(raw_cursor.parent, "bucket");
+    assert_eq!(raw_cursor.entries_seen, 1);
+    assert!(raw_cursor.last_entry.is_some());
+    assert_ne!(raw_cursor.page_digest, [0; 32]);
+    assert_eq!(partial_cache.validated_raw_enumeration_cursor(), Some(raw_cursor));
+    assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Runtime));
+}
+
 #[tokio::test]
 #[serial]
 async fn test_scan_data_folder_reports_invalid_checkpoint_ignored_once() {

crates/scanner/src/scanner_folder/tests/enumeration_restart.rs

@@ -40,7 +40,7 @@ pub(in crate::scanner_folder) fn observe_raw_entry(dir: &str, name: &std::ffi::O
     }
 }
 
-struct ObservationGuard;
+pub(in crate::scanner_folder) struct ObservationGuard;
 
 impl Drop for ObservationGuard {
     fn drop(&mut self) {
@@ -48,6 +48,18 @@ impl Drop for ObservationGuard {
     }
 }
 
+pub(in crate::scanner_folder) fn install_raw_entry_budget(root: PathBuf, limit: u64) -> ObservationGuard {
+    *OBSERVATION.lock().expect("install raw-entry observation") = Some(Observation {
+        root,
+        limit,
+        entries: 0,
+        name_bytes: 0,
+        first_entry: None,
+        last_entry: None,
+    });
+    ObservationGuard
+}
+
 #[derive(serde::Deserialize)]
 #[serde(deny_unknown_fields)]
 struct Request {
@@ -84,8 +96,21 @@ async fn round(request: &Request) -> serde_json::Value {
         }
         let mut initial = DataUsageCache::default();
         initial.info.name = "bucket".to_string();
+        let source = crate::data_usage_define::DataUsageCacheSource::new(0, 0);
+        let plan = crate::data_usage_define::DataUsageScanPlanDigest([31; 32]);
+        let identity = crate::data_usage_define::DataUsageScanIdentity {
+            version: 1,
+            bucket_incarnation: Uuid::from_u128(31),
+            set_layout: crate::data_usage_define::DataUsageScanPlanDigest([32; 32]),
+            publication_epoch: 1,
+            tier_registry_generation: 0,
+            scan_mode: HealScanMode::Normal,
+        };
+        assert_eq!(
+            initial.prepare_bucket_checkpoint("bucket", 1, 0, source, plan, identity),
+            crate::data_usage_define::DataUsageCachePrepareOutcome::Reset
+        );
         initial.info.skip_healing = true;
-        initial.info.snapshot_complete = false;
         initial.replace("bucket", "", DataUsageEntry::default());
         tokio::fs::write(&cache_path, initial.marshal_msg().expect("initial cache codec"))
             .await
@@ -106,15 +131,7 @@ async fn round(request: &Request) -> serde_json::Value {
     .expect("open synthetic disk in this process");
     let parent = CancellationToken::new();
     let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default());
-    *OBSERVATION.lock().expect("install observation") = Some(Observation {
-        root: disk.path(),
-        limit: request.raw_entry_budget,
-        entries: 0,
-        name_bytes: 0,
-        first_entry: None,
-        last_entry: None,
-    });
-    let _observation_guard = ObservationGuard;
+    let _observation_guard = install_raw_entry_budget(disk.path(), request.raw_entry_budget);
     let result = scan_data_folder(
         budget.token(),
         budget.clone(),

docs/operations/oidc-console-integration.md

@@ -39,7 +39,7 @@ Every provider key can be set as `RUSTFS_IDENTITY_OPENID_<KEY>` in the process e
 | `email_claim`, `username_claim` | `RUSTFS_IDENTITY_OPENID_EMAIL_CLAIM`, `RUSTFS_IDENTITY_OPENID_USERNAME_CLAIM` | Identity claims shown in the Console. |
 | `role_policy` | `RUSTFS_IDENTITY_OPENID_ROLE_POLICY` | One fixed policy for every login from this provider. Connectivity testing only. |
 | `display_name` | `RUSTFS_IDENTITY_OPENID_DISPLAY_NAME` | Login button label. |
-| `hide_from_ui` | `RUSTFS_IDENTITY_OPENID_HIDE_FROM_UI` | Hides the provider from `/oidc/providers`. |
+| `hide_from_ui` | `RUSTFS_IDENTITY_OPENID_HIDE_FROM_UI` | Hides the provider from `/oidc/providers`. Required (`on`) for STS workload issuers whose discovery document omits `authorization_endpoint`; see [workload provider requirements](oidc-provider-requirements.md#sts-workload-providers). Complete hidden providers still support direct Console login. |
 
 Process-level settings (environment only, never suffixed per provider):
 

docs/operations/oidc-provider-requirements.md

@@ -7,6 +7,8 @@ RustFS is a standard OpenID Connect relying party using the authorization-code f
 
 ## Requirements
 
+The table below describes Console login. For endpoint-free STS issuers, use the [workload contract](#sts-workload-providers) below.
+
 | # | Requirement | Details | Code anchor |
 | --- | --- | --- | --- |
 | 1 | Discovery document | `RUSTFS_IDENTITY_OPENID_CONFIG_URL` names the provider (issuer base or full discovery URL); RustFS fetches `{issuer}/.well-known/openid-configuration` and needs `issuer`, `authorization_endpoint`, `token_endpoint`, `jwks_uri`, and the standard `*_supported` arrays. When `RUSTFS_IDENTITY_OPENID_ISSUER` is set, the document's `issuer` must equal it exactly; otherwise RustFS tries the issuer candidates derived from the config URL. | `crates/iam/src/oidc.rs` `discover_provider`, `discover_provider_from_config_url` |
@@ -20,6 +22,16 @@ RustFS is a standard OpenID Connect relying party using the authorization-code f
 | 9 | Registered redirect URI | The provider must accept the callback `{public-origin}/rustfs/admin/v3/oidc/callback/{provider_id}`. RustFS picks the origin in this order: the provider's `redirect_uri` (`RUSTFS_IDENTITY_OPENID_REDIRECT_URI`), then `RUSTFS_BROWSER_REDIRECT_URL`, then the request's own scheme and host — the last only when `RUSTFS_IDENTITY_OPENID_REDIRECT_URI_DYNAMIC` is enabled. | `rustfs/src/admin/handlers/oidc.rs` `derive_callback_uri_with_provider_config`, `browser_redirect_url` |
 | 10 | Logout endpoint (optional) | When discovery advertises `end_session_endpoint`, RustFS builds an RP-initiated logout URL with `id_token_hint`, `client_id`, and `post_logout_redirect_uri`. Without it, logout falls back to the console login page. | `crates/iam/src/oidc.rs` `build_logout_url` (reads `end_session_endpoint` from `ProviderMetadataWithLogout`) |
 
+## STS workload providers
+
+For workload identity tokens exchanged through `AssumeRoleWithWebIdentity` (for example Kubernetes service-account tokens), set `RUSTFS_IDENTITY_OPENID_HIDE_FROM_UI=on` (`hide_from_ui=on` in persisted configuration, or `hide_from_ui: true` in the admin JSON API) when discovery omits `authorization_endpoint`. The default is off; without this setting, RustFS applies the Console discovery contract and rejects the missing endpoint.
+
+Endpoint-free workload discovery requires `issuer`, `jwks_uri`, and `id_token_signing_alg_values_supported`. `token_endpoint` is optional; browser authorization endpoints, `response_types_supported`, and `subject_types_supported` are not required for this path. An explicitly null, empty, or malformed `authorization_endpoint` is rejected rather than treated as absent. Configuration validation returns `authorization_endpoint: null` for an accepted workload-only provider.
+
+Set `RUSTFS_IDENTITY_OPENID_CONFIG_URL` to the issuer/discovery URL and `RUSTFS_IDENTITY_OPENID_CLIENT_ID` to the intended token audience. Configure policy claims or `RUSTFS_IDENTITY_OPENID_ROLE_POLICY` to grant the required RustFS permissions. Signature, issuer, audience, and expiration verification remain enforced, and private endpoints still require the outbound allowlist. JWKS requests accept both `application/json` and `application/jwk-set+json`.
+
+Workload-only providers cannot perform Console authorization-code login. Hiding a provider with complete discovery metadata only hides its listing; its existing direct Console login remains available.
+
 ## Deployment notes
 
 - Behind a load balancer, authorize and callback requests must reach the same RustFS node while the `state` is in flight, or set `RUSTFS_BROWSER_REDIRECT_URL` so the callback URL is stable; the callback error text names both remedies.

rustfs/src/admin/handlers/oidc.rs

@@ -465,7 +465,7 @@ impl Operation for ValidateOidcConfigHandler {
                 valid: true,
                 message: "OIDC configuration is valid".to_string(),
                 issuer: Some(validation.issuer),
-                authorization_endpoint: Some(validation.authorization_endpoint),
+                authorization_endpoint: validation.authorization_endpoint,
                 token_endpoint: validation.token_endpoint,
             },
         )
@@ -1296,6 +1296,101 @@ mod tests {
     use http::{Extensions, HeaderMap, HeaderValue, Uri};
     use temp_env::with_var;
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn validate_handler_preserves_workload_null_and_console_endpoints() {
+        use crate::admin::runtime_sources::{AppContext, publish_test_app_context};
+        use http_body_util::BodyExt as _;
+        use rustfs_iam::store::{Store as _, object::IAM_CONFIG_PREFIX};
+        use std::sync::Arc;
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        // The admin URL boundary rejects literal loopback hosts. A local proxy
+        // serves the public-shaped test origin without external DNS or traffic.
+        let proxy = format!("http://{}", listener.local_addr().unwrap());
+        let base = "http://oidc-handler.example.invalid".to_string();
+        temp_env::async_with_vars(
+            [("RUSTFS_OUTBOUND_ALLOW_ORIGINS", Some(base.as_str())),
+             ("HTTP_PROXY", Some(proxy.as_str())), ("http_proxy", Some(proxy.as_str())),
+             ("HTTPS_PROXY", None), ("https_proxy", None), ("ALL_PROXY", None), ("all_proxy", None),
+             ("NO_PROXY", Some("")), ("no_proxy", Some(""))],
+            async {
+                let _ = rustfs_credentials::init_global_action_credentials(Some("OIDCVALIDATEROOT".into()), Some("oidcValidateRootSecret123".into()));
+                let env = rustfs_test_utils::TestECStoreEnv::builder().prefix("oidc_validate_handler")
+                    .disk_count(1).init_bucket_metadata(false).build().await;
+                rustfs_iam::store::object::ObjectStore::new(Arc::clone(&env.ecstore))
+                    .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX)).await.unwrap();
+                let iam = rustfs_iam::init_iam_sys(Arc::clone(&env.ecstore)).await.unwrap();
+                publish_test_app_context(Arc::new(AppContext::with_default_interfaces(
+                    Arc::clone(&env.ecstore), iam, Arc::new(rustfs_kms::KmsServiceManager::new()),
+                )));
+                let server_base = base.clone();
+                let server = tokio::spawn(async move {
+                    for path in ["/.well-known/openid-configuration", "/jwks", "/.well-known/openid-configuration", "/complete/.well-known/openid-configuration", "/complete/jwks"] {
+                        let (mut stream, _) = tokio::time::timeout(std::time::Duration::from_secs(15), listener.accept()).await.unwrap().unwrap();
+                        let mut request = Vec::new();
+                        while !request.ends_with(b"\r\n\r\n") {
+                            request.push(stream.read_u8().await.unwrap());
+                            assert!(request.len() < 8192);
+                        }
+                        let request = String::from_utf8(request).unwrap();
+                        let target = Url::parse(request.lines().next().unwrap().split_whitespace().nth(1).unwrap()).unwrap();
+                        assert_eq!(target.origin().ascii_serialization(), server_base);
+                        assert_eq!(target.path(), path);
+                        let mut body = if path.ends_with("/jwks") { serde_json::json!({"keys": []}) } else {
+                            serde_json::json!({"issuer": server_base, "jwks_uri": format!("{server_base}/jwks"), "id_token_signing_alg_values_supported": ["RS256"]})
+                        };
+                        if path == "/complete/.well-known/openid-configuration" {
+                            body["authorization_endpoint"] = serde_json::json!(format!("{server_base}/authorize"));
+                            body["token_endpoint"] = serde_json::json!(format!("{server_base}/token"));
+                            body["response_types_supported"] = serde_json::json!(["code"]);
+                            body["subject_types_supported"] = serde_json::json!(["public"]);
+                        }
+                        let body = body.to_string();
+                        stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap();
+                    }
+                });
+                for (hidden, complete) in [(true, false), (false, false), (false, true)] {
+                    let document = serde_json::json!({"provider_id": "workload", "client_id": "rustfs-test", "issuer": base,
+                        "config_url": format!("{base}{}/.well-known/openid-configuration", if complete { "/complete" } else { "" }), "hide_from_ui": hidden});
+                    let request = || {
+                        let mut req = build_oidc_request("/rustfs/admin/v3/oidc/validate", None, None);
+                        req.method = Method::POST;
+                        req.input = Body::from(document.to_string());
+                        req
+                    };
+                    let denied = ValidateOidcConfigHandler {}.call(request(), Params::new()).await.unwrap_err();
+                    assert_eq!(denied.code(), &S3ErrorCode::InvalidRequest);
+                    assert_eq!(denied.message(), Some("authentication required"));
+                    let mut req = request();
+                    req.credentials = Some(s3s::auth::Credentials { access_key: "OIDCVALIDATEROOT".into(), secret_key: "oidcValidateRootSecret123".into() });
+                    let result = ValidateOidcConfigHandler {}.call(req, Params::new()).await;
+                    if !hidden && !complete {
+                        let err = result.unwrap_err();
+                        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+                        assert!(err.message().unwrap().contains("authorization_endpoint"));
+                        continue;
+                    }
+                    let (status, body) = result.unwrap().output;
+                    assert_eq!(status, StatusCode::OK);
+                    let body = body.collect().await.unwrap().to_bytes();
+                    let response: serde_json::Value = serde_json::from_slice(&body).unwrap();
+                    assert_eq!(response["valid"], true);
+                    assert_eq!(response["issuer"], base);
+                    if complete {
+                        assert_eq!(response["authorization_endpoint"], format!("{base}/authorize"));
+                        assert_eq!(response["token_endpoint"], format!("{base}/token"));
+                    } else {
+                        assert_eq!(response.get("authorization_endpoint"), Some(&serde_json::Value::Null));
+                        assert_eq!(response.get("token_endpoint"), Some(&serde_json::Value::Null));
+                    }
+                }
+                server.await.unwrap();
+            },
+        ).await;
+    }
+
     fn build_oidc_request(
         uri: &'static str,
         host: Option<&'static str>,